apiVersion: v1 kind: ConfigMap metadata: name: starter-contract-stub data: contract_stub.py: | import datetime import hashlib import json import os import threading import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer LEDGER = os.environ["CP9_LEDGER_FILE"] LOCK = threading.Lock() VERSION = "cp9.starter.contract-stub/v1" def write_ledger(contract, handler, accepted, body=b""): parsed = urllib.parse.urlsplit(handler.path) event = { "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), "contract_version": VERSION, "contract": contract, "method": handler.command, "path": parsed.path, "query_keys": sorted(urllib.parse.parse_qs(parsed.query, keep_blank_values=True)), "body_bytes": len(body), "body_sha256": hashlib.sha256(body).hexdigest() if body else "", "accepted": accepted, } with LOCK: with open(LEDGER, "a", encoding="utf-8") as ledger: ledger.write(json.dumps(event, separators=(",", ":")) + "\n") class Handler(BaseHTTPRequestHandler): def do_GET(self): parsed = urllib.parse.urlsplit(self.path) if parsed.path == "/healthz": self.respond(200, {"status": "ok", "contract_version": VERSION}) return if parsed.path == "/ledger": with LOCK: try: with open(LEDGER, encoding="utf-8") as ledger: body = ledger.read().encode() except FileNotFoundError: body = b"" self.respond_bytes(200, body, "application/x-ndjson") return if parsed.path == "/metrics": self.respond_bytes(200, b"", "text/plain; version=0.0.4") return if self.server.server_port == 9090 and parsed.path in ("/api/v1/query", "/api/v1/query_range"): query = urllib.parse.parse_qs(parsed.query, keep_blank_values=True) required = {"query"} if parsed.path == "/api/v1/query" else {"query", "start", "step"} if any(not query.get(key, [""])[0] for key in required): write_ledger("o11y", self, False) self.respond(400, {"status": "error", "errorType": "bad_data", "error": "missing required query fields"}) return write_ledger("o11y", self, True) result_type = "vector" if parsed.path == "/api/v1/query" else "matrix" self.respond(200, {"status": "success", "data": {"resultType": result_type, "result": []}}) return self.reject() def do_PUT(self): parsed = urllib.parse.urlsplit(self.path) parts = parsed.path.removeprefix("/metrics/job/").split("/") if (self.server.server_port != 9091 or not parsed.path.startswith("/metrics/job/") or len(parts) % 2 != 1 or any(not part for part in parts)): self.reject() return try: length = int(self.headers.get("Content-Length", "0")) except ValueError: length = -1 if length < 0 or length > 1024 * 1024: write_ledger("metering", self, False) self.respond(400 if length < 0 else 413, {"status": "error"}) return body = self.rfile.read(length) write_ledger("metering", self, True, body) self.respond(202, {"status": "accepted"}) def do_POST(self): self.reject() def do_DELETE(self): self.reject() def do_PATCH(self): self.reject() def do_HEAD(self): self.reject() def do_OPTIONS(self): self.reject() def reject(self): contract = "o11y" if self.server.server_port == 9090 else "metering" write_ledger(contract, self, False) self.respond(404, {"status": "error", "error": "unknown contract"}) def respond(self, status, value): self.respond_bytes(status, json.dumps(value, separators=(",", ":")).encode() + b"\n") def respond_bytes(self, status, body, content_type="application/json"): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, format, *args): parsed = urllib.parse.urlsplit(self.path) print(json.dumps({"method": self.command, "path": parsed.path, "status": args[1]}), flush=True) def serve(port): ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever() threading.Thread(target=serve, args=(9090,), daemon=True).start() serve(9091) --- apiVersion: v1 kind: Service metadata: name: starter-o11y-stub spec: selector: app.kubernetes.io/name: starter-contract-stub ports: - name: http port: 9090 targetPort: o11y --- apiVersion: v1 kind: Service metadata: name: starter-metering-stub spec: selector: app.kubernetes.io/name: starter-contract-stub ports: - name: http port: 9091 targetPort: metering --- apiVersion: apps/v1 kind: Deployment metadata: name: starter-contract-stub spec: replicas: 1 selector: matchLabels: app.kubernetes.io/name: starter-contract-stub template: metadata: labels: app.kubernetes.io/name: starter-contract-stub spec: securityContext: runAsNonRoot: true runAsUser: 65532 runAsGroup: 65532 fsGroup: 65532 seccompProfile: type: RuntimeDefault containers: - name: contract-stub image: public.ecr.aws/docker/library/python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df command: - python3 - /app/contract_stub.py env: - name: CP9_LEDGER_FILE value: /var/lib/cp9-stub/calls.jsonl - name: PYTHONDONTWRITEBYTECODE value: "1" ports: - name: o11y containerPort: 9090 - name: metering containerPort: 9091 readinessProbe: httpGet: path: /healthz port: o11y periodSeconds: 2 failureThreshold: 30 livenessProbe: httpGet: path: /healthz port: metering periodSeconds: 10 resources: requests: cpu: 25m memory: 32Mi limits: cpu: 250m memory: 128Mi securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL volumeMounts: - name: source mountPath: /app readOnly: true - name: ledger mountPath: /var/lib/cp9-stub volumes: - name: source configMap: name: starter-contract-stub - name: ledger emptyDir: {}