diff --git a/README.md b/README.md index 1478b45..9f85c94 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,212 @@ # daena -Collect it. Keep it. Ship it. That’s Daena! 🀠 + +Collect, buffer, and forward logs and events from Linux to central observability backends. + +## Overview + +Daena is a modular, extensible Linux service that: + +- Collects logs, events, and telemetry from pluggable sources +- Normalizes and processes records through a configurable pipeline +- Persists data locally in LMDB for durability, replay, and buffering +- Forwards records reliably to one or more remote sinks +- Exposes a REST API for management and inspection +- Provides a CLI for operational tasks +- Degrades gracefully under failures with retry, backoff, and dead-letter queues + +## Architecture + +``` +Sources β†’ [Processors] β†’ Pipeline β†’ Persistence (LMDB) β†’ Sink Dispatcher β†’ Remote + ↻ retry queue +``` + +Key design decisions: + +- **Plugin architecture**: Sources, processors, and sinks are loaded via + Python entry points. Add new ones by installing a package or registering + in `pyproject.toml`. +- **Pluggable persistence**: The `Backend` protocol abstracts storage. + MVP uses LMDB. Future: NATS JetStream. +- **Reliability-first**: Records survive restarts, sink failures trigger + exponential backoff, and permanently failed records land in a dead-letter + queue. + +## Quick start + +```bash +# Prerequisites: pyenv with Python 3.13, uv installed + +# Clone and set up +git clone && cd daena +pyenv local 3.13.5 +uv sync --extra dev + +# Run with dummy source/sink (no external dependencies) +uv run python -c " +import asyncio +from daena.config import DaenaConfig, SourceConfig, SinkConfig +from daena.core.runtime import DaenaRuntime + +async def main(): + config = DaenaConfig() + config.sources = [SourceConfig(name='test', type='dummy', + options={'interval': 1.0, 'message': 'hello'})] + config.sinks = [SinkConfig(name='log', type='dummy')] + config.persistence.lmdb.path = '/tmp/daena_data' + + runtime = DaenaRuntime(config) + await runtime.start() + await asyncio.sleep(5) + print(f'Records: {runtime.backend.count()}') + await runtime.stop() + +asyncio.run(main()) +" + +# Run tests +uv run pytest tests/ -v + +# Use the CLI (point at a running daena API) +uv run daena status +uv run daena health +uv run daena queue +``` + +## Project structure + +``` +daena/ +β”œβ”€β”€ pyproject.toml # Project metadata, dependencies, entry points +β”œβ”€β”€ config/daena.yaml # Example configuration +β”œβ”€β”€ deploy/daena.service # systemd unit file +β”œβ”€β”€ src/daena/ +β”‚ β”œβ”€β”€ app.py # `daena-serve` entry point (uvicorn) +β”‚ β”œβ”€β”€ config.py # YAML β†’ pydantic config +β”‚ β”œβ”€β”€ logutil.py # structlog setup +β”‚ β”œβ”€β”€ exceptions.py # Domain exceptions +β”‚ β”œβ”€β”€ api/ # FastAPI REST API (v1 versioned) +β”‚ β”œβ”€β”€ cli/ # Typer CLI +β”‚ β”œβ”€β”€ core/ +β”‚ β”‚ β”œβ”€β”€ runtime.py # Service lifecycle +β”‚ β”‚ └── pipeline.py # Collection β†’ persist β†’ dispatch +β”‚ β”œβ”€β”€ persistence/ +β”‚ β”‚ β”œβ”€β”€ backend.py # Abstract Backend protocol +β”‚ β”‚ β”œβ”€β”€ models.py # RecordState, StoredRecord +β”‚ β”‚ └── lmdb.py # LMDB implementation +β”‚ β”œβ”€β”€ pipeline/ +β”‚ β”‚ β”œβ”€β”€ base.py # Source/Processor/Sink ABCs +β”‚ β”‚ β”œβ”€β”€ record.py # In-memory Record dataclass +β”‚ β”‚ β”œβ”€β”€ sources/ # Built-in sources (dummy) +β”‚ β”‚ β”œβ”€β”€ sinks/ # Built-in sinks (dummy, http) +β”‚ β”‚ └── processors/ # Built-in processors (passthrough) +β”‚ └── plugins/ +β”‚ β”œβ”€β”€ loader.py # Entry-point discovery +β”‚ └── registry.py # Plugin registry +└── tests/ +``` + +## Configuration + +Daena loads configuration from `/etc/daena/daena.yaml`, `./daena.yaml`, or +`./config/daena.yaml`. See `config/daena.yaml` for a full example. + +```yaml +logging: + level: info + format: json + +api: + http: + host: 127.0.0.1 + port: 8642 + +persistence: + backend: lmdb + lmdb: + path: /var/lib/daena/data + map_size: 104857600 + +sources: + - name: system_logs + type: journald # future + options: {} + +processors: + - name: add_hostname + type: enricher # future + options: {} + +sinks: + - name: central + type: http + options: + url: https://logs.example.com/ingest +``` + +## REST API + +The API runs on `http://127.0.0.1:8642/api/v1`: + +| Endpoint | Description | +|----------|-------------| +| `GET /system/status` | Service status | +| `GET /system/health` | Health check | +| `GET /system/config` | Loaded configuration | +| `GET /plugins` | Registered plugins | +| `GET /queue` | Queue state counts | +| `GET /queue/pending` | Pending records | +| `GET /queue/dead` | Dead-letter records | +| `GET /sinks` | Configured sinks | + +## Deployment + +### systemd + +```bash +# Create daena user +sudo useradd -r -s /sbin/nologin -d /var/lib/daena daena + +# Install package +uv pip install . + +# Set up directories +sudo mkdir -p /etc/daena /var/lib/daena /var/log/daena /run/daena +sudo cp config/daena.yaml /etc/daena/ +sudo cp deploy/daena.service /etc/systemd/system/ +sudo systemctl daemon-reload +sudo systemctl enable --now daena +``` + +## Extending + +Add a new source by creating a class that inherits `daena.pipeline.base.Source` +and registering it as an entry point: + +```python +# my_sources.py +from daena.pipeline.base import Source +from daena.pipeline.record import Record + +class MySource(Source): + async def start(self): + pass + async def stop(self): + pass +``` + +```toml +# pyproject.toml +[project.entry-points."daena.sources"] +my_source = "my_sources:MySource" +``` + +Same pattern applies for `Processor` and `Sink` base classes. + +## Branch workflow + +- `main` β€” stable integration branch +- `dev` β€” active development branch + +Feature work happens on `dev` or topic branches. Merges to `main` represent +stable release checkpoints. diff --git a/config/daena.yaml b/config/daena.yaml new file mode 100644 index 0000000..2e7093f --- /dev/null +++ b/config/daena.yaml @@ -0,0 +1,47 @@ +# daena configuration +# Path: /etc/daena/daena.yaml + +data_dir: /var/lib/daena +log_dir: /var/log/daena +pid_file: /run/daena/daena.pid + +logging: + level: info + format: json + +api: + http: + host: 127.0.0.1 + port: 8642 + workers: 1 + +pipeline: + batch_size: 100 + flush_interval: 5.0 + max_queue_size: 10000 + +persistence: + backend: lmdb + lmdb: + path: /var/lib/daena/data + map_size: 104857600 + max_databases: 16 + +sources: + - name: dummy + type: dummy + enabled: true + options: + interval: 5.0 + message: "daena heartbeat" + +processors: + - name: passthrough + type: passthrough + enabled: true + +sinks: + - name: log + type: dummy + enabled: true + options: {} diff --git a/deploy/daena.service b/deploy/daena.service new file mode 100644 index 0000000..05fa31b --- /dev/null +++ b/deploy/daena.service @@ -0,0 +1,31 @@ +[Unit] +Description=Daena - Linux log and event collector +Documentation=https://github.com/anomalyco/daena +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=daena +Group=daena + +ExecStart=/usr/local/bin/daena-serve +ExecReload=/bin/kill -HUP $MAINPID +Restart=on-failure +RestartSec=5 + +AmbientCapabilities=CAP_SYS_PTRACE +NoNewPrivileges=true +ProtectSystem=full +ProtectHome=true +PrivateTmp=true + +StateDirectory=daena +LogsDirectory=daena +RuntimeDirectory=daena + +# Environment +Environment=DAENA_CONFIG_PATH=/etc/daena/daena.yaml + +[Install] +WantedBy=multi-user.target diff --git a/src/daena/__init__.py b/src/daena/__init__.py index 3f4c88c..af6676c 100644 --- a/src/daena/__init__.py +++ b/src/daena/__init__.py @@ -1 +1 @@ -from daena._version import __version__ +from daena._version import __version__ as __version__ diff --git a/src/daena/__main__.py b/src/daena/__main__.py new file mode 100644 index 0000000..a42abdb --- /dev/null +++ b/src/daena/__main__.py @@ -0,0 +1,5 @@ +from __future__ import annotations + +from daena.cli.main import app + +app() diff --git a/src/daena/api/__init__.py b/src/daena/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/api/router.py b/src/daena/api/router.py new file mode 100644 index 0000000..b3896c7 --- /dev/null +++ b/src/daena/api/router.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +from fastapi import APIRouter + +from daena.api.routes.plugins import router as plugins_router +from daena.api.routes.queue import router as queue_router +from daena.api.routes.sinks import router as sinks_router +from daena.api.routes.system import router as system_router + +api_router = APIRouter() + +api_router.include_router(system_router, prefix="/system", tags=["system"]) +api_router.include_router(plugins_router, prefix="/plugins", tags=["plugins"]) +api_router.include_router(queue_router, prefix="/queue", tags=["queue"]) +api_router.include_router(sinks_router, prefix="/sinks", tags=["sinks"]) diff --git a/src/daena/api/routes/__init__.py b/src/daena/api/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/api/routes/plugins.py b/src/daena/api/routes/plugins.py new file mode 100644 index 0000000..222352e --- /dev/null +++ b/src/daena/api/routes/plugins.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from fastapi import APIRouter, Request + +router = APIRouter() + + +@router.get("") +async def list_plugins(request: Request) -> dict: + runtime = request.app.state.runtime + registry = runtime.registry + plugins = registry.all_plugins() + result: dict[str, list[dict]] = {} + for category, entries in plugins.items(): + result[category] = [{"name": name, "class": cls.__name__} for name, cls in entries.items()] + return {"plugins": result} diff --git a/src/daena/api/routes/queue.py b/src/daena/api/routes/queue.py new file mode 100644 index 0000000..bf89155 --- /dev/null +++ b/src/daena/api/routes/queue.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from fastapi import APIRouter, Query, Request + +from daena.persistence.models import RecordState + +router = APIRouter() + + +@router.get("") +async def get_queue_state(request: Request) -> dict: + runtime = request.app.state.runtime + backend = runtime.backend + if backend is None: + return {"error": "backend not initialized"} + return { + "pending": backend.count(RecordState.pending), + "delivered": backend.count(RecordState.delivered), + "failed": backend.count(RecordState.failed), + "dead": backend.count(RecordState.dead), + "total": backend.count(), + } + + +@router.get("/pending") +async def get_pending( + request: Request, + limit: int = Query(default=100, le=1000), + offset: int = Query(default=0, ge=0), +) -> dict: + runtime = request.app.state.runtime + backend = runtime.backend + if backend is None: + return {"error": "backend not initialized", "records": []} + records = backend.list_by_state(RecordState.pending, limit=limit, offset=offset) + return {"records": [r.to_dict() for r in records], "count": len(records)} + + +@router.get("/dead") +async def get_dead( + request: Request, + limit: int = Query(default=100, le=1000), + offset: int = Query(default=0, ge=0), +) -> dict: + runtime = request.app.state.runtime + backend = runtime.backend + if backend is None: + return {"error": "backend not initialized", "records": []} + records = backend.list_by_state(RecordState.dead, limit=limit, offset=offset) + return {"records": [r.to_dict() for r in records], "count": len(records)} diff --git a/src/daena/api/routes/sinks.py b/src/daena/api/routes/sinks.py new file mode 100644 index 0000000..bfd9818 --- /dev/null +++ b/src/daena/api/routes/sinks.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from fastapi import APIRouter, Request + +router = APIRouter() + + +@router.get("") +async def list_sinks(request: Request) -> dict: + runtime = request.app.state.runtime + pipeline = runtime.pipeline + if pipeline is None: + return {"sinks": []} + sinks = pipeline.sinks + return { + "sinks": [{"name": s.name, "type": type(s).__name__, "config": s.config} for s in sinks] + } diff --git a/src/daena/api/routes/system.py b/src/daena/api/routes/system.py new file mode 100644 index 0000000..095cc36 --- /dev/null +++ b/src/daena/api/routes/system.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from fastapi import APIRouter, Request + +from daena.core.runtime import DaenaRuntime + +router = APIRouter() + + +def _resolve_runtime(request: Request) -> DaenaRuntime: + return request.app.state.runtime # type: ignore[no-any-return] + + +@router.get("/status") +async def get_status(request: Request) -> dict: + runtime = _resolve_runtime(request) + return { + "service": "daena", + "version": "0.1.0", + "running": runtime._running, + "uptime": None, + } + + +@router.get("/health") +async def get_health(request: Request) -> dict: + runtime = _resolve_runtime(request) + healthy = runtime._running + return { + "healthy": healthy, + "running": runtime._running, + "backend": runtime.backend is not None, + "plugins_loaded": bool(runtime.registry.list_sources()), + } + + +@router.get("/config") +async def get_config(request: Request) -> dict: + runtime = _resolve_runtime(request) + cfg = runtime.config + return { + "data_dir": cfg.data_dir, + "log_dir": cfg.log_dir, + "api": cfg.api.model_dump(), + "pipeline": cfg.pipeline.model_dump(), + "persistence": cfg.persistence.model_dump(), + "sources": [s.model_dump() for s in cfg.sources], + "processors": [p.model_dump() for p in cfg.processors], + "sinks": [s.model_dump() for s in cfg.sinks], + } diff --git a/src/daena/api/server.py b/src/daena/api/server.py new file mode 100644 index 0000000..962f370 --- /dev/null +++ b/src/daena/api/server.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from daena.api.router import api_router +from daena.core.runtime import DaenaRuntime +from daena.logutil import get_logger + +log = get_logger("daena.api") + + +class DaenaAPI: + """Wraps the FastAPI application with daena runtime integration.""" + + def __init__(self, runtime: DaenaRuntime) -> None: + self._runtime = runtime + self._app: FastAPI | None = None + + def build(self) -> FastAPI: + runtime = self._runtime + + @asynccontextmanager + async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + log.info("api_starting") + await runtime.start() + yield + log.info("api_stopping") + await runtime.stop() + + app = FastAPI( + title="Daena", + description="Daena service management API", + version="0.1.0", + lifespan=lifespan, + ) + + app.state.runtime = runtime + app.include_router(api_router, prefix="/api/v1") + + self._app = app + return app + + @property + def app(self) -> FastAPI: + if self._app is None: + return self.build() + return self._app diff --git a/src/daena/app.py b/src/daena/app.py new file mode 100644 index 0000000..2d09d10 --- /dev/null +++ b/src/daena/app.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from pathlib import Path + +import uvicorn + +from daena.api.server import DaenaAPI +from daena.config import load_config +from daena.logutil import get_logger, setup_logging + +log = get_logger("daena.app") + + +def serve() -> None: + """Entry point for `daena-serve` (systemd/uvicorn mode).""" + config_path = _resolve_config_path() + config = load_config(config_path) + setup_logging( + level=config.logging.level, + fmt=config.logging.format, + log_file=config.logging.file, + ) + + runtime = _build_runtime(config) + api = DaenaAPI(runtime) + app = api.build() + + log.info( + "starting_server", + host=config.api.http.host, + port=config.api.http.port, + ) + uvicorn.run( + app, + host=config.api.http.host, + port=config.api.http.port, + workers=config.api.http.workers, + log_level=config.logging.level, + ) + + +def _resolve_config_path() -> str | None: + for candidate in ( + "/etc/daena/daena.yaml", + "./daena.yaml", + "./config/daena.yaml", + ): + p = Path(candidate) + if p.exists(): + return str(p) + return None + + +def _build_runtime(config): + """Lazy import to prevent premature backend init.""" + from daena.core.runtime import DaenaRuntime + + return DaenaRuntime(config) + + +if __name__ == "__main__": + serve() diff --git a/src/daena/cli/__init__.py b/src/daena/cli/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/cli/main.py b/src/daena/cli/main.py new file mode 100644 index 0000000..e4c2128 --- /dev/null +++ b/src/daena/cli/main.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import httpx +import typer +from rich.console import Console +from rich.table import Table + +from daena._version import __version__ + +app = typer.Typer( + name="daena", + help="Daena: collect, buffer, and forward logs and events.", +) +console = Console() + +DEFAULT_API_URL = "http://127.0.0.1:8642/api/v1" + + +def _client(timeout: float = 10.0) -> httpx.Client: + return httpx.Client(timeout=httpx.Timeout(timeout)) + + +def _api_url() -> str: + return DEFAULT_API_URL # TODO: make configurable via env/flag + + +@app.command() +def status() -> None: + """Show daemon status.""" + try: + with _client(5.0) as c: + resp = c.get(f"{_api_url()}/system/status") + resp.raise_for_status() + data = resp.json() + console.print(f"[bold]Daena[/bold] v{data.get('version', '?')}") + console.print(f"Running: {data.get('running', '?')}") + except httpx.ConnectError: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None + + +@app.command() +def health() -> None: + """Check service health.""" + try: + with _client(5.0) as c: + resp = c.get(f"{_api_url()}/system/health") + resp.raise_for_status() + data = resp.json() + if data.get("healthy"): + console.print("[green]Healthy[/green]") + else: + console.print("[red]Unhealthy[/red]") + raise typer.Exit(1) + except httpx.ConnectError: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None + + +@app.command() +def queue() -> None: + """Inspect queue state.""" + try: + with _client() as c: + resp = c.get(f"{_api_url()}/queue") + resp.raise_for_status() + data = resp.json() + table = Table("State", "Count") + for key in ("pending", "delivered", "failed", "dead", "total"): + table.add_row(key, str(data.get(key, 0))) + console.print(table) + except httpx.ConnectError: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None + + +@app.command() +def sinks() -> None: + """List configured sinks.""" + try: + with _client() as c: + resp = c.get(f"{_api_url()}/sinks") + resp.raise_for_status() + data = resp.json() + sinks_list = data.get("sinks", []) + if not sinks_list: + console.print("No sinks configured") + return + table = Table("Name", "Type") + for s in sinks_list: + table.add_row(s["name"], s["type"]) + console.print(table) + except httpx.ConnectError: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None + + +@app.command() +def plugins() -> None: + """List loaded plugins.""" + try: + with _client() as c: + resp = c.get(f"{_api_url()}/plugins") + resp.raise_for_status() + data = resp.json() + plugins_data = data.get("plugins", {}) + for category, items in plugins_data.items(): + if items: + table = Table(f"{category.title()}") + for item in items: + table.add_row(f"{item['name']} ({item['class']})") + console.print(table) + console.print() + except httpx.ConnectError: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None + + +@app.command() +def config() -> None: + """Show loaded service configuration.""" + try: + with _client() as c: + resp = c.get(f"{_api_url()}/system/config") + resp.raise_for_status() + data = resp.json() + console.print(data) + except httpx.ConnectError: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None + + +@app.command() +def version() -> None: + """Show version.""" + console.print(f"daena v{__version__}") + + +if __name__ == "__main__": + app() diff --git a/src/daena/config.py b/src/daena/config.py new file mode 100644 index 0000000..28b5b0f --- /dev/null +++ b/src/daena/config.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import yaml +from pydantic import Field, field_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class LoggingConfig(BaseSettings): + level: str = "info" + format: str = "json" + file: str | None = None + + +class APIHTTPConfig(BaseSettings): + host: str = "127.0.0.1" + port: int = 8642 + workers: int = 1 + + +class APIConfig(BaseSettings): + http: APIHTTPConfig = APIHTTPConfig() + + +class SinkRetryConfig(BaseSettings): + max_attempts: int = 5 + min_delay: float = 1.0 + max_delay: float = 120.0 + multiplier: float = 2.0 + + +class SinkConfig(BaseSettings): + name: str + type: str + enabled: bool = True + retry: SinkRetryConfig = SinkRetryConfig() + options: dict[str, Any] = Field(default_factory=dict) + + +class SourceConfig(BaseSettings): + name: str + type: str + enabled: bool = True + options: dict[str, Any] = Field(default_factory=dict) + + +class ProcessorConfig(BaseSettings): + name: str + type: str + enabled: bool = True + options: dict[str, Any] = Field(default_factory=dict) + + +class LMDBConfig(BaseSettings): + path: str = "/var/lib/daena/data" + map_size: int = 104_857_600 + max_databases: int = 16 + + +class PersistenceConfig(BaseSettings): + backend: str = "lmdb" + lmdb: LMDBConfig = LMDBConfig() + options: dict[str, Any] = Field(default_factory=dict) + + +class PipelineConfig(BaseSettings): + batch_size: int = 100 + flush_interval: float = 5.0 + max_queue_size: int = 10_000 + + +class DaenaConfig(BaseSettings): + model_config = SettingsConfigDict(env_prefix="DAENA_", extra="ignore") + + data_dir: str = "/var/lib/daena" + log_dir: str = "/var/log/daena" + pid_file: str = "/run/daena/daena.pid" + + logging: LoggingConfig = LoggingConfig() + api: APIConfig = APIConfig() + pipeline: PipelineConfig = PipelineConfig() + persistence: PersistenceConfig = PersistenceConfig() + + sources: list[SourceConfig] = Field(default_factory=list) + processors: list[ProcessorConfig] = Field(default_factory=list) + sinks: list[SinkConfig] = Field(default_factory=list) + + @field_validator("data_dir", "log_dir", mode="before") + @classmethod + def ensure_trailing(cls, v: str) -> str: + return v.rstrip("/") + + +def load_config(path: str | Path | None = None) -> DaenaConfig: + config = DaenaConfig() + if path: + p = Path(path) + if not p.exists(): + raise FileNotFoundError(f"Configuration file not found: {p}") + with open(p) as f: + raw: dict[str, Any] = yaml.safe_load(f) or {} + merged = config.model_dump() + merged.update(raw) + config = DaenaConfig(**merged) + return config diff --git a/src/daena/core/__init__.py b/src/daena/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/core/di.py b/src/daena/core/di.py new file mode 100644 index 0000000..8002997 --- /dev/null +++ b/src/daena/core/di.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from typing import Any + + +class Container: + """Minimal dependency injection container. + + Provides singleton and factory resolution for service components. + """ + + def __init__(self) -> None: + self._instances: dict[str, Any] = {} + self._factories: dict[str, callable] = {} + + def register(self, key: str, instance: Any) -> None: + self._instances[key] = instance + + def register_factory(self, key: str, factory: callable) -> None: + self._factories[key] = factory + + def resolve(self, key: str) -> Any: + if key in self._instances: + return self._instances[key] + if key in self._factories: + instance = self._factories[key]() + self._instances[key] = instance + return instance + raise KeyError(f"No component registered for: {key}") + + def has(self, key: str) -> bool: + return key in self._instances or key in self._factories diff --git a/src/daena/core/pipeline.py b/src/daena/core/pipeline.py new file mode 100644 index 0000000..f3a0e46 --- /dev/null +++ b/src/daena/core/pipeline.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import asyncio +import contextlib +from collections.abc import Callable +from typing import Any + +from daena.config import DaenaConfig +from daena.logutil import get_logger +from daena.persistence.backend import Backend +from daena.persistence.models import StoredRecord +from daena.pipeline.base import DeliveryResult, Processor, Sink, Source +from daena.pipeline.record import Record +from daena.plugins.registry import PluginRegistry + +log = get_logger("daena.pipeline") + + +class SourceRunner: + """Wraps a Source and handles record ingestion into the pipeline.""" + + def __init__( + self, + source: Source, + on_record: Callable[[Record], Any], + ) -> None: + self._source = source + self._on_record = on_record + + async def start(self) -> None: + if hasattr(self._source, "set_callback"): + self._source.set_callback(self._on_record) + log.info("source_starting", source=self._source.name) + await self._source.start() + + async def stop(self) -> None: + log.info("source_stopping", source=self._source.name) + await self._source.stop() + + +class SinkDispatcher: + """Manages delivery of records to a sink with retry logic.""" + + def __init__( + self, + sink: Sink, + backend: Backend, + config: DaenaConfig, + ) -> None: + self._sink = sink + self._backend = backend + self._config = config + self._logger = log.bind(sink=sink.name) + + async def start(self) -> None: + self._logger.info("sink_starting") + await self._sink.start() + + async def stop(self) -> None: + self._logger.info("sink_stopping") + await self._sink.stop() + + async def deliver(self, record: Record) -> DeliveryResult: + return await self._sink.deliver(record) + + +class Pipeline: + """Orchestrates the full collection β†’ persist β†’ forward data flow. + + Lifecycle: + 1. Sources emit Records + 2. Processors transform/filter Records + 3. Records are persisted to the backend + 4. SinkDispatcher reads pending records and delivers them + 5. Delivery results update record state (ack/nack/dead) + """ + + def __init__( + self, + config: DaenaConfig, + backend: Backend, + registry: PluginRegistry, + ) -> None: + self._config = config + self._backend = backend + self._registry = registry + self._sources: list[Source] = [] + self._processors: list[Processor] = [] + self._dispatchers: list[SinkDispatcher] = [] + self._source_runners: list[SourceRunner] = [] + self._dispatch_task: asyncio.Task[None] | None = None + self._running = False + self._queue: asyncio.Queue[Record] = asyncio.Queue(maxsize=config.pipeline.max_queue_size) + + async def start(self) -> None: + self._running = True + self._build_pipeline() + for runner in self._source_runners: + await runner.start() + for dispatcher in self._dispatchers: + await dispatcher.start() + self._dispatch_task = asyncio.create_task(self._dispatch_loop()) + log.info( + "pipeline_started", + sources=len(self._source_runners), + sinks=len(self._dispatchers), + processors=len(self._processors), + ) + + async def stop(self) -> None: + self._running = False + if self._dispatch_task: + self._dispatch_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._dispatch_task + for dispatcher in self._dispatchers: + await dispatcher.stop() + for runner in reversed(self._source_runners): + await runner.stop() + log.info("pipeline_stopped") + + def _build_pipeline(self) -> None: + for sc in self._config.sources: + cls = self._registry.get_source(sc.type) + if not cls: + log.warning("source_type_not_found", type=sc.type) + continue + source = cls(name=sc.name, config=sc.options) + self._sources.append(source) + runner = SourceRunner(source, self._ingest) + self._source_runners.append(runner) + + for pc in self._config.processors: + cls = self._registry.get_processor(pc.type) + if not cls: + log.warning("processor_type_not_found", type=pc.type) + continue + proc = cls(name=pc.name, config=pc.options) + self._processors.append(proc) + + for sc in self._config.sinks: + cls = self._registry.get_sink(sc.type) + if not cls: + log.warning("sink_type_not_found", type=sc.type) + continue + sink = cls(name=sc.name, config=sc.options) + self._dispatchers.append(SinkDispatcher(sink, self._backend, self._config)) + + async def _ingest(self, record: Record) -> None: + """Called by sources. Runs processors and persists.""" + for proc in self._processors: + result = await proc.process(record) + if result is None: + return + record = result + + stored = StoredRecord( + record_id=record.record_id, + source=record.source, + record_type=record.record_type, + body=record.body, + metadata=record.metadata, + destination=record.destination, + created_at=record.timestamp, + ) + self._backend.put(stored) + log.debug("record_persisted", record_id=record.record_id) + + async def _dispatch_loop(self) -> None: + """Background loop: pull pending records from backend and deliver.""" + while self._running: + try: + await self._dispatch_batch() + except asyncio.CancelledError: + break + except Exception: + log.exception("dispatch_loop_error") + await asyncio.sleep(1) + + async def _dispatch_batch(self) -> None: + pending = self._backend.pending(limit=self._config.pipeline.batch_size) + retry = self._backend.ready_for_retry(limit=self._config.pipeline.batch_size) + records = pending + retry + + if not records: + await asyncio.sleep(self._config.pipeline.flush_interval) + return + + for stored in records: + if not self._running: + break + record = Record( + source=stored.source, + record_type=stored.record_type, + body=stored.body, + metadata=stored.metadata, + destination=stored.destination, + record_id=stored.record_id, + timestamp=stored.created_at, + ) + for dispatcher in self._dispatchers: + await self._deliver_with_retry(dispatcher, record) + + if len(records) < self._config.pipeline.batch_size: + await asyncio.sleep(self._config.pipeline.flush_interval) + + async def _deliver_with_retry(self, dispatcher: SinkDispatcher, record: Record) -> None: + retry_cfg = None + for sc in self._config.sinks: + if sc.name == dispatcher._sink.name: + retry_cfg = sc.retry + break + + max_attempts = retry_cfg.max_attempts if retry_cfg else 3 + min_delay = retry_cfg.min_delay if retry_cfg else 1.0 + max_delay = retry_cfg.max_delay if retry_cfg else 120.0 + + attempt = 0 + while attempt < max_attempts: + attempt += 1 + try: + result = await dispatcher.deliver(record) + except Exception as exc: + result = DeliveryResult( + success=False, + record_id=record.record_id, + error=str(exc), + ) + + if result.success: + self._backend.ack(record.record_id) + return + + if result.permanent: + self._backend.dead(record.record_id, result.error or "permanent failure") + return + + backoff = min(min_delay * (2 ** (attempt - 1)), max_delay) + stored = self._backend.get(record.record_id) + if stored: + self._backend.nack( + record.record_id, + error=result.error or "unknown", + retry_delay=backoff, + ) + + if attempt < max_attempts: + await asyncio.sleep(backoff) + + self._backend.dead( + record.record_id, + f"max retries ({max_attempts}) exceeded: {result.error}", + ) + + @property + def sources(self) -> list[Source]: + return list(self._sources) + + @property + def sinks(self) -> list[Sink]: + return [d._sink for d in self._dispatchers] + + @property + def queue_size(self) -> int: + return self._queue.qsize() diff --git a/src/daena/core/runtime.py b/src/daena/core/runtime.py new file mode 100644 index 0000000..98200ee --- /dev/null +++ b/src/daena/core/runtime.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +from daena.config import DaenaConfig +from daena.core.pipeline import Pipeline +from daena.logutil import get_logger +from daena.persistence.backend import Backend +from daena.persistence.lmdb import LMDBBackend +from daena.plugins.loader import discover_plugins +from daena.plugins.registry import PluginRegistry, get_registry + +log = get_logger("daena.runtime") + + +class DaenaRuntime: + """Top-level service runtime. + + Owns configuration, persistence backend, pipeline, and lifecycle. + Designed to run under asyncio with clean signal-based shutdown. + """ + + def __init__(self, config: DaenaConfig) -> None: + self._config = config + self._registry = get_registry() + self._backend: Backend | None = None + self._pipeline: Pipeline | None = None + self._running = False + + @property + def config(self) -> DaenaConfig: + return self._config + + @property + def registry(self) -> PluginRegistry: + return self._registry + + @property + def backend(self) -> Backend | None: + return self._backend + + @property + def pipeline(self) -> Pipeline | None: + return self._pipeline + + async def start(self) -> None: + log.info("runtime_starting") + + discover_plugins() + log.info( + "plugins_discovered", + sources=list(self._registry.list_sources().keys()), + processors=list(self._registry.list_processors().keys()), + sinks=list(self._registry.list_sinks().keys()), + ) + + persistence_cfg = self._config.persistence + if persistence_cfg.backend == "lmdb": + self._backend = LMDBBackend( + path=persistence_cfg.lmdb.path, + map_size=persistence_cfg.lmdb.map_size, + max_databases=persistence_cfg.lmdb.max_databases, + ) + else: + raise ValueError(f"Unknown backend: {persistence_cfg.backend}") + + self._backend.open() + log.info("backend_opened", backend=persistence_cfg.backend) + + self._pipeline = Pipeline(self._config, self._backend, self._registry) + await self._pipeline.start() + self._running = True + log.info("runtime_started") + + async def stop(self) -> None: + log.info("runtime_stopping") + self._running = False + + if self._pipeline: + await self._pipeline.stop() + + if self._backend: + self._backend.close() + log.info("backend_closed") + + log.info("runtime_stopped") + + async def reload(self) -> None: + log.info("runtime_reloading") + await self.stop() + await self.start() + log.info("runtime_reloaded") diff --git a/src/daena/exceptions.py b/src/daena/exceptions.py new file mode 100644 index 0000000..513cfce --- /dev/null +++ b/src/daena/exceptions.py @@ -0,0 +1,37 @@ +from __future__ import annotations + + +class DaenaError(Exception): + """Base exception for all daena errors.""" + + +class ConfigurationError(DaenaError): + """Configuration loading or validation failed.""" + + +class BackendError(DaenaError): + """Persistence backend operation failed.""" + + +class PipelineError(DaenaError): + """Pipeline processing error.""" + + +class SinkError(DaenaError): + """Sink delivery failed.""" + + +class SinkPermanentError(SinkError): + """Non-retryable sink failure.""" + + +class SourceError(DaenaError): + """Source collection error.""" + + +class PluginError(DaenaError): + """Plugin loading or registration error.""" + + +class RecordInvalidError(DaenaError): + """Record validation failed.""" diff --git a/src/daena/logutil.py b/src/daena/logutil.py new file mode 100644 index 0000000..221b90f --- /dev/null +++ b/src/daena/logutil.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import logging +import sys +from pathlib import Path + +import structlog + + +def setup_logging( + level: str = "info", + fmt: str = "json", + log_file: str | None = None, +) -> None: + level_map: dict[str, int] = { + "debug": logging.DEBUG, + "info": logging.INFO, + "warning": logging.WARNING, + "error": logging.ERROR, + "critical": logging.CRITICAL, + } + log_level = level_map.get(level.lower(), logging.INFO) + + processors: list[structlog.typing.Processor] = [ + structlog.stdlib.add_log_level, + structlog.stdlib.add_logger_name, + structlog.processors.TimeStamper(fmt="iso"), + structlog.processors.StackInfoRenderer(), + structlog.dev.set_exc_info, + structlog.processors.format_exc_info, + ] + + if fmt == "json": + processors.append(structlog.processors.JSONRenderer()) + else: + processors.append(structlog.dev.ConsoleRenderer()) + + structlog.configure( + processors=processors, + wrapper_class=structlog.stdlib.BoundLogger, + context_class=dict, + logger_factory=structlog.stdlib.LoggerFactory(), + cache_logger_on_first_use=True, + ) + + logging.basicConfig( + format="%(message)s", + stream=sys.stdout, + level=log_level, + force=True, + ) + + if log_file: + p = Path(log_file) + p.parent.mkdir(parents=True, exist_ok=True) + handler = logging.FileHandler(str(p)) + handler.setLevel(log_level) + handler.setFormatter(logging.Formatter("%(message)s")) + logging.getLogger().addHandler(handler) + + +def get_logger(name: str | None = None) -> structlog.stdlib.BoundLogger: + return structlog.get_logger(name or __name__) diff --git a/src/daena/persistence/__init__.py b/src/daena/persistence/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/persistence/backend.py b/src/daena/persistence/backend.py new file mode 100644 index 0000000..b3cc60a --- /dev/null +++ b/src/daena/persistence/backend.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Protocol + +from daena.persistence.models import RecordState, StoredRecord + + +class Backend(Protocol): + """Abstract interface for persistence backends. + + This Protocol defines the contract that all storage backends must fulfill. + The MVP implementation uses LMDB; future implementations may target + NATS JetStream, filesystem-backed stores, or cloud queues. + """ + + def open(self) -> None: + """Open or initialize the backend.""" + + def close(self) -> None: + """Close the backend and release resources.""" + + def put(self, record: StoredRecord) -> None: + """Store a new record.""" + + def get(self, record_id: str) -> StoredRecord | None: + """Retrieve a record by ID.""" + + def update_state( + self, + record_id: str, + state: RecordState, + last_error: str | None = None, + next_retry_at: str | None = None, + ) -> None: + """Update a record's delivery state.""" + + def pending( + self, + limit: int = 100, + older_than: str | None = None, + ) -> list[StoredRecord]: + """Return pending records for delivery, oldest first.""" + + def ready_for_retry( + self, + limit: int = 100, + now: str | None = None, + ) -> list[StoredRecord]: + """Return failed records whose retry time has arrived.""" + + def ack(self, record_id: str) -> None: + """Mark a record as successfully delivered.""" + + def nack( + self, + record_id: str, + error: str, + retry_delay: float = 10.0, + ) -> None: + """Mark delivery failed and schedule retry.""" + + def dead(self, record_id: str, error: str) -> None: + """Move a record to the dead-letter queue.""" + + def count(self, state: RecordState | None = None) -> int: + """Count records, optionally filtered by state.""" + + def list_by_state( + self, + state: RecordState, + limit: int = 100, + offset: int = 0, + ) -> list[StoredRecord]: + """List records by state with pagination.""" + + def iter_pending(self) -> Iterator[StoredRecord]: + """Iterate over all pending records (for recovery).""" + + def delete_old( + self, + before: str, + state: RecordState | None = None, + ) -> int: + """Delete records older than a timestamp. Returns count.""" + + def cleanup(self) -> None: + """Perform backend-specific maintenance.""" diff --git a/src/daena/persistence/lmdb.py b/src/daena/persistence/lmdb.py new file mode 100644 index 0000000..0b15460 --- /dev/null +++ b/src/daena/persistence/lmdb.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import json +from collections.abc import Iterator +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import lmdb + +from daena.exceptions import BackendError +from daena.persistence.models import RecordState, StoredRecord + +_REC_PREFIX = b"rec:" +_IDX_PENDING = b"idx:pending:" +_IDX_DELIVERED = b"idx:delivered:" +_IDX_FAILED = b"idx:failed:" +_IDX_DEAD = b"idx:dead:" +_IDX_RETRY = b"idx:retry:" + +_PREFIX_MAP: dict[RecordState, bytes] = { + RecordState.pending: _IDX_PENDING, + RecordState.delivered: _IDX_DELIVERED, + RecordState.failed: _IDX_FAILED, + RecordState.dead: _IDX_DEAD, +} + + +class LMDBBackend: + """LMDB-based persistence backend using key-prefix partitioning. + + Uses a single LMDB database with key prefixes to emulate named + sub-databases. + + Key design: + - rec:{record_id} -> serialized record JSON + - idx:pending:{ts}\x00{id} -> "" (ordered FIFO queue) + - idx:delivered:{ts}\x00{id} -> "" + - idx:failed:{ts}\x00{id} -> "" + - idx:dead:{ts}\x00{id} -> "" + - idx:retry:{next}\x00{id} -> "" + """ + + def __init__( + self, + path: str | Path = "/var/lib/daena/data", + map_size: int = 104_857_600, + max_databases: int = 16, + ) -> None: + self._path = Path(path) + self._map_size = map_size + self._max_databases = max_databases + self._env: lmdb.Environment | None = None + + def _ensure_env(self) -> lmdb.Environment: + if self._env is None: + raise BackendError("Backend not opened") + return self._env + + def open(self) -> None: + self._path.mkdir(parents=True, exist_ok=True) + self._env = lmdb.open(str(self._path), self._map_size, self._max_databases) + + def close(self) -> None: + if self._env is not None: + self._env.close() + self._env = None + + def _rec_key(self, record_id: str) -> bytes: + return _REC_PREFIX + record_id.encode() + + def _idx_key(self, prefix: bytes, record_id: str) -> bytes: + ts = datetime.now(UTC).isoformat() + return prefix + f"{ts}\x00{record_id}".encode() + + def _record_id_from_idx(self, key: bytes, prefix: bytes) -> str: + suffix = key[len(prefix) :].decode() + return suffix.split("\x00", 1)[1] + + def _read_record(self, txn: lmdb.Transaction, record_id: str) -> StoredRecord | None: + raw = txn.get(self._rec_key(record_id)) + if raw is None: + return None + return StoredRecord.from_dict(json.loads(raw)) + + def put(self, record: StoredRecord) -> None: + env = self._ensure_env() + rec_key = self._rec_key(record.record_id) + value = json.dumps(record.to_dict()).encode() + idx_key = self._idx_key(_IDX_PENDING, record.record_id) + + with env.begin(write=True) as txn: + if txn.get(rec_key) is not None: + return + txn.put(rec_key, value) + txn.put(idx_key, b"") + + def get(self, record_id: str) -> StoredRecord | None: + env = self._ensure_env() + key = self._rec_key(record_id) + with env.begin() as txn: + value = txn.get(key) + if value is None: + return None + return StoredRecord.from_dict(json.loads(value)) + + def update_state( + self, + record_id: str, + state: RecordState, + last_error: str | None = None, + next_retry_at: str | None = None, + ) -> None: + env = self._ensure_env() + rec_key = self._rec_key(record_id) + now = datetime.now(UTC).isoformat() + + with env.begin(write=True) as txn: + raw = txn.get(rec_key) + if raw is None: + return + record_dict: dict[str, Any] = json.loads(raw) + record_dict["state"] = state.value + record_dict["updated_at"] = now + if last_error is not None: + record_dict["last_error"] = last_error + if next_retry_at is not None: + record_dict["next_retry_at"] = next_retry_at + txn.put(rec_key, json.dumps(record_dict).encode()) + + target_prefix = _PREFIX_MAP.get(state) + if target_prefix and state != RecordState.pending: + new_idx = self._idx_key(target_prefix, record_id) + txn.put(new_idx, b"") + + if state == RecordState.failed and next_retry_at: + retry_key = _IDX_RETRY + f"{next_retry_at}\x00{record_id}".encode() + txn.put(retry_key, b"") + + def pending(self, limit: int = 100, older_than: str | None = None) -> list[StoredRecord]: + _ = older_than + return self._list_by_state_prefix(_IDX_PENDING, RecordState.pending, limit) + + def ready_for_retry(self, limit: int = 100, now: str | None = None) -> list[StoredRecord]: + env = self._ensure_env() + now_str = now or datetime.now(UTC).isoformat() + results: list[StoredRecord] = [] + seen: set[str] = set() + with env.begin() as txn: + cursor = txn.cursor() + if not cursor.set_range(_IDX_RETRY): + return results + for key, _ in cursor: + if not key.startswith(_IDX_RETRY): + break + record_id = self._record_id_from_idx(key, _IDX_RETRY) + retry_ts = key[len(_IDX_RETRY) :].decode().split("\x00", 1)[0] + if retry_ts > now_str: + break + if record_id in seen: + continue + seen.add(record_id) + record = self._read_record(txn, record_id) + if record and record.state == RecordState.failed: + results.append(record) + if len(results) >= limit: + break + return results + + def ack(self, record_id: str) -> None: + self.update_state(record_id, RecordState.delivered) + + def nack(self, record_id: str, error: str, retry_delay: float = 10.0) -> None: + from datetime import timedelta + + next_retry = (datetime.now(UTC) + timedelta(seconds=retry_delay)).isoformat() + self.update_state( + record_id, + RecordState.failed, + last_error=error, + next_retry_at=next_retry, + ) + + def dead(self, record_id: str, error: str) -> None: + self.update_state(record_id, RecordState.dead, last_error=error) + + def count(self, state: RecordState | None = None) -> int: + env = self._ensure_env() + with env.begin() as txn: + if state is None: + cursor = txn.cursor() + count = 0 + if cursor.set_range(_REC_PREFIX): + for key, _ in cursor: + if not key.startswith(_REC_PREFIX): + break + count += 1 + return count + prefix = _PREFIX_MAP.get(state) + if prefix is None: + return 0 + cursor = txn.cursor() + count = 0 + if cursor.set_range(prefix): + for key, _ in cursor: + if not key.startswith(prefix): + break + record_id = self._record_id_from_idx(key, prefix) + record = self._read_record(txn, record_id) + if record and record.state == state: + count += 1 + return count + + def list_by_state( + self, state: RecordState, limit: int = 100, offset: int = 0 + ) -> list[StoredRecord]: + prefix = _PREFIX_MAP.get(state) + if prefix is None: + return [] + return self._list_by_state_prefix(prefix, state, limit, offset) + + def iter_pending(self) -> Iterator[StoredRecord]: + seen: set[str] = set() + env = self._ensure_env() + with env.begin() as txn: + cursor = txn.cursor() + if cursor.set_range(_IDX_PENDING): + for key, _ in cursor: + if not key.startswith(_IDX_PENDING): + break + record_id = self._record_id_from_idx(key, _IDX_PENDING) + if record_id in seen: + continue + seen.add(record_id) + record = self._read_record(txn, record_id) + if record and record.state == RecordState.pending: + yield record + + def delete_old(self, before: str, state: RecordState | None = None) -> int: + env = self._ensure_env() + removed = 0 + + prefixes = [_IDX_DELIVERED, _IDX_DEAD] + if state: + p = _PREFIX_MAP.get(state) + prefixes = [p] if p else [] + + with env.begin(write=True) as txn: + for prefix in prefixes: + cursor = txn.cursor() + if not cursor.set_range(prefix): + continue + seen_del: set[str] = set() + for key, _ in cursor: + if not key.startswith(prefix): + break + suffix = key[len(prefix) :].decode() + ts_part = suffix.split("\x00", 1)[0] + if ts_part > before: + break + record_id = suffix.split("\x00", 1)[1] + if record_id in seen_del: + continue + seen_del.add(record_id) + txn.delete(self._rec_key(record_id)) + txn.delete(key) + removed += 1 + return removed + + def cleanup(self) -> None: + pass + + def _list_by_state_prefix( + self, + prefix: bytes, + state: RecordState, + limit: int = 100, + offset: int = 0, + ) -> list[StoredRecord]: + seen: set[str] = set() + env = self._ensure_env() + results: list[StoredRecord] = [] + with env.begin() as txn: + cursor = txn.cursor() + if not cursor.set_range(prefix): + return results + skipped = 0 + for key, _ in cursor: + if not key.startswith(prefix): + break + record_id = self._record_id_from_idx(key, prefix) + if record_id in seen: + continue + seen.add(record_id) + record = self._read_record(txn, record_id) + if record is None or record.state != state: + continue + if skipped < offset: + skipped += 1 + continue + results.append(record) + if len(results) >= limit: + break + return results diff --git a/src/daena/persistence/models.py b/src/daena/persistence/models.py new file mode 100644 index 0000000..74e1af7 --- /dev/null +++ b/src/daena/persistence/models.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from enum import StrEnum +from typing import Any + + +class RecordState(StrEnum): + pending = "pending" + delivering = "delivering" + delivered = "delivered" + failed = "failed" + dead = "dead" + + +class StoredRecord: + """A record as stored in the persistence backend.""" + + def __init__( + self, + record_id: str | None = None, + source: str = "", + record_type: str = "", + body: dict[str, Any] | None = None, + metadata: dict[str, Any] | None = None, + state: RecordState = RecordState.pending, + created_at: str | None = None, + updated_at: str | None = None, + retry_count: int = 0, + next_retry_at: str | None = None, + last_error: str | None = None, + destination: str | None = None, + ) -> None: + self.record_id = record_id or str(uuid.uuid4()) + self.source = source + self.record_type = record_type + self.body = body or {} + self.metadata = metadata or {} + self.state = state + self.created_at = created_at or datetime.now(UTC).isoformat() + self.updated_at = updated_at or self.created_at + self.retry_count = retry_count + self.next_retry_at = next_retry_at + self.last_error = last_error + self.destination = destination + + def to_dict(self) -> dict[str, Any]: + return { + "record_id": self.record_id, + "source": self.source, + "record_type": self.record_type, + "body": self.body, + "metadata": self.metadata, + "state": self.state.value, + "created_at": self.created_at, + "updated_at": self.updated_at, + "retry_count": self.retry_count, + "next_retry_at": self.next_retry_at, + "last_error": self.last_error, + "destination": self.destination, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> StoredRecord: + return cls( + record_id=data.get("record_id"), + source=data.get("source", ""), + record_type=data.get("record_type", ""), + body=data.get("body", {}), + metadata=data.get("metadata", {}), + state=RecordState(data.get("state", "pending")), + created_at=data.get("created_at"), + updated_at=data.get("updated_at"), + retry_count=data.get("retry_count", 0), + next_retry_at=data.get("next_retry_at"), + last_error=data.get("last_error"), + destination=data.get("destination"), + ) diff --git a/src/daena/pipeline/__init__.py b/src/daena/pipeline/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/pipeline/base.py b/src/daena/pipeline/base.py new file mode 100644 index 0000000..65a2f22 --- /dev/null +++ b/src/daena/pipeline/base.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any + +from daena.pipeline.record import Record + + +@dataclass +class DeliveryResult: + success: bool + record_id: str + error: str | None = None + permanent: bool = False + + +class Source(ABC): + """Base class for all record sources. + + Sources collect or receive data and push Records into the pipeline. + """ + + name: str + + def __init__(self, name: str, config: dict[str, Any] | None = None) -> None: + self.name = name + self.config = config or {} + + @abstractmethod + async def start(self) -> None: + """Start collecting records. Called during service startup.""" + + @abstractmethod + async def stop(self) -> None: + """Stop collecting. Called during shutdown.""" + + +class Processor(ABC): + """Base class for record processors. + + Processors transform, filter, or enrich Records passing through the pipeline. + Return None to drop the record. + """ + + name: str + + def __init__(self, name: str, config: dict[str, Any] | None = None) -> None: + self.name = name + self.config = config or {} + + @abstractmethod + async def process(self, record: Record) -> Record | None: ... + + +class Sink(ABC): + """Base class for record sinks. + + Sinks deliver Records to external destinations. + """ + + name: str + + def __init__(self, name: str, config: dict[str, Any] | None = None) -> None: + self.name = name + self.config = config or {} + + @abstractmethod + async def deliver(self, record: Record) -> DeliveryResult: ... + + async def start(self) -> None: + return + + async def stop(self) -> None: + return diff --git a/src/daena/pipeline/processors/__init__.py b/src/daena/pipeline/processors/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/pipeline/processors/passthrough.py b/src/daena/pipeline/processors/passthrough.py new file mode 100644 index 0000000..64679a4 --- /dev/null +++ b/src/daena/pipeline/processors/passthrough.py @@ -0,0 +1,11 @@ +from __future__ import annotations + +from daena.pipeline.base import Processor +from daena.pipeline.record import Record + + +class PassthroughProcessor(Processor): + """Passes records through without modification.""" + + async def process(self, record: Record) -> Record | None: + return record diff --git a/src/daena/pipeline/record.py b/src/daena/pipeline/record.py new file mode 100644 index 0000000..474f9d0 --- /dev/null +++ b/src/daena/pipeline/record.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + + +@dataclass +class Record: + """In-memory representation of a collected event or log.""" + + source: str + record_type: str + body: dict[str, Any] + metadata: dict[str, Any] = field(default_factory=dict) + destination: str | None = None + record_id: str = field(default_factory=lambda: str(uuid.uuid4())) + timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) diff --git a/src/daena/pipeline/sinks/__init__.py b/src/daena/pipeline/sinks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/pipeline/sinks/dummy.py b/src/daena/pipeline/sinks/dummy.py new file mode 100644 index 0000000..2c4bbc6 --- /dev/null +++ b/src/daena/pipeline/sinks/dummy.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from daena.pipeline.base import DeliveryResult, Sink +from daena.pipeline.record import Record + + +class DummySink(Sink): + """Logs delivered records without sending anywhere. + + Useful for testing and development. + """ + + async def deliver(self, record: Record) -> DeliveryResult: + return DeliveryResult(success=True, record_id=record.record_id) diff --git a/src/daena/pipeline/sinks/http.py b/src/daena/pipeline/sinks/http.py new file mode 100644 index 0000000..18da020 --- /dev/null +++ b/src/daena/pipeline/sinks/http.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from typing import Any + +import httpx + +from daena.pipeline.base import DeliveryResult, Sink +from daena.pipeline.record import Record + + +class HTTPSink(Sink): + """Delivers records to a remote HTTP endpoint. + + Supports configurable URL, method, headers, timeouts, and batching. + """ + + def __init__(self, name: str, config: dict[str, Any] | None = None) -> None: + super().__init__(name, config) + cfg = config or {} + self._url: str = cfg.get("url", "http://localhost:9999/ingest") + self._method: str = cfg.get("method", "POST").upper() + self._headers: dict[str, str] = cfg.get("headers", {}) + self._timeout: float = float(cfg.get("timeout", 30.0)) + self._client: httpx.AsyncClient | None = None + + async def start(self) -> None: + self._client = httpx.AsyncClient( + timeout=httpx.Timeout(self._timeout), + headers=self._headers, + ) + + async def stop(self) -> None: + if self._client: + await self._client.aclose() + self._client = None + + async def deliver(self, record: Record) -> DeliveryResult: + if self._client is None: + return DeliveryResult( + success=False, + record_id=record.record_id, + error="HTTP sink not initialized", + ) + + payload = { + "id": record.record_id, + "source": record.source, + "type": record.record_type, + "timestamp": record.timestamp, + "body": record.body, + "metadata": record.metadata, + } + + try: + response = await self._client.request( + method=self._method, + url=self._url, + json=payload, + ) + if response.is_success: + return DeliveryResult(success=True, record_id=record.record_id) + if 400 <= response.status_code < 500 and response.status_code not in (429,): + return DeliveryResult( + success=False, + record_id=record.record_id, + error=f"HTTP {response.status_code}: {response.text[:200]}", + permanent=True, + ) + return DeliveryResult( + success=False, + record_id=record.record_id, + error=f"HTTP {response.status_code}: {response.text[:200]}", + ) + except httpx.TimeoutException as exc: + return DeliveryResult( + success=False, + record_id=record.record_id, + error=f"Timeout: {exc}", + ) + except httpx.RequestError as exc: + return DeliveryResult( + success=False, + record_id=record.record_id, + error=str(exc), + ) diff --git a/src/daena/pipeline/sources/__init__.py b/src/daena/pipeline/sources/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/pipeline/sources/dummy.py b/src/daena/pipeline/sources/dummy.py new file mode 100644 index 0000000..0ef3cf1 --- /dev/null +++ b/src/daena/pipeline/sources/dummy.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import asyncio +import contextlib +from typing import Any + +from daena.pipeline.base import Source +from daena.pipeline.record import Record + + +class DummySource(Source): + """Generates test records at a configurable interval. + + Useful for development, testing, and demo purposes. + """ + + def __init__(self, name: str, config: dict[str, Any] | None = None) -> None: + super().__init__(name, config) + self._interval: float = float(config.get("interval", 5.0) if config else 5.0) + self._message: str = ( + config.get("message", "daena test event") if config else "daena test event" + ) + self._task: asyncio.Task[None] | None = None + self._running = False + self._on_record: callable | None = None + + def set_callback(self, cb: callable) -> None: + self._on_record = cb + + async def start(self) -> None: + self._running = True + self._task = asyncio.create_task(self._emit_loop()) + + async def stop(self) -> None: + self._running = False + if self._task: + self._task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._task + + async def _emit_loop(self) -> None: + counter = 0 + while self._running: + counter += 1 + record = Record( + source=self.name, + record_type="test", + body={ + "message": f"{self._message} #{counter}", + "counter": counter, + }, + metadata={"dummy": True}, + ) + if self._on_record: + await self._on_record(record) + await asyncio.sleep(self._interval) diff --git a/src/daena/plugins/__init__.py b/src/daena/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/daena/plugins/loader.py b/src/daena/plugins/loader.py new file mode 100644 index 0000000..435197e --- /dev/null +++ b/src/daena/plugins/loader.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from importlib.metadata import entry_points + +from daena.exceptions import PluginError +from daena.plugins.registry import get_registry + +SOURCE_EP = "daena.sources" +PROCESSOR_EP = "daena.processors" +SINK_EP = "daena.sinks" + + +def discover_plugins() -> None: + """Discover plugins via package entry points and register them.""" + registry = get_registry() + errors: list[str] = [] + + for ep in entry_points(group=SOURCE_EP): + try: + cls = ep.load() + registry.register_source(ep.name, cls) + except Exception as exc: + errors.append(f"source {ep.name}: {exc}") + + for ep in entry_points(group=PROCESSOR_EP): + try: + cls = ep.load() + registry.register_processor(ep.name, cls) + except Exception as exc: + errors.append(f"processor {ep.name}: {exc}") + + for ep in entry_points(group=SINK_EP): + try: + cls = ep.load() + registry.register_sink(ep.name, cls) + except Exception as exc: + errors.append(f"sink {ep.name}: {exc}") + + if errors: + raise PluginError("Plugin loading encountered errors:\n " + "\n ".join(errors)) diff --git a/src/daena/plugins/registry.py b/src/daena/plugins/registry.py new file mode 100644 index 0000000..6660f36 --- /dev/null +++ b/src/daena/plugins/registry.py @@ -0,0 +1,55 @@ +from __future__ import annotations + + +class PluginRegistry: + """Registry for plugin factories. + + Maintains separate maps for sources, processors, and sinks. + Each entry maps plugin type name β†’ factory callable. + """ + + def __init__(self) -> None: + self._sources: dict[str, type] = {} + self._processors: dict[str, type] = {} + self._sinks: dict[str, type] = {} + + def register_source(self, name: str, cls: type) -> None: + self._sources[name] = cls + + def register_processor(self, name: str, cls: type) -> None: + self._processors[name] = cls + + def register_sink(self, name: str, cls: type) -> None: + self._sinks[name] = cls + + def get_source(self, name: str) -> type | None: + return self._sources.get(name) + + def get_processor(self, name: str) -> type | None: + return self._processors.get(name) + + def get_sink(self, name: str) -> type | None: + return self._sinks.get(name) + + def list_sources(self) -> dict[str, type]: + return dict(self._sources) + + def list_processors(self) -> dict[str, type]: + return dict(self._processors) + + def list_sinks(self) -> dict[str, type]: + return dict(self._sinks) + + def all_plugins(self) -> dict[str, dict[str, type]]: + return { + "sources": self.list_sources(), + "processors": self.list_processors(), + "sinks": self.list_sinks(), + } + + +_registry = PluginRegistry() + + +def get_registry() -> PluginRegistry: + return _registry diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..f7efefa --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path + +import pytest +import yaml + +from daena.config import DaenaConfig +from daena.persistence.lmdb import LMDBBackend +from daena.persistence.models import StoredRecord + + +@pytest.fixture +def tmp_db_path(tmp_path: Path) -> Path: + return tmp_path / "lmdb" + + +@pytest.fixture +def lmdb_backend(tmp_db_path: Path) -> Iterator[LMDBBackend]: + backend = LMDBBackend( + path=str(tmp_db_path), + map_size=10_485_760, + ) + backend.open() + yield backend + backend.close() + + +@pytest.fixture +def sample_record() -> StoredRecord: + return StoredRecord( + source="test_source", + record_type="test", + body={"message": "hello"}, + metadata={"env": "test"}, + ) + + +@pytest.fixture +def minimal_config() -> DaenaConfig: + return DaenaConfig() + + +@pytest.fixture +def config_file(tmp_path: Path) -> Path: + path = tmp_path / "daena.yaml" + data = { + "data_dir": "/tmp/daena_data", + "logging": {"level": "debug", "format": "json"}, + "sources": [ + { + "name": "test", + "type": "dummy", + "enabled": True, + "options": {"interval": 1.0}, + } + ], + "sinks": [ + { + "name": "output", + "type": "dummy", + "enabled": True, + "options": {}, + } + ], + } + with open(path, "w") as f: + yaml.dump(data, f) + return path diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..ce21d8f --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from pathlib import Path + +from daena.config import DaenaConfig, load_config + + +def test_minimal_config_has_defaults() -> None: + config = DaenaConfig() + assert config.data_dir == "/var/lib/daena" + assert config.logging.level == "info" + assert config.api.http.port == 8642 + assert config.pipeline.batch_size == 100 + + +def test_load_config_from_file(config_file: Path) -> None: + config = load_config(str(config_file)) + assert config.data_dir == "/tmp/daena_data" + assert config.logging.level == "debug" + assert len(config.sources) == 1 + assert config.sources[0].name == "test" + assert config.sources[0].type == "dummy" + + +def test_load_config_nonexistent_file() -> None: + try: + load_config("/nonexistent/daena.yaml") + raise AssertionError("Expected FileNotFoundError") # noqa: B011 + except FileNotFoundError: + pass + + +def test_config_can_set_persistence_backend() -> None: + config = DaenaConfig() + assert config.persistence.backend == "lmdb" + assert config.persistence.lmdb.map_size == 104_857_600 + + +def test_config_source_options(minimal_config: DaenaConfig) -> None: + cfg = minimal_config + assert cfg.sources == [] + assert cfg.processors == [] + assert cfg.sinks == [] diff --git a/tests/test_persistence.py b/tests/test_persistence.py new file mode 100644 index 0000000..7244e90 --- /dev/null +++ b/tests/test_persistence.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +from daena.persistence.lmdb import LMDBBackend +from daena.persistence.models import RecordState, StoredRecord + + +def test_put_and_get(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + retrieved = lmdb_backend.get(sample_record.record_id) + assert retrieved is not None + assert retrieved.record_id == sample_record.record_id + assert retrieved.source == "test_source" + assert retrieved.body == {"message": "hello"} + + +def test_pending_returns_new_records( + lmdb_backend: LMDBBackend, sample_record: StoredRecord +) -> None: + lmdb_backend.put(sample_record) + pending = lmdb_backend.pending(limit=10) + assert len(pending) == 1 + assert pending[0].record_id == sample_record.record_id + + +def test_ack(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + lmdb_backend.ack(sample_record.record_id) + retrieved = lmdb_backend.get(sample_record.record_id) + assert retrieved is not None + assert retrieved.state == RecordState.delivered + pending = lmdb_backend.pending() + assert all(r.record_id != sample_record.record_id for r in pending) + + +def test_nack(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + lmdb_backend.nack(sample_record.record_id, error="test error", retry_delay=1.0) + retrieved = lmdb_backend.get(sample_record.record_id) + assert retrieved is not None + assert retrieved.state == RecordState.failed + assert retrieved.last_error == "test error" + assert retrieved.next_retry_at is not None + + +def test_dead(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + lmdb_backend.dead(sample_record.record_id, error="permanent failure") + retrieved = lmdb_backend.get(sample_record.record_id) + assert retrieved is not None + assert retrieved.state == RecordState.dead + + +def test_count(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + assert lmdb_backend.count() == 0 + lmdb_backend.put(sample_record) + assert lmdb_backend.count() == 1 + assert lmdb_backend.count(RecordState.pending) == 1 + + +def test_list_by_state(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + pending = lmdb_backend.list_by_state(RecordState.pending) + assert len(pending) == 1 + lmdb_backend.ack(sample_record.record_id) + delivered = lmdb_backend.list_by_state(RecordState.delivered) + assert len(delivered) == 1 + + +def test_delete_old(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + lmdb_backend.ack(sample_record.record_id) + before = "2100-01-01T00:00:00" + removed = lmdb_backend.delete_old(before, state=RecordState.delivered) + assert removed == 1 + assert lmdb_backend.get(sample_record.record_id) is None + + +def test_iter_pending(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + records = list(lmdb_backend.iter_pending()) + assert len(records) == 1 + assert records[0].record_id == sample_record.record_id + + +def test_update_state(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + lmdb_backend.update_state(sample_record.record_id, RecordState.failed, last_error="oops") + retrieved = lmdb_backend.get(sample_record.record_id) + assert retrieved is not None + assert retrieved.state == RecordState.failed + assert retrieved.last_error == "oops" + + +def test_stored_record_to_from_dict(sample_record: StoredRecord) -> None: + d = sample_record.to_dict() + restored = StoredRecord.from_dict(d) + assert restored.record_id == sample_record.record_id + assert restored.source == sample_record.source + assert restored.body == sample_record.body + assert restored.state == sample_record.state + + +def test_ready_for_retry(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None: + lmdb_backend.put(sample_record) + lmdb_backend.nack(sample_record.record_id, "retry later", retry_delay=0) + ready = lmdb_backend.ready_for_retry(limit=10) + assert len(ready) == 1 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..f35488b --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import pytest + +from daena.pipeline.base import DeliveryResult, Processor +from daena.pipeline.processors.passthrough import PassthroughProcessor +from daena.pipeline.record import Record +from daena.pipeline.sinks.dummy import DummySink +from daena.pipeline.sinks.http import HTTPSink +from daena.pipeline.sources.dummy import DummySource + + +def test_dummy_source_config() -> None: + source = DummySource( + name="test", + config={"interval": 1.0, "message": "hello"}, + ) + assert source.name == "test" + assert source._interval == 1.0 + assert source._message == "hello" + + +@pytest.mark.asyncio +async def test_dummy_sink_delivery() -> None: + sink = DummySink(name="test") + record = Record(source="test", record_type="test", body={"key": "value"}) + result = await sink.deliver(record) + assert isinstance(result, DeliveryResult) + assert result.success is True + assert result.record_id == record.record_id + + +@pytest.mark.asyncio +async def test_passthrough_processor() -> None: + proc = PassthroughProcessor(name="test") + record = Record(source="test", record_type="test", body={"key": "value"}) + result = await proc.process(record) + assert result is not None + assert result.record_id == record.record_id + assert result.body == {"key": "value"} + + +def test_http_sink_requires_url() -> None: + sink = HTTPSink(name="test", config={"url": "http://example.com/ingest"}) + assert sink._url == "http://example.com/ingest" + assert sink._method == "POST" + + +def test_record_has_id() -> None: + r1 = Record(source="s1", record_type="t1", body={}) + r2 = Record(source="s2", record_type="t2", body={}) + assert r1.record_id != r2.record_id + assert r1.timestamp is not None + + +@pytest.mark.asyncio +async def test_processor_drop_record() -> None: + class DropProcessor(Processor): + async def process(self, record: Record) -> Record | None: # noqa: ARG002 + return None + + proc = DropProcessor(name="dropper") + record = Record(source="test", record_type="test", body={}) + result = await proc.process(record) + assert result is None