feat: Kubernetes log export ingestion

.cursor/commands/code-review.md

@@ -15,7 +15,7 @@ If it's clean, say so — don't invent problems.
   `src/core/llm/provider.py` and the `noop` provider still works; schema changes
   add a new Alembic migration and leave applied ones untouched.
 - **Config & security**: no hard-coded secrets/URLs/paths (config from
-  `src/config/settings.py`); no secrets in the diff; untrusted
+  env vars via `src/config/settings.py`); no secrets in the diff; untrusted
   log input never builds SQL/shell strings unsafely.
 - **Tests & style**: functional changes include tests; normalization changes
   include a before/after case; type annotations present. Run `make lint` and

.cursor/rules/100-python.mdc

@@ -10,8 +10,8 @@ alwaysApply: false
   attributes.
 - Async everywhere DB or HTTP is involved: SQLAlchemy async sessions
   (`src/db/session.py`), `httpx.AsyncClient`. Don't block the event loop.
-- Read configuration from `src/config/settings.py` (pydantic-settings)
-  — no hard-coded URLs, keys, or paths.
+- Read configuration from `src/config/settings.py` (pydantic-settings,
+  `get_settings()`) — no hard-coded URLs, keys, or paths.
 - Log with `structlog`; render user-facing CLI output with `rich`.
 - Wrap external calls (LLM, network) with `tenacity` retries, as existing code
   does. All model access goes through `src/core/llm/provider.py`.

.env.example

@@ -72,6 +72,28 @@ WORKER_POLL_INTERVAL=2
 # IRSA in-cluster) — never set an access key here.
 ADAPTER_CLOUDWATCH_REGION=us-east-1
 
+# Grafana Loki source adapter (raglogs ingest --adapter loki).
+# URL is required for /health to report the adapter as available. Auth is optional
+# (open Loki) — use Bearer *or* basic, not both. Tenant sets X-Scope-OrgID
+# (Grafana Cloud / multi-tenant Loki). Default query is used when ingest params
+# omit `query`.
+# LOKI_URL=http://localhost:3100
+# LOKI_TENANT=
+# LOKI_BEARER_TOKEN=
+# LOKI_USERNAME=
+# LOKI_PASSWORD=
+# LOKI_QUERY={job=~".+"}
+
+# Datadog Logs Search API (raglogs ingest --adapter datadog). Both keys are required;
+# the application key needs logs_read_data. Site is the Datadog site hostname
+# (datadoghq.com, us3.datadoghq.com, datadoghq.eu, ...). page_size max is 1000
+# (Datadog API limit). max_rows caps how many events one ingest run will pull.
+DATADOG_API_KEY=
+DATADOG_APP_KEY=
+DATADOG_SITE=datadoghq.com
+DATADOG_PAGE_SIZE=1000
+DATADOG_MAX_ROWS=10000
+
 
 # ── Cluster severity weights ──────────────────────────────────────────────────
 # These control how clusters are ranked by importance.
@@ -81,3 +103,4 @@ SEVERITY_WEIGHT_ERROR=4.0
 SEVERITY_WEIGHT_WARN=3.0
 SEVERITY_WEIGHT_INFO=1.0
 SEVERITY_WEIGHT_DEBUG=0.5
+

CONTRIBUTING.md

@@ -100,7 +100,7 @@ The normalization step (`src/core/normalization/patterns.py`) determines cluster
 
 **Log source adapters**
 
-New adapters go in `src/adapters/`. Each adapter yields `ParsedLogLine` objects. The rest of the pipeline is fully source-agnostic. Useful adapters: Datadog, Loki, Kubernetes pod logs, CloudWatch.
+New adapters go in `src/adapters/`. Each adapter yields `ParsedLogLine` objects. The rest of the pipeline is fully source-agnostic. CloudWatch, Datadog, Loki, and Kubernetes log exports (`k8s`) are already supported.
 
 **Trigger patterns**
 

README.md

@@ -237,7 +237,7 @@ raglogs init --no-migrate   # skip migrations
 
 ### `raglogs ingest`
 
