Quick attempt at Prometheus (#1848)

Close https://github.com/zalando/patroni/issues/318

patroni/api.py

@@ -45,7 +45,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
         self.wfile.write(body.encode('utf-8'))
 
     def _write_json_response(self, status_code, response):
-        self._write_response(status_code, json.dumps(response), content_type='application/json')
+        self._write_response(status_code, json.dumps(response, default=str), content_type='application/json')
 
     def check_auth(func):
         """Decorator function to check authorization header or client certificates
@@ -180,6 +180,83 @@ class RestApiHandler(BaseHTTPRequestHandler):
         else:
             self.send_error(502)
 
+    def do_GET_metrics(self):
+        postgres = self.get_postgresql_status(True)
+        patroni = self.server.patroni
+        epoch = datetime.datetime(1970, 1, 1, tzinfo=tzutc)
+
+        metrics = []
+
+        metrics.append("# HELP patroni_version Patroni semver without periods.")
+        metrics.append("# TYPE patroni_version gauge")
+        padded_semver = ''.join([x.zfill(2) for x in patroni.version.split('.')])  # 2.0.2 => 020002
+        metrics.append("patroni_version {0}".format(padded_semver))
+
+        metrics.append("# HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise.")
+        metrics.append("# TYPE patroni_postgres_running gauge")
+        metrics.append("patroni_postgres_running {0}".format(int(postgres['state'] == 'running')))
+
+        metrics.append("# HELP patroni_postmaster_start_time Epoch seconds since Postgres started.")
+        metrics.append("# TYPE patroni_postmaster_start_time gauge")
+        postmaster_start_time = postgres.get('postmaster_start_time')
+        postmaster_start_time = (postmaster_start_time - epoch).total_seconds() if postmaster_start_time else 0
+        metrics.append("patroni_postmaster_start_time {0}".format(postmaster_start_time))
+
+        metrics.append("# HELP patroni_master Value is 1 if this node is the leader, 0 otherwise.")
+        metrics.append("# TYPE patroni_master gauge")
+        metrics.append("patroni_master {0}".format(int(postgres['role'] == 'master')))
+
+        metrics.append("# HELP patroni_xlog_location Current location of the Postgres"
+                       " transaction log, 0 if this node is not the leader.")
+        metrics.append("# TYPE patroni_xlog_location counter")
+        metrics.append("patroni_xlog_location {0}".format(postgres.get('xlog', {}).get('location', 0)))
+
+        metrics.append("# HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise.")
+        metrics.append("# TYPE patroni_standby_leader gauge")
+        metrics.append("patroni_standby_leader {0}".format(int(postgres['role'] == 'standby_leader')))
+
+        metrics.append("# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.")
+        metrics.append("# TYPE patroni_replica gauge")
+        metrics.append("patroni_replica {0}".format(int(postgres['role'] == 'replica')))
+
+        metrics.append("# HELP patroni_xlog_received_location Current location of the received"
+                       " Postgres transaction log, 0 if this node is not a replica.")
+        metrics.append("# TYPE patroni_xlog_received_location counter")
+        metrics.append("patroni_xlog_received_location {0}".format(
+                        postgres.get('xlog', {}).get('received_location', 0)))
+
+        metrics.append("# HELP patroni_xlog_replayed_location Current location of the replayed"
+                       " Postgres transaction log, 0 if this node is not a replica.")
+        metrics.append("# TYPE patroni_xlog_replayed_location counter")
+        metrics.append("patroni_xlog_replayed_location {0}".format(
+                        postgres.get('xlog', {}).get('replayed_location', 0)))
+
+        metrics.append("# HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed"
+                       " Postgres transaction log, 0 if null.")
+        metrics.append("# TYPE patroni_xlog_replayed_timestamp gauge")
+        replayed_timestamp = postgres.get('xlog', {}).get('replayed_timestamp')
+        replayed_timestamp = (replayed_timestamp - epoch).total_seconds() if replayed_timestamp else 0
+        metrics.append("patroni_xlog_replayed_timestamp {0}".format(replayed_timestamp))
+
+        metrics.append("# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.")
+        metrics.append("# TYPE patroni_xlog_paused gauge")
+        metrics.append("patroni_xlog_paused {0}".format(
+                        int(postgres.get('xlog', {}).get('paused', False) is True)))
+
+        metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.")
+        metrics.append("# TYPE patroni_postgres_server_version gauge")
+        metrics.append("patroni_postgres_server_version {0}".format(postgres.get('server_version', 0)))
+
+        metrics.append("# HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked.")
+        metrics.append("# TYPE patroni_cluster_unlocked gauge")
+        metrics.append("patroni_cluster_unlocked {0}".format(int(postgres['cluster_unlocked'])))
+
+        metrics.append("# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.")
+        metrics.append("# TYPE patroni_postgres_timeline counter")
+        metrics.append("patroni_postgres_timeline {0}".format(postgres.get('timeline', 0)))
+
+        self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain')
+
     def _read_json_content(self, body_is_optional=False):
         if 'content-length' not in self.headers:
             return self.send_error(411) if not body_is_optional else {}
@@ -476,7 +553,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             if postgresql.state not in ('running', 'restarting', 'starting'):
                 raise RetryFailedError('')
             stmt = ("SELECT " + postgresql.POSTMASTER_START_TIME + ", " + postgresql.TL_LSN + ","
-                    " pg_catalog.to_char(pg_catalog.pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),"
+                    " pg_catalog.pg_last_xact_replay_timestamp(),"
                     " pg_catalog.array_to_json(pg_catalog.array_agg(pg_catalog.row_to_json(ri))) "
                     "FROM (SELECT (SELECT rolname FROM pg_authid WHERE oid = usesysid) AS usename,"
                     " application_name, client_addr, w.state, sync_state, sync_priority"

patroni/postgresql/__init__.py

@@ -48,7 +48,7 @@ def null_context():
 
 class Postgresql(object):
 
-    POSTMASTER_START_TIME = "pg_catalog.to_char(pg_catalog.pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ')"
+    POSTMASTER_START_TIME = "pg_catalog.pg_postmaster_start_time()"
     TL_LSN = ("CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 "
               "ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name("
               "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, "  # master timeline
@@ -865,10 +865,10 @@ class Postgresql(object):
         try:
             query = "SELECT " + self.POSTMASTER_START_TIME
             if current_thread().ident == self.__thread_ident:
-                return self.query(query).fetchone()[0]
+                return self.query(query).fetchone()[0].isoformat(sep=' ')
             with self.connection().cursor() as cursor:
                 cursor.execute(query)
-                return cursor.fetchone()[0]
+                return cursor.fetchone()[0].isoformat(sep=' ')
         except psycopg2.Error:
             return None
 

tests/__init__.py

@@ -1,3 +1,4 @@
+import datetime
 import os
 import shutil
 import unittest
@@ -10,7 +11,7 @@ import urllib3
 from patroni.dcs import Leader, Member
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
-from patroni.utils import RetryFailedError
+from patroni.utils import RetryFailedError, tzutc
 
 
 class SleepException(Exception):
@@ -95,10 +96,11 @@ class MockCursor(object):
             self.results = [(1, 2, 1, 0, False, 1, 1, None, None)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
             self.results = [(False, 2)]
-        elif sql.startswith('SELECT pg_catalog.to_char'):
+        elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'):
             replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\
                                '"state":"streaming","sync_state":"async","sync_priority":0}]'
-            self.results = [('', 0, '', 0, '', '', False, replication_info)]
+            now = datetime.datetime.now(tzutc)
+            self.results = [(now, 0, '', 0, '', False, now, replication_info)]
         elif sql.startswith('SELECT name, setting'):
             self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
                             ('wal_block_size', '8192', None, 'integer', 'internal'),

tests/test_api.py

@@ -30,7 +30,7 @@ class MockPostgresql(object):
     pending_restart = True
     wal_name = 'wal'
     lsn_name = 'lsn'
-    POSTMASTER_START_TIME = 'pg_catalog.to_char(pg_catalog.pg_postmaster_start_time'
+    POSTMASTER_START_TIME = 'pg_catalog.pg_postmaster_start_time()'
     TL_LSN = 'CASE WHEN pg_catalog.pg_is_in_recovery()'
 
     @staticmethod
@@ -39,7 +39,7 @@ class MockPostgresql(object):
 
     @staticmethod
     def postmaster_start_time():
-        return str(postmaster_start_time)
+        return postmaster_start_time
 
     @staticmethod
     def replica_cached_timeline(_):
@@ -236,6 +236,10 @@ class TestRestApiHandler(unittest.TestCase):
         mock_dcs.cluster.config = None
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /config'))
 
+    @patch.object(MockPatroni, 'dcs')
+    def test_do_GET_metrics(self, mock_dcs):
+        self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /metrics'))
+
     @patch.object(MockPatroni, 'dcs')
     def test_do_PATCH_config(self, mock_dcs):
         config = {'postgresql': {'use_slots': False, 'use_pg_rewind': True, 'parameters': {'wal_level': 'logical'}}}

tests/test_postgresql.py

@@ -1,3 +1,4 @@
+import datetime
 import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
 import psutil
@@ -522,8 +523,9 @@ class TestPostgresql(BaseTestPostgresql):
             self.assertEqual(self.p.get_major_version(), 0)
 
     def test_postmaster_start_time(self):
-        with patch.object(MockCursor, "fetchone", Mock(return_value=('foo', True, '', '', '', '', False))):
-            self.assertEqual(self.p.postmaster_start_time(), 'foo')
+        now = datetime.datetime.now()
+        with patch.object(MockCursor, "fetchone", Mock(return_value=(now, True, '', '', '', '', False))):
+            self.assertEqual(self.p.postmaster_start_time(), now.isoformat(sep=' '))
             t = Thread(target=self.p.postmaster_start_time)
             t.start()
             t.join()