""" Simple Python API service. Exposes: GET / -> basic info GET /health -> liveness/readiness probe GET /metrics -> runtime metrics consumed by the monitoring dashboard """ import os import time import threading from datetime import datetime, timezone from flask import Flask, jsonify app = Flask(__name__) START_TIME = time.time() # --- In-memory metrics (thread-safe) -------------------------------------- _lock = threading.Lock() _metrics = { "requests_total": 0, "total_response_time_ms": 0.0, } @app.before_request def _start_timer(): from flask import g g.start_time = time.perf_counter() @app.after_request def _record_metrics(response): from flask import g elapsed_ms = (time.perf_counter() - getattr(g, "start_time", time.perf_counter())) * 1000 with _lock: _metrics["requests_total"] += 1 _metrics["total_response_time_ms"] += elapsed_ms # Allow the standalone dashboard.html to call this API from any origin response.headers["Access-Control-Allow-Origin"] = "*" return response @app.route("/") def index(): return jsonify({ "service": "sample-api", "message": "API is running. See /health and /metrics." }) @app.route("/health") def health(): """Used by the container platform's health checks and the monitor.py script.""" return jsonify({ "status": "healthy", "timestamp": datetime.now(timezone.utc).isoformat(), "uptime_seconds": round(time.time() - START_TIME, 2), }), 200 @app.route("/metrics") def metrics(): """Used by dashboard.html to render live stats.""" with _lock: total = _metrics["requests_total"] total_time = _metrics["total_response_time_ms"] avg_ms = round(total_time / total, 2) if total else 0.0 return jsonify({ "status": "healthy", "uptime_seconds": round(time.time() - START_TIME, 2), "requests_total": total, "avg_response_time_ms": avg_ms, "timestamp": datetime.now(timezone.utc).isoformat(), }) if __name__ == "__main__": port = int(os.environ.get("PORT", 8080)) app.run(host="0.0.0.0", port=port)