v2.2.2: add cfs plugin

Author: vadv

mamonsu/__init__.py

@@ -1,7 +1,7 @@
 __author__ = 'Dmitry Vasilyev'
 __author_email__ = '[email protected]'
 __description__ = 'Monitoring agent for PostgreSQL'
-__version__ = '2.2.1'
+__version__ = '2.2.2'
 __licence__ = 'BSD'
 
 __url__ = 'https://github.com/postgrespro/mamonsu'

mamonsu/lib/sender.py

@@ -3,6 +3,7 @@
 import time
 
 from mamonsu.lib.plugin import Plugin
+import mamonsu.lib.platform as platform
 
 
 class Sender():
@@ -25,15 +26,18 @@ def add_sender(self, sender):
         self._senders.append(sender)
 
     # resend all values to senders
-    def send(self, key, value, delta=None, host=None, clock=None):
+    def send(self, key, value, delta=None, host=None, clock=None, only_positive_speed=False):
 
         if clock is None:
             clock = int(time.time())
 
         hash_key = self._hash(key, host)
         if delta is not None:
-            if isinstance(value, float) or isinstance(value, int):
+            if isinstance(value, float) or isinstance(value, platform.INTEGER_TYPES):
                 if hash_key in self._last_values:
+                    if only_positive_speed and (self._last_values[hash_key][0] > value):
+                            self._last_values[hash_key] = (value, clock)
+                            return
                     last_value, last_time = self._last_values[hash_key]
                     self._last_values[hash_key] = (value, clock)
                     if delta == Plugin.DELTA.speed_per_second:

mamonsu/plugins/pgsql/__init__.py

@@ -2,5 +2,6 @@
 __all__ += ['health', 'instance', 'xlog']
 __all__ += ['pg_stat_statement', 'pg_buffercache', 'pg_wait_sampling']
 __all__ += ['checkpoint', 'oldest', 'pg_locks']
+__all__ += ['cfs']
 
 from . import *

mamonsu/plugins/pgsql/cfs.py

