#!/usr/bin/env python3
"""Stdlib-only metrics fixture for the fictional Larkspur Grocers catalog.
`python app.py --dump` writes Prometheus text to stdout and exits.
`python app.py` serves /metrics and /healthz for the compose file.
The process is not a production service and does not invent traffic.
"""
from __future__ import annotations
import argparse
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
try:
import yaml
except ImportError: # compose image has no PyYAML; the catalog beside the app is JSON-free YAML we parse minimally
yaml = None
ROOT = Path(__file__).resolve().parents[1]
def catalog_path() -> Path:
override = os.environ.get("CATALOG_PATH")
if override:
return Path(override)
return ROOT / "metrics" / "catalog.yml"
def load_catalog(path: Path):
text = path.read_text(encoding="utf-8")
if yaml is not None:
return yaml.safe_load(text)
# The compose image does not install PyYAML. Accept the committed catalog
# with a tiny loader that understands the mapping this file uses.
return _tiny_catalog(text)
def _tiny_catalog(text: str):
metrics = []
current = None
key = None
for raw in text.splitlines():
if not raw.strip() or raw.lstrip().startswith("#"):
if current and key == "help" and raw.strip().startswith("#"):
continue
continue
if raw.startswith(" - name:"):
if current:
metrics.append(current)
current = {"name": raw.split(":", 1)[1].strip(), "labels": []}
key = "name"
continue
if current is None:
continue
if raw.startswith(" type:"):
current["type"] = raw.split(":", 1)[1].strip()
elif raw.startswith(" help:"):
current["help"] = raw.split(":", 1)[1].strip()
elif raw.startswith(" labels:"):
body = raw.split(":", 1)[1].strip()
current["labels"] = [part.strip() for part in body.strip("[]").split(",") if part.strip()]
elif raw.startswith(" buckets:"):
body = raw.split(":", 1)[1].strip()
current["buckets"] = [part.strip().strip('"') for part in body.strip("[]").split(",") if part.strip()]
if current:
metrics.append(current)
return {"metrics": metrics}
def _sample_labels(metric: dict) -> dict:
"""One legal sample per family. Values are placeholders, not measurements."""
name = metric["name"]
chosen = {
"http_requests_total": {
"service": "orders-api",
"code": "200",
"route": "/orders",
"slo": "availability",
},
"http_request_duration_seconds": {
"service": "orders-api",
"route": "/orders",
"slo": "latency",
},
"payments_queue_oldest_age_seconds": {"service": "payments-worker"},
"payments_queue_depth": {"service": "payments-worker"},
"up": {"job": "storefront", "instance": "storefront-a.larkspur.example:8080"},
"node_cpu_seconds_total": {
"instance": "node-a.larkspur.example:9100",
"cpu": "0",
"mode": "idle",
},
"node_memory_available_bytes": {"instance": "node-a.larkspur.example:9100"},
"node_memory_total_bytes": {"instance": "node-a.larkspur.example:9100"},
"node_filesystem_avail_bytes": {
"instance": "node-a.larkspur.example:9100",
"mountpoint": "/",
"fstype": "ext4",
"device": "/dev/vda1",
},
"node_filesystem_size_bytes": {
"instance": "node-a.larkspur.example:9100",
"mountpoint": "/",
"fstype": "ext4",
"device": "/dev/vda1",
},
"pg_up": {"job": "postgres", "instance": "postgres-a.larkspur.example:9187"},
"pg_stat_activity_connections": {
"job": "postgres",
"instance": "postgres-a.larkspur.example:9187",
},
"pg_settings_max_connections": {
"job": "postgres",
"instance": "postgres-a.larkspur.example:9187",
},
"probe_success": {"job": "blackbox-http", "instance": "https://www.larkspur.example/"},
"probe_ssl_earliest_cert_expiry": {
"job": "blackbox-tls",
"instance": "www.larkspur.example:443",
},
"prometheus_rule_evaluation_failures_total": {
"job": "prometheus",
"instance": "prometheus.larkspur.example:9090",
},
"alertmanager_notifications_failed_total": {"service": "alertmanager"},
"alertmanager_notifications_total": {"service": "alertmanager"},
"ALERTS": {
"alertname": "Watchdog",
"alertstate": "firing",
"severity": "info",
"service": "prometheus",
"team": "platform",
},
}
labels = chosen[name]
missing = [label for label in metric.get("labels") or [] if label not in labels]
if missing:
raise SystemExit("demo-app has no sample labels for {0}: {1}".format(name, missing))
return labels
def _fmt_labels(labels: dict) -> str:
parts = []
for key in sorted(labels):
value = str(labels[key]).replace("\\", "\\\\").replace('"', '\\"')
parts.append('{0}="{1}"'.format(key, value))
return ",".join(parts)
def render_exposition(catalog: dict) -> str:
lines = []
for metric in catalog["metrics"]:
name = metric["name"]
lines.append("# HELP {0} {1}".format(name, metric["help"]))
lines.append("# TYPE {0} {1}".format(name, metric["type"]))
labels = _sample_labels(metric)
rendered = _fmt_labels(labels)
if metric["type"] == "histogram":
# One observation in the first bucket. Cumulative counts stay
# non-decreasing and the +Inf bucket equals _count.
for bucket in metric["buckets"]:
lines.append(
'{0}_bucket{{{1},le="{2}"}} 1'.format(name, rendered, bucket)
)
lines.append("{0}_sum{{{1}}} 0.05".format(name, rendered))
lines.append("{0}_count{{{1}}} 1".format(name, rendered))
elif metric["type"] == "counter":
lines.append("{0}{{{1}}} 1".format(name, rendered))
else:
lines.append("{0}{{{1}}} 1".format(name, rendered))
lines.append("")
return "\n".join(lines)
def exposition() -> str:
return render_exposition(load_catalog(catalog_path()))
class Handler(BaseHTTPRequestHandler):
def do_GET(self): # noqa: N802
path = self.path.split("?", 1)[0]
if path == "/metrics":
body = exposition().encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
if path == "/healthz":
body = b"ok\n"
self.send_response(200)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
self.send_response(404)
self.end_headers()
def log_message(self, fmt, *args):
return
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--dump", action="store_true", help="write exposition to stdout and exit")
parser.add_argument("--port", type=int, default=int(os.environ.get("PORT", "8080")))
args = parser.parse_args()
if args.dump:
print(exposition(), end="")
return
server = ThreadingHTTPServer(("0.0.0.0", args.port), Handler)
server.serve_forever()
if __name__ == "__main__":
main()