From 09f2f579d79c8afe01afc09b6b5ea3496912cc8d Mon Sep 17 00:00:00 2001 From: Mark Mercado Date: Thu, 4 Mar 2021 06:37:29 -0500 Subject: [PATCH] Quick attempt at Prometheus (#1848) Close https://github.com/zalando/patroni/issues/318 --- patroni/api.py | 81 +++++++++++++++++++++++++++++++++- patroni/postgresql/__init__.py | 6 +-- tests/__init__.py | 8 ++-- tests/test_api.py | 8 +++- tests/test_postgresql.py | 6 ++- 5 files changed, 97 insertions(+), 12 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 06be43ef..e02d5acf 100644 --- a/patroni/api.py +++ b/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" diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 7c5eb1f8..aced2d7c 100644 --- a/patroni/postgresql/__init__.py +++ b/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 diff --git a/tests/__init__.py b/tests/__init__.py index 32c65cc4..23861881 100644 --- a/tests/__init__.py +++ b/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'), diff --git a/tests/test_api.py b/tests/test_api.py index 75ef8ed1..ec91267e 100644 --- a/tests/test_api.py +++ b/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'}}} diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index b22d1a03..3a3937d4 100644 --- a/tests/test_postgresql.py +++ b/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()