197 lines
9.3 KiB
Python
197 lines
9.3 KiB
Python
import asyncio
|
|
import json
|
|
import signal
|
|
|
|
from app.api.metrics import api_metrics
|
|
from app.api.producer import ConfluentKafkaProducer
|
|
from app.api.stats import get_aggregated_pipeline_stats
|
|
from app.common.config import config
|
|
from app.common.logging import setup_logger
|
|
from app.common.validation import PayloadError, parse_payload, validate_structure
|
|
|
|
logger = setup_logger("ingestion-api")
|
|
|
|
|
|
class HTTPError(ValueError):
|
|
def __init__(self, status, reason, detail):
|
|
super().__init__(detail)
|
|
self.status, self.reason = status, reason
|
|
|
|
|
|
class IngestionAPIServer:
|
|
def __init__(self, host=None, port=None, queue_max_size=None, producer=None,
|
|
worker_stats_url=None, max_body_bytes=None, read_timeout=None,
|
|
delivery_timeout=None):
|
|
self.host = host if host is not None else config.api.host
|
|
self.port = port if port is not None else config.api.port
|
|
self.queue_max_size = queue_max_size if queue_max_size is not None else config.api.queue_max_size
|
|
self.max_body_bytes = max_body_bytes if max_body_bytes is not None else config.api.max_body_bytes
|
|
self.read_timeout = read_timeout if read_timeout is not None else config.api.read_timeout_seconds
|
|
self.delivery_timeout = delivery_timeout if delivery_timeout is not None else config.kafka.delivery_timeout_seconds
|
|
if min(self.queue_max_size, self.max_body_bytes, self.read_timeout, self.delivery_timeout) <= 0:
|
|
raise ValueError("API capacities and timeouts must be positive")
|
|
self.worker_stats_url = worker_stats_url or config.api.worker_stats_url
|
|
self.producer = producer if producer is not None else ConfluentKafkaProducer()
|
|
self._pending = set()
|
|
self._clients = set()
|
|
self._server = None
|
|
self._stopping = False
|
|
|
|
async def start(self):
|
|
self._server = await asyncio.start_server(self._handle_client, self.host, self.port)
|
|
self.port = self._server.sockets[0].getsockname()[1]
|
|
logger.info(f"Listening on {self.host}:{self.port}", extra={"event": "server_started"})
|
|
|
|
async def stop(self):
|
|
self._stopping = True
|
|
if self._server:
|
|
self._server.close()
|
|
await self._server.wait_closed()
|
|
deadline = asyncio.get_running_loop().time() + config.api.shutdown_timeout_seconds
|
|
outstanding = self._pending | self._clients
|
|
if outstanding:
|
|
_, remaining = await asyncio.wait(outstanding, timeout=max(0, deadline-asyncio.get_running_loop().time()))
|
|
for task in remaining:
|
|
if isinstance(task, asyncio.Task):
|
|
task.cancel()
|
|
await asyncio.gather(*(t for t in remaining if isinstance(t, asyncio.Task)), return_exceptions=True)
|
|
await asyncio.to_thread(self.producer.close, max(0, deadline-asyncio.get_running_loop().time()))
|
|
logger.info("API drained and stopped", extra={"event": "server_stopped"})
|
|
|
|
async def _read_request(self, reader):
|
|
line = await reader.readline()
|
|
if not line:
|
|
return None
|
|
try:
|
|
method, path, version = line.decode("ascii").strip().split()
|
|
except (ValueError, UnicodeError) as exc:
|
|
raise HTTPError(400, "bad_request", "Invalid request line") from exc
|
|
if version not in ("HTTP/1.0", "HTTP/1.1"):
|
|
raise HTTPError(400, "bad_request", "Unsupported HTTP version")
|
|
headers = {}
|
|
header_bytes = len(line)
|
|
while True:
|
|
line = await reader.readline()
|
|
header_bytes += len(line)
|
|
if header_bytes > 16384:
|
|
raise HTTPError(400, "bad_request", "Headers exceed 16384 bytes")
|
|
if line in (b"\r\n", b"\n"):
|
|
break
|
|
if not line or b":" not in line:
|
|
raise HTTPError(400, "bad_request", "Invalid header framing")
|
|
key, value = line.decode("latin-1").split(":", 1)
|
|
key, value = key.strip().lower(), value.strip()
|
|
if key in headers and key in ("content-length", "transfer-encoding"):
|
|
raise HTTPError(400, "bad_request", "Duplicate framing header")
|
|
headers[key] = value
|
|
if "transfer-encoding" in headers:
|
|
raise HTTPError(400, "bad_request", "Transfer-Encoding is unsupported; use Content-Length")
|
|
length = headers.get("content-length", "0")
|
|
if not length.isascii() or not length.isdecimal():
|
|
raise HTTPError(400, "bad_request", "Invalid Content-Length")
|
|
if len(length) > 10 or int(length) > self.max_body_bytes:
|
|
raise HTTPError(413, "payload_too_large", "Request body exceeds configured limit")
|
|
return method.upper(), path, headers, await reader.readexactly(int(length))
|
|
|
|
async def _handle_client(self, reader, writer):
|
|
task = asyncio.current_task()
|
|
self._clients.add(task)
|
|
try:
|
|
async with asyncio.timeout(self.read_timeout):
|
|
request = await self._read_request(reader)
|
|
if request:
|
|
await self._dispatch_route(writer, *request)
|
|
except TimeoutError:
|
|
await self._error(writer, 408, "request_timeout", "Request read deadline exceeded")
|
|
except HTTPError as exc:
|
|
await self._error(writer, exc.status, exc.reason, str(exc))
|
|
except (asyncio.IncompleteReadError, ValueError) as exc:
|
|
await self._error(writer, 400, "bad_request", str(exc))
|
|
except (ConnectionError, BrokenPipeError):
|
|
pass
|
|
except Exception:
|
|
logger.exception("HTTP handler failure", extra={"event": "request_handler_error"})
|
|
await self._error(writer, 500, "internal_error", "Request failed")
|
|
finally:
|
|
self._clients.discard(task)
|
|
writer.close()
|
|
try:
|
|
await writer.wait_closed()
|
|
except ConnectionError:
|
|
pass
|
|
|
|
async def _dispatch_route(self, writer, method, path, headers, body):
|
|
if method == "GET" and path in ("/health", "/healthz", "/live"):
|
|
await self._json(writer, 200, {"status": "ok", "in_flight": len(self._pending)})
|
|
elif method == "GET" and path == "/metrics":
|
|
await self._send_response(writer, 200, "text/plain; version=0.0.4; charset=utf-8", api_metrics.get_prometheus_metrics().encode())
|
|
elif method == "GET" and path == "/stats":
|
|
await self._json(writer, 200, await get_aggregated_pipeline_stats(self.worker_stats_url))
|
|
elif method == "POST" and path == "/api/v1/cdr/raw":
|
|
await self._handle_raw_cdr_ingestion(writer, body)
|
|
else:
|
|
await self._error(writer, 404, "not_found", f"Cannot {method} {path}")
|
|
|
|
async def _handle_raw_cdr_ingestion(self, writer, body):
|
|
try:
|
|
payload = parse_payload(body)
|
|
validate_structure(payload)
|
|
except PayloadError as exc:
|
|
await self._error(writer, 400, exc.reason, str(exc))
|
|
return
|
|
if self._stopping:
|
|
await self._error(writer, 503, "shutting_down", "API is draining")
|
|
return
|
|
if len(self._pending) >= self.queue_max_size:
|
|
await self._error(writer, 429, "queue_full", "Outstanding publication capacity exhausted")
|
|
return
|
|
try:
|
|
delivery = asyncio.wrap_future(self.producer.produce(config.kafka.raw_topic, payload["raw_id"], body))
|
|
self._pending.add(delivery)
|
|
def completed(future):
|
|
self._pending.discard(future)
|
|
if not future.cancelled():
|
|
future.exception() # Consume failures after an HTTP timeout/disconnect.
|
|
delivery.add_done_callback(completed)
|
|
await asyncio.wait_for(asyncio.shield(delivery), self.delivery_timeout)
|
|
except Exception as exc:
|
|
logger.warning(str(exc), extra={"raw_id": payload["raw_id"], "event": "kafka_delivery_failed"})
|
|
await self._error(writer, 503, "broker_unavailable", "Delivery unconfirmed; retry with the same raw_id and payload")
|
|
return
|
|
api_metrics.increment_received(1)
|
|
await self._json(writer, 202, {"status": "accepted", "raw_id": payload["raw_id"], "queued": True})
|
|
|
|
async def _error(self, writer, status, error, detail):
|
|
await self._json(writer, status, {"error": error, "detail": detail})
|
|
|
|
async def _json(self, writer, status, payload):
|
|
await self._send_response(writer, status, "application/json", json.dumps(payload).encode())
|
|
|
|
async def _send_response(self, writer, status_code, content_type, body):
|
|
from http import HTTPStatus
|
|
headers = (f"HTTP/1.1 {status_code} {HTTPStatus(status_code).phrase}\r\n"
|
|
f"Content-Type: {content_type}\r\nContent-Length: {len(body)}\r\n"
|
|
"Connection: close\r\n\r\n")
|
|
writer.write(headers.encode() + body)
|
|
await writer.drain()
|
|
|
|
|
|
async def main():
|
|
server = IngestionAPIServer()
|
|
stopping = asyncio.Event()
|
|
loop = asyncio.get_running_loop()
|
|
for sig in (signal.SIGTERM, signal.SIGINT):
|
|
try:
|
|
loop.add_signal_handler(sig, stopping.set)
|
|
except NotImplementedError:
|
|
signal.signal(sig, lambda *_: loop.call_soon_threadsafe(stopping.set))
|
|
await server.start()
|
|
try:
|
|
await stopping.wait()
|
|
finally:
|
|
await server.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|