initial

Author: integrii

Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.11-alpine
+
+WORKDIR /app
+COPY client.py .
+
+CMD ["python3", "/app/client.py"]

Makefile

@@ -0,0 +1,9 @@
+IMG ?= example-python-check:latest
+
+.PHONY: build push
+
+build:
+	docker build -t $(IMG) .
+
+push: build
+	docker push $(IMG)

README.md

@@ -0,0 +1,48 @@
+# Python Kuberhealthy Client
+
+This directory contains a minimal Python application that demonstrates how to
+report status back to [Kuberhealthy](https://github.com/kuberhealthy/kuberhealthy).
+The example loads the `KH_REPORTING_URL` and `KH_RUN_UUID` environment variables
+provided to checker pods and includes commented calls to `report_ok` and
+`report_error`.
+
+## Running the example
+
+Set the `KH_REPORTING_URL` and `KH_RUN_UUID` environment variables, add your
+check logic to `client.py`, and then run:
+
+```bash
+python3 client.py
+```
+
+Within the `main` function, uncomment either `report_ok()` or
+`report_error("message")` after your logic depending on the result.
+
+## Building and pushing the check
+
+Use the provided `Makefile` and `Dockerfile` to build and publish the check
+image.
+
+```bash
+make build IMG=myrepo/example-check:latest
+make push IMG=myrepo/example-check:latest
+```
+
+## Using in your own checks
+
+1. Add your check logic to `client.py` by replacing the placeholder in `main`.
+   Call `report_ok()` when the check succeeds or `report_error("message")`
+   when it fails.
+2. Build and push your image as shown above.
+3. Create a `KuberhealthyCheck` resource pointing at your image and apply it to any
+   cluster where Kuberhealthy runs:
+
+```yaml
+apiVersion: kuberhealthy.github.io/v2
+kind: KuberhealthyCheck
+metadata:
+  name: example-python-check
+spec:
+  image: myrepo/example-check:latest
+  runInterval: 1m
+```

client.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Example Kuberhealthy client in Python."""
+
+import json
+import os
+import urllib.request
+
+
+KH_REPORTING_URL = "KH_REPORTING_URL"
+KH_RUN_UUID = "KH_RUN_UUID"
+
+
+def _get_env(name: str) -> str:
+    """Return the value of the environment variable *name* or raise an error."""
+    value = os.getenv(name)
+    if not value:
+        raise EnvironmentError(f"{name} must be set")
+    return value
+
+
+def _post_status(payload: dict) -> None:
+    """Send *payload* to the Kuberhealthy reporting URL."""
+    url = _get_env(KH_REPORTING_URL)
+    run_uuid = _get_env(KH_RUN_UUID)
+    data = json.dumps(payload).encode("utf-8")
+    request = urllib.request.Request(
+        url,
+        data=data,
+        headers={"content-type": "application/json", "kh-run-uuid": run_uuid},
+    )
+    with urllib.request.urlopen(request, timeout=10) as response:  # nosec B310
+        response.read()
+
+
+def report_ok() -> None:
+    """Report a successful check to Kuberhealthy."""
+    _post_status({"OK": True, "Errors": []})
+
+
+def report_error(message: str) -> None:
+    """Report a failure to Kuberhealthy with *message* as the error."""
+    _post_status({"OK": False, "Errors": [message]})
+
+
+def main() -> None:
+    """Run the example client."""
+    # INSERT YOUR CHECK LOGIC HERE
+    # report_ok()
+    # report_error("something went wrong")
+    pass
+
+
+if __name__ == "__main__":
+    main()