@@ -0,0 +1,150 @@
+# -*- coding: utf-8 -*-
+
+from mamonsu.plugins.pgsql.plugin import PgsqlPlugin as Plugin
+from .pool import Pooler
+
+
+class Cfs(Plugin):
+
+    Interval = 5 * 60
+
+    DEFAULT_CONFIG = {'force_enable': str(False)}
+
+    compressed_ratio_sql = """
+select
+    n.nspname || '.' || c.relname as table_name,
+    cfs_compression_ratio(c.oid::regclass) as ratio,
+    (pg_catalog.pg_total_relation_size(c.oid::regclass) - pg_catalog.pg_indexes_size(c.oid::regclass)) as compressed_size
+from
+    pg_catalog.pg_class as c
+    left join pg_catalog.pg_namespace n on n.oid = c.relnamespace
+where c.reltablespace in (select oid from pg_catalog.pg_tablespace where spcoptions::text ~ 'compression')
+    and c.relkind IN ('r','v','m','S','f','')
+
+union all
+
+select
+    n.nspname || '.' || c.relname as table_name,
+    cfs_compression_ratio(c.oid::regclass) as ratio,
+    pg_catalog.pg_total_relation_size(c.oid::regclass) as compressed_size -- pg_toast included
+from
+    pg_catalog.pg_class as c
+    left join pg_catalog.pg_namespace n on n.oid = c.relnamespace
+where c.reltablespace in (select oid from pg_catalog.pg_tablespace where spcoptions::text ~ 'compression')
+    and c.relkind = 'i' and n.nspname <> 'pg_toast';
+"""
+
+    activity_sql = """
+select
+    cfs_gc_activity_processed_bytes(), -- writed
+    cfs_gc_activity_processed_pages() * 8 * 1024, -- scanned
+    cfs_gc_activity_processed_files(), -- writed
+    cfs_gc_activity_scanned_files(); -- scanned
+"""
+
+    prev = {}
+
+    def run(self, zbx):
+
+        if self.plugin_config('force_enable') == 'False':
+            self.disable_and_exit_if_not_pgpro_ee()
+
+        relations, compressed_size, non_compressed_size = [], 0, 0
+        for db in Pooler.databases:
+            for row in Pooler.query(self.compressed_ratio_sql, db):
+                relation_name = '{0}.{1}'.format(db, row[0])
+                relations.append({'{#COMPRESSED_RELATION}': relation_name})
+                compressed_size += row[2]
+                non_compressed_size += row[2] * row[1]
+                zbx.send('pgsql.cfs.relation[{0}]'.format(row[0]), row[1])
+        zbx.send('pgsql.cfs.discovery_compressed_relations[]', zbx.json({'data': relations}))
+        zbx.send('pgsql.cfs.activity[total_compress_ratio]', non_compressed_size / compressed_size)
+        del(relations, compressed_size, non_compressed_size)
+
+        info = Pooler.query(self.activity_sql)[0]
+        zbx.send('pgsql.cfs.activity[writed_bytes]', info[0], delta=self.DELTA_SPEED, only_positive_speed=True)
+        zbx.send('pgsql.cfs.activity[scanned_bytes]', info[1], delta=self.DELTA_SPEED, only_positive_speed=True)
+
+        # calculate current compress ratio
+        if ('writed_bytes' in self.prev) and ('scanned_bytes' in self.prev):
+            if info[0] > self.prev['writed_bytes'] and info[1] > self.prev['scanned_bytes']:
+                val = (self.prev['scanned_bytes'] - info[1]) / ((self.prev['writed_bytes'] - info[0]) * self.Interval)
+                zbx.send('pgsql.cfs.activity[current_compress_ratio]', val)
+        self.prev['writed_bytes'] = info[0]
+        self.prev['scanned_bytes'] = info[1]
+
+        zbx.send('pgsql.cfs.activity[compressed_files]', info[2], delta=self.DELTA_SPEED, only_positive_speed=True)
+        zbx.send('pgsql.cfs.activity[scanned_files]', info[3], delta=self.DELTA_SPEED, only_positive_speed=True)
+
+    def items(self, template):
+        return template.item({
+            'name': 'PostgreSQL cfs compression: Writed byte/s',
+            'key': 'pgsql.cfs.activity[writed_bytes]',
+            'delay': self.Interval
+        }) + template.item({
+            'name': 'PostgreSQL cfs compression: Scanned byte/s',
+            'key': 'pgsql.cfs.activity[scanned_bytes]',
+            'delay': self.Interval
+        }) + template.item({
+            'name': 'PostgreSQL cfs compression: compressed files/s',
+            'key': 'pgsql.cfs.activity[compressed_files]',
+            'delay': self.Interval
+        }) + template.item({
+            'name': 'PostgreSQL cfs compression: scanned files/s',
+            'key': 'pgsql.cfs.activity[scanned_files]',
+            'delay': self.Interval
+        }) + template.item({
+            'name': 'PostgreSQL cfs compression: current ratio',
+            'key': 'pgsql.cfs.activity[current_compress_ratio]',
+            'delay': self.Interval
+        }) + template.item({
+            'name': 'PostgreSQL cfs compression: total ratio',
+            'key': 'pgsql.cfs.activity[total_compress_ratio]',
+            'delay': self.Interval
+        })
+
+    def graphs(self, template):
+        result = template.graph({
+            'name': 'PostgreSQL cfs compression: current ratio',
+            'items': [{
+                'key': 'pgsql.cfs.activity[current_compress_ratio]',
+                'color': '00CC00'
+            }]
+        })
+        result += template.graph({
+            'name': 'PostgreSQL cfs compression: compressed files',
+            'items': [{
+                'key': 'pgsql.cfs.activity[compressed_files]',
+                'color': '00CC00'
+            }]
+        })
+        result += template.graph({
+            'name': 'PostgreSQL cfs compression: writed bytes',
+            'items': [{
+                'key': 'pgsql.cfs.activity[writed_bytes]',
+                'color': '00CC00'
+            }]
+        })
+        return result
+
+    def discovery_rules(self, template):
+        rule = {
+            'name': 'Compressed relations discovery',
+            'key': 'pgsql.cfs.discovery_compressed_relations[]',
+            'filter': '{#COMPRESSED_RELATION}:.*'
+        }
+        items = [
+            {'key': 'pgsql.cfs.relation[{#COMPRESSED_RELATION}]',
+                'name': 'Relation {#COMPRESSED_RELATION}: compression ratio',
+                'delay': self.Interval}
+        ]
+        graphs = [
+            {
+                'name': 'Relation {#COMPRESSED_RELATION}: compression ratio',
+                'delay': self.Interval,
+                'items': [
+                    {'color': '00CC00',
+                        'key': 'pgsql.cfs.relation[{#COMPRESSED_RELATION}]'}]
+            },
+        ]
+        return template.discovery_rule(rule=rule, items=items, graphs=graphs)

mamonsu/plugins/pgsql/driver/_connection.py

@@ -10,15 +10,15 @@
 class ConnectionInfo(object):
 
     def __init__(self, info={}):
