129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""Plugin lifecycle: owns the periodic tasks and shared transport/buffer state.
|
|
|
|
Instantiated and driven by router.py's startup/shutdown events. The plugin is
|
|
non-critical to the proxy: any failure here is logged and contained, never
|
|
propagated into request routing.
|
|
"""
|
|
import asyncio
|
|
import time
|
|
from typing import Optional
|
|
|
|
from compliance import scheduler
|
|
from compliance.buffer import BufferStore
|
|
from compliance.log import log
|
|
from compliance.settings import ComplianceSettings
|
|
from compliance.transport import Transport
|
|
|
|
|
|
class AissurancePlugin:
|
|
def __init__(self, settings: ComplianceSettings):
|
|
self._settings = settings
|
|
self.started_at = time.monotonic()
|
|
self.stopping = False
|
|
self.sent_total = 0
|
|
|
|
self.transport: Optional[Transport] = None
|
|
self.buffer = BufferStore(settings)
|
|
|
|
self._stop_event = asyncio.Event()
|
|
self._tasks: list[asyncio.Task] = []
|
|
self._auth_failed = False
|
|
|
|
# -- accessors used by the scheduler -------------------------------------
|
|
@property
|
|
def cfg(self) -> ComplianceSettings:
|
|
return self._settings
|
|
|
|
@property
|
|
def cfg_obj(self):
|
|
"""Live Router Config, so the collector sees current endpoints."""
|
|
from config import get_config
|
|
return get_config()
|
|
|
|
async def sleep(self, seconds: float) -> None:
|
|
"""Sleep up to ``seconds``, returning early if the plugin is stopping."""
|
|
try:
|
|
await asyncio.wait_for(self._stop_event.wait(), timeout=seconds)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
async def handle_auth_failure(self) -> None:
|
|
"""401/403: stop retrying, clear the buffer, alert, and go dormant."""
|
|
if self._auth_failed:
|
|
return
|
|
self._auth_failed = True
|
|
self.buffer.clear()
|
|
log("auth_failed", level="error", router_id=self._settings.router_id,
|
|
message="AISSURANCE_KEY rejected — compliance reporting halted")
|
|
self.stopping = True
|
|
self._stop_event.set()
|
|
|
|
# -- lifecycle -----------------------------------------------------------
|
|
async def start(self) -> None:
|
|
errors = self._settings.validation_errors()
|
|
if errors:
|
|
log("plugin_disabled", level="error", reasons=errors)
|
|
return
|
|
|
|
self.transport = Transport(self._settings)
|
|
self.started_at = time.monotonic()
|
|
|
|
if not self._settings.tenant_id:
|
|
log("tenant_id_missing", level="warning",
|
|
message="No AISSURANCE_TENANT_ID / compliance.tenant_id; payload tenant_id will be null")
|
|
|
|
# Startup handshake: connectivity check + config report (best effort).
|
|
try:
|
|
await scheduler.run_health_cycle(self)
|
|
await scheduler.run_config_report(self)
|
|
except Exception as exc: # never let handshake failure block startup
|
|
log("handshake_failed", level="warning", error=str(exc))
|
|
|
|
self._tasks = [
|
|
asyncio.create_task(self._poll_loop()),
|
|
asyncio.create_task(self._health_loop()),
|
|
]
|
|
log("plugin_started", router_id=self._settings.router_id,
|
|
polling_interval=self._settings.polling_interval,
|
|
buffered=self.buffer.count())
|
|
|
|
async def _poll_loop(self) -> None:
|
|
# Emit an initial snapshot immediately so evidence appears without
|
|
# waiting a full interval, then settle into the configured cadence.
|
|
while not self.stopping:
|
|
try:
|
|
await scheduler.run_discovery_cycle(self)
|
|
except Exception as exc: # contain — plugin is non-critical
|
|
log("cycle_error", level="error", error=str(exc))
|
|
if self.stopping:
|
|
break
|
|
await self.sleep(self._settings.polling_interval)
|
|
|
|
async def _health_loop(self) -> None:
|
|
while not self.stopping:
|
|
await self.sleep(self._settings.health_interval)
|
|
if self.stopping:
|
|
break
|
|
try:
|
|
await scheduler.run_health_cycle(self)
|
|
except Exception as exc:
|
|
log("health_error", level="warning", error=str(exc))
|
|
|
|
async def stop(self) -> None:
|
|
self.stopping = True
|
|
self._stop_event.set()
|
|
for t in self._tasks:
|
|
t.cancel()
|
|
try:
|
|
await t
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
self._tasks = []
|
|
if self.transport is not None:
|
|
try:
|
|
await self.transport.aclose()
|
|
except Exception:
|
|
pass
|
|
self.transport = None
|
|
log("plugin_stopped", router_id=self._settings.router_id,
|
|
buffered=self.buffer.count(), successful=self.sent_total)
|