feat: log aggregation — Loki/Elasticsearch shipping with documented label schema

backend/app/config.py

@@ -43,6 +43,15 @@ class Settings(BaseSettings):
     # If absent, derived from app_secret_key (dev/test only — not production-safe).
     notification_encryption_key: str = ""
 
+    # Log aggregation (leave blank to disable each shipper).
+    # Loki: set to the base URL, e.g. http://loki:3100
+    # Elasticsearch: set to the base URL, e.g. http://es:9200
+    loki_url: str = ""
+    elasticsearch_url: str = ""
+    # Identifies the service in Loki stream labels; override to "worker"
+    # in the Celery worker container.
+    log_service_name: str = "backend"
+
     model_config = SettingsConfigDict(
         env_file=".env", env_file_encoding="utf-8", extra="ignore"
     )

backend/app/main.py

@@ -17,7 +17,7 @@
 from app.config import settings
 from app.database import async_session_factory, engine
 from app.models.base import Base
-from app.observability import configure_logging
+from app.observability import configure_log_shippers, configure_logging
 from app.services.auth import bootstrap_admin
 from app.services.scheduler import (
     get_scheduler,
@@ -27,6 +27,15 @@
 from app.services.workspace import get_or_create_default_workspace
 
 configure_logging(log_level=settings.log_level, env=settings.app_env)
+configure_log_shippers(
+    loki_url=settings.loki_url,
+    elasticsearch_url=settings.elasticsearch_url,
+    labels={
+        "app": "autoflowops",
+        "service": settings.log_service_name,
+        "env": settings.app_env,
+    },
+)
 logger = structlog.get_logger(__name__)
 
 

backend/app/observability.py

@@ -2,9 +2,15 @@
 
 from __future__ import annotations
 
+import json
 import logging
+import queue
 import sys
+import threading
+from datetime import UTC, datetime
+from typing import Any
 
+import httpx
 import structlog
 from prometheus_client import Counter
 
@@ -77,3 +83,186 @@ def configure_logging(log_level: str = "INFO", env: str = "development") -> None
         wrapper_class=structlog.stdlib.BoundLogger,
         cache_logger_on_first_use=True,
     )
