"""
OneAgent PostgreSQL plugin
==========================
Gathers PostgreSQL stats.
The full description of stats is described at https://www.postgresql.org/docs/current/static/monitoring-stats.html
Plugin collects subset of:
* database level metrics available at *pg_stat_database* view. Only non-template databases are taken into account (those that have *datistemplate* set to *false*)
* aggregated table level metrics read from *pg_stat_all_tables*
This is a singleton type plugin - one instance will be created even if multiple PostgreSQL process
groups instances are detected. To enable plugin following configuration options need to be specified:
* user credentials - role with *md5* authentication method should be provided
* port - access to the PostgreSQL via localhost interface is required
* database name
"""
import collections
from ruxit.api.data import PluginMeasurement, PluginProperty
from ruxit.api.base_plugin import BasePlugin
from ruxit.api.exceptions import AuthException, ConfigException
from ruxit.api.selectors import ListenPortSelector
from contextlib import closing
from contextlib import suppress
import pg8000
import logging
from pg8000.core import NULL_BYTE, RESPONSE_CODE, InterfaceError, ProgrammingError
DatabasePerSecondStats = collections.namedtuple("DatabasePerSecondStats", ['tup_inserted', 'tup_updated', 'tup_deleted',
'xact_commit', 'xact_rollback', 'blks_hit',
'blks_read'])
"""
List of metrics that are collected on database level as per second stats:
"""
TablePerSecondStats = collections.namedtuple("TablePerSecondStats", ["idx_tup_fetch", "seq_tup_read", "idx_scan"])
"""
List of metrics that are collected on table level as per second stats:
"""
log = logging.getLogger(__name__)
[docs]def handle_ERROR_RESPONSE_patch(self, data, ps):
"""
Patch for handle_ERROR_RESPONSE function from pg8000 package ensuring that plugin will not crash
while decoding error response in the situation when proper encoding is not set yet.
"""
responses = tuple(
(s[0:1], s[1:].decode(self._client_encoding, "ignore")) for s in
data.split(NULL_BYTE))
msg_dict = dict(responses)
if msg_dict[RESPONSE_CODE] == "28000":
self.error = InterfaceError("md5 password authentication failed")
else:
self.error = ProgrammingError(*tuple(v for k, v in responses))
pg8000.Connection.handle_ERROR_RESPONSE = handle_ERROR_RESPONSE_patch
[docs]class PostgresqlPlugin(BasePlugin):
"""
PostgreSQL plugin class. This plugin is stateful - initialize method is defined
in order to maintain one connection to database engine for the lifetime of the plugin.
"""
postgres_auth_errors = ['28P01']
"""
PostgreSQL authentication error code
"""
postgres_auth_msgs = ['md5 password authentication failed',
'server requesting password authentication, but no password was provided',
'server requesting MD5 password authentication, but no password was provided'
]
"""
List of messages returned by PostgreSQL that raises authorization exception if encountered while initializing database connection
"""
[docs] def initialize(self, **kwargs):
"""
Plugin initialization is limited to parsing configuration.
"""
config = kwargs['config']
self.process_types = kwargs["activation_context"].value.process_types
if config['port'] == '':
raise ConfigException('"Port" field cannot be empty')
self.port = config['port']
if config['database'] == '':
raise ConfigException('"Database name" field cannot be empty')
self.database = config['database']
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" field is not present in configuration')
self.databases = {}
def _createpm(self, *args, **kwargs):
return PluginMeasurement(*args, entity_selector=ListenPortSelector(self.port, self.process_types), **kwargs)
[docs] def get_databases(self, cursor):
"""
Collects and stores list of existing database.
"""
cursor.execute('SELECT datname FROM pg_database WHERE datistemplate = false')
self.databases = {row[0] for row in cursor.fetchall()}
[docs] def get_database_level_statistics(self, cursor):
"""
Performs query that selects requested database metrics and sends gathered results.
"""
cursor.execute('SELECT datname,numbackends,tup_inserted,tup_updated,tup_deleted,xact_commit,xact_rollback,'
'blks_hit,blks_read FROM pg_stat_database')
results = cursor.fetchall()
for row in results:
# numbackends++ for requested db
datname = row[0]
numbackends = row[1]
stats = DatabasePerSecondStats(*row[2:])
if datname in self.databases:
dim = {'database': datname}
for field in stats._fields:
self.results_builder.add_per_second_result(self._createpm(
dimensions=dim,
key=field,
value=getattr(stats, field))
)
self.results_builder.add_absolute_result(self._createpm(
dimensions=dim,
key='numbackends',
value=numbackends)
)
with suppress(ZeroDivisionError):
cache_hit_ratio = (100*stats.blks_hit/(stats.blks_read+stats.blks_hit))
# calculated cache hit ratio
self.results_builder.add_absolute_result(self._createpm(
dimensions=dim,
key='cache_hit_ratio',
value=cache_hit_ratio)
)
[docs] def get_table_level_statistics(self):
"""
Method that tries connecting to every database and gets its stats.
"""
for database in self.databases:
try:
with closing(pg8000.connect(
user=self.user,
password=self.password,
database=database,
port=int(self.port))) as conn, \
closing(conn.cursor()) as cursor:
cursor.execute('SELECT SUM(idx_tup_fetch),SUM(seq_tup_read),SUM(idx_scan) FROM pg_stat_all_tables')
results = cursor.fetchall()
stats = TablePerSecondStats(*results[0])
dim = {'database': database}
for field in stats._fields:
self.results_builder.add_per_second_result(
self._createpm(dimensions=dim, key=field, value=getattr(stats, field))
)
except:
log.info(self, exc_info=1)
[docs] def query(self, **kwargs):
"""
Main method that connects to PostgreSQL using configuration from global variable.
Raises:
ruxit.api.exceptions.AuthException: PostgreSQL responded with 28P01
ruxit.api.exceptions.ConfigException: connection and data parsing errors
"""
try:
with closing(pg8000.connect(
user=self.user,
password=self.password,
database=self.database,
port=int(self.port))) as conn, \
closing(conn.cursor()) as cursor:
self.get_databases(cursor)
self.get_database_level_statistics(cursor)
except pg8000.ProgrammingError as ex:
args_len = len(ex.args)
if args_len > 1 and ex.args[1] in self.postgres_auth_errors:
raise AuthException from ex
elif args_len > 2:
raise ConfigException(ex.args[2]) from ex
else:
raise ConfigException(ex.args[0]) from ex
except pg8000.InterfaceError as ex:
if ex.args[0] in self.postgres_auth_msgs:
raise AuthException from ex
else:
raise ConfigException(ex.args[0]) from ex
self.get_table_level_statistics()
