'Discovers the mongo cluster and creates a thread for each primary.'

@log_fatal_exceptions
def run(self):

    update_mininum_mongodb_version(None)
    self.main_conn = self.create_authed_client()
    LOG.always('Source    MongoDB    version:    %s', self.main_conn.admin.command('buildInfo')['version'])
    for dm in self.doc_managers:
        name = dm.__class__.__module__
        module = sys.modules[name]
        version = 'unknown'
        if hasattr(module, '__version__'):
            version = module.__version__
        elif hasattr(module, 'version'):
            version = module.version
        LOG.always('Target    DocManager:    %s    version:    %s', name, version)
    self.read_oplog_progress()
    conn_type = None
    try:
        self.main_conn.admin.command('isdbgrid')
    except pymongo.errors.OperationFailure:
        conn_type = 'REPLSET'
    if (conn_type == 'REPLSET'):
        is_master = self.main_conn.admin.command('isMaster')
        if ('setName' not in is_master):
            LOG.error(('No    replica    set    at    "%s"!    A    replica    set    is    required    to    run    mongo-connector.    Shutting    down...' % self.address))
            return
        self.main_conn.close()
        self.main_conn = self.create_authed_client(replicaSet=is_master['setName'])
        self.update_version_from_client(self.main_conn)
        oplog = OplogThread(self.main_conn, self.doc_managers, self.oplog_progress, self.namespace_config, **self.kwargs)
        self.shard_set[0] = oplog
        LOG.info(('MongoConnector:    Starting    connection    thread    %s' % self.main_conn))
        oplog.start()
        while self.can_run:
            shard_thread = self.shard_set[0]
            if (not (shard_thread.running and shard_thread.is_alive())):
                LOG.error(('MongoConnector:    OplogThread    %s    unexpectedly    stopped!    Shutting    down' % str(self.shard_set[0])))
                self.oplog_thread_join()
                for dm in self.doc_managers:
                    dm.stop()
                return
            self.write_oplog_progress()
            time.sleep(1)
    else:
        while self.can_run:
            for shard_doc in retry_until_ok((lambda : list(self.main_conn.config.shards.find()))):
                shard_id = shard_doc['_id']
                if (shard_id in self.shard_set):
                    shard_thread = self.shard_set[shard_id]
                    if (not (shard_thread.running and shard_thread.is_alive())):
                        LOG.error(('MongoConnector:    OplogThread    %s    unexpectedly    stopped!    Shutting    down' % str(self.shard_set[shard_id])))
                        self.oplog_thread_join()
                        for dm in self.doc_managers:
                            dm.stop()
                        return
                    self.write_oplog_progress()
                    time.sleep(1)
                    continue
                try:
                    (repl_set, hosts) = shard_doc['host'].split('/')
                except ValueError:
                    cause = 'The    system    only    uses    replica    sets!'
                    LOG.exception('MongoConnector:    %s', cause)
                    self.oplog_thread_join()
                    for dm in self.doc_managers:
                        dm.stop()
                    return
                shard_conn = self.create_authed_client(hosts, replicaSet=repl_set)
                self.update_version_from_client(shard_conn)
                oplog = OplogThread(shard_conn, self.doc_managers, self.oplog_progress, self.namespace_config, mongos_client=self.main_conn, **self.kwargs)
                self.shard_set[shard_id] = oplog
                msg = 'Starting    connection    thread'
                LOG.info(('MongoConnector:    %s    %s' % (msg, shard_conn)))
                oplog.start()
    if (self.signal is not None):
        LOG.info('recieved    signal    %s:    shutting    down...', self.signal)
    self.oplog_thread_join()
    self.write_oplog_progress()
