Add MariaDB HTTP readiness service

.dockerignore

@@ -0,0 +1,8 @@
+.git
+.gitignore
+.dockerignore
+.venv
+__pycache__
+*.pyc
+.env
+tests

.gitignore

@@ -0,0 +1,4 @@
+.venv/
+__pycache__/
+*.py[cod]
+.env

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PORT=8080
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt \
+    && useradd --system --uid 10001 --create-home app
+
+COPY --chown=10001:10001 server.py .
+
+USER 10001:10001
+EXPOSE 8080
+
+CMD ["python", "server.py"]

README.md

@@ -0,0 +1,16 @@
+# MariaDB HTTP Check
+
+Small HTTP service for checking MariaDB connectivity from the Nubes HTTP runtime.
+
+Configuration is read from `DB_HOST`, `DB_PORT`, `DB_NAME`, `DB_USER`, and `DB_PASSWORD`.
+The default listening port is `8080`; it can be changed with `PORT`.
+
+- `GET /healthz` checks that the process is serving requests.
+- `GET /readyz` checks MariaDB with `SELECT 1` and returns `503` while the database is unavailable.
+- `GET /` returns the same readiness status as `/readyz`.
+
+Build the container image with:
+
+```sh
+docker build -t naeel/mariadb-check:0.1 .
+```

requirements.txt

@@ -0,0 +1 @@
+PyMySQL>=1.1,<2

server.py

@@ -0,0 +1,79 @@
+import json
+import os
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import urlsplit
+
+import pymysql
+
+
+PORT = int(os.environ.get("PORT", "8080"))
+DB_CONFIG = {
+    "host": os.environ.get("DB_HOST", "127.0.0.1"),
+    "port": int(os.environ.get("DB_PORT", "3306")),
+    "database": os.environ.get("DB_NAME", "maria"),
+    "user": os.environ.get("DB_USER", ""),
+    "password": os.environ.get("DB_PASSWORD", ""),
+    "charset": "utf8mb4",
+    "connect_timeout": 3,
+    "read_timeout": 3,
+    "write_timeout": 3,
+    "autocommit": True,
+}
+
+
+def database_ready():
+    connection = None
+    cursor = None
+    try:
+        connection = pymysql.connect(**DB_CONFIG)
+        cursor = connection.cursor()
+        cursor.execute("SELECT 1")
+        return cursor.fetchone() == (1,)
+    except (pymysql.MySQLError, OSError):
+        return False
+    finally:
+        if cursor is not None:
+            cursor.close()
+        if connection is not None:
+            connection.close()
+
+
+class Handler(BaseHTTPRequestHandler):
+    server_version = "MariaDBCheck/1.0"
+    sys_version = ""
+
+    def send_json(self, status, payload):
+        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        path = urlsplit(self.path).path
+        if path == "/healthz":
+            self.send_json(HTTPStatus.OK, {"status": "ok"})
+            return
+
+        if path in ("/", "/readyz"):
+            ready = database_ready()
+            status = HTTPStatus.OK if ready else HTTPStatus.SERVICE_UNAVAILABLE
+            self.send_json(
+                status,
+                {
+                    "service": "maria-http",
+                    "status": "ready" if ready else "unavailable",
+                    "database": "connected" if ready else "unavailable",
+                },
+            )
+            return
+
+        self.send_json(HTTPStatus.NOT_FOUND, {"error": "not_found"})
+
+
+if __name__ == "__main__":
+    server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
+    server.serve_forever()

tests/test_server.py

@@ -0,0 +1,29 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pymysql
+
+import server
+
+
+class DatabaseReadyTests(unittest.TestCase):
+    @patch("server.pymysql.connect")
+    def test_returns_true_for_select_one(self, connect):
+        connection = MagicMock()
+        cursor = connection.cursor.return_value
+        cursor.fetchone.return_value = (1,)
+        connect.return_value = connection
+
+        self.assertTrue(server.database_ready())
+        cursor.execute.assert_called_once_with("SELECT 1")
+        cursor.close.assert_called_once()
+        connection.close.assert_called_once()
+
+    @patch("server.pymysql.connect", side_effect=pymysql.MySQLError("unavailable"))
+    def test_returns_false_when_database_is_unavailable(self, connect):
+        self.assertFalse(server.database_ready())
+        connect.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()