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,75 @@
 )
 from sqlalchemy.exc import OperationalError, ProgrammingError, SQLAlchemyError
 
+try:
+    from http.server import HTTPServer, BaseHTTPRequestHandler
+    from threading import Thread
+except ImportError:
+    HTTPServer = None
+
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 MAX_RETRIES = 3
 logger = logging.getLogger(__name__)
+engine = None
+
+if HTTPServer is not None:
+
+    class HealthHandler(BaseHTTPRequestHandler):
+        def log_message(self, format, *args):
+            logger.debug("HTTP: %s", format % args)
+
+        def do_GET(self):
+            if self.path == "/ready":
+                self.handle_ready()
+            elif self.path == "/live":
+                self.handle_live()
+            else:
+                self.send_response(404)
+                self.end_headers()
+
+        def handle_ready(self):
+            status = {"database": "healthy"}
+            all_healthy = True
+
+            try:
+                if engine is not None:
+                    with engine.connect() as conn:
+                        conn.execute(sqlalchemy.text("SELECT 1"))
+            except Exception:
+                status["database"] = "unreachable"
+                all_healthy = False
+
+            if all_healthy:
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                self.wfile.write(json.dumps({"status": "ready", **status}).encode())
+            else:
+                self.send_response(503)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                self.wfile.write(json.dumps({"status": "not ready", **status}).encode())
+
+        def handle_live(self):
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(json.dumps({"status": "live"}).encode())
+
+    health_server = None
+    health_thread = None
+
+    def start_health_server(port: int):
+        global health_server, health_thread
+        health_server = HTTPServer(("0.0.0.0", port), HealthHandler)
+        health_thread = Thread(target=health_server.serve_forever, daemon=True)
+        health_thread.start()
+        logger.info("Health server started on port %d", port)
+
+    def stop_health_server():
+        global health_server
+        if health_server:
+            health_server.shutdown()
 
 
 @click.command(context_settings=CONTEXT_SETTINGS)
@@ -62,13 +130,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 +157,7 @@ def cli(
 
     metadata = MetaData()
     try:
+        global engine
         engine = create_engine(
             db_uri,
             pool_size=5,
@@ -97,6 +173,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
@@ -109,7 +188,11 @@ def cli(
 
     sub.add_callback(name="database", function=callback_function)
 
-    sub.start()
+    try:
+        sub.start()
+    finally:
+        if health_port is not None:
+            stop_health_server()
 
 
 def process_chain(chain, engine, issues_table):

ers-protobuf-dbwriter/entrypoint.sh

@@ -15,4 +15,5 @@ exec python3 ./dbwriter.py --subscriber-bootstrap "${ERS_DBWRITER_KAFKA_BOOTSTRA
                       --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,73 @@
 import opmonlib.opmon_entry_pb2 as opmon_schema
 from influxdb import InfluxDBClient
 
+try:
+    from http.server import HTTPServer, BaseHTTPRequestHandler
+    from threading import Thread
+except ImportError:
+    HTTPServer = None
+
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 logger = logging.getLogger(__name__)
+influx = None
+
+if HTTPServer is not None:
+
+    class HealthHandler(BaseHTTPRequestHandler):
+        def log_message(self, format, *args):
+            logger.debug("HTTP: %s", format % args)
+
+        def do_GET(self):
+            if self.path == "/ready":
+                self.handle_ready()
+            elif self.path == "/live":
+                self.handle_live()
+            else:
+                self.send_response(404)
+                self.end_headers()
+
+        def handle_ready(self):
+            status = {"influxdb": "healthy"}
+            all_healthy = True
+
+            try:
+                if influx is not None:
+                    influx.get_list_database()
+            except Exception:
+                status["influxdb"] = "unreachable"
+                all_healthy = False
+
+            if all_healthy:
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                self.wfile.write(json.dumps({"status": "ready", **status}).encode())
+            else:
+                self.send_response(503)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                self.wfile.write(json.dumps({"status": "not ready", **status}).encode())
+
+        def handle_live(self):
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(json.dumps({"status": "live"}).encode())
+
+    health_server = None
+    health_thread = None
+
+    def start_health_server(port: int):
+        global health_server, health_thread
+        health_server = HTTPServer(("0.0.0.0", port), HealthHandler)
+        health_thread = Thread(target=health_server.serve_forever, daemon=True)
+        health_thread.start()
+        logger.info("Health server started on port %d", port)
+
+    def stop_health_server():
+        global health_server
+        if health_server:
+            health_server.shutdown()
 
 
 @click.command(context_settings=CONTEXT_SETTINGS)
@@ -67,6 +133,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 +148,7 @@ def cli(
     influxdb_create,
     influxdb_timeout,
     debug,
+    health_port,
 ):
     logging.basicConfig(
         format="%(asctime)s %(levelname)-8s %(message)s",
@@ -115,14 +188,21 @@ 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(
         target=consume, daemon=True, args=(q, influxdb_timeout, influx)
     )
     thread.start()
 
-    sub.start()
+    try:
+        sub.start()
+    finally:
+        if health_port is not None:
+            stop_health_server()
 
 
 def consume(q: queue.Queue, timeout_ms, influx: InfluxDBClient = None):