replica & async rest API health check enhancement (#1599)

- ``GET /replica?lag=<max-lag>``: replica check endpoint.
- ``GET /asynchronous?lag=<max-lag>`` or ``GET /async&lag=<max-lag>``: asynchronous standby check endpoint.

Checks replication latency and returns status code **200** only when the latency is below a specified value. The key leader_optime from DCS is used for the leader WAL position and compute latency on the replica for performance reasons. Please note that the value in leader_optime might be a couple of seconds old (based on loop_wait).

Co-authored-by: Alexander Kukushkin <[email protected]>

docs/rest_api.rst

@@ -19,6 +19,13 @@ For all health check ``GET`` requests Patroni returns a JSON document with the s
 
 - ``GET /replica``: replica health check endpoint. It returns HTTP status code **200** only when the Patroni node is in the state ``running``, the role is ``replica`` and ``noloadbalance`` tag is not set.
 
+- ``GET /replica?lag=<max-lag>``: replica check endpoint. In addition to checks from ``replica``, it also checks replication latency and returns status code **200** only when it is below specified value. The key cluster.last_leader_operation from DCS is used for Leader wal position and compute latency on replica for performance reasons. max-lag can be specified in bytes (integer) or in human readable values, for e.g. 16kB, 64MB, 1GB.
+
+  - ``GET /replica?lag=1048576``
+  - ``GET /replica?lag=1024kB``
+  - ``GET /replica?lag=10MB``
+  - ``GET /replica?lag=1GB``
+
 - ``GET /read-only``: like the above endpoint, but also includes the primary.
 
 - ``GET /standby-leader``: returns HTTP status code **200** only when the Patroni node is running as the leader in a :ref:`standby cluster <standby_cluster>`.
@@ -27,6 +34,13 @@ For all health check ``GET`` requests Patroni returns a JSON document with the s
 
 - ``GET /asynchronous`` or ``GET /async``: returns HTTP status code **200** only when the Patroni node is running as an asynchronous standby.
 
+- ``GET /asynchronous?lag=<max-lag>`` or ``GET /async?lag=<max-lag>``: asynchronous standby check endpoint. In addition to checks from ``asynchronous`` or ``async``, it also checks replication latency and returns status code **200** only when it is below specified value. The key cluster.last_leader_operation from DCS is used for Leader wal position and compute latency on replica for performance reasons. max-lag can be specified in bytes (integer) or in human readable values, for e.g. 16kB, 64MB, 1GB.
+
+  - ``GET /async?lag=1048576``
+  - ``GET /async?lag=1024kB``
+  - ``GET /async?lag=10MB``
+  - ``GET /async?lag=1GB``
+
 - ``GET /health``: returns HTTP status code **200** only when PostgreSQL is up and running.
 
 - ``GET /liveness``: always returns HTTP status code **200** what only indicates that Patroni is running. Could be used for ``livenessProbe``.

patroni/api.py

@@ -9,9 +9,11 @@ import datetime
 import os
 import six
 import socket
+import sys
 
 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
 from six.moves.socketserver import ThreadingMixIn
+from six.moves.urllib_parse import urlparse, parse_qs
 from threading import Thread
 
 from .exceptions import PostgresConnectionException, PostgresException
@@ -92,7 +94,14 @@ class RestApiHandler(BaseHTTPRequestHandler):
         patroni = self.server.patroni
         cluster = patroni.dcs.cluster
 
-        replica_status_code = 200 if not patroni.noloadbalance and \
+        leader_optime = cluster and cluster.last_leader_operation or 0
+        replayed_location = response.get('xlog', {}).get('replayed_location', 0)
+        max_replica_lag = parse_int(self.path_query.get('lag', [sys.maxsize])[0], 'B')
+        if max_replica_lag is None:
+            max_replica_lag = sys.maxsize
+        is_lagging = leader_optime and leader_optime > replayed_location + max_replica_lag
+
+        replica_status_code = 200 if not patroni.noloadbalance and not is_lagging and \
             response.get('role') == 'replica' and response.get('state') == 'running' else 503
 
         if not cluster and patroni.ha.is_paused():
@@ -439,6 +448,9 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         ret = BaseHTTPRequestHandler.parse_request(self)
         if ret:
+            urlpath = urlparse(self.path)
+            self.path = urlpath.path
+            self.path_query = parse_qs(urlpath.query) or {}
             mname = self.path.lstrip('/').split('/')[0]
             mname = self.command + ('_' + mname if mname else '')
             if hasattr(self, 'do_' + mname):

tests/__init__.py

@@ -93,7 +93,7 @@ class MockCursor(object):
         elif sql.startswith('SELECT pg_catalog.to_char'):
             replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\
                                '"state":"streaming","sync_state":"async","sync_priority":0}]'
-            self.results = [('', 0, '', '', '', '', False, replication_info)]
+            self.results = [('', 0, '', 0, '', '', False, 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

@@ -161,7 +161,11 @@ class TestRestApiHandler(unittest.TestCase):
     _authorization = '\nAuthorization: Basic dGVzdDp0ZXN0'
 
     def test_do_GET(self):
+        MockPatroni.dcs.cluster.last_leader_operation = 20
         MockRestApiServer(RestApiHandler, 'GET /replica')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=10MB')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=10485760')
         MockRestApiServer(RestApiHandler, 'GET /read-only')
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})):
             MockRestApiServer(RestApiHandler, 'GET /replica')