+
+
+# ---------------------------------------------------------------------------
+# Remote log shippers (optional, activated via LOKI_URL / ELASTICSEARCH_URL)
+# ---------------------------------------------------------------------------
+
+_log = structlog.get_logger(__name__)
+
+_JSON_FORMATTER = structlog.stdlib.ProcessorFormatter(
+    processors=[
+        structlog.contextvars.merge_contextvars,
+        structlog.stdlib.add_log_level,
+        structlog.stdlib.add_logger_name,
+        structlog.processors.TimeStamper(fmt="iso"),
+        structlog.processors.StackInfoRenderer(),
+        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+        structlog.processors.JSONRenderer(),
+    ]
+)
+
+
+class _LokiHandler(logging.Handler):
+    """Batched push handler for Grafana Loki's HTTP push API.
+
+    Runs a background daemon thread that drains a queue and flushes
+    entries every FLUSH_INTERVAL seconds or when BATCH_SIZE is reached.
+    Uses synchronous httpx so the handler never touches the asyncio loop.
+    Network errors are silently swallowed to avoid log recursion.
+    """
+
+    FLUSH_INTERVAL: float = 2.0
+    BATCH_SIZE: int = 20
+
+    def __init__(self, url: str, labels: dict[str, str]) -> None:
+        super().__init__()
+        self._push_url = url.rstrip("/") + "/loki/api/v1/push"
+        self._labels = labels
+        self._queue: queue.SimpleQueue[tuple[str, str]] = queue.SimpleQueue()
+        self._client = httpx.Client(timeout=5.0)
+        self._thread = threading.Thread(
+            target=self._run, daemon=True, name="loki-shipper"
+        )
+        self._thread.start()
+
+    def emit(self, record: logging.LogRecord) -> None:
+        try:
+            ns = str(int(record.created * 1_000_000_000))
+            self._queue.put_nowait((ns, self.format(record)))
+        except Exception:  # noqa: BLE001
+            self.handleError(record)
+
+    def _run(self) -> None:
+        while True:
+            batch: list[tuple[str, str]] = []
+            try:
+                batch.append(self._queue.get(timeout=self.FLUSH_INTERVAL))
+            except queue.Empty:
+                continue
+            while len(batch) < self.BATCH_SIZE:
+                try:
+                    batch.append(self._queue.get_nowait())
+                except queue.Empty:
+                    break
+            self._push(batch)
+
+    def _push(self, entries: list[tuple[str, str]]) -> None:
+        payload: dict[str, Any] = {
+            "streams": [
+                {
+                    "stream": self._labels,
+                    "values": [[ts, line] for ts, line in entries],
+                }
+            ]
+        }
+        try:
+            self._client.post(self._push_url, json=payload)
+        except Exception:  # noqa: BLE001
+            pass
+
+
+class _ElasticsearchHandler(logging.Handler):
+    """Batched bulk-index handler for Elasticsearch.
+
+    Indexes documents into ``{index_prefix}-YYYY.MM.DD`` using the
+    ``_bulk`` API.  Network errors are silently swallowed.
+    """
+
+    FLUSH_INTERVAL: float = 2.0
+    BATCH_SIZE: int = 20
+
+    def __init__(self, url: str, index_prefix: str = "autoflowops") -> None:
+        super().__init__()
+        self._bulk_url = url.rstrip("/") + "/_bulk"
+        self._index_prefix = index_prefix
+        self._queue: queue.SimpleQueue[tuple[str, dict[str, Any]]] = queue.SimpleQueue()
+        self._client = httpx.Client(timeout=5.0)
+        self._thread = threading.Thread(
+            target=self._run, daemon=True, name="es-shipper"
+        )
+        self._thread.start()
+
+    def emit(self, record: logging.LogRecord) -> None:
+        try:
+            doc: dict[str, Any] = json.loads(self.format(record))
+            now = datetime.now(UTC)
+            doc["@timestamp"] = doc.pop("timestamp", now.isoformat())
+            date_str = now.strftime("%Y.%m.%d")
+            self._queue.put_nowait((date_str, doc))
+        except Exception:  # noqa: BLE001
+            self.handleError(record)
+
+    def _run(self) -> None:
+        while True:
+            batch: list[tuple[str, dict[str, Any]]] = []
+            try:
+                batch.append(self._queue.get(timeout=self.FLUSH_INTERVAL))
+            except queue.Empty:
+                continue
+            while len(batch) < self.BATCH_SIZE:
+                try:
+                    batch.append(self._queue.get_nowait())
+                except queue.Empty:
+                    break
+            self._push(batch)
+
+    def _push(self, entries: list[tuple[str, dict[str, Any]]]) -> None:
+        lines: list[str] = []
+        for date_str, doc in entries:
+            idx = f"{self._index_prefix}-{date_str}"
+            lines.append(json.dumps({"index": {"_index": idx}}))
+            lines.append(json.dumps(doc))
+        body = "\n".join(lines) + "\n"
+        try:
+            self._client.post(
+                self._bulk_url,
+                content=body.encode(),
+                headers={"Content-Type": "application/x-ndjson"},
+            )
+        except Exception:  # noqa: BLE001
+            pass
+
+
+def configure_log_shippers(
+    *,
+    loki_url: str = "",
+    elasticsearch_url: str = "",
+    labels: dict[str, str] | None = None,
+    index_prefix: str = "autoflowops",
+) -> None:
+    """Attach optional remote log shippers to the root logger.
+
+    Both shippers use ``_JSON_FORMATTER`` so log lines are always shipped
+    as JSON regardless of the console renderer configured by
+    ``configure_logging()``.
+
+    Args:
+        loki_url: Full base URL of the Loki instance
+            (e.g. ``http://loki:3100``).  Leave blank to disable.
+        elasticsearch_url: Full base URL of the Elasticsearch cluster
+            (e.g. ``http://es:9200``).  Leave blank to disable.
+        labels: Loki stream labels attached to every pushed entry.
+            Defaults to ``{"app": "autoflowops"}``.
+        index_prefix: Prefix for Elasticsearch daily indices.
+    """
+    if not loki_url and not elasticsearch_url:
+        return
+
+    effective_labels = labels if labels is not None else {"app": "autoflowops"}
+    root = logging.getLogger()
+
+    if loki_url:
+        loki_handler = _LokiHandler(url=loki_url, labels=effective_labels)
+        loki_handler.setFormatter(_JSON_FORMATTER)
+        root.addHandler(loki_handler)
+        _log.info("loki log shipping enabled", loki_url=loki_url)
+
+    if elasticsearch_url:
+        es_handler = _ElasticsearchHandler(
+            url=elasticsearch_url, index_prefix=index_prefix
+        )
+        es_handler.setFormatter(_JSON_FORMATTER)
+        root.addHandler(es_handler)
+        _log.info("elasticsearch log shipping enabled", es_url=elasticsearch_url)

backend/app/worker/celery_app.py

@@ -1,4 +1,5 @@
 from celery import Celery
+from celery.signals import worker_process_init
 
 from app.config import settings
 
@@ -16,3 +17,19 @@
     task_track_started=True,
     timezone=settings.default_timezone,
 )
+
+
+@worker_process_init.connect
+def _configure_worker_logging(**_kwargs: object) -> None:
+    from app.observability import configure_log_shippers, configure_logging
+
+    configure_logging(log_level=settings.log_level, env=settings.app_env)
+    configure_log_shippers(
+        loki_url=settings.loki_url,
+        elasticsearch_url=settings.elasticsearch_url,
+        labels={
+            "app": "autoflowops",
+            "service": "worker",
+            "env": settings.app_env,
+        },
+    )

docker-compose.observability.yml

@@ -27,13 +27,40 @@ services:
       - "3001:3000"
     depends_on:
       - prometheus
+      - loki
+    networks:
+      - autoflowops_default
+    restart: unless-stopped
+
+  loki:
+    image: grafana/loki:3.1.0
+    volumes:
+      - ./observability/loki-config.yml:/etc/loki/local-config.yaml:ro
+      - loki_data:/loki
+    command: -config.file=/etc/loki/local-config.yaml
+    ports:
+      - "3100:3100"
+    networks:
+      - autoflowops_default
+    restart: unless-stopped
+
+  promtail:
+    image: grafana/promtail:3.1.0
+    volumes:
+      - ./observability/promtail-config.yml:/etc/promtail/config.yml:ro
+      - /var/lib/docker/containers:/var/lib/docker/containers:ro
+      - /var/run/docker.sock:/var/run/docker.sock:ro
+    command: -config.file=/etc/promtail/config.yml
+    depends_on:
+      - loki
     networks:
       - autoflowops_default
     restart: unless-stopped
 
 volumes:
   prometheus_data:
   grafana_data:
+  loki_data:
 
 networks:
   autoflowops_default: