From 542d2467959e0109cb75d9f9c1eb09be2229c35e Mon Sep 17 00:00:00 2001 From: btcneves Date: Wed, 20 May 2026 22:54:14 -0300 Subject: [PATCH] =?UTF-8?q?feat:=20add=20log=20aggregation=20=E2=80=94=20L?= =?UTF-8?q?oki/Elasticsearch=20shipping=20with=20documented=20label=20sche?= =?UTF-8?q?ma?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add Loki + Promtail services to docker-compose.observability.yml so `make obs-up` starts the full observability stack (Prometheus, Grafana, Loki, Promtail) in one command - observability/loki-config.yml: single-node filesystem storage (v13 schema) - observability/promtail-config.yml: Docker SD, compose-project filter, JSON pipeline stage; promotes level + logger to stream labels; maps request_id / workspace_id / user_id to structured metadata - observability/grafana/provisioning/datasources/loki.yml: Loki datasource auto-provisioned alongside Prometheus - backend/app/observability.py: optional _LokiHandler and _ElasticsearchHandler — batched background-thread push for environments without a sidecar agent; activated by LOKI_URL / ELASTICSEARCH_URL env vars - backend/app/config.py: loki_url, elasticsearch_url, log_service_name settings - backend/app/main.py: wire configure_log_shippers() on startup - backend/app/worker/celery_app.py: wire configure_log_shippers() via worker_process_init signal so the Celery worker ships logs independently - docs/log-aggregation.md: label schema reference, LogQL query examples, env var table, Promtail project filter instructions --- backend/app/config.py | 9 + backend/app/main.py | 11 +- backend/app/observability.py | 189 ++++++++++++++++++ backend/app/worker/celery_app.py | 17 ++ docker-compose.observability.yml | 27 +++ docs/log-aggregation.md | 160 +++++++++++++++ .../grafana/provisioning/datasources/loki.yml | 11 + observability/loki-config.yml | 42 ++++ observability/promtail-config.yml | 67 +++++++ 9 files changed, 532 insertions(+), 1 deletion(-) create mode 100644 docs/log-aggregation.md create mode 100644 observability/grafana/provisioning/datasources/loki.yml create mode 100644 observability/loki-config.yml create mode 100644 observability/promtail-config.yml diff --git a/backend/app/config.py b/backend/app/config.py index 705ee76..4505f09 100644 --- a/backend/app/config.py +++ b/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" ) diff --git a/backend/app/main.py b/backend/app/main.py index 6d18088..c3ad2b5 100644 --- a/backend/app/main.py +++ b/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__) diff --git a/backend/app/observability.py b/backend/app/observability.py index f842c8d..7a11967 100644 --- a/backend/app/observability.py +++ b/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) diff --git a/backend/app/worker/celery_app.py b/backend/app/worker/celery_app.py index 8ea625c..21adcd1 100644 --- a/backend/app/worker/celery_app.py +++ b/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, + }, + ) diff --git a/docker-compose.observability.yml b/docker-compose.observability.yml index 38412ee..09b7f9f 100644 --- a/docker-compose.observability.yml +++ b/docker-compose.observability.yml @@ -27,6 +27,32 @@ 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 @@ -34,6 +60,7 @@ services: volumes: prometheus_data: grafana_data: + loki_data: networks: autoflowops_default: diff --git a/docs/log-aggregation.md b/docs/log-aggregation.md new file mode 100644 index 0000000..ede7ad5 --- /dev/null +++ b/docs/log-aggregation.md @@ -0,0 +1,160 @@ +# Log Aggregation + +AutoFlowOps ships structured JSON logs to external aggregation backends. +Two shipping modes are available: + +| Mode | How it works | When to use | +|---|---|---| +| **Agent-based (recommended)** | Promtail scrapes Docker container stdout/stderr and pushes to Loki | Standard self-hosted deployments with Docker | +| **Direct push** | The application pushes each log line over HTTP to Loki or Elasticsearch | Kubernetes / cloud environments where a sidecar agent is not available | + +Both modes can be active simultaneously. + +--- + +## Label schema + +Every log stream carries a consistent set of labels. + +### Stream labels (low cardinality — used for stream selection) + +| Label | Source | Example values | +|---|---|---| +| `app` | constant | `autoflowops` | +| `service` | compose service name / `LOG_SERVICE_NAME` env var | `backend`, `worker` | +| `env` | `APP_ENV` environment variable | `production`, `development` | +| `level` | structlog `level` field | `info`, `warning`, `error`, `critical` | +| `logger` | structlog `logger` field (module path) | `app.services.notifications` | + +### Structured metadata (high cardinality — available for filtering, not for stream selection) + +| Field | Description | Emitted when | +|---|---|---| +| `request_id` | Per-request UUID injected by the ASGI middleware | Inside any HTTP request | +| `workspace_id` | Active workspace UUID | After workspace resolution in the auth dependency | +| `user_id` | Authenticated user UUID | After JWT validation | +| `job_id` | Job UUID | Job execution tasks | +| `execution_id` | Execution UUID | Job execution tasks | + +High-cardinality fields (`request_id`, `workspace_id`, `user_id`) are **not** promoted to Loki stream labels. They are extracted by the Promtail pipeline as structured metadata and are queryable with LogQL label filters. + +--- + +## Quick start — Loki (agent-based, recommended) + +### Prerequisites + +- The main AutoFlowOps stack must be running (`make up`) +- Docker socket access for Promtail (`/var/run/docker.sock`) +- The Docker compose project name must be `autoflowops` (the default when the repo directory is `AutoFlowOps`) + +### Start + +```bash +make obs-up +``` + +This starts Prometheus, Grafana, Loki, and Promtail in a single overlay network. + +| Service | URL | Credentials | +|---|---|---| +| Grafana | http://localhost:3001 | admin / admin | +| Loki API | http://localhost:3100 | — | +| Prometheus | http://localhost:9090 | — | + +Grafana is pre-provisioned with both the Prometheus and Loki datasources. + +### Stop + +```bash +make obs-down +``` + +### How Promtail works + +Promtail connects to the Docker daemon via `/var/run/docker.sock`, discovers containers belonging to the `autoflowops` compose project, and reads their log files from `/var/lib/docker/containers`. It applies the label pipeline defined in `observability/promtail-config.yml` and pushes streams to Loki. + +The JSON fields emitted by structlog in `APP_ENV=production` are parsed by the pipeline stage and promoted to structured metadata. + +--- + +## Quick start — Loki (direct push, optional) + +For environments without a sidecar agent, the backend can push logs directly to Loki over HTTP. + +Set the following environment variables before starting the backend: + +```env +LOKI_URL=http://loki:3100 +LOG_SERVICE_NAME=backend # override to "worker" in the worker container +``` + +The backend attaches a `_LokiHandler` to the root logger. +Entries are batched (up to 20) and flushed every 2 seconds by a background daemon thread. + +> **Note**: Direct push and agent-based scraping can both be active. Loki deduplicates streams by labels + nanosecond timestamp, so no double-counting occurs. + +--- + +## Quick start — Elasticsearch (direct push) + +Set the following environment variable before starting the backend: + +```env +ELASTICSEARCH_URL=http://es:9200 +``` + +Logs are indexed into daily indices named `autoflowops-YYYY.MM.DD` using the `_bulk` API. The `timestamp` field from structlog is mapped to `@timestamp` (Kibana-compatible). + +No Docker Compose configuration is provided for Elasticsearch, as cluster topologies vary significantly. Configure your own cluster and point `ELASTICSEARCH_URL` at it. A typical Filebeat + Logstash pipeline reading from Docker `json-file` logs works equally well without any code changes (structlog already outputs JSON in production). + +--- + +## Example LogQL queries + +```logql +# All backend errors in the last hour +{app="autoflowops", service="backend", level="error"} + +# Errors from a specific module +{app="autoflowops", level="error", logger="app.services.notifications"} + +# Trace a specific request across backend + worker logs +{app="autoflowops"} | json | request_id="550e8400-e29b-41d4-a716-446655440000" + +# All logs for a specific workspace +{app="autoflowops"} | json | workspace_id="" + +# Job execution failures in the last 24h +{app="autoflowops", service="worker", level="error"} |= "job_id" + +# Rate of error logs per minute (for alerting) +sum(rate({app="autoflowops", level="error"}[1m])) +``` + +--- + +## Environment variable reference + +| Variable | Default | Description | +|---|---|---| +| `LOKI_URL` | `` (disabled) | Base URL of Loki instance for direct push | +| `ELASTICSEARCH_URL` | `` (disabled) | Base URL of Elasticsearch cluster for direct push | +| `LOG_SERVICE_NAME` | `backend` | Value of the `service` Loki stream label. Set to `worker` in the Celery worker container | +| `APP_ENV` | `development` | Controls log renderer (JSON in production, coloured console otherwise) and the `env` label | +| `LOG_LEVEL` | `INFO` | Root log level | + +--- + +## Configuring the Promtail compose project filter + +If your Docker compose project name differs from `autoflowops` (e.g. you renamed the repository directory), update the filter in `observability/promtail-config.yml`: + +```yaml +filters: + - name: label + values: ["com.docker.compose.project="] +``` + +The project name is the lowercase directory name where `docker-compose.yml` lives. +Run `docker compose ps` and look at the container names — the prefix before the first `-` is the project name. diff --git a/observability/grafana/provisioning/datasources/loki.yml b/observability/grafana/provisioning/datasources/loki.yml new file mode 100644 index 0000000..6317d67 --- /dev/null +++ b/observability/grafana/provisioning/datasources/loki.yml @@ -0,0 +1,11 @@ +apiVersion: 1 + +datasources: + - name: Loki + type: loki + access: proxy + url: http://loki:3100 + isDefault: false + editable: false + jsonData: + maxLines: 1000 diff --git a/observability/loki-config.yml b/observability/loki-config.yml new file mode 100644 index 0000000..31441d2 --- /dev/null +++ b/observability/loki-config.yml @@ -0,0 +1,42 @@ +auth_enabled: false + +server: + http_listen_port: 3100 + grpc_listen_port: 0 + +common: + instance_addr: 127.0.0.1 + path_prefix: /loki + storage: + filesystem: + chunks_directory: /loki/chunks + rules_directory: /loki/rules + replication_factor: 1 + ring: + kvstore: + store: inmemory + +query_range: + results_cache: + cache: + embedded_cache: + enabled: true + max_size_mb: 100 + +schema_config: + configs: + - from: 2024-01-01 + store: tsdb + object_store: filesystem + schema: v13 + index: + prefix: index_ + period: 24h + +limits_config: + reject_old_samples: true + reject_old_samples_max_age: 168h + allow_structured_metadata: true + +analytics: + reporting_enabled: false diff --git a/observability/promtail-config.yml b/observability/promtail-config.yml new file mode 100644 index 0000000..2eb8569 --- /dev/null +++ b/observability/promtail-config.yml @@ -0,0 +1,67 @@ +# Promtail configuration for AutoFlowOps. +# +# Label schema applied to every log stream: +# app = "autoflowops" (constant) +# service = compose service name (backend | worker | frontend | db | redis) +# env = APP_ENV value (development | production) +# level = structlog level field (debug | info | warning | error | critical) +# logger = structlog logger name (module path, e.g. app.services.notifications) +# +# Structured fields extracted from JSON log lines and available for LogQL +# label-filter / line-filter queries (but NOT promoted to stream labels to +# keep cardinality low): +# request_id, workspace_id, user_id +# +# These fields are emitted by structlog contextvars middleware and are +# available on every log line that was produced inside a request context. + +server: + http_listen_port: 9080 + grpc_listen_port: 0 + +positions: + filename: /tmp/positions.yaml + +clients: + - url: http://loki:3100/loki/api/v1/push + +scrape_configs: + - job_name: autoflowops-containers + docker_sd_configs: + - host: unix:///var/run/docker.sock + refresh_interval: 5s + filters: + - name: label + values: ["com.docker.compose.project=autoflowops"] + relabel_configs: + # Promote compose service name → service label + - source_labels: ["__meta_docker_container_label_com_docker_compose_service"] + target_label: service + # Constant app label + - replacement: autoflowops + target_label: app + # Carry the APP_ENV value if it was injected as a container env label; + # falls back to an empty string if absent — set explicitly in compose. + - source_labels: ["__meta_docker_container_env_APP_ENV"] + target_label: env + pipeline_stages: + # Extract structured fields from the JSON produced by structlog in + # production mode (APP_ENV=production). In development mode the log + # lines are human-readable text; the json stage is a no-op for those. + - json: + expressions: + level: level + logger: logger + request_id: request_id + workspace_id: workspace_id + user_id: user_id + # Promote level + logger to Loki stream labels (low cardinality). + # request_id / workspace_id / user_id are kept as structured metadata + # only — they are high-cardinality and must not become stream labels. + - labels: + level: + logger: + - structured_metadata: + request_id: + workspace_id: + user_id: