diff --git a/.env.example b/.env.example index 9c332b2..00377a2 100644 --- a/.env.example +++ b/.env.example @@ -72,6 +72,16 @@ WORKER_POLL_INTERVAL=2 # IRSA in-cluster) — never set an access key here. ADAPTER_CLOUDWATCH_REGION=us-east-1 +# 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. diff --git a/README.md b/README.md index b9ba0e5..c5f9a32 100644 --- a/README.md +++ b/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 or a pull adapter (CloudWatch, Datadog). File mode supports JSON and plain-text formats, single files, directories, and glob patterns. ```bash raglogs ingest ./logs/app.log @@ -246,6 +246,12 @@ 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 ``` | Flag | Description | @@ -256,6 +262,11 @@ raglogs ingest ./logs/ --format json | `--env` | Default environment | | `--format` | `json`, `text`, or `auto` (default) | | `--with-embeddings` | Generate vector embeddings (requires embeddings provider) | +| `--adapter` | Source adapter: `file` (default), `cloudwatch`, or `datadog` | +| `--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 | **Output** @@ -557,6 +568,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 +781,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 +897,7 @@ make clean ``` raglogs/ ├── src/ -│ ├── adapters/file/ File discovery and line reading +│ ├── adapters/ Log source adapters (file, cloudwatch, datadog) │ ├── api/routes/ FastAPI route handlers │ ├── cli/commands/ Typer CLI commands │ ├── config/ Pydantic settings @@ -905,13 +922,22 @@ 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./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 - Semantic cluster merging via pgvector diff --git a/docs/raglogs-hardening-and-integration-design.md b/docs/raglogs-hardening-and-integration-design.md index 7dff7a2..79eda9c 100644 --- a/docs/raglogs-hardening-and-integration-design.md +++ b/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_*`, 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`. diff --git a/src/adapters/datadog/adapter.py b/src/adapters/datadog/adapter.py new file mode 100644 index 0000000..1d8aa17 --- /dev/null +++ b/src/adapters/datadog/adapter.py @@ -0,0 +1,397 @@ +from datetime import datetime, timezone +from typing import Any, Iterable, Iterator, Optional +from urllib.parse import urlparse + +import httpx +import orjson +from dateutil import parser as dateutil_parser +from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential + +from src.adapters.base import LogStreamRef, RawLogLine, SourceSpec, TimeWindow +from src.core.errors import AdapterUnavailableError + +PAGE_SIZE_MAX = 1000 +PAGE_SIZE_DEFAULT = 1000 +MAX_ROWS_DEFAULT = 10000 +SEARCH_PATH = "/api/v2/logs/events/search" + +# Canonical Datadog site hostnames (DD_SITE). Always talk to https://api.. +KNOWN_DATADOG_SITES = frozenset( + { + "datadoghq.com", + "us3.datadoghq.com", + "us5.datadoghq.com", + "datadoghq.eu", + "ap1.datadoghq.com", + "ap2.datadoghq.com", + "uk1.datadoghq.com", + "ddog-gov.com", + "us1.ddog-gov.com", + "us2.ddog-gov.com", + } +) + +_TRACE_KEYS = ( + "trace_id", + "traceId", + "dd.trace_id", + "dd.traceId", +) +_REQUEST_KEYS = ( + "request_id", + "requestId", + "req_id", + "correlation_id", + "correlationId", + "http.request_id", + "http.requestId", +) + + +def normalize_site(site: str) -> str: + """Return a canonical Datadog site hostname, or raise AdapterUnavailableError. + + Accepts ``datadoghq.com``, ``us3.datadoghq.com``, ``app.datadoghq.eu``, + ``api.datadoghq.com``, or ``https://api.datadoghq.com``. Arbitrary origins + are rejected so API keys are never POSTed off-Datadog. + """ + raw = (site or "").strip() + if not raw: + return "datadoghq.com" + + if "://" in raw: + parsed = urlparse(raw) + if parsed.scheme != "https": + raise AdapterUnavailableError( + f"unsupported Datadog site {raw!r}: only https Datadog sites are allowed" + ) + host = (parsed.hostname or "").lower().rstrip(".") + else: + if "/" in raw or "@" in raw: + raise AdapterUnavailableError(f"unsupported Datadog site {raw!r}") + host = raw.lower().rstrip(".") + + if host.startswith("app.") or host.startswith("api."): + host = host[4:] + if host not in KNOWN_DATADOG_SITES: + raise AdapterUnavailableError( + f"unsupported Datadog site {raw!r}; expected a Datadog site hostname " + f"(e.g. datadoghq.com, us3.datadoghq.com, datadoghq.eu)" + ) + return host + + +def api_base_url(site: str) -> str: + """Build ``https://api.`` from a validated Datadog site hostname.""" + return f"https://api.{normalize_site(site)}" + + +def _lookup(data: dict[str, Any], *keys: str) -> Any: + """Return the first present key, including dotted paths (``dd.trace_id``).""" + for key in keys: + if key in data: + return data[key] + if "." not in key: + continue + current: Any = data + found = True + for part in key.split("."): + if isinstance(current, dict) and part in current: + current = current[part] + else: + found = False + break + if found: + return current + return None + + +def _tag_value(tags: list[Any], name: str) -> Optional[str]: + prefix = f"{name}:" + for tag in tags: + if isinstance(tag, str) and tag.startswith(prefix): + value = tag[len(prefix) :] + if value: + return value + return None + + +def _as_str(value: Any) -> Optional[str]: + if value is None: + return None + text = str(value).strip() + return text or None + + +def map_datadog_event(event: dict[str, Any]) -> dict[str, Any]: + """Map a Datadog v2 log event onto raglogs JSON field aliases. + + Datadog reserved attributes (``status``, ``service``, ``host``, ``message``, + ``timestamp``) plus ``env`` / trace / request IDs from tags or nested + custom attributes. Unknown envelope fields are dropped so core parsing + stays source-agnostic. + """ + raw_attrs = event.get("attributes") + attrs = raw_attrs if isinstance(raw_attrs, dict) else {} + raw_nested = attrs.get("attributes") + nested = raw_nested if isinstance(raw_nested, dict) else {} + raw_tags = attrs.get("tags") + tags = raw_tags if isinstance(raw_tags, list) else [] + + env = ( + _as_str(attrs.get("env")) + or _tag_value(tags, "env") + or _tag_value(tags, "environment") + or _as_str(_lookup(nested, "env", "environment")) + ) + payload: dict[str, Any] = {} + timestamp = attrs.get("timestamp") + if timestamp is not None: + payload["timestamp"] = timestamp + message = attrs.get("message") + if message is not None: + payload["message"] = message + status = attrs.get("status") + if status is not None: + payload["level"] = status + service = attrs.get("service") + if service is not None: + payload["service"] = service + host = attrs.get("host") + if host is not None: + payload["host"] = host + if env is not None: + payload["env"] = env + trace_id = _as_str( + _lookup(nested, *_TRACE_KEYS) or _lookup(attrs, *_TRACE_KEYS) + ) + if trace_id is not None: + payload["trace_id"] = trace_id + request_id = _as_str( + _lookup(nested, *_REQUEST_KEYS) or _lookup(attrs, *_REQUEST_KEYS) + ) + if request_id is not None: + payload["request_id"] = request_id + return payload + + +def parse_event_timestamp(event: dict[str, Any]) -> Optional[datetime]: + """Parse the Datadog event timestamp for RawLogLine.received_at.""" + raw_attrs = event.get("attributes") + attrs = raw_attrs if isinstance(raw_attrs, dict) else {} + raw = attrs.get("timestamp") + if raw is None: + return None + if isinstance(raw, (int, float)): + ts = float(raw) + if ts > 1e12: + ts /= 1000.0 + try: + return datetime.fromtimestamp(ts, tz=timezone.utc) + except (ValueError, OverflowError, OSError): + return None + if isinstance(raw, str): + try: + dt = dateutil_parser.parse(raw) + except (ValueError, OverflowError): + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + return None + + +def _iso8601(dt: datetime) -> str: + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" + + +def _int_param(params: dict[str, Any], key: str, default: int) -> int: + raw = params.get(key, default) + try: + return int(raw) + except (TypeError, ValueError): + return default + + +def _clamp_page_size(value: int) -> int: + return max(1, min(value, PAGE_SIZE_MAX)) + + +def _indexes_param(params: dict[str, Any]) -> list[str]: + raw = params.get("indexes") + if isinstance(raw, str): + return [part.strip() for part in raw.split(",") if part.strip()] + if isinstance(raw, list): + return [str(part).strip() for part in raw if str(part).strip()] + return [] + + +def _is_retryable(exc: BaseException) -> bool: + """Retry 429 / 5xx and transport failures — not 4xx auth or validation errors.""" + if isinstance(exc, httpx.HTTPStatusError): + return exc.response.status_code == 429 or exc.response.status_code >= 500 + return isinstance(exc, (httpx.TimeoutException, httpx.NetworkError)) + + +def _client() -> httpx.Client: + return httpx.Client(timeout=60.0) + + +class DatadogSourceAdapter: + """SourceAdapter over the Datadog Logs Search API (v2). + + Auth via ``DATADOG_API_KEY`` + ``DATADOG_APP_KEY`` + — keys are never accepted as SourceSpec params. + """ + + name = "datadog" + + def __init__( + self, + api_key: str = "", + app_key: str = "", + site: str = "datadoghq.com", + page_size: int = PAGE_SIZE_DEFAULT, + max_rows: int = MAX_ROWS_DEFAULT, + ): + self.api_key = api_key + self.app_key = app_key + self.site = normalize_site(site) + self.page_size = _clamp_page_size(page_size) + self.max_rows = max(1, max_rows) + + def check_available(self) -> None: + """Raise AdapterUnavailableError if API/app keys are missing. + + Local-only — used by /health. Does not call Datadog. + """ + if not self.api_key or not self.app_key: + raise AdapterUnavailableError( + "datadog adapter requires DATADOG_API_KEY " + "and DATADOG_APP_KEY" + ) + + def discover(self, spec: SourceSpec) -> Iterable[LogStreamRef]: + query = spec.params.get("query") + if query is None or (isinstance(query, str) and not query.strip()): + query = "*" + else: + query = str(query) + indexes = _indexes_param(spec.params) + page_size = _clamp_page_size( + _int_param(spec.params, "page_size", self.page_size) + ) + max_rows = max(1, _int_param(spec.params, "max_rows", self.max_rows)) + site = normalize_site(str(spec.params.get("site") or self.site)) + stream_id = f"{','.join(indexes)}|{query}" if indexes else query + yield LogStreamRef( + adapter=self.name, + stream_id=stream_id, + metadata={ + "query": query, + "indexes": indexes, + "page_size": page_size, + "max_rows": max_rows, + "site": site, + }, + ) + + def read(self, ref: LogStreamRef, window: TimeWindow) -> Iterator[RawLogLine]: + self.check_available() + query = ref.metadata.get("query") or ref.stream_id or "*" + indexes = ref.metadata.get("indexes") or [] + page_size = _clamp_page_size( + int(ref.metadata.get("page_size") or self.page_size) + ) + max_rows = max(1, int(ref.metadata.get("max_rows") or self.max_rows)) + site = ref.metadata.get("site") or self.site + url = api_base_url(str(site)) + SEARCH_PATH + cursor = ref.cursor + yielded = 0 + + client = _client() + try: + while yielded < max_rows: + remaining = max_rows - yielded + payload: dict[str, Any] = { + "filter": { + "query": query, + "from": _iso8601(window.start), + "to": _iso8601(window.end), + }, + "sort": "timestamp", + "page": {"limit": min(page_size, remaining)}, + } + if indexes: + payload["filter"]["indexes"] = indexes + if cursor: + payload["page"]["cursor"] = cursor + + try: + page = self._search_logs(client, url, payload) + except httpx.HTTPError as e: + raise AdapterUnavailableError( + f"Datadog Logs API unavailable for {ref.stream_id}: {e}" + ) from e + + events = page.get("data") + if not events: + ref.cursor = None + break + + for event in events: + if not isinstance(event, dict): + continue + mapped = map_datadog_event(event) + text = orjson.dumps(mapped).decode() + yield RawLogLine( + text=text, + source_ref=ref.stream_id, + received_at=parse_event_timestamp(event), + ) + yielded += 1 + if yielded >= max_rows: + break + + after = (page.get("meta") or {}).get("page", {}).get("after") + if not after or after == cursor or yielded >= max_rows: + # Keep the next-page cursor when we hit max_rows so --resume-job + # can continue; clear it when the window is exhausted. + keep_cursor = yielded >= max_rows and after and after != cursor + ref.cursor = after if keep_cursor else None + break + cursor = after + ref.cursor = cursor + finally: + client.close() + + def _headers(self) -> dict[str, str]: + return { + "DD-API-KEY": self.api_key, + "DD-APPLICATION-KEY": self.app_key, + "Content-Type": "application/json", + "Accept": "application/json", + } + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=2, max=10), + retry=retry_if_exception(_is_retryable), + reraise=True, + ) + def _search_logs( + self, client: httpx.Client, url: str, payload: dict[str, Any] + ) -> dict: + response = client.post(url, headers=self._headers(), json=payload) + response.raise_for_status() + try: + data = response.json() + except ValueError as e: + raise AdapterUnavailableError( + f"Datadog Logs API returned invalid JSON: {e}" + ) from e + if not isinstance(data, dict): + raise AdapterUnavailableError("Datadog Logs API returned a non-object body") + return data diff --git a/src/adapters/registry.py b/src/adapters/registry.py index 3f219ed..3bb7d86 100644 --- a/src/adapters/registry.py +++ b/src/adapters/registry.py @@ -1,7 +1,7 @@ from src.adapters.base import SourceAdapter from src.core.errors import AdapterUnavailableError -ADAPTER_NAMES = ("file", "cloudwatch") +ADAPTER_NAMES = ("file", "cloudwatch", "datadog") def get_adapter(name: str, settings) -> SourceAdapter: @@ -15,4 +15,14 @@ def get_adapter(name: str, settings) -> SourceAdapter: from src.adapters.cloudwatch.adapter import CloudWatchSourceAdapter return CloudWatchSourceAdapter(region=settings.adapter_cloudwatch_region) + if name == "datadog": + from src.adapters.datadog.adapter import DatadogSourceAdapter + + return DatadogSourceAdapter( + api_key=settings.datadog_api_key, + app_key=settings.datadog_app_key, + site=settings.datadog_site, + page_size=settings.datadog_page_size, + max_rows=settings.datadog_max_rows, + ) raise AdapterUnavailableError(f"Unknown adapter: {name!r}") diff --git a/src/api/routes/ingestions.py b/src/api/routes/ingestions.py index df776d2..9fd5945 100644 --- a/src/api/routes/ingestions.py +++ b/src/api/routes/ingestions.py @@ -118,7 +118,7 @@ def create_ingestion(request: IngestRequest): try: adapter = get_adapter(request.adapter, get_settings()) # Fail fast on missing/invalid params (e.g. no log_group) — discover() for - # cloudwatch is local-only validation, no network call. + # cloudwatch/datadog is local-only validation, no network call. list(adapter.discover(SourceSpec(adapter=request.adapter, params=request.params))) except AdapterUnavailableError as e: raise HTTPException( diff --git a/src/cli/commands/ingest.py b/src/cli/commands/ingest.py index b5a3133..5474ee6 100644 --- a/src/cli/commands/ingest.py +++ b/src/cli/commands/ingest.py @@ -24,7 +24,7 @@ def ingest_cmd( env: Optional[str] = typer.Option(None, "--env", help="Default environment"), fmt: str = typer.Option("auto", "--format", help="Log format: json|text|auto"), with_embeddings: bool = typer.Option(False, "--with-embeddings/--no-embeddings", help="Generate embeddings"), - adapter: str = typer.Option("file", "--adapter", help="Source adapter: file|cloudwatch"), + adapter: str = typer.Option("file", "--adapter", help="Source adapter: file|cloudwatch|datadog"), param: Optional[List[str]] = typer.Option(None, "--param", help="Adapter param as key=value (repeatable)"), since: Optional[str] = typer.Option(None, "--since", help="Window, e.g. 30m, 1h, 24h (non-file adapters)"), from_time: Optional[str] = typer.Option(None, "--from", help="Window start, ISO 8601 (non-file adapters)"), diff --git a/src/config/settings.py b/src/config/settings.py index 54536c3..e2a48e2 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -31,6 +31,11 @@ class Settings(BaseSettings): # Adapters adapter_cloudwatch_region: str = "us-east-1" + datadog_api_key: str = "" + datadog_app_key: str = "" + datadog_site: str = "datadoghq.com" + datadog_page_size: int = 1000 + datadog_max_rows: int = 10000 # Cluster scoring weights severity_weight_fatal: float = 5.0 diff --git a/tests/unit/test_adapters.py b/tests/unit/test_adapters.py index 0811fbe..e2073e4 100644 --- a/tests/unit/test_adapters.py +++ b/tests/unit/test_adapters.py @@ -1,16 +1,17 @@ """ Tests for src.adapters.base / src.adapters.file.adapter / src.adapters.cloudwatch.adapter -/ src.adapters.registry. +/ src.adapters.datadog.adapter / src.adapters.registry. File adapter tests check that the new SourceAdapter-conforming class produces identical output to the pre-existing free functions (discover_files/detect_format/read_lines). CloudWatch adapter tests mock the boto3 client boundary, not internals — same convention -as tests/unit/test_worker.py. +as tests/unit/test_worker.py. Datadog adapter tests mock the httpx client boundary. """ from datetime import datetime, timezone from pathlib import Path from unittest.mock import MagicMock, patch +import orjson import pytest from src.adapters.base import LogStreamRef, SourceAdapter, SourceSpec, TimeWindow @@ -290,6 +291,331 @@ def test_check_available_ok_with_credentials(self): adapter.check_available() # does not raise +# ── DatadogSourceAdapter ───────────────────────────────────────────────────── + +def _dd_event( + message: str = "hello", + status: str = "error", + timestamp: str = "2026-01-01T00:00:00.000Z", + **attrs: object, +) -> dict: + attributes = { + "message": message, + "status": status, + "timestamp": timestamp, + **attrs, + } + return {"id": "abc", "type": "log", "attributes": attributes} + + +def _dd_page(events: list, after: str | None = None) -> dict: + page: dict = {"data": events} + if after is not None: + page["meta"] = {"page": {"after": after}} + return page + + +def _http_status_error(status_code: int) -> Exception: + import httpx + + request = httpx.Request("POST", "https://api.datadoghq.com/api/v2/logs/events/search") + response = httpx.Response(status_code, request=request) + return httpx.HTTPStatusError("error", request=request, response=response) + + +class TestDatadogMapping: + def test_maps_reserved_attributes_to_json_aliases(self): + from src.adapters.datadog.adapter import map_datadog_event + + event = _dd_event( + message="Stripe webhook failed", + status="error", + service="billing-worker", + host="i-abc", + tags=["env:prod", "version:1.2"], + ) + mapped = map_datadog_event(event) + assert mapped["message"] == "Stripe webhook failed" + assert mapped["level"] == "error" + assert mapped["service"] == "billing-worker" + assert mapped["host"] == "i-abc" + assert mapped["env"] == "prod" + assert mapped["timestamp"] == "2026-01-01T00:00:00.000Z" + + def test_maps_nested_trace_and_request_ids(self): + from src.adapters.datadog.adapter import map_datadog_event + + event = _dd_event( + attributes={ + "dd": {"trace_id": "trace-1"}, + "http": {"request_id": "req-9"}, + } + ) + mapped = map_datadog_event(event) + assert mapped["trace_id"] == "trace-1" + assert mapped["request_id"] == "req-9" + + def test_env_falls_back_to_nested_attribute(self): + from src.adapters.datadog.adapter import map_datadog_event + + event = _dd_event(attributes={"env": "staging"}) + assert map_datadog_event(event)["env"] == "staging" + + def test_mapped_event_is_parseable_as_json_log_line(self): + import orjson + + from src.adapters.datadog.adapter import map_datadog_event + from src.core.parsing.json_parser import parse_json_line + + event = _dd_event( + message="Stripe webhook failed", + status="error", + service="billing-worker", + host="i-abc", + tags=["env:prod"], + attributes={"dd": {"trace_id": "t-1"}, "request_id": "r-1"}, + ) + parsed = parse_json_line(orjson.dumps(map_datadog_event(event)).decode()) + assert parsed is not None + assert parsed.parse_error is None + assert parsed.message == "Stripe webhook failed" + assert parsed.level == "error" + assert parsed.service == "billing-worker" + assert parsed.environment == "prod" + assert parsed.host == "i-abc" + assert parsed.trace_id == "t-1" + assert parsed.request_id == "r-1" + + +class TestDatadogSourceAdapter: + def test_conforms_to_protocol(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + assert isinstance(DatadogSourceAdapter(api_key="k", app_key="a"), SourceAdapter) + + def test_discover_defaults_query_to_star(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + refs = list(DatadogSourceAdapter().discover(SourceSpec(adapter="datadog", params={}))) + assert [r.stream_id for r in refs] == ["*"] + assert refs[0].metadata["query"] == "*" + + def test_discover_uses_query_and_indexes(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + spec = SourceSpec( + adapter="datadog", + params={"query": "service:api", "indexes": "main,pci"}, + ) + refs = list(DatadogSourceAdapter().discover(spec)) + assert refs[0].stream_id == "main,pci|service:api" + assert refs[0].metadata["indexes"] == ["main", "pci"] + + def test_discover_clamps_page_size_and_coerces_string_ints(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + spec = SourceSpec( + adapter="datadog", + params={"page_size": "5000", "max_rows": "250"}, + ) + refs = list(DatadogSourceAdapter().discover(spec)) + assert refs[0].metadata["page_size"] == 1000 + assert refs[0].metadata["max_rows"] == 250 + + def test_api_base_url_variants(self): + from src.adapters.datadog.adapter import api_base_url + + assert api_base_url("datadoghq.com") == "https://api.datadoghq.com" + assert api_base_url("us3.datadoghq.com") == "https://api.us3.datadoghq.com" + assert api_base_url("app.datadoghq.eu") == "https://api.datadoghq.eu" + assert api_base_url("https://api.datadoghq.com") == "https://api.datadoghq.com" + + def test_api_base_url_rejects_non_datadog_origins(self): + from src.adapters.datadog.adapter import api_base_url + + for site in ( + "http://evil.example", + "https://evil.example", + "https://api.datadoghq.com.evil.example", + "evil.example", + ): + with pytest.raises(AdapterUnavailableError, match="unsupported Datadog site"): + api_base_url(site) + + def test_discover_rejects_non_datadog_site_param(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + adapter = DatadogSourceAdapter(api_key="k", app_key="a") + spec = SourceSpec( + adapter="datadog", + params={"site": "http://evil.example"}, + ) + with pytest.raises(AdapterUnavailableError, match="unsupported Datadog site"): + list(adapter.discover(spec)) + + def test_constructor_rejects_non_datadog_site(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + with pytest.raises(AdapterUnavailableError, match="unsupported Datadog site"): + DatadogSourceAdapter(api_key="k", app_key="a", site="https://evil.example") + + def _adapter(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + return DatadogSourceAdapter(api_key="api", app_key="app", site="datadoghq.com") + + def _no_retry_sleep(self) -> None: + from src.adapters.datadog.adapter import DatadogSourceAdapter + + DatadogSourceAdapter._search_logs.retry.sleep = lambda *_: None + + def test_read_maps_events_and_sets_received_at(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + adapter = self._adapter() + ref = LogStreamRef(adapter="datadog", stream_id="*", metadata={"query": "*"}) + page = _dd_page([_dd_event(message="line1", timestamp="2026-01-01T00:00:00Z")]) + + with patch.object(DatadogSourceAdapter, "_search_logs", return_value=page): + lines = list(adapter.read(ref, _WINDOW)) + + assert len(lines) == 1 + assert '"message":"line1"' in lines[0].text.replace(" ", "") + assert '"level":"error"' in lines[0].text.replace(" ", "") + assert lines[0].received_at == datetime(2026, 1, 1, tzinfo=timezone.utc) + assert lines[0].source_ref == "*" + assert ref.cursor is None + + def test_read_paginates_until_no_after_cursor(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + adapter = self._adapter() + ref = LogStreamRef(adapter="datadog", stream_id="service:api", metadata={"query": "service:api"}) + pages = [ + _dd_page([_dd_event(message="a"), _dd_event(message="b")], after="tok1"), + _dd_page([_dd_event(message="c")]), + ] + + with patch.object(DatadogSourceAdapter, "_search_logs", side_effect=pages) as mock_search: + messages = [orjson.loads(raw.text)["message"] for raw in adapter.read(ref, _WINDOW)] + + assert messages == ["a", "b", "c"] + assert mock_search.call_count == 2 + second_payload = mock_search.call_args_list[1].args[2] + assert second_payload["page"]["cursor"] == "tok1" + assert ref.cursor is None + + def test_read_honors_ref_cursor_on_first_request(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + adapter = self._adapter() + ref = LogStreamRef( + adapter="datadog", + stream_id="*", + cursor="resume-tok", + metadata={"query": "*"}, + ) + + with patch.object(DatadogSourceAdapter, "_search_logs", return_value=_dd_page([])) as mock_search: + list(adapter.read(ref, _WINDOW)) + + payload = mock_search.call_args.args[2] + assert payload["page"]["cursor"] == "resume-tok" + + def test_read_stops_at_max_rows_and_keeps_next_cursor(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + adapter = self._adapter() + ref = LogStreamRef( + adapter="datadog", + stream_id="*", + metadata={"query": "*", "max_rows": 2, "page_size": 10}, + ) + page = _dd_page( + [_dd_event(message="a"), _dd_event(message="b"), _dd_event(message="c")], + after="next", + ) + + with patch.object(DatadogSourceAdapter, "_search_logs", return_value=page) as mock_search: + lines = list(adapter.read(ref, _WINDOW)) + + assert len(lines) == 2 + assert ref.cursor == "next" + payload = mock_search.call_args.args[2] + assert payload["page"]["limit"] == 2 # remaining, not full page_size + + def test_read_sends_absolute_window_and_indexes(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + adapter = self._adapter() + ref = LogStreamRef( + adapter="datadog", + stream_id="main|*", + metadata={"query": "service:api", "indexes": ["main"], "site": "us3.datadoghq.com"}, + ) + + with patch.object(DatadogSourceAdapter, "_search_logs", return_value=_dd_page([])) as mock_search: + list(adapter.read(ref, _WINDOW)) + + url = mock_search.call_args.args[1] + payload = mock_search.call_args.args[2] + assert url == "https://api.us3.datadoghq.com/api/v2/logs/events/search" + assert payload["filter"]["from"] == "2026-01-01T00:00:00.000Z" + assert payload["filter"]["to"] == "2026-01-02T00:00:00.000Z" + assert payload["filter"]["query"] == "service:api" + assert payload["filter"]["indexes"] == ["main"] + assert payload["sort"] == "timestamp" + + def test_read_raises_without_keys(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + adapter = DatadogSourceAdapter() + ref = LogStreamRef(adapter="datadog", stream_id="*") + with pytest.raises(AdapterUnavailableError): + list(adapter.read(ref, _WINDOW)) + + def test_check_available_raises_without_keys(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + with pytest.raises(AdapterUnavailableError): + DatadogSourceAdapter(api_key="", app_key="").check_available() + + def test_check_available_ok_with_keys(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + DatadogSourceAdapter(api_key="k", app_key="a").check_available() + + def test_retry_skips_non_retryable_client_errors(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + self._no_retry_sleep() + mock_client = MagicMock() + mock_client.post.side_effect = _http_status_error(403) + adapter = self._adapter() + ref = LogStreamRef(adapter="datadog", stream_id="*", metadata={"query": "*"}) + + with patch("src.adapters.datadog.adapter._client", return_value=mock_client): + with pytest.raises(AdapterUnavailableError): + list(adapter.read(ref, _WINDOW)) + + assert mock_client.post.call_count == 1 + + def test_retry_retries_rate_limit_errors(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + + self._no_retry_sleep() + mock_client = MagicMock() + mock_client.post.side_effect = _http_status_error(429) + adapter = self._adapter() + ref = LogStreamRef(adapter="datadog", stream_id="*", metadata={"query": "*"}) + + with patch("src.adapters.datadog.adapter._client", return_value=mock_client): + with pytest.raises(AdapterUnavailableError): + list(adapter.read(ref, _WINDOW)) + + assert mock_client.post.call_count == 3 + + # ── registry ────────────────────────────────────────────────────────────────── class TestRegistry: @@ -309,6 +635,14 @@ def test_get_adapter_cloudwatch(self): adapter = get_adapter("cloudwatch", get_settings()) assert isinstance(adapter, CloudWatchSourceAdapter) + def test_get_adapter_datadog(self): + from src.adapters.datadog.adapter import DatadogSourceAdapter + from src.adapters.registry import get_adapter + from src.config import get_settings + + adapter = get_adapter("datadog", get_settings()) + assert isinstance(adapter, DatadogSourceAdapter) + def test_get_adapter_unknown_raises(self): from src.adapters.registry import get_adapter from src.config import get_settings diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index 3462d8e..f50461e 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -119,6 +119,13 @@ def test_health_reports_cloudwatch_unavailable_without_credentials(self): assert resp.json()["adapters"]["cloudwatch"].startswith("unavailable:") + def test_health_reports_datadog_unavailable_without_keys(self): + with patch("src.db.session.check_connection", return_value=True), \ + _patch_get_db(execute_scalar=3): + resp = client.get("/health") + + assert resp.json()["adapters"]["datadog"].startswith("unavailable:") + # ── POST /ingestions ────────────────────────────────────────────────────────── @@ -226,6 +233,42 @@ def capture_add(obj): assert resp.status_code == 202 assert resp.json()["status"] == "pending" + def test_202_for_datadog_adapter_with_default_query(self): + """ + Datadog discover() is local-only and defaults query to '*', so enqueue + should succeed without keys or an explicit query param. + """ + wj_id = str(uuid.uuid4()) + mock_db = _ctx_db() + + def capture_add(obj): + obj.id = uuid.UUID(wj_id) + + mock_db.add.side_effect = capture_add + + with patch("src.db.session.get_db", side_effect=lambda: mock_db): + resp = client.post("/ingestions", json={ + "paths": [], + "adapter": "datadog", + "params": {}, + "since": "1h", + }) + + assert resp.status_code == 202 + assert resp.json()["status"] == "pending" + + def test_400_for_datadog_non_datadog_site(self): + """params.site must be a Datadog hostname — a full URL must not leak keys.""" + resp = client.post("/ingestions", json={ + "paths": [], + "adapter": "datadog", + "params": {"site": "http://evil.example"}, + "since": "1h", + }) + + assert resp.status_code == 400 + assert resp.json()["detail"]["error_code"] == "ADAPTER_UNAVAILABLE" + # ── GET /ingestions/jobs/{id} ─────────────────────────────────────────────────