Ensure workable application health probes

elisa-logbook/elisa-logbook-deployment.yaml

@@ -71,20 +71,29 @@ spec:
               name: http
               protocol: TCP
           startupProbe:
-            tcpSocket:
+            httpGet:
+              path: /ready
               port: http
             initialDelaySeconds: 15
             periodSeconds: 20
+            timeoutSeconds: 5
+            failureThreshold: 6
           livenessProbe:
-            tcpSocket:
+            httpGet:
+              path: /live
               port: http
-            initialDelaySeconds: 15
-            periodSeconds: 20
+            initialDelaySeconds: 60
+            periodSeconds: 30
+            timeoutSeconds: 10
+            failureThreshold: 3
           readinessProbe:
-            tcpSocket:
+            httpGet:
+              path: /ready
               port: http
-            initialDelaySeconds: 15
-            periodSeconds: 20
+            initialDelaySeconds: 5
+            periodSeconds: 5
+            timeoutSeconds: 5
+            failureThreshold: 3
           resources:
             limits:
               memory: 1Gi

elisa-logbook/logbook.py

@@ -83,6 +83,31 @@ def sanitize_run_type(run_type: str) -> str:
 # The first type of logging is fileLogbook, which writes logs to a given file in the current working directory.
 
 
+@app.route("/ready")
+def ready():
+    status = {"file_system": "healthy"}
+    all_healthy = True
+
+    try:
+        if not Path(app.config["PATH"]).exists():
+            status["file_system"] = "unreachable"
+            all_healthy = False
+
+    except Exception:
+        status["file_system"] = "error"
+        all_healthy = False
+
+    if all_healthy:
+        return jsonify({"status": "ready", **status}), 200
+    else:
+        return jsonify({"status": "not ready", **status}), 503
+
+
+@app.route("/live")
+def live():
+    return jsonify({"status": "live"}), 200
+
+
 @app.route("/")
 def index():
     return "<h1>Welcome to the logbook API!</h1>"

ers-protobuf-dbwriter/dbwriter.py

@@ -4,6 +4,7 @@
 #  received with this code.
 #
 
+import json
 import logging
 import re
 import sys
@@ -13,6 +14,7 @@
 import click
 import erskafka.ERSSubscriber as erssub
 import google.protobuf.json_format as pb_json
+import sqlalchemy
 from sqlalchemy import (
     BigInteger,
     Column,
@@ -25,9 +27,76 @@
 )
 from sqlalchemy.exc import OperationalError, ProgrammingError, SQLAlchemyError
 
+import socket as _socket
+from threading import Thread
+
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 MAX_RETRIES = 3
 logger = logging.getLogger(__name__)
+engine = None
+
+_health_server_socket = None
+
+
+def _handle_health_client(conn):
+    try:
+        data = b""
+        conn.settimeout(5)
+        while b"\r\n\r\n" not in data:
+            chunk = conn.recv(4096)
+            if not chunk:
+                break
+            data += chunk
+
+        if b"GET /ready" in data:
+            status = {"database": "healthy"}
+            all_healthy = True
+            try:
+                if engine is not None:
+                    with engine.connect() as c:
+                        c.execute(sqlalchemy.text("SELECT 1"))
+            except Exception:
+                status["database"] = "unreachable"
+                all_healthy = False
+            code, phrase = (200, "OK") if all_healthy else (503, "Service Unavailable")
+            body = json.dumps({"status": "ready" if all_healthy else "not ready", **status}).encode()
+        elif b"GET /live" in data:
+            code, phrase = 200, "OK"
+            body = json.dumps({"status": "live"}).encode()
+        else:
+            code, phrase = 404, "Not Found"
+            body = b"Not Found"
+
+        response = (
+            f"HTTP/1.0 {code} {phrase}\r\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n"
+        ).encode() + body
+        conn.sendall(response)
+    except Exception:
+        pass
+    finally:
+        try:
+            conn.close()
+        except Exception:
+            pass
+
+
+def _health_accept_loop(sock):
+    while True:
+        try:
+            conn, _ = sock.accept()
+            Thread(target=_handle_health_client, args=(conn,), daemon=True).start()
+        except OSError:
+            break
+
+
+def start_health_server(port: int):
+    global _health_server_socket
+    _health_server_socket = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
+    _health_server_socket.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1)
+    _health_server_socket.bind(("0.0.0.0", port))
+    _health_server_socket.listen(10)
+    Thread(target=_health_accept_loop, args=(_health_server_socket,), daemon=True).start()
+    logger.info("Health server started on port %d", port)
 
 
 @click.command(context_settings=CONTEXT_SETTINGS)