-Ingests one or more log files into the database. Supports JSON and plain-text formats, single files, directories, and glob patterns.
+Ingests logs into the database from local files, a pull adapter (CloudWatch, Datadog, Loki), or Kubernetes log exports. File mode supports JSON and plain-text formats, single files, directories, and glob patterns.
 
 ```bash
 raglogs ingest ./logs/app.log
@@ -246,6 +246,25 @@ raglogs ingest ./logs/*.log
 raglogs ingest ./logs/ --recursive
 raglogs ingest ./logs/ --service api --env production
 raglogs ingest ./logs/ --format json
+
+# CloudWatch Logs (AWS credential chain; no keys on the CLI)
+raglogs ingest --adapter cloudwatch --param log_group=/aws/lambda/my-service --since 1h
+
+# Datadog Logs Search API (keys via env; see Configuration)
+raglogs ingest --adapter datadog --param query='service:billing-worker status:error' --since 1h
+```
+
+**Kubernetes log exports** (`--adapter k8s`) — concatenated `kubectl logs`, Fluent Bit / Vector JSON lines, CRI (kubelet) node logs, `.gz` files, and tarballs. Namespace / pod / container map to environment / host / service.
+
+```bash
+# kubectl capture (prefix + timestamps give pod/container and event time)
+kubectl logs -n production -l app=billing-worker --all-containers \
+  --prefix --timestamps --since=1h > /tmp/billing-export.log
+raglogs ingest --adapter k8s /tmp/billing-export.log
+
+# kubelet /var/log/pods dump (path supplies namespace, pod, container)
+raglogs ingest --adapter k8s --recursive ./var/log/pods
+raglogs ingest --adapter k8s ./node-logs.tar.gz
 ```
 
 | Flag | Description |
@@ -256,6 +275,42 @@ raglogs ingest ./logs/ --format json
 | `--env` | Default environment |
 | `--format` | `json`, `text`, or `auto` (default) |
 | `--with-embeddings` | Generate vector embeddings (requires embeddings provider) |
+| `--adapter` | `file` (default), `cloudwatch`, `datadog`, `loki`, or `k8s` |
+| `--param` | Adapter param as `key=value` (repeatable) |
+| `--since` | Window for pull adapters, e.g. `30m`, `1h`, `24h` (default last 1h) |
+| `--from` / `--to` | Explicit ISO 8601 window bounds (pull adapters) |
+| `--resume-job` | Prior ingestion job UUID to resume pagination cursors from |
+
+**Loki**
+
+Pull a bounded window from Grafana Loki via LogQL — no intermediate files.
+Auth and the Loki origin come from env (`LOKI_*`); tenant and query can be
+passed per ingest. Stream labels are mapped onto `service` / `environment` /
+`host` (`app`/`service`/`job`, `namespace`/`env`, `pod`/`instance`).
+
+`query_range`'s `limit` is global across matching streams. Paginating by
+advancing `start` to the latest timestamp in a full page can skip earlier
+lines from other streams in a wide selector — prefer a narrow LogQL query
+when completeness matters.
+
+```bash
+export LOKI_URL=http://localhost:3100
+raglogs ingest --adapter loki --param query='{app="api"}' --since 1h
+raglogs ingest --adapter loki --param query='{namespace="prod"}' \
+  --from 2026-03-12T22:00:00+00:00 --to 2026-03-12T22:30:00+00:00
+```
+
+```bash
+curl -X POST http://localhost:8000/ingestions \
+  -H "Content-Type: application/json" \
+  -d '{"adapter":"loki","params":{"query":"{app=\"api\"}"},"since":"1h"}'
+```
+
+| Param | Description |
+|---|---|
+| `query` / `queries` | LogQL selector (required unless `LOKI_QUERY` is set) |
+| `tenant` | Override `LOKI_TENANT` (`X-Scope-OrgID`) |
+| `limit` | Page size for `query_range` (default 5000, Loki max) |
 
 **Output**
 
@@ -557,6 +612,12 @@ All settings are read from `.env`, environment variables, or CLI flags. Priority
 | `DEFAULT_BASELINE_WINDOW` | `24h` | How far back to compare for baseline |
 | `MAX_CLUSTERS_FOR_EXPLAIN` | `10` | Max clusters sent to the explain pipeline |
 | `MAX_EVIDENCE_ITEMS` | `8` | Max evidence lines in output |
+| `ADAPTER_CLOUDWATCH_REGION` | `us-east-1` | AWS region for `--adapter cloudwatch` |
+| `DATADOG_API_KEY` | _(empty)_ | Datadog API key (`logs_read_data`) |
+| `DATADOG_APP_KEY` | _(empty)_ | Datadog application key |
+| `DATADOG_SITE` | `datadoghq.com` | Datadog site (`us3.datadoghq.com`, `datadoghq.eu`, …) |
+| `DATADOG_PAGE_SIZE` | `1000` | Logs per Datadog page (API max 1000) |
+| `DATADOG_MAX_ROWS` | `10000` | Max events pulled in one Datadog ingest run |
 
 ---
 
@@ -764,7 +825,7 @@ make api
 | Method | Endpoint | Description |
 |---|---|---|
 | `GET` | `/health` | Service and DB health check |
-| `POST` | `/ingestions` | Ingest log files |
+| `POST` | `/ingestions` | Enqueue an ingest job (`adapter`: `file`, `cloudwatch`, or `datadog`) |
 | `GET` | `/ingestions` | List recent completed ingestion jobs, newest first |
 | `GET` | `/ingestions/{job_id}` | Poll ingestion job status |
 | `GET` | `/ingestions/latest` | ID of the most recently completed ingestion job, if any |
