python.d/postgres.chart.py

   1 # -*- coding: utf-8 -*-
   2 # Description: example netdata python.d module
   3 # Author: Pawel Krupa (paulfantom)
   4
   5 import psycopg2
   6 from base import SimpleService
   7 from psycopg2.extras import DictCursor
   8 from psycopg2 import extensions
   9 from psycopg2 import extras
  10
  11 NAME = "test"
  12
  13 # default module values
  14 update_every = 1
  15 priority = 90000
  16 retries = 60
  17
  18 # Default Config options.
  19 # {
  20 #    'database': None,
  21 #    'user': 'postgres',
  22 #    'password': None,
  23 #    'host': 'localhost',
  24 #    'port': 5432
  25 # }
  26
  27 ARCHIVE = """
  28 SELECT
  29     CAST(COUNT(*) AS INT) AS file_count,
  30     CAST(COALESCE(SUM(CAST(archive_file ~ $r$\.ready$$r$ as INT)), 0) AS INT) AS ready_count,
  31     CAST(COALESCE(SUM(CAST(archive_file ~ $r$\.done$$r$ AS INT)), 0) AS INT) AS done_count
  32 FROM
  33     pg_catalog.pg_ls_dir('pg_xlog/archive_status') AS archive_files (archive_file);
  34 """
  35
  36 BACKENDS = """
  37 SELECT
  38     count(*) - (SELECT count(*) FROM pg_stat_activity WHERE state = 'idle') AS backends_active,
  39     (SELECT count(*) FROM pg_stat_activity WHERE state = 'idle' ) AS backends_idle
  40 FROM
  41     pg_stat_activity;
  42 """
  43
  44 TABLE_SIZE_ON_DISK = """
  45 SELECT ((sum(relpages)* 8) * 1024) AS size_relations FROM pg_class WHERE relkind IN ('r', 't');
  46 """
  47
  48 TABLE_COUNT = """
  49 SELECT count(1) as relations FROM pg_class WHERE relkind IN ('r', 't');
  50 """
  51
  52 INDEX_SIZE_ON_DISK = """
  53 SELECT ((sum(relpages)* 8) * 1024) AS size_indexes FROM pg_class WHERE relkind = 'i';
  54 """
  55
  56 INDEX_COUNT = """
  57 SELECT count(1) as indexes FROM pg_class WHERE relkind = 'i';
  58 """
  59
  60 DATABASE = """
  61 SELECT
  62     sum(xact_commit) AS xact_commit,
  63     sum(xact_rollback) AS xact_rollback,
  64     sum(blks_read) AS blks_read,
  65     sum(blks_hit) AS blks_hit,
  66     sum(tup_returned) AS tup_returned,
  67     sum(tup_fetched) AS tup_fetched,
  68     sum(tup_inserted) AS tup_inserted,
  69     sum(tup_updated) AS tup_updated,
  70     sum(tup_deleted) AS tup_deleted,
  71     sum(conflicts) AS conflicts
  72 FROM pg_stat_database;
  73 """
  74
  75 STATIO = """
  76 SELECT
  77     sum(heap_blks_read) AS heap_blocks_read,
  78     sum(heap_blks_hit) AS heap_blocks_hit,
  79     sum(idx_blks_read) AS index_blocks_read,
  80     sum(idx_blks_hit) AS index_blocks_hit,
  81     sum(toast_blks_read) AS toast_blocks_read,
  82     sum(toast_blks_hit) AS toast_blocks_hit,
  83     sum(tidx_blks_read) AS toastindex_blocks_read,
  84     sum(tidx_blks_hit) AS toastindex_blocks_hit
  85 FROM
  86     pg_statio_all_tables
  87 WHERE
  88     schemaname <> 'pg_catalog';
  89 """
  90 BGWRITER = 'SELECT * FROM pg_stat_bgwriter;'
  91 LOCKS = 'SELECT mode, count(mode) AS count FROM pg_locks GROUP BY mode ORDER BY mode;'
  92 REPLICATION = """
  93 SELECT
  94     client_hostname,
  95     client_addr,
  96     state,
  97     sent_offset - (
  98         replay_offset - (sent_xlog - replay_xlog) * 255 * 16 ^ 6 ) AS byte_lag
  99 FROM (
 100     SELECT
 101         client_addr, client_hostname, state,
 102         ('x' || lpad(split_part(sent_location,   '/', 1), 8, '0'))::bit(32)::bigint AS sent_xlog,
 103         ('x' || lpad(split_part(replay_location, '/', 1), 8, '0'))::bit(32)::bigint AS replay_xlog,
 104         ('x' || lpad(split_part(sent_location,   '/', 2), 8, '0'))::bit(32)::bigint AS sent_offset,
 105         ('x' || lpad(split_part(replay_location, '/', 2), 8, '0'))::bit(32)::bigint AS replay_offset
 106     FROM pg_stat_replication
 107 ) AS s;
 108 """
 109
 110 LOCK_MAP = {'AccessExclusiveLock': 'lock_access_exclusive',
 111             'AccessShareLock': 'lock_access_share',
 112             'ExclusiveLock': 'lock_exclusive',
 113             'RowExclusiveLock': 'lock_row_exclusive',
 114             'RowShareLock': 'lock_row_share',
 115             'ShareUpdateExclusiveLock': 'lock_update_exclusive_lock',
 116             'ShareLock': 'lock_share',
 117             'ShareRowExclusiveLock': 'lock_share_row_exclusive',
 118             'SIReadLock': 'lock_si_read'}
 119
 120 ORDER = ['db_stat_transactions', 'db_stat_tuple_read', 'db_stat_tuple_returned', 'db_stat_tuple_write', 'backend_process', 'index_count', 'index_size', 'table_count', 'table_size', 'locks', 'wal', 'operations_heap', 'operations_index', 'operations_toast', 'operations_toast_index', 'background_writer']
 121
 122 CHARTS = {
 123     'db_stat_transactions': {
 124         'options': [None, 'Transactions', 'Count', 'Database Stat', 'postgres.db_stat_transactions', 'line'],
 125         'lines': [
 126             ['db_stat_xact_commit', 'Committed', 'absolute'],
 127             ['db_stat_xact_rollback', 'Rolled Back', 'absolute']
 128         ]},
 129     'db_stat_tuple_read': {
 130         'options': [None, 'Tuple read', 'Count', 'Database Stat', 'postgres.db_stat_tuple_read', 'line'],
 131         'lines': [
 132             ['db_stat_blks_read', 'Disk', 'absolute'],
 133             ['db_stat_blks_hit', 'Cache', 'absolute']
 134         ]},
 135     'db_stat_tuple_returned': {
 136         'options': [None, 'Tuple returned', 'Count', 'Database Stat', 'postgres.db_stat_tuple_returned', 'line'],
 137         'lines': [
 138             ['db_stat_tup_returned', 'Sequential', 'absolute'],
 139             ['db_stat_tup_fetched', 'Bitmap', 'absolute']
 140         ]},
 141     'db_stat_tuple_write': {
 142         'options': [None, 'Tuple write', 'Count', 'Database Stat', 'postgres.db_stat_tuple_write', 'line'],
 143         'lines': [
 144             ['db_stat_tup_inserted', 'Inserted', 'absolute'],
 145             ['db_stat_tup_updated', 'Updated', 'absolute'],
 146             ['db_stat_tup_deleted', 'Deleted', 'absolute'],
 147             ['db_stat_conflicts', 'Conflicts', 'absolute']
 148         ]},
 149     'backend_process': {
 150         'options': [None, 'Backend processes', 'Count', 'Backend processes', 'postgres.backend_process', 'line'],
 151         'lines': [
 152             ['backend_process_active', 'Active', 'absolute'],
 153             ['backend_process_idle', 'Idle', 'absolute']
 154         ]},
 155     'index_count': {
 156         'options': [None, 'Total index', 'Count', 'Index', 'postgres.index_count', 'line'],
 157         'lines': [
 158             ['index_count', 'Total index', 'absolute']
 159         ]},
 160     'index_size': {
 161         'options': [None, 'Index size', 'MB', 'Index', 'postgres.index_size', 'line'],
 162         'lines': [
 163             ['index_size', 'Size', 'absolute', 1, 1024*1024]
 164         ]},
 165     'table_count': {
 166         'options': [None, 'Total table', 'Count', 'Table', 'postgres.table_count', 'line'],
 167         'lines': [
 168             ['table_count', 'Total table', 'absolute']
 169         ]},
 170     'table_size': {
 171         'options': [None, 'Table size', 'MB', 'Table', 'postgres.table_size', 'line'],
 172         'lines': [
 173             ['table_size', 'Size', 'absolute', 1, 1024*1024]
 174         ]},
 175     'locks': {
 176         'options': [None, 'Table size', 'Count', 'Locks', 'postgres.locks', 'line'],
 177         'lines': [
 178             ['lock_access_exclusive', 'Access Exclusive', 'absolute'],
 179             ['lock_access_share', 'Access Share', 'absolute'],
 180             ['lock_exclusive', 'Exclusive', 'absolute'],
 181             ['lock_row_exclusive', 'Row Exclusive', 'absolute'],
 182             ['lock_row_share', 'Row Share', 'absolute'],
 183             ['lock_update_exclusive_lock', 'Update Exclusive Lock', 'absolute'],
 184             ['lock_share', 'Share', 'absolute'],
 185             ['lock_share_row_exclusive', 'Share Row Exclusive', 'absolute'],
 186             ['lock_si_read', 'SI Read', 'absolute']
 187         ]},
 188     'wal': {
 189         'options': [None, 'WAL stats', 'Files', 'WAL', 'postgres.wal', 'line'],
 190         'lines': [
 191             ['wal_total', 'Total', 'absolute'],
 192             ['wal_ready', 'Ready', 'absolute'],
 193             ['wal_done', 'Done', 'absolute']
 194         ]},
 195     'operations_heap': {
 196         'options': [None, 'Heap', 'iops', 'IO Operations', 'postgres.operations_heap', 'line'],
 197         'lines': [
 198             ['operations_heap_blocks_read', 'Read', 'absolute'],
 199             ['operations_heap_blocks_hit', 'Hit', 'absolute']
 200         ]},
 201     'operations_index': {
 202         'options': [None, 'Index', 'iops', 'IO Operations', 'postgres.operations_index', 'line'],
 203         'lines': [
 204             ['operations_index_blocks_read', 'Read', 'absolute'],
 205             ['operations_index_blocks_hit', 'Hit', 'absolute']
 206         ]},
 207     'operations_toast': {
 208         'options': [None, 'Toast', 'iops', 'IO Operations', 'postgres.operations_toast', 'line'],
 209         'lines': [
 210             ['operations_toast_blocks_read', 'Read', 'absolute'],
 211             ['operations_toast_blocks_hit', 'Hit', 'absolute']
 212         ]},
 213     'operations_toast_index': {
 214         'options': [None, 'Toast index', 'iops', 'IO Operations', 'postgres.operations_toast_index', 'line'],
 215         'lines': [
 216             ['operations_toastindex_blocks_read', 'Read', 'absolute'],
 217             ['operations_toastindex_blocks_hit', 'Hit', 'absolute']
 218         ]},
 219     'background_writer': {
 220         'options': [None, 'Checkpoints', 'Count', 'Background Writer', 'postgres.background_writer', 'line'],
 221         'lines': [
 222             ['background_writer_scheduled', 'Scheduled', 'absolute'],
 223             ['background_writer_requested', 'Requested', 'absolute']
 224         ]}
 225 }
 226
 227 class Service(SimpleService):
 228     def __init__(self, configuration=None, name=None):
 229         super(self.__class__, self).__init__(configuration=configuration, name=name)
 230         self.order = ORDER
 231         self.definitions = CHARTS
 232         self.configuration = configuration
 233         self.connection = None
 234         self.data = {}
 235         self.old_data = {}
 236
 237     def connect(self):
 238         params = dict(user='postgres',
 239                       database=None,
 240                       password=None,
 241                       host='localhost',
 242                       port=5432)
 243         params.update(self.configuration)
 244         self.connection = psycopg2.connect(**params)
 245         self.connection.set_isolation_level(extensions.ISOLATION_LEVEL_AUTOCOMMIT)
 246
 247     def check(self):
 248         try:
 249             self.connect()
 250             return True
 251         except Exception as e:
 252             self.error(e)
 253             return False
 254
 255     def _get_data(self):
 256         self.connect()
 257
 258         cursor = self.connection.cursor(cursor_factory=DictCursor)
 259         self.add_stats(cursor)
 260
 261         cursor.close()
 262         self.connection.close()
 263
 264         return self.data;
 265
 266     def add_stats(self, cursor):
 267         self.add_database_stats(cursor)
 268         self.add_backend_stats(cursor)
 269         self.add_index_stats(cursor)
 270         self.add_table_stats(cursor)
 271         self.add_lock_stats(cursor)
 272         self.add_statio_stats(cursor)
 273         self.add_bgwriter_stats(cursor)
 274
 275         #self.add_replication_stats(cursor)
 276
 277         # add_wal_metrics needs superuser to get directory listings
 278         #if self.config.get('superuser', True):
 279         #self.add_wal_stats(cursor)
 280
 281     def add_database_stats(self, cursor):
 282         cursor.execute(DATABASE)
 283         temp = cursor.fetchone()
 284
 285         self.add_derive_value('db_stat_xact_commit', int(temp.get('xact_commit', 0)))
 286         self.add_derive_value('db_stat_xact_rollback', int(temp.get('xact_rollback', 0)))
 287         self.add_derive_value('db_stat_blks_read', int(temp.get('blks_read', 0)))
 288         self.add_derive_value('db_stat_blks_hit', int(temp.get('blks_hit', 0)))
 289         self.add_derive_value('db_stat_tup_returned', int(temp.get('tup_returned', 0)))
 290         self.add_derive_value('db_stat_tup_fetched', int(temp.get('tup_fetched', 0)))
 291         self.add_derive_value('db_stat_tup_inserted', int(temp.get('tup_inserted', 0)))
 292         self.add_derive_value('db_stat_tup_updated', int(temp.get('tup_updated', 0)))
 293         self.add_derive_value('db_stat_tup_deleted', int(temp.get('tup_deleted', 0)))
 294         self.add_derive_value('db_stat_conflicts', int(temp.get('conflicts', 0)))
 295
 296     def add_backend_stats(self, cursor):
 297         cursor.execute(BACKENDS)
 298         temp = cursor.fetchone()
 299
 300         self.data['backend_process_active'] = int(temp.get('backends_active', 0))
 301         self.data['backend_process_idle'] = int(temp.get('backends_idle', 0))
 302
 303     def add_index_stats(self, cursor):
 304         cursor.execute(INDEX_COUNT)
 305         temp = cursor.fetchone()
 306         self.data['index_count'] = int(temp.get('indexes', 0))
 307
 308         cursor.execute(INDEX_SIZE_ON_DISK)
 309         temp = cursor.fetchone()
 310         self.data['index_size'] = int(temp.get('size_indexes', 0))
 311
 312     def add_table_stats(self, cursor):
 313         cursor.execute(TABLE_COUNT)
 314         temp = cursor.fetchone()
 315         self.data['table_count'] = int(temp.get('relations', 0))
 316
 317         cursor.execute(TABLE_SIZE_ON_DISK)
 318         temp = cursor.fetchone()
 319         self.data['table_size'] = int(temp.get('size_relations', 0))
 320
 321     def add_lock_stats(self, cursor):
 322         cursor.execute(LOCKS)
 323         temp = cursor.fetchall()
 324         for key in LOCK_MAP:
 325             found = False
 326             for row in temp:
 327                 if row['mode'] == key:
 328                     found = True
 329                     self.data[LOCK_MAP[key]] = int(row['count'])
 330
 331             if not found:
 332                     self.data[LOCK_MAP[key]] = 0
 333
 334     def add_wal_stats(self, cursor):
 335         cursor.execute(ARCHIVE)
 336         temp = cursor.fetchone()
 337         self.add_derive_value('wal_total', int(temp.get('file_count', 0)))
 338         self.add_derive_value('wal_ready', int(temp.get('ready_count', 0)))
 339         self.add_derive_value('wal_done', int(temp.get('done_count', 0)))
 340
 341     def add_statio_stats(self, cursor):
 342         cursor.execute(STATIO)
 343         temp = cursor.fetchone()
 344         self.add_derive_value('operations_heap_blocks_read', int(temp.get('heap_blocks_read', 0)))
 345         self.add_derive_value('operations_heap_blocks_hit', int(temp.get('heap_blocks_hit', 0)))
 346         self.add_derive_value('operations_index_blocks_read', int(temp.get('index_blocks_read', 0)))
 347         self.add_derive_value('operations_index_blocks_hit', int(temp.get('index_blocks_hit', 0)))
 348         self.add_derive_value('operations_toast_blocks_read', int(temp.get('toast_blocks_read', 0)))
 349         self.add_derive_value('operations_toast_blocks_hit', int(temp.get('toast_blocks_hit', 0)))
 350         self.add_derive_value('operations_toastindex_blocks_read', int(temp.get('toastindex_blocks_read', 0)))
 351         self.add_derive_value('operations_toastindex_blocks_hit', int(temp.get('toastindex_blocks_hit', 0)))
 352
 353     def add_bgwriter_stats(self, cursor):
 354         cursor.execute(BGWRITER)
 355         temp = cursor.fetchone()
 356
 357         self.add_derive_value('background_writer_scheduled', temp.get('checkpoints_timed', 0))
 358         self.add_derive_value('background_writer_requested', temp.get('checkpoints_requests', 0))
 359
 360     def add_derive_value(self, key, value):
 361         if key not in self.old_data.keys():
 362             self.data[key] = 0
 363         else:
 364             self.data[key] = value - self.old_data[key]
 365
 366         self.old_data[key] = value
 367
 368
 369 '''
 370     def add_replication_stats(self, cursor):
 371         cursor.execute(REPLICATION)
 372         temp = cursor.fetchall()
 373         for row in temp:
 374             self.add_gauge_value('Replication/%s' % row.get('client_addr', 'Unknown'),
 375                                  'byte_lag',
 376                                  int(row.get('byte_lag', 0)))
 377 '''