@@ -62,13 +131,20 @@
     help="name of table used in the database",
 )
 @click.option("--debug", type=click.BOOL, default=True, help="Set debug print levels")
+@click.option(
+    "--health-port",
+    type=click.INT,
+    default=None,
+    help="Port for HTTP health endpoint (if not set, no health endpoint)",
+)
 def cli(
     subscriber_bootstrap,
     subscriber_group,
     subscriber_timeout,
     db_uri,
     db_table,
     debug,
+    health_port,
 ):
     logging.basicConfig(
         format="%(asctime)s %(levelname)-8s %(message)s",
@@ -82,6 +158,7 @@ def cli(
 
     metadata = MetaData()
     try:
+        global engine
         engine = create_engine(
             db_uri,
             pool_size=5,
@@ -97,6 +174,9 @@ def cli(
 
     check_tables(engine=engine)
 
+    if health_port is not None:
+        start_health_server(health_port)
+
     subscriber_conf = {}
     subscriber_conf["bootstrap"] = subscriber_bootstrap
     subscriber_conf["timeout"] = subscriber_timeout

ers-protobuf-dbwriter/entrypoint.sh

@@ -8,11 +8,12 @@ if [[ ! -f ../entrypoint_functions.sh ]]; then
 fi
 source ../entrypoint_functions.sh
 
-ensure_required_variables "ERS_DBWRITER_KAFKA_BOOTSTRAP_SERVER ERS_DBWRITER_KAFKA_TIMEOUT_MS ERS_DBWRITER_KAFKA_GROUP DATABASE_URI ERS_DBWRITER_DB_TABLENAME"
+ensure_required_variables "ERS_DBWRITER_KAFKA_BOOTSTRAP_SERVER ERS_DBWRITER_KAFKA_TIMEOUT_MS ERS_DBWRITER_KAFKA_GROUP DATABASE_URI ERS_DBWRITER_DB_TABLENAME HEALTH_PORT"
 
 exec python3 ./dbwriter.py --subscriber-bootstrap "${ERS_DBWRITER_KAFKA_BOOTSTRAP_SERVER}" \
                       --subscriber-group "${ERS_DBWRITER_KAFKA_GROUP}" \
                       --subscriber-timeout "${ERS_DBWRITER_KAFKA_TIMEOUT_MS}" \
                       --db-uri "${DATABASE_URI}" \
                       --db-table "${ERS_DBWRITER_DB_TABLENAME}" \
+                      --health-port "${HEALTH_PORT}" \
                       --debug False

ers-protobuf-dbwriter/ers-dbwriter-deployment.yaml

@@ -50,6 +50,10 @@ spec:
         - image: ghcr.io/dune-daq/microservices:9685
           imagePullPolicy: Always
           name: ers-protobuf-dbwriter
+          ports:
+            - containerPort: 8080
+              protocol: TCP
+              name: health
           env:
             - name: MICROSERVICE
               value: ers-protobuf-dbwriter
@@ -66,6 +70,8 @@ spec:
                 secretKeyRef:
                   key: uri
                   name: ers-postgresql-svcbind-custom-user
+            - name: HEALTH_PORT
+              value: "8080"
           resources:
             limits:
               memory: 1Gi
@@ -86,6 +92,30 @@ spec:
           volumeMounts:
             - name: tmp-volume
               mountPath: /tmp
+          startupProbe:
+            httpGet:
+              path: /ready
+              port: health
+            initialDelaySeconds: 15
+            periodSeconds: 20
+            timeoutSeconds: 5
+            failureThreshold: 6
+          livenessProbe:
+            httpGet:
+              path: /live
+              port: health
+            initialDelaySeconds: 60
+            periodSeconds: 30
+            timeoutSeconds: 10
+            failureThreshold: 3
+          readinessProbe:
+            httpGet:
+              path: /ready
+              port: health
+            initialDelaySeconds: 5
+            periodSeconds: 5
+            timeoutSeconds: 5
+            failureThreshold: 3
       volumes:
         - name: tmp-volume
           emptyDir:

opmon-protobuf-dbwriter/dbwriter.py

@@ -4,6 +4,7 @@
 #  received with this code.
 #
 
+import json
 import logging
 import queue
 import threading
@@ -16,8 +17,74 @@
 import opmonlib.opmon_entry_pb2 as opmon_schema
 from influxdb import InfluxDBClient
 
+import socket as _socket
+from threading import Thread
+
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 logger = logging.getLogger(__name__)
+influx = None
+
+_health_server_socket = None
+
+
+def _handle_health_client(conn):
+    try:
+        data = b""
+        conn.settimeout(5)
+        while b"\r\n\r\n" not in data:
+            chunk = conn.recv(4096)
+            if not chunk:
+                break
+            data += chunk
+
+        if b"GET /ready" in data:
+            status = {"influxdb": "healthy"}
+            all_healthy = True
+            try:
+                if influx is not None:
+                    influx.ping()
+            except Exception:
+                status["influxdb"] = "unreachable"
+                all_healthy = False
+            code, phrase = (200, "OK") if all_healthy else (503, "Service Unavailable")
+            body = json.dumps({"status": "ready" if all_healthy else "not ready", **status}).encode()
+        elif b"GET /live" in data:
+            code, phrase = 200, "OK"
+            body = json.dumps({"status": "live"}).encode()
+        else:
+            code, phrase = 404, "Not Found"
+            body = b"Not Found"
+
+        response = (
+            f"HTTP/1.0 {code} {phrase}\r\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n"
+        ).encode() + body
+        conn.sendall(response)
+    except Exception:
+        pass
+    finally:
+        try:
+            conn.close()
+        except Exception:
+            pass
+
+
+def _health_accept_loop(sock):
+    while True:
+        try:
+            conn, _ = sock.accept()
+            Thread(target=_handle_health_client, args=(conn,), daemon=True).start()
+        except OSError:
+            break
+
+
+def start_health_server(port: int):
+    global _health_server_socket
+    _health_server_socket = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
+    _health_server_socket.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1)
+    _health_server_socket.bind(("0.0.0.0", port))
+    _health_server_socket.listen(10)
+    Thread(target=_health_accept_loop, args=(_health_server_socket,), daemon=True).start()
+    logger.info("Health server started on port %d", port)
 
 
 @click.command(context_settings=CONTEXT_SETTINGS)
@@ -67,6 +134,12 @@
     help="Size in ms of the batches sent to influx",
 )
 @click.option("--debug", type=click.BOOL, default=True, help="Set debug print levels")
+@click.option(
+    "--health-port",
+    type=click.INT,
+    default=None,
+    help="Port for HTTP health endpoint (if not set, no health endpoint)",
+)
 def cli(
     subscriber_bootstrap,
     subscriber_group,
@@ -76,6 +149,7 @@ def cli(
     influxdb_create,
     influxdb_timeout,
     debug,
+    health_port,
 ):
     logging.basicConfig(
         format="%(asctime)s %(levelname)-8s %(message)s",
@@ -85,6 +159,7 @@ def cli(
 
     # Create InfluxDB client using from_dsn for URI-based connection
     # The from_dsn method extracts database name from the URI path
+    global influx
     influx = InfluxDBClient.from_dsn(influxdb_uri)
 
     # Extract database name from URI using urlparse
@@ -115,6 +190,9 @@ def cli(
 
     callback_function = partial(process_entry, q=q)
 
+    if health_port is not None:
+        start_health_server(health_port)
+
     sub.add_callback(name="to_influx", function=callback_function)
 
     thread = threading.Thread(