@@ -880,7 +941,7 @@ make clean
 ```
 raglogs/
 ├── src/
-│   ├── adapters/file/       File discovery and line reading
+│   ├── adapters/            Log source adapters (file, cloudwatch, datadog, loki, k8s)
 │   ├── api/routes/          FastAPI route handlers
 │   ├── cli/commands/        Typer CLI commands
 │   ├── config/              Pydantic settings
@@ -905,15 +966,24 @@ raglogs/
 
 **Adding a log source adapter**
 
-New source adapters go in `raglogs/adapters/`. Each adapter yields `ParsedLogLine` objects. The normalization, fingerprinting, storage, clustering, and explain pipeline is fully source-agnostic.
+New source adapters go in `src/adapters/` and implement `SourceAdapter` (`discover` / `read`), yielding `RawLogLine` objects that the existing parser maps onto `ParsedLogLine`. The normalization, fingerprinting, storage, clustering, and explain pipeline is fully source-agnostic.
+
+**Datadog adapter limits**
+
+- Auth: `DD-API-KEY` + `DD-APPLICATION-KEY` (application key needs `logs_read_data`). Keys come from env only — never CLI `--param`.
+- Endpoint: `POST https://api.<site>/api/v2/logs/events/search` with an absolute `from`/`to` window (relative ranges drop events while paginating).
+- Pagination: cursor from `meta.page.after`; resume with `--resume-job`.
+- Page size: default 1000, Datadog hard max 1000 (`--param page_size=N` or `DATADOG_PAGE_SIZE`).
+- Max rows per run: default 10000 (`--param max_rows=N` or `DATADOG_MAX_ROWS`). Hitting the cap saves the next cursor for resume.
+- Rate limits: HTTP 429 and 5xx are retried up to 3 times with exponential backoff; a persistent failure marks the job `ADAPTER_UNAVAILABLE` (or `partial: true` if some events already landed).
+- Field mapping: Datadog `status` → `level`; `service` / `host` / `message` / `timestamp` pass through; `env` from the `env:` tag or attributes; `trace_id` / `request_id` from nested custom attributes when present. Other nested Datadog attributes are dropped so core parsing stays source-agnostic.
 
 ---
 
 ## Roadmap
 
-- Datadog adapter
-- Loki adapter
-- Kubernetes log export ingestion
+## Roadmap
+
 - Semantic cluster merging via pgvector
 - Markdown incident report export (`raglogs explain --format markdown > postmortem.md`)
 

docs/raglogs-hardening-and-integration-design.md

@@ -119,7 +119,7 @@ Adapters to build, in priority order:
 
 **Data model.** Add `source_adapter` and `source_ref` columns to the ingestion job + log rows for provenance ("which stream did this cluster come from").
 
-**Config.** Per-adapter blocks under `ADAPTER_CLOUDWATCH_*`, etc. Adapters are optional; absent credentials disable the adapter and surface a clear `/health` sub-status.
+**Config.** Per-adapter blocks under `ADAPTER_CLOUDWATCH_*`, `DATADOG_*`, `LOKI_*`, etc. Adapters are optional; absent credentials disable the adapter and surface a clear `/health` sub-status.
 
 **Failure modes.** Adapter unreachable → job fails with a typed error (`ADAPTER_UNAVAILABLE`) and does not partially ingest silently. Partial reads (rate-limited AWS API) → resumable via `LogStreamRef` cursor; job reports `partial: true`.
 

src/adapters/base.py

@@ -6,6 +6,7 @@
 @dataclass
 class SourceSpec:
     """What integrators send instead of file paths — identifies an adapter and its params."""
+
     adapter: str
     params: dict[str, Any] = field(default_factory=dict)
     service: Optional[str] = None
@@ -21,6 +22,7 @@ class TimeWindow:
 @dataclass
 class LogStreamRef:
     """A concrete stream resolved from a SourceSpec (a log group, label set, or file)."""
+
     adapter: str
     stream_id: str
     cursor: Optional[str] = None
@@ -29,10 +31,21 @@ class LogStreamRef:
 
 @dataclass
 class RawLogLine:
-    """One unparsed line plus its provenance, ready for the parser → normalization pipeline."""
+    """One unparsed line plus its provenance, ready for the parser → normalization pipeline.
+
+    Adapters may fill `default_service` / `default_environment` / `default_host` from
+    source metadata (Loki stream labels, k8s pod fields). Ingestion uses them only
+    when the parsed line and SourceSpec do not already set those fields — core stays
+    source-agnostic.
+    """
+
     text: str
     source_ref: str
     received_at: Optional[datetime] = None
+    default_service: Optional[str] = None
+    default_environment: Optional[str] = None
+    default_host: Optional[str] = None
+    extra: dict[str, Any] = field(default_factory=dict)
 
 
 @runtime_checkable