def check(self, instance):
        """
        Returns a dictionary that looks a lot like what's sent back by
        db.serverStatus()
        """

        def total_seconds(td):
            """
            Returns total seconds of a timedelta in a way that's safe for
            Python < 2.7
            """
            if hasattr(td, 'total_seconds'):
                return td.total_seconds()
            else:
                return (lag.microseconds + (lag.seconds + lag.days * 24 * 3600) * 10 ** 6) / 10.0 ** 6

        if 'server' not in instance:
            raise Exception("Missing 'server' in mongo config")

        # x.509 authentication
        ssl_params = {
            'ssl': instance.get('ssl', None),
            'ssl_keyfile': instance.get('ssl_keyfile', None),
            'ssl_certfile': instance.get('ssl_certfile', None),
            'ssl_cert_reqs': instance.get('ssl_cert_reqs', None),
            'ssl_ca_certs': instance.get('ssl_ca_certs', None),
        }

        for key, param in list(iteritems(ssl_params)):
            if param is None:
                del ssl_params[key]

        server = instance['server']
        username, password, db_name, nodelist, clean_server_name, auth_source = self._parse_uri(
            server, sanitize_username=bool(ssl_params)
        )

        additional_metrics = instance.get('additional_metrics', [])

        # Get the list of metrics to collect
        collect_tcmalloc_metrics = 'tcmalloc' in additional_metrics
        metrics_to_collect = self._get_metrics_to_collect(server, additional_metrics)

        # Tagging
        tags = instance.get('tags', [])
        # ...de-dupe tags to avoid a memory leak
        tags = list(set(tags))

        if not db_name:
            self.log.info('No MongoDB database found in URI. Defaulting to admin.')
            db_name = 'admin'

        service_check_tags = ["db:%s" % db_name]
        service_check_tags.extend(tags)

        # ...add the `server` tag to the metrics' tags only
        # (it's added in the backend for service checks)
        tags.append('server:%s' % clean_server_name)

        if nodelist:
            host = nodelist[0][0]
            port = nodelist[0][1]
            service_check_tags = service_check_tags + ["host:%s" % host, "port:%s" % port]

        timeout = float(instance.get('timeout', DEFAULT_TIMEOUT)) * 1000
        try:
            cli = pymongo.mongo_client.MongoClient(
                server,
                socketTimeoutMS=timeout,
                connectTimeoutMS=timeout,
                serverSelectionTimeoutMS=timeout,
                read_preference=pymongo.ReadPreference.PRIMARY_PREFERRED,
                **ssl_params
            )
            # some commands can only go against the admin DB
            admindb = cli['admin']
            db = cli[db_name]
        except Exception:
            self.service_check(self.SERVICE_CHECK_NAME, AgentCheck.CRITICAL, tags=service_check_tags)
            raise

        # Authenticate
        do_auth = True
        use_x509 = ssl_params and not password

        if not username:
            self.log.debug(u"A username is required to authenticate to `%s`", server)
            do_auth = False

        if do_auth:
            if auth_source:
                msg = "authSource was specified in the the server URL: using '%s' as the authentication database"
                self.log.info(msg, auth_source)
                self._authenticate(
                    cli[auth_source], username, password, use_x509, clean_server_name, service_check_tags
                )
            else:
                self._authenticate(db, username, password, use_x509, clean_server_name, service_check_tags)

        try:
            status = db.command('serverStatus', tcmalloc=collect_tcmalloc_metrics)
        except Exception:
            self.service_check(self.SERVICE_CHECK_NAME, AgentCheck.CRITICAL, tags=service_check_tags)
            raise
        else:
            self.service_check(self.SERVICE_CHECK_NAME, AgentCheck.OK, tags=service_check_tags)

        if status['ok'] == 0:
            raise Exception(status['errmsg'].__str__())

        ops = db.current_op()
        status['fsyncLocked'] = 1 if ops.get('fsyncLock') else 0

        status['stats'] = db.command('dbstats')
        dbstats = {db_name: {'stats': status['stats']}}

        # Handle replica data, if any
        # See
        # http://www.mongodb.org/display/DOCS/Replica+Set+Commands#ReplicaSetCommands-replSetGetStatus  # noqa
        if is_affirmative(instance.get('replica_check', True)):
            try:
                data = {}

                replSet = admindb.command('replSetGetStatus')
                if replSet:
                    primary = None
                    current = None

                    # need a new connection to deal with replica sets
                    setname = replSet.get('set')
                    cli_rs = pymongo.mongo_client.MongoClient(
                        server,
                        socketTimeoutMS=timeout,
                        connectTimeoutMS=timeout,
                        serverSelectionTimeoutMS=timeout,
                        replicaset=setname,
                        read_preference=pymongo.ReadPreference.NEAREST,
                        **ssl_params
                    )

                    if do_auth:
                        if auth_source:
                            self._authenticate(
                                cli_rs[auth_source], username, password, use_x509, server, service_check_tags
                            )
                        else:
                            self._authenticate(
                                cli_rs[db_name], username, password, use_x509, server, service_check_tags
                            )

                    # Replication set information
                    replset_name = replSet['set']
                    replset_state = self.get_state_name(replSet['myState']).lower()

                    tags.extend([u"replset_name:{0}".format(replset_name), u"replset_state:{0}".format(replset_state)])

                    # Find nodes: master and current node (ourself)
                    for member in replSet.get('members'):
                        if member.get('self'):
                            current = member
                        if int(member.get('state')) == 1:
                            primary = member

                    # Compute a lag time
                    if current is not None and primary is not None:
                        if 'optimeDate' in primary and 'optimeDate' in current:
                            lag = primary['optimeDate'] - current['optimeDate']
                            data['replicationLag'] = total_seconds(lag)

                    if current is not None:
                        data['health'] = current['health']

                    data['state'] = replSet['myState']

                    if current is not None:
                        total = 0.0
                        cfg = cli_rs['local']['system.replset'].find_one()
                        for member in cfg.get('members'):
                            total += member.get('votes', 1)
                            if member['_id'] == current['_id']:
                                data['votes'] = member.get('votes', 1)
                        data['voteFraction'] = data['votes'] / total

                    status['replSet'] = data

                    # Submit events
                    self._report_replica_set_state(data['state'], clean_server_name, replset_name)

            except Exception as e:
                if "OperationFailure" in repr(e) and (
                    "not running with --replSet" in str(e) or "replSetGetStatus" in str(e)
                ):
                    pass
                else:
                    raise e

        # If these keys exist, remove them for now as they cannot be serialized
        try:
            status['backgroundFlushing'].pop('last_finished')
        except KeyError:
            pass
        try:
            status.pop('localTime')
        except KeyError:
            pass

        dbnames = cli.database_names()
        self.gauge('mongodb.dbs', len(dbnames), tags=tags)

        for db_n in dbnames:
            db_aux = cli[db_n]
            dbstats[db_n] = {'stats': db_aux.command('dbstats')}

        # Go through the metrics and save the values
        for metric_name in metrics_to_collect:
            # each metric is of the form: x.y.z with z optional
            # and can be found at status[x][y][z]
            value = status

            if metric_name.startswith('stats'):
                continue
            else:
                try:
                    for c in metric_name.split("."):
                        value = value[c]
                except KeyError:
                    continue

            # value is now status[x][y][z]
            if not isinstance(value, (int, long, float)):
                raise TypeError(
                    u"{0} value is a {1}, it should be an int, a float or a long instead.".format(
                        metric_name, type(value)
                    )
                )

            # Submit the metric
            submit_method, metric_name_alias = self._resolve_metric(metric_name, metrics_to_collect)
            submit_method(self, metric_name_alias, value, tags=tags)

        for st, value in iteritems(dbstats):
            for metric_name in metrics_to_collect:
                if not metric_name.startswith('stats.'):
                    continue

                try:
                    val = value['stats'][metric_name.split('.')[1]]
                except KeyError:
                    continue

                # value is now status[x][y][z]
                if not isinstance(val, (int, long, float)):
                    raise TypeError(
                        u"{0} value is a {1}, it should be an int, a float or a long instead.".format(
                            metric_name, type(val)
                        )
                    )

                # Submit the metric
                metrics_tags = tags + [
                    u"cluster:db:{0}".format(st),  # FIXME 6.0 - keep for backward compatibility
                    u"db:{0}".format(st),
                ]

                submit_method, metric_name_alias = self._resolve_metric(metric_name, metrics_to_collect)
                submit_method(self, metric_name_alias, val, tags=metrics_tags)

        if is_affirmative(instance.get('collections_indexes_stats')):
            mongo_version = cli.server_info().get('version', '0.0')
            if LooseVersion(mongo_version) >= LooseVersion("3.2"):
                self._collect_indexes_stats(instance, db, tags)
            else:
                msg = "'collections_indexes_stats' is only available starting from mongo 3.2: your mongo version is %s"
                self.log.error(msg, mongo_version)

        # Report the usage metrics for dbs/collections
        if 'top' in additional_metrics:
            try:
                dbtop = admindb.command('top')
                for ns, ns_metrics in iteritems(dbtop['totals']):
                    if "." not in ns:
                        continue

                    # configure tags for db name and collection name
                    dbname, collname = ns.split(".", 1)
                    ns_tags = tags + ["db:%s" % dbname, "collection:%s" % collname]

                    # iterate over DBTOP metrics
                    for m in self.TOP_METRICS:
                        # each metric is of the form: x.y.z with z optional
                        # and can be found at ns_metrics[x][y][z]
                        value = ns_metrics
                        try:
                            for c in m.split("."):
                                value = value[c]
                        except Exception:
                            continue

                        # value is now status[x][y][z]
                        if not isinstance(value, (int, long, float)):
                            raise TypeError(
                                u"{0} value is a {1}, it should be an int, a float or a long instead.".format(
                                    m, type(value)
                                )
                            )

                        # Submit the metric
                        submit_method, metric_name_alias = self._resolve_metric(m, metrics_to_collect, prefix="usage")
                        submit_method(self, metric_name_alias, value, tags=ns_tags)
                        # Keep old incorrect metric
                        if metric_name_alias.endswith('countps'):
                            GAUGE(self, metric_name_alias[:-2], value, tags=ns_tags)
            except Exception as e:
                self.log.warning('Failed to record `top` metrics %s' % str(e))

        if 'local' in dbnames:  # it might not be if we are connectiing through mongos
            # Fetch information analogous to Mongo's db.getReplicationInfo()
            localdb = cli['local']

            oplog_data = {}

            for ol_collection_name in ("oplog.rs", "oplog.$main"):
                ol_options = localdb[ol_collection_name].options()
                if ol_options:
                    break

            if ol_options:
                try:
                    oplog_data['logSizeMB'] = round_value(ol_options['size'] / 2.0 ** 20, 2)

                    oplog = localdb[ol_collection_name]

                    oplog_data['usedSizeMB'] = round_value(
                        localdb.command("collstats", ol_collection_name)['size'] / 2.0 ** 20, 2
                    )

                    op_asc_cursor = oplog.find({"ts": {"$exists": 1}}).sort("$natural", pymongo.ASCENDING).limit(1)
                    op_dsc_cursor = oplog.find({"ts": {"$exists": 1}}).sort("$natural", pymongo.DESCENDING).limit(1)

                    try:
                        first_timestamp = op_asc_cursor[0]['ts'].as_datetime()
                        last_timestamp = op_dsc_cursor[0]['ts'].as_datetime()
                        oplog_data['timeDiff'] = total_seconds(last_timestamp - first_timestamp)
                    except (IndexError, KeyError):
                        # if the oplog collection doesn't have any entries
                        # if an object in the collection doesn't have a ts value, we ignore it
                        pass
                except KeyError:
                    # encountered an error trying to access options.size for the oplog collection
                    self.log.warning(u"Failed to record `ReplicationInfo` metrics.")

            for m, value in iteritems(oplog_data):
                submit_method, metric_name_alias = self._resolve_metric('oplog.%s' % m, metrics_to_collect)
                submit_method(self, metric_name_alias, value, tags=tags)

        else:
            self.log.debug('"local" database not in dbnames. Not collecting ReplicationInfo metrics')

        # get collection level stats
        try:
            # Ensure that you're on the right db
            db = cli[db_name]
            # grab the collections from the configutation
            coll_names = instance.get('collections', [])
            # loop through the collections
            for coll_name in coll_names:
                # grab the stats from the collection
                stats = db.command("collstats", coll_name)
                # loop through the metrics
                for m in self.collection_metrics_names:
                    coll_tags = tags + ["db:%s" % db_name, "collection:%s" % coll_name]
                    value = stats.get(m, None)
                    if not value:
                        continue

                    # if it's the index sizes, then it's a dict.
                    if m == 'indexSizes':
                        submit_method, metric_name_alias = self._resolve_metric(
                            'collection.%s' % m, self.COLLECTION_METRICS
                        )
                        # loop through the indexes
                        for idx, val in iteritems(value):
                            # we tag the index
                            idx_tags = coll_tags + ["index:%s" % idx]
                            submit_method(self, metric_name_alias, val, tags=idx_tags)
                    else:
                        submit_method, metric_name_alias = self._resolve_metric(
                            'collection.%s' % m, self.COLLECTION_METRICS
                        )
                        submit_method(self, metric_name_alias, value, tags=coll_tags)
        except Exception as e:
            self.log.warning(u"Failed to record `collection` metrics.")
            self.log.exception(e)
