diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..aad5be5 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,38 @@ +name: CI + +on: + push: + branches: [main, dev] + pull_request: + branches: [main] + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Install uv + run: pip install uv --no-cache-dir + - name: Install dependencies + run: uv sync --extra dev + - name: Ruff check + run: uv run ruff check src/ tests/ + - name: Ruff format + run: uv run ruff format --check src/ tests/ + + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Install uv + run: pip install uv --no-cache-dir + - name: Install dependencies + run: uv sync --extra dev + - name: Run tests + run: uv run pytest tests/ -v diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..c68daba --- /dev/null +++ b/Dockerfile @@ -0,0 +1,27 @@ +FROM python:3.13-slim AS builder + +RUN pip install uv --no-cache-dir + +WORKDIR /app +COPY pyproject.toml . +RUN uv sync --no-dev --frozen + +FROM python:3.13-slim + +RUN apt-get update && apt-get install -y --no-install-recommends \ + liblmdb0 \ + && rm -rf /var/lib/apt/lists/* + +RUN groupadd -r daena && useradd -r -g daena -d /var/lib/daena daena + +COPY --from=builder /app /app +COPY src/ /app/src/ +COPY proto/ /app/proto/ + +ENV PATH="/app/.venv/bin:$PATH" \ + DAENA_CONFIG_PATH=/etc/daena/daena.yaml + +EXPOSE 8642 + +USER daena +ENTRYPOINT ["daena-serve"] diff --git a/README.md b/README.md index 9f85c94..ab344de 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Daena is a modular, extensible Linux service that: - 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 +- Exposes a gRPC API for management and inspection - Provides a CLI for operational tasks - Degrades gracefully under failures with retry, backoff, and dead-letter queues @@ -67,7 +67,10 @@ asyncio.run(main()) # Run tests uv run pytest tests/ -v -# Use the CLI (point at a running daena API) +# Start gRPC server +uv run daena-serve + +# Use the CLI (point at a running daena gRPC API) uv run daena status uv run daena health uv run daena queue @@ -78,15 +81,20 @@ uv run daena queue ``` daena/ ├── pyproject.toml # Project metadata, dependencies, entry points +├── Dockerfile # Container image +├── proto/daena/v1/daena.proto # gRPC service definition ├── config/daena.yaml # Example configuration ├── deploy/daena.service # systemd unit file +├── .github/workflows/ci.yml # CI pipeline ├── src/daena/ -│ ├── app.py # `daena-serve` entry point (uvicorn) +│ ├── app.py # `daena-serve` entry point (async gRPC) │ ├── config.py # YAML → pydantic config │ ├── logutil.py # structlog setup │ ├── exceptions.py # Domain exceptions -│ ├── api/ # FastAPI REST API (v1 versioned) -│ ├── cli/ # Typer CLI +│ ├── api/ +│ │ ├── grpc_server.py # DaenaServicer + DaenaGRPCServer +│ │ └── pb2/ # Generated gRPC stubs +│ ├── cli/ # Typer CLI (gRPC client) │ ├── core/ │ │ ├── runtime.py # Service lifecycle │ │ └── pipeline.py # Collection → persist → dispatch @@ -117,9 +125,10 @@ logging: format: json api: - http: + grpc: host: 127.0.0.1 port: 8642 + max_workers: 10 persistence: backend: lmdb @@ -144,20 +153,43 @@ sinks: url: https://logs.example.com/ingest ``` -## REST API +## gRPC API -The API runs on `http://127.0.0.1:8642/api/v1`: +The gRPC API runs on `127.0.0.1:8642` with the following RPCs: -| 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 | +| RPC | Description | +|-----|-------------| +| `GetStatus` | Service status and version | +| `GetHealth` | Health check (backend, plugins, runtime) | +| `GetConfig` | Loaded configuration as JSON | +| `ListPlugins` | Registered sources, processors, and sinks | +| `GetQueueState` | Queue state counts (pending, delivered, failed, dead) | +| `ListPendingRecords` | Pending records with pagination | +| `ListDeadRecords` | Dead-letter records with pagination | +| `ListSinks` | Configured sinks | +| `ReloadConfig` | Reload service configuration at runtime | + +### Reflect the gRPC service + +```bash +# Install grpcurl to inspect the service +grpcurl -plaintext 127.0.0.1:8642 list +grpcurl -plaintext 127.0.0.1:8642 describe daena.v1.Daena +``` + +## CLI + +Daena provides a CLI that connects to the gRPC server: + +```bash +daena status # Show service status +daena health # Check health +daena queue # Inspect queue state +daena sinks # List configured sinks +daena plugins # List registered plugins +daena config # Show loaded configuration +daena version # Show version +``` ## Deployment @@ -178,6 +210,21 @@ sudo systemctl daemon-reload sudo systemctl enable --now daena ``` +### Docker + +```bash +# Build image +docker build -t daena:latest . + +# Run with config +docker run -d \ + --name daena \ + -v /path/to/daena.yaml:/etc/daena/daena.yaml:ro \ + -v daena-data:/var/lib/daena \ + -p 8642:8642 \ + daena:latest +``` + ## Extending Add a new source by creating a class that inherits `daena.pipeline.base.Source` diff --git a/tests/smoke_test.py b/tests/smoke_test.py new file mode 100644 index 0000000..d231940 --- /dev/null +++ b/tests/smoke_test.py @@ -0,0 +1,175 @@ +"""End-to-end smoke test: gRPC server + pipeline + CLI commands.""" + +from __future__ import annotations + +import asyncio +import json +import sys +import tempfile +from pathlib import Path + +import grpc.aio + +from daena.api.grpc_server import DaenaGRPCServer +from daena.api.pb2 import daena_pb2 as pb2 +from daena.api.pb2 import daena_pb2_grpc as pb2_grpc +from daena.config import DaenaConfig, ProcessorConfig, SinkConfig, SourceConfig +from daena.core.runtime import DaenaRuntime +from daena.logutil import setup_logging + +setup_logging(level="error", fmt="json") + + +async def smoke_test() -> int: + with tempfile.TemporaryDirectory() as tmpdir: + data_dir = Path(tmpdir) / "data" + data_dir.mkdir(parents=True, exist_ok=True) + + config = DaenaConfig() + config.data_dir = str(data_dir) + config.log_dir = str(tmpdir) + config.persistence.lmdb.path = str(data_dir / "lmdb") + config.sources = [ + SourceConfig( + name="smoke_source", + type="dummy", + options={"interval": 0.5, "message": "smoke_test"}, + ) + ] + config.processors = [ + ProcessorConfig(name="passthrough", type="passthrough"), + ] + config.sinks = [ + SinkConfig(name="log", type="dummy"), + ] + + runtime = DaenaRuntime(config) + server = DaenaGRPCServer(runtime, host="127.0.0.1", port=18642) + + await runtime.start() + await server.start() + + channel = grpc.aio.insecure_channel("127.0.0.1:18642") + stub = pb2_grpc.DaenaStub(channel) + + await asyncio.sleep(2.5) + + failed = 0 + + # 1. GetStatus + try: + resp = await stub.GetStatus(pb2.Empty(), timeout=5) + assert resp.service == "daena" + assert resp.version == "0.1.0" + assert resp.running is True + print(f"[PASS] GetStatus: service={resp.service}, running={resp.running}") + except Exception as e: + print(f"[FAIL] GetStatus: {e}") + failed += 1 + + # 2. GetHealth + try: + resp = await stub.GetHealth(pb2.Empty(), timeout=5) + assert resp.healthy is True + assert resp.running is True + assert resp.backend is True + assert resp.plugins_loaded is True + print(f"[PASS] GetHealth: healthy={resp.healthy}, backend={resp.backend}") + except Exception as e: + print(f"[FAIL] GetHealth: {e}") + failed += 1 + + # 3. GetConfig + try: + resp = await stub.GetConfig(pb2.Empty(), timeout=5) + cfg = json.loads(resp.config_json) + assert "sources" in cfg + assert cfg["sources"][0]["name"] == "smoke_source" + print(f"[PASS] GetConfig: sources={len(cfg['sources'])}") + except Exception as e: + print(f"[FAIL] GetConfig: {e}") + failed += 1 + + # 4. ListPlugins + try: + resp = await stub.ListPlugins(pb2.Empty(), timeout=5) + assert len(resp.sources.items) > 0 + assert len(resp.processors.items) > 0 + assert len(resp.sinks.items) > 0 + print( + f"[PASS] ListPlugins: sources={len(resp.sources.items)}, " + f"processors={len(resp.processors.items)}, sinks={len(resp.sinks.items)}" + ) + except Exception as e: + print(f"[FAIL] ListPlugins: {e}") + failed += 1 + + # 5. GetQueueState + try: + resp = await stub.GetQueueState(pb2.Empty(), timeout=5) + print( + f"[PASS] GetQueueState: pending={resp.pending}, " + f"delivered={resp.delivered}, total={resp.total}" + ) + except Exception as e: + print(f"[FAIL] GetQueueState: {e}") + failed += 1 + + # 6. ListPendingRecords + try: + resp = await stub.ListPendingRecords(pb2.Empty(), timeout=5) + print(f"[PASS] ListPendingRecords: count={resp.count}") + except Exception as e: + print(f"[FAIL] ListPendingRecords: {e}") + failed += 1 + + # 7. ListDeadRecords + try: + resp = await stub.ListDeadRecords(pb2.Empty(), timeout=5) + print(f"[PASS] ListDeadRecords: count={resp.count}") + except Exception as e: + print(f"[FAIL] ListDeadRecords: {e}") + failed += 1 + + # 8. ListSinks + try: + resp = await stub.ListSinks(pb2.Empty(), timeout=5) + assert len(resp.sinks) > 0 + assert resp.sinks[0].name == "log" + print(f"[PASS] ListSinks: sinks={len(resp.sinks)}") + except Exception as e: + print(f"[FAIL] ListSinks: {e}") + failed += 1 + + # 9. Pipeline produced records + total = runtime.backend.count() + if total > 0: + print(f"[PASS] Pipeline production: {total} records created in 2.5s") + else: + print("[FAIL] Pipeline production: no records produced") + failed += 1 + + # 10. ReloadConfig + try: + resp = await stub.ReloadConfig(pb2.Empty(), timeout=10) + assert resp.success is True, f"Reload failed: {resp.message}" + print("[PASS] ReloadConfig: success=True") + except Exception as e: + print(f"[FAIL] ReloadConfig: {e}") + failed += 1 + + await channel.close() + await server.stop() + await runtime.stop() + + print(f"\n{'=' * 40}") + if failed: + print(f"[RESULT] {failed} test(s) FAILED") + else: + print("[RESULT] All smoke tests PASSED") + return failed + + +if __name__ == "__main__": + ret = asyncio.run(smoke_test()) + sys.exit(ret) diff --git a/tests/test_http_sink.py b/tests/test_http_sink.py new file mode 100644 index 0000000..3746ecc --- /dev/null +++ b/tests/test_http_sink.py @@ -0,0 +1,195 @@ +"""End-to-end test for HTTP sink data forwarding.""" + +from __future__ import annotations + +import asyncio +import json +import tempfile +from pathlib import Path +from typing import Any + +import pytest + +from daena.config import DaenaConfig, SinkConfig, SourceConfig +from daena.core.runtime import DaenaRuntime +from daena.logutil import setup_logging +from daena.persistence.models import RecordState + +setup_logging(level="error", fmt="json") + +pytestmark = pytest.mark.asyncio + + +class _TestHTTPServer: + """Minimal async HTTP server to receive forwarded records.""" + + def __init__(self, status: int = 200, body: str = "OK") -> None: + self.received: list[dict[str, Any]] = [] + self._status = status + self._body = body + self._server: asyncio.AbstractServer | None = None + + async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + data = await reader.readuntil(b"\r\n\r\n") + raw = data.decode() + content_length = 0 + for line in raw.split("\r\n"): + if line.lower().startswith("content-length:"): + content_length = int(line.split(":")[1].strip()) + if content_length: + body = await reader.readexactly(content_length) + try: + payload = json.loads(body) + self.received.append(payload) + except json.JSONDecodeError: + pass + resp_body = self._body.encode() + writer.write( + f"HTTP/1.1 {self._status} OK\r\nContent-Length: {len(resp_body)}\r\n\r\n".encode() + + resp_body + ) + await writer.drain() + writer.close() + + async def start(self, host: str = "127.0.0.1", port: int = 0) -> int: + self._server = await asyncio.start_server(self._handle, host, port) + addr = self._server.sockets[0].getsockname() + return addr[1] + + async def stop(self) -> None: + if self._server: + self._server.close() + await self._server.wait_closed() + + +async def test_http_sink_forwards_records() -> None: + server = _TestHTTPServer(status=200) + port = await server.start() + + with tempfile.TemporaryDirectory() as tmpdir: + data_dir = Path(tmpdir) / "data" + data_dir.mkdir(parents=True) + + config = DaenaConfig() + config.data_dir = str(tmpdir) + config.log_dir = str(tmpdir) + config.persistence.lmdb.path = str(data_dir / "lmdb") + config.sources = [ + SourceConfig( + name="test_source", + type="dummy", + options={"interval": 0.2, "message": "http_sink_test"}, + ) + ] + config.sinks = [ + SinkConfig( + name="test_sink", + type="http", + options={"url": f"http://127.0.0.1:{port}/ingest", "timeout": 5.0}, + ) + ] + + runtime = DaenaRuntime(config) + await runtime.start() + await asyncio.sleep(3.0) + await runtime.stop() + + await server.stop() + + assert len(server.received) > 0, f"Expected at least 1 record, got {len(server.received)}" + + first = server.received[0] + assert "id" in first + assert "source" in first + assert first["source"] == "test_source" + assert "body" in first + assert "timestamp" in first + + +async def test_http_sink_permanent_failure_on_404() -> None: + """4xx errors (non-429) should permanently fail the record.""" + server = _TestHTTPServer(status=404, body="Not Found") + port = await server.start() + + with tempfile.TemporaryDirectory() as tmpdir: + data_dir = Path(tmpdir) / "data" + data_dir.mkdir(parents=True) + + config = DaenaConfig() + config.data_dir = str(tmpdir) + config.log_dir = str(tmpdir) + config.persistence.lmdb.path = str(data_dir / "lmdb") + config.sources = [ + SourceConfig( + name="test_source", + type="dummy", + options={"interval": 0.2, "message": "http_404_test"}, + ) + ] + config.sinks = [ + SinkConfig( + name="test_sink", + type="http", + options={ + "url": f"http://127.0.0.1:{port}/ingest", + "timeout": 5.0, + "retry": {"max_attempts": 1, "min_delay": 0.1}, + }, + ) + ] + + runtime = DaenaRuntime(config) + await runtime.start() + await asyncio.sleep(3.0) + + dead = runtime.backend.count(RecordState.dead) + failed = runtime.backend.count(RecordState.failed) + total = runtime.backend.count() + + await runtime.stop() + + await server.stop() + + assert dead > 0 or failed > 0, ( + f"Expected records to fail, dead={dead} failed={failed} total={total}" + ) + + +async def test_http_sink_delivery_result() -> None: + """Successful HTTP delivery should create delivered records.""" + server = _TestHTTPServer(status=200) + port = await server.start() + + with tempfile.TemporaryDirectory() as tmpdir: + data_dir = Path(tmpdir) / "data" + data_dir.mkdir(parents=True) + + config = DaenaConfig() + config.data_dir = str(tmpdir) + config.log_dir = str(tmpdir) + config.persistence.lmdb.path = str(data_dir / "lmdb") + config.sources = [ + SourceConfig( + name="test_source", + type="dummy", + options={"interval": 0.2, "message": "http_delivery_test"}, + ) + ] + config.sinks = [ + SinkConfig( + name="test_sink", + type="http", + options={"url": f"http://127.0.0.1:{port}/ingest", "timeout": 5.0}, + ) + ] + + runtime = DaenaRuntime(config) + await runtime.start() + await asyncio.sleep(3.0) + + delivered = runtime.backend.count(RecordState.delivered) + await runtime.stop() + + await server.stop() + + assert delivered > 0, f"Expected delivered records, got {delivered}"