"""
OneAgent MongoDB plugin
=======================
Gathers MongoDB stats using two MongoDB commands:
* dbStats - https://docs.mongodb.com/manual/reference/command/dbStats/
* serverStatus - https://docs.mongodb.com/manual/reference/command/serverStatus/
Running the plugin requires specifying:
* user credentials - *clusterMonitor* role is required. If authorization is disabled on MongoDB then leave *user* and *password* fields empty
* port - access to the MongoDB via localhost interface is required
This is a singleton type plugin - one instance will be created even if multiple MongoDB process
groups instances are detected.
"""
import pymongo
from ruxit.api.exceptions import AuthException, ConfigException
from ruxit.api.data import PluginMeasurement, PluginProperty, MEAttribute
from ruxit.api.base_plugin import BasePlugin
from ruxit.api.selectors import ListenPortSelector
import re
from collections import namedtuple
from contextlib import closing
from ruxit.api.snapshot import parse_port_bindings
import logging
log = logging.getLogger(__name__)
QueryDesc = namedtuple("QueryDesc", ["timeseries_key", "per_second_result_flag"])
"""
Namedtuple used for storing informations about timeseries metric, including corresponding timeseries key and information about metric type.
"""
[docs]class MongodbPlugin(BasePlugin):
"""
MongoDB plugin class. This plugin is stateful, however, the state is limited to
processing of JSON and received configuration together with maintaining database connection.
"""
[docs] def initialize(self, **kwargs):
config = kwargs['config']
self.timeseries = kwargs['json_config']['metrics']
self.process_types = kwargs['activation_context'].value.process_types
self.process_snapshot = kwargs['process_snapshot']
if config['port'] == '':
raise ConfigException("port is empty")
self.port = config['port']
self.auth = None
if 'auth_user' in config and 'auth_password' in config:
self.user = config['auth_user']
self.password = config['auth_password']
else:
raise ConfigException("user or password is empty")
self.server_status_metrics = {}
self.db_stats_metrics = {}
self.regex_pattern = "\[\'([^\']*)\'\]"
self.mongodb_auth_error_codes = [13, 18]
self.initialize_metrics()
[docs] def initialize_metrics(self):
"""
On plugin creation corresponding plugin.json file is processed in order to extract
statistics that are related to server status and particular database.
"""
server_status_metrics = {}
db_stats_metrics = {}
try:
for metric in self.timeseries:
query = metric['source']['query']
timeseries_key = metric['timeseries']['key']
per_second_result_flag = metric.get('source', {}).get('persecond', False)
query_start_index = query.find('[')
if query.startswith('server_status'):
if query[query_start_index:] in server_status_metrics and not query[query_start_index:].endswith('2'):
server_status_metrics[query[query_start_index:]] = QueryDesc(timeseries_key, per_second_result_flag)
else:
server_status_metrics[query[query_start_index:]] = QueryDesc(timeseries_key, per_second_result_flag)
elif query.startswith('db_stats'):
if query[query_start_index:] in db_stats_metrics and not query[query_start_index:].endswith('2'):
db_stats_metrics[query[query_start_index:]] = QueryDesc(timeseries_key, per_second_result_flag)
else:
db_stats_metrics[query[query_start_index:]] = QueryDesc(timeseries_key, per_second_result_flag)
except KeyError:
log.info(self, exc_info=1)
self.server_status_metrics = server_status_metrics
self.db_stats_metrics = db_stats_metrics
def _createpm(self, *args, **kwargs):
return PluginMeasurement(*args, entity_selector=ListenPortSelector(self.port, self.process_types), **kwargs)
def _add_result(self, query_desc, value, dim=None):
if value is not None:
if query_desc.per_second_result_flag == 1:
self.results_builder.add_per_second_result(self._createpm(
key=query_desc.timeseries_key,
value=value,
dimensions=dim)
)
else:
self.results_builder.add_absolute_result(self._createpm(
key=query_desc.timeseries_key,
value=value,
dimensions=dim)
)
else:
log.info('Metric ' + query_desc.timeseries_key + ' is not available')
[docs] def query(self, **kwargs):
"""
Main method that creates connection to MongoDB engine based on provided credatials and controls plugin
execution by invoking methods responsible for data collection.
Raises:
ruxit.api.exceptions.AuthException: MongoDB server returned either 13 or 18 error indicating auth issues
ruxit.api.exceptions.ConfigException: unable to connect for other reasons
"""
uri = self._create_uri()
try:
with closing(pymongo.MongoClient(uri)) as client:
databases = client.database_names()
self.get_server_metrics(client)
self.get_database_metrics(client, databases)
except pymongo.errors.OperationFailure as ex:
if ex.code in self.mongodb_auth_error_codes:
raise AuthException from ex
except pymongo.errors.ServerSelectionTimeoutError as ex:
raise ConfigException(ex.args[0]) from ex
def _create_uri(self):
for ps_entry in self.process_snapshot.entries:
if ps_entry.process_type == 21:
port_bindings = parse_port_bindings(ps_entry)
for binding in port_bindings:
if str(binding[1]) == self.port:
return self._url(binding[0])
log.info('PortBindings data: %s', port_bindings)
raise ConfigException('Unable to detect MongoDB binding')
def _url(self, ip):
if not self.user or not self.password:
return 'mongodb://{address}:{port}'.format(address=ip, port=self.port)
else:
return 'mongodb://{user}:{pwd}@{address}:{port}'.format(user=self.user,
pwd=self.password,
address=ip,
port=self.port)
[docs] def get_server_metrics(self, client):
"""
Collect MongoDB server level metrics and sends results.
"""
server_status = client.db.command("serverStatus")
for query, query_desc in self.server_status_metrics.items():
indices = re.findall(self.regex_pattern, query)
value = self.get_value_of_indices_chain(indices, server_status)
self._add_result(query_desc, value)
[docs] def get_database_metrics(self, client, databases):
"""
Iterates over a list of database received in main query, collects and sends database level metrics.
"""
for database in databases:
db = client[database]
db_stats = db.command('dbstats')
dim = {'database': database}
for query, query_desc in self.db_stats_metrics.items():
indices = re.findall(self.regex_pattern, query)
value = self.get_value_of_indices_chain(indices, db_stats)
self._add_result(query_desc, value, dim)
[docs] def get_value_of_indices_chain(self, indices, dict):
for index in indices:
if index not in dict:
return None
dict = dict[index]
return dict