def check ( self , instance ) : def total_seconds ( td ) : if hasattr ( td , 'total_seconds' ) : return td . total_seconds ( ) else : return ( lag . microseconds + ( lag . seconds + lag . days * 24 * 3600 ) * 10 ** 6 ) / 10.0 ** 6 if 'server' not in instance : raise Exception ( "Missing 'server' in mongo config" ) ssl_params = { 'ssl' : instance . get ( 'ssl' , None ) , 'ssl_keyfile' : instance . get ( 'ssl_keyfile' , None ) , 'ssl_certfile' : instance . get ( 'ssl_certfile' , None ) , 'ssl_cert_reqs' : instance . get ( 'ssl_cert_reqs' , None ) , 'ssl_ca_certs' : instance . get ( 'ssl_ca_certs' , None ) , } for key , param in list ( iteritems ( ssl_params ) ) : if param is None : del ssl_params [ key ] server = instance [ 'server' ] username , password , db_name , nodelist , clean_server_name , auth_source = self . _parse_uri ( server , sanitize_username = bool ( ssl_params ) ) additional_metrics = instance . get ( 'additional_metrics' , [ ] ) collect_tcmalloc_metrics = 'tcmalloc' in additional_metrics metrics_to_collect = self . _get_metrics_to_collect ( server , additional_metrics ) tags = instance . get ( 'tags' , [ ] ) tags = list ( set ( tags ) ) if not db_name : self . log . info ( 'No MongoDB database found in URI. Defaulting to admin.' ) db_name = 'admin' service_check_tags = [ "db:%s" % db_name ] service_check_tags . extend ( tags ) tags . append ( 'server:%s' % clean_server_name ) if nodelist : host = nodelist [ 0 ] [ 0 ] port = nodelist [ 0 ] [ 1 ] service_check_tags = service_check_tags + [ "host:%s" % host , "port:%s" % port ] timeout = float ( instance . get ( 'timeout' , DEFAULT_TIMEOUT ) ) * 1000 try : cli = pymongo . mongo_client . MongoClient ( server , socketTimeoutMS = timeout , connectTimeoutMS = timeout , serverSelectionTimeoutMS = timeout , read_preference = pymongo . ReadPreference . PRIMARY_PREFERRED , ** ssl_params ) admindb = cli [ 'admin' ] db = cli [ db_name ] except Exception : self . service_check ( self . SERVICE_CHECK_NAME , AgentCheck . CRITICAL , tags = service_check_tags ) raise do_auth = True use_x509 = ssl_params and not password if not username : self . log . debug ( u"A username is required to authenticate to `%s`" , server ) do_auth = False if do_auth : if auth_source : msg = "authSource was specified in the the server URL: using '%s' as the authentication database" self . log . info ( msg , auth_source ) self . _authenticate ( cli [ auth_source ] , username , password , use_x509 , clean_server_name , service_check_tags ) else : self . _authenticate ( db , username , password , use_x509 , clean_server_name , service_check_tags ) try : status = db . command ( 'serverStatus' , tcmalloc = collect_tcmalloc_metrics ) except Exception : self . service_check ( self . SERVICE_CHECK_NAME , AgentCheck . CRITICAL , tags = service_check_tags ) raise else : self . service_check ( self . SERVICE_CHECK_NAME , AgentCheck . OK , tags = service_check_tags ) if status [ 'ok' ] == 0 : raise Exception ( status [ 'errmsg' ] . __str__ ( ) ) ops = db . current_op ( ) status [ 'fsyncLocked' ] = 1 if ops . get ( 'fsyncLock' ) else 0 status [ 'stats' ] = db . command ( 'dbstats' ) dbstats = { db_name : { 'stats' : status [ 'stats' ] } } if is_affirmative ( instance . get ( 'replica_check' , True ) ) : try : data = { } replSet = admindb . command ( 'replSetGetStatus' ) if replSet : primary = None current = None setname = replSet . get ( 'set' ) cli_rs = pymongo . mongo_client . MongoClient ( server , socketTimeoutMS = timeout , connectTimeoutMS = timeout , serverSelectionTimeoutMS = timeout , replicaset = setname , read_preference = pymongo . ReadPreference . NEAREST , ** ssl_params ) if do_auth : if auth_source : self . _authenticate ( cli_rs [ auth_source ] , username , password , use_x509 , server , service_check_tags ) else : self . _authenticate ( cli_rs [ db_name ] , username , password , use_x509 , server , service_check_tags ) replset_name = replSet [ 'set' ] replset_state = self . get_state_name ( replSet [ 'myState' ] ) . lower ( ) tags . extend ( [ u"replset_name:{0}" . format ( replset_name ) , u"replset_state:{0}" . format ( replset_state ) ] ) for member in replSet . get ( 'members' ) : if member . get ( 'self' ) : current = member if int ( member . get ( 'state' ) ) == 1 : primary = member if current is not None and primary is not None : if 'optimeDate' in primary and 'optimeDate' in current : lag = primary [ 'optimeDate' ] - current [ 'optimeDate' ] data [ 'replicationLag' ] = total_seconds ( lag ) if current is not None : data [ 'health' ] = current [ 'health' ] data [ 'state' ] = replSet [ 'myState' ] if current is not None : total = 0.0 cfg = cli_rs [ 'local' ] [ 'system.replset' ] . find_one ( ) for member in cfg . get ( 'members' ) : total += member . get ( 'votes' , 1 ) if member [ '_id' ] == current [ '_id' ] : data [ 'votes' ] = member . get ( 'votes' , 1 ) data [ 'voteFraction' ] = data [ 'votes' ] / total status [ 'replSet' ] = data self . _report_replica_set_state ( data [ 'state' ] , clean_server_name , replset_name ) except Exception as e : if "OperationFailure" in repr ( e ) and ( "not running with --replSet" in str ( e ) or "replSetGetStatus" in str ( e ) ) : pass else : raise e try : status [ 'backgroundFlushing' ] . pop ( 'last_finished' ) except KeyError : pass try : status . pop ( 'localTime' ) except KeyError : pass dbnames = cli . database_names ( ) self . gauge ( 'mongodb.dbs' , len ( dbnames ) , tags = tags ) for db_n in dbnames : db_aux = cli [ db_n ] dbstats [ db_n ] = { 'stats' : db_aux . command ( 'dbstats' ) } for metric_name in metrics_to_collect : value = status if metric_name . startswith ( 'stats' ) : continue else : try : for c in metric_name . split ( "." ) : value = value [ c ] except KeyError : continue if not isinstance ( value , ( int , long , float ) ) : raise TypeError ( u"{0} value is a {1}, it should be an int, a float or a long instead." . format ( metric_name , type ( value ) ) ) submit_method , metric_name_alias = self . _resolve_metric ( metric_name , metrics_to_collect ) submit_method ( self , metric_name_alias , value , tags = tags ) for st , value in iteritems ( dbstats ) : for metric_name in metrics_to_collect : if not metric_name . startswith ( 'stats.' ) : continue try : val = value [ 'stats' ] [ metric_name . split ( '.' ) [ 1 ] ] except KeyError : continue if not isinstance ( val , ( int , long , float ) ) : raise TypeError ( u"{0} value is a {1}, it should be an int, a float or a long instead." . format ( metric_name , type ( val ) ) ) metrics_tags = tags + [ u"cluster:db:{0}" . format ( st ) , u"db:{0}" . format ( st ) , ] submit_method , metric_name_alias = self . _resolve_metric ( metric_name , metrics_to_collect ) submit_method ( self , metric_name_alias , val , tags = metrics_tags ) if is_affirmative ( instance . get ( 'collections_indexes_stats' ) ) : mongo_version = cli . server_info ( ) . get ( 'version' , '0.0' ) if LooseVersion ( mongo_version ) >= LooseVersion ( "3.2" ) : self . _collect_indexes_stats ( instance , db , tags ) else : msg = "'collections_indexes_stats' is only available starting from mongo 3.2: your mongo version is %s" self . log . error ( msg , mongo_version ) if 'top' in additional_metrics : try : dbtop = admindb . command ( 'top' ) for ns , ns_metrics in iteritems ( dbtop [ 'totals' ] ) : if "." not in ns : continue dbname , collname = ns . split ( "." , 1 ) ns_tags = tags + [ "db:%s" % dbname , "collection:%s" % collname ] for m in self . TOP_METRICS : value = ns_metrics try : for c in m . split ( "." ) : value = value [ c ] except Exception : continue if not isinstance ( value , ( int , long , float ) ) : raise TypeError ( u"{0} value is a {1}, it should be an int, a float or a long instead." . format ( m , type ( value ) ) ) submit_method , metric_name_alias = self . _resolve_metric ( m , metrics_to_collect , prefix = "usage" ) submit_method ( self , metric_name_alias , value , tags = ns_tags ) if metric_name_alias . endswith ( 'countps' ) : GAUGE ( self , metric_name_alias [ : - 2 ] , value , tags = ns_tags ) except Exception as e : self . log . warning ( 'Failed to record `top` metrics %s' % str ( e ) ) if 'local' in dbnames : localdb = cli [ 'local' ] oplog_data = { } for ol_collection_name in ( "oplog.rs" , "oplog.$main" ) : ol_options = localdb [ ol_collection_name ] . options ( ) if ol_options : break if ol_options : try : oplog_data [ 'logSizeMB' ] = round_value ( ol_options [ 'size' ] / 2.0 ** 20 , 2 ) oplog = localdb [ ol_collection_name ] oplog_data [ 'usedSizeMB' ] = round_value ( localdb . command ( "collstats" , ol_collection_name ) [ 'size' ] / 2.0 ** 20 , 2 ) op_asc_cursor = oplog . find ( { "ts" : { "$exists" : 1 } } ) . sort ( "$natural" , pymongo . ASCENDING ) . limit ( 1 ) op_dsc_cursor = oplog . find ( { "ts" : { "$exists" : 1 } } ) . sort ( "$natural" , pymongo . DESCENDING ) . limit ( 1 ) try : first_timestamp = op_asc_cursor [ 0 ] [ 'ts' ] . as_datetime ( ) last_timestamp = op_dsc_cursor [ 0 ] [ 'ts' ] . as_datetime ( ) oplog_data [ 'timeDiff' ] = total_seconds ( last_timestamp - first_timestamp ) except ( IndexError , KeyError ) : pass except KeyError : self . log . warning ( u"Failed to record `ReplicationInfo` metrics." ) for m , value in iteritems ( oplog_data ) : submit_method , metric_name_alias = self . _resolve_metric ( 'oplog.%s' % m , metrics_to_collect ) submit_method ( self , metric_name_alias , value , tags = tags ) else : self . log . debug ( '"local" database not in dbnames. Not collecting ReplicationInfo metrics' ) try : db = cli [ db_name ] coll_names = instance . get ( 'collections' , [ ] ) for coll_name in coll_names : stats = db . command ( "collstats" , coll_name ) for m in self . collection_metrics_names : coll_tags = tags + [ "db:%s" % db_name , "collection:%s" % coll_name ] value = stats . get ( m , None ) if not value : continue if m == 'indexSizes' : submit_method , metric_name_alias = self . _resolve_metric ( 'collection.%s' % m , self . COLLECTION_METRICS ) for idx , val in iteritems ( value ) : idx_tags = coll_tags + [ "index:%s" % idx ] submit_method ( self , metric_name_alias , val , tags = idx_tags ) else : submit_method , metric_name_alias = self . _resolve_metric ( 'collection.%s' % m , self . COLLECTION_METRICS ) submit_method ( self , metric_name_alias , value , tags = coll_tags ) except Exception as e : self . log . warning ( u"Failed to record `collection` metrics." ) self . log . exception ( e )
Returns a dictionary that looks a lot like what's sent back by
        db.serverStatus()