import logging
import socket
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import TimeoutError as FutureTimeout
from datetime import datetime
from pathlib import Path
from typing import Any, cast
from .collectors.base import Collector, CollectorContext
from .collectors.cron import CronCollector
from .collectors.docker import DockerCollector
from .collectors.generic_worker import GenericWorkerCollector
from .collectors.process import ProcessCollector
from .collectors.systemd import SystemdCollector
from .config import AppConfig
from .db.connection import connect
from .db.migrations import apply_migrations
from .db.repositories import WatcherRepository
from .models import (
AgentRegistration,
CollectorError,
CollectorResult,
RemoteEvent,
WatcherCycleResult,
)
from .remote import observation_from_remote_event
from .timeutil import iso, utc_now
LOGGER = logging.getLogger(__name__)
def build_collector(config_type: str) -> Collector:
collectors: dict[str, Any] = {"cron": CronCollector, "docker": DockerCollector, "systemd": SystemdCollector,
"process": ProcessCollector, "generic_worker": GenericWorkerCollector,
"generic-worker": GenericWorkerCollector}
if config_type not in collectors:
raise ValueError(f"unsupported collector type: {config_type}")
return cast(Collector, collectors[config_type]())
class Watcher:
def __init__(self, config: AppConfig, base_dir: Path | None = None) -> None:
self.config = config
self.base_dir = base_dir or Path.cwd()
self.connection = connect(config.watcher.database_path)
self.schema_version = apply_migrations(self.connection)
self.started_at = utc_now()
self.last_cycle_started_at: datetime | None = None
self.last_cycle_finished_at: datetime | None = None
self.last_successful_cycle_at: datetime | None = None
self.last_cycle_duration_ms: int | None = None
self.last_results: dict[str, CollectorResult] = {}
@property
def repository(self) -> WatcherRepository:
return WatcherRepository(self.connection)
def run_once(self) -> WatcherCycleResult:
started = utc_now()
self.last_cycle_started_at = started
enabled = [item for item in self.config.collectors if item.enabled]
results: list[CollectorResult] = []
context_host = socket.gethostname()
def execute(item: Any) -> CollectorResult:
collector = build_collector(item.type)
context = CollectorContext(started, context_host, item, self.base_dir)
return collector.collect(context)
for item in enabled:
try:
with ThreadPoolExecutor(max_workers=1, thread_name_prefix=f"collector-{item.name}") as executor:
future = executor.submit(execute, item)
try:
current = future.result(timeout=item.timeout_seconds)
except FutureTimeout:
current = CollectorResult(item.name, "unknown", started, utc_now(), False, (),
(CollectorError("timeout", f"collector timed out after {item.timeout_seconds}s"),), {})
except Exception as exc: # noqa: BLE001 - collector isolation boundary
current = CollectorResult(item.name, "unknown", started, utc_now(), False, (),
(CollectorError(type(exc).__name__, str(exc)),), {})
except Exception as exc: # noqa: BLE001 - collector construction isolation boundary
current = CollectorResult(item.name, "unknown", started, utc_now(), False, (),
(CollectorError(type(exc).__name__, str(exc)),), {})
results.append(current)
self._persist_result(current)
finished = utc_now()
self.last_cycle_finished_at = finished
self.last_cycle_duration_ms = int((finished - started).total_seconds() * 1000)
self.last_results = {result.collector_name: result for result in results}
if results and all(result.success for result in results):
self.last_successful_cycle_at = finished
self.connection.commit()
return WatcherCycleResult(started, finished, len(results), sum(result.success for result in results),
sum(not result.success for result in results), sum(len(result.observations) for result in results))
def _persist_result(self, result: CollectorResult) -> None:
repository = self.repository
observations_created = 0
with self.connection:
for observation in result.observations:
repository.persist_observation(observation, None, result.finished_at)
observations_created += 1
repository.save_collector_run(result, observations_created)
if not result.success:
for error in result.errors:
repository.save_watcher_event("collector_failed", "warning", error.message, result.collector_name,
{"error_type": error.error_type, "instance_key": error.instance_key})
def register_remote_agent(self, registration: AgentRegistration) -> int:
with self.connection:
return self.repository.ensure_agent(registration, utc_now())
def ingest_remote_event(self, event: RemoteEvent) -> dict[str, Any]:
received_at = utc_now()
repository = self.repository
with self.connection:
agent_id = repository.ensure_agent(event.agent, received_at)
remote_event_id, created = repository.record_remote_event(event, agent_id, received_at)
if not created:
return {"accepted": True, "duplicate": True, "remote_event_id": remote_event_id}
observation = observation_from_remote_event(event)
observation_id, changed = repository.persist_observation(observation, None, received_at)
repository.mark_remote_event(remote_event_id, True)
return {"accepted": True, "duplicate": False, "remote_event_id": remote_event_id,
"observation_id": observation_id, "status_changed": changed}
def health(self) -> dict[str, Any]:
self.repository.refresh_remote_staleness(utc_now(), self.config.watcher.remote_stale_after_seconds)
summary = self.repository.summary()
successful = sum(1 for result in self.last_results.values() if result.success)
failed = sum(1 for result in self.last_results.values() if not result.success)
database_status = "ok"
try:
self.connection.execute("SELECT 1").fetchone()
except Exception: # noqa: BLE001 - health must report database failure
database_status = "error"
job_issue = any(summary[key] > 0 for key in ("jobs_degraded", "jobs_failed", "jobs_stale", "jobs_unknown"))
watcher_status = "healthy" if database_status == "ok" and failed == 0 and not job_issue else "degraded"
enabled_collectors = sum(1 for item in self.config.collectors if item.enabled)
return {"watcher_status": watcher_status, "application_version": "0.1.0", "schema_version": self.schema_version,
"current_time_utc": iso(utc_now()), "started_at": iso(self.started_at),
"last_cycle_started_at": iso(self.last_cycle_started_at), "last_cycle_finished_at": iso(self.last_cycle_finished_at),
"last_successful_cycle_at": iso(self.last_successful_cycle_at), "last_cycle_duration_ms": self.last_cycle_duration_ms,
"database_status": database_status, "collectors_total": len(self.config.collectors),
"collectors_enabled": enabled_collectors, "collectors_disabled": len(self.config.collectors) - enabled_collectors,
"collectors_successful": successful, "collectors_failed": failed, **summary}
def close(self) -> None:
self.connection.close()