116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
import json
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import Optional
|
|
|
|
from app.common.config import config
|
|
from app.common.logging import setup_logger
|
|
from app.worker.metrics import worker_metrics
|
|
|
|
logger = setup_logger("processing-worker")
|
|
|
|
|
|
class WorkerHTTPRequestHandler(BaseHTTPRequestHandler):
|
|
"""
|
|
Lightweight HTTP request handler for worker internal metrics and stats.
|
|
"""
|
|
error_content_type = "application/json"
|
|
|
|
def log_message(self, format: str, *args) -> None:
|
|
# Suppress default noisy stderr logging
|
|
pass
|
|
|
|
def send_error(self, code: int, message: Optional[str] = None, explain: Optional[str] = None) -> None:
|
|
from http import HTTPStatus
|
|
try:
|
|
if hasattr(self, "headers") and self.headers:
|
|
content_length = int(self.headers.get("Content-Length", 0))
|
|
if content_length > 0:
|
|
self.rfile.read(min(content_length, 65536))
|
|
except Exception:
|
|
pass
|
|
try:
|
|
status = HTTPStatus(code)
|
|
short_msg = status.phrase
|
|
except ValueError:
|
|
short_msg = "error"
|
|
err_msg = message or short_msg
|
|
detail = explain or err_msg
|
|
body = json.dumps({"error": err_msg, "detail": detail}).encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Connection", "close")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self) -> None:
|
|
if self.path == "/internal/stats":
|
|
stats = worker_metrics.get_internal_stats()
|
|
body = json.dumps(stats, indent=2).encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
return
|
|
|
|
if self.path == "/metrics":
|
|
metrics_text = worker_metrics.get_prometheus_metrics()
|
|
body = metrics_text.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 self.path in ("/health", "/healthz", "/live"):
|
|
body = b'{"status": "ok"}'
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
return
|
|
|
|
self.send_error(404, "not_found", f"Cannot GET {self.path}")
|
|
|
|
|
|
class WorkerInternalServer:
|
|
"""
|
|
Background HTTP server serving /internal/stats and /metrics on worker port (8001).
|
|
"""
|
|
|
|
def __init__(self, host: Optional[str] = None, port: Optional[int] = None):
|
|
self.host = host or config.worker.stats_host
|
|
self.port = port or config.worker.stats_port
|
|
self._server: Optional[ThreadingHTTPServer] = None
|
|
self._thread: Optional[threading.Thread] = None
|
|
|
|
def start(self) -> None:
|
|
self._server = ThreadingHTTPServer(
|
|
(self.host, self.port), WorkerHTTPRequestHandler
|
|
)
|
|
self._thread = threading.Thread(
|
|
target=self._server.serve_forever, daemon=True, name="WorkerInternalHTTP"
|
|
)
|
|
self._thread.start()
|
|
logger.info(
|
|
f"Worker internal HTTP server started on http://{self.host}:{self.port}",
|
|
extra={"event": "worker_http_started"},
|
|
)
|
|
|
|
def stop(self) -> None:
|
|
if self._server:
|
|
self._server.shutdown()
|
|
self._server.server_close()
|
|
if self._thread:
|
|
self._thread.join(timeout=2.0)
|
|
logger.info(
|
|
"Worker internal HTTP server stopped cleanly",
|
|
extra={"event": "worker_http_stopped"},
|
|
)
|