-        self._connection_info = info
-        self.host = info.get('host') or os.environ.get('PGHOST')
-        self.port = info.get('port') or int(os.environ.get('PGPORT') or 5432)
-        self.user = info.get('user') or os.environ.get('PGUSER')
-        self.passwd = info.get('passwd') or os.environ.get('PGPASSWORD')
-        self.db = info.get('db') or os.environ.get('PGDATABASE')
-        self.timeout = info.get('timeout') or int(
+        self.info = info
+        self.host = self.info.get('host') or os.environ.get('PGHOST')
+        self.port = self.info.get('port') or int(os.environ.get('PGPORT') or 5432)
+        self.user = self.info.get('user') or os.environ.get('PGUSER')
+        self.passwd = self.info.get('passwd') or os.environ.get('PGPASSWORD')
+        self.db = self.info.get('db') or os.environ.get('PGDATABASE')
+        self.timeout = self.info.get('timeout') or int(
             os.environ.get('PGTIMEOUT') or 1)
-        self.appname = info.get('appname') or os.environ.get('PGAPPNAME')
+        self.appname = self.info.get('appname') or os.environ.get('PGAPPNAME')
         self.log = logging.getLogger('PGSQL-({0})'.format(self.conn_str()))
 
     def conn_str(self):

mamonsu/plugins/pgsql/driver/pool.py

@@ -50,13 +50,13 @@ def __init__(self):
         }
 
     def connection_string(self, db=None):
-        self._init_conn_(db)
+        self._init_connection(db)
         return self.all_connections[db].conn_str()
 
     def query(self, query, db=None):
         if db is None:
             db = self.db
-        self._init_conn_(db)
+        self._init_connection(db)
         return self.all_connections[db].query(query)
 
     def server_version(self, db=None):
@@ -115,7 +115,7 @@ def is_pgpro(self, db=None):
         return self._cache['pgpro'][db]
 
     def is_pgpro_ee(self, db=None):
-        if not self.is_pgpro(self, db):
+        if not self.is_pgpro(db):
             return False
         if db in self._cache['pgproee']:
             return self._cache['pgproee'][db]
@@ -149,9 +149,9 @@ def get_sql(self, typ, db=None):
     def run_sql_type(self, typ, db=None):
         return self.query(self.get_sql(typ, db), db)
 
-    def _init_conn_(self, db):
-        conn = self.all_connections.get(db)
-        if conn is None:
-            info = self._connection_info
+    def _init_connection(self, db):
+        if db not in self.all_connections:
+            # create new connection
+            info = self.info
             info['db'] = db
             self.all_connections[db] = Connection(info)

mamonsu/plugins/pgsql/oldest.py

@@ -30,22 +30,20 @@ def run(self, zbx):
         zbx.send('pgsql.oldest[query_time]', query)
 
     def graphs(self, template):
-        graph = {
+        result = template.graph({
             'name': 'PostgreSQL oldest query running time',
             'items': [{
                 'key': 'pgsql.oldest[query_time]',
                 'color': '00CC00'
             }]
-        }
-        result = template.graph(graph)
-        graph = {
+        })
+        result += template.graph({
             'name': 'PostgreSQL age of oldest xid',
             'items': [{
                 'key': 'pgsql.oldest[xid_age]',
                 'color': '00CC00'
             }]
-        }
-        result += template.graph(graph)
+        })
         return result
 
     def items(self, template):

mamonsu/plugins/pgsql/plugin.py

@@ -44,5 +44,5 @@ def disable_and_exit_if_extension_is_not_installed(self, ext, db=None):
     def disable_and_exit_if_not_pgpro_ee(self, db=None):
         if not Pooler.is_pgpro_ee(db):
             raise PluginDisableException("""Disable plugin and exit, because \
-PostgresPro Enterprise Edition is not detected [instance: '{1}']
+PostgresPro Enterprise Edition is not detected [instance: '{0}']
 """.format(Pooler.connection_string(db)))

mamonsu/plugins/pgsql/pool.py

@@ -1,4 +1,4 @@
 # -*- coding: utf-8 -*-
 
 from mamonsu.plugins.pgsql.driver.pool import Pool
-Pooler = Pool()
+Pooler = Pool()  # create connection with environment var

mamonsu/plugins/system/linux/disk_stats.py

@@ -41,7 +41,7 @@ def run(self, zbx):
                 zbx.send('system.disk.write[{0}]'.format(
                     dev), write, self.DELTA_SPEED)
                 zbx.send('system.disk.utilization[{0}]'.format(
-                    dev), ticks/10, self.DELTA_SPEED)
+                    dev), ticks / 10, self.DELTA_SPEED)
 
             zbx.send('system.disk.all_read[]', all_read, self.DELTA_SPEED)
             zbx.send('system.disk.all_write[]', all_write, self.DELTA_SPEED)