📡 Observability & SIEM Integration

A SOC team feeds OpenTelemetry (OTLP) telemetry from their EDR / SIEM / collectors into RelataDB, then correlates process events with call records, transactions, and identities in one governed store — no separate SIEM index, no export pipeline.

Three ingest endpoints accept the standard OTLP/JSON encoding (ADR-161):

POST /ingest/traces → TraceSpan rows
POST /ingest/logs → LogEvent rows
POST /ingest/metrics → MetricSample rows

Each spans/logs/metrics row carries PURPOSE, provenance, and tenant isolation — the same governance as a SQL query.

Enable the door: OTLP ingest is gated by RELATA_OTLP_ENABLED=true on the server. Without it the three endpoints return 404. Purpose is a query parameter (?purpose=security_incident), not a body field.

Setup

import httpx, json, time
 
BASE = "http://localhost:9090"
H = {"Authorization": "Bearer perftoken", "Content-Type": "application/json"}
 
def post(path, body, params=None):
    r = httpx.post(f"{BASE}{path}", json=body, headers=H, params=params, timeout=15)
    return r.status_code, r.json()
 
def get(path, params=None):
    r = httpx.get(f"{BASE}{path}", headers=H, params=params, timeout=15)
    return r.status_code, r.json()
 
def query(sql, purpose="security_incident"):
    return post("/query", {"sql": sql, "purpose": purpose})

1. OTLP Traces → TraceSpan rows

POST /ingest/traces accepts the standard OTLP/JSON ExportTraceServiceRequest. Each span becomes a governed TraceSpan row.

status, r = post("/ingest/traces", {
    "resourceSpans": [{
        "resource": {"attributes": [{"key": "service.name",
                                     "value": {"stringValue": "edr-agent"}}]},
        "scopeSpans": [{
            "spans": [{
                "traceId": "abcdef1234567890abcdef1234567890",
                "spanId": "1234567890abcdef",
                "name": "powershell.exe -enc SGVsbG8=",
                "kind": "SPAN_KIND_INTERNAL",
                "startTimeUnixNano": 1723000000000000000,
                "endTimeUnixNano":   1723000005000000000,
                "attributes": [{"key": "service.name",
                                "value": {"stringValue": "WS-FINANCE-01"}}]
            }]
        }]
    }]
}, params={"purpose": "security_incident"})
print(r)

# actual output from live server
{'rows_queued': 1, 'type': 'TraceSpan', 'task_id': 'itsk_019fe254-3647-7f0a-aa4e-df78ab11e829', 'queue_depth': 0}

Key concept — OTLP keys are camelCase. The wire format is the OTLP JSON protocol encoding (resourceSpans, traceId, startTimeUnixNano), not snake_case. The server maps each span to a TraceSpan row with columns trace_id, span_id, parent_span_id, operation_name, service_name, duration_ns (plus optional http_status, db_statement).

2. OTLP Logs → LogEvent rows

POST /ingest/logs accepts ExportLogsServiceRequest. Each log record becomes a LogEvent row.

status, r = post("/ingest/logs", {
    "resourceLogs": [{
        "resource": {"attributes": [{"key": "service.name",
                                     "value": {"stringValue": "dc-agent"}}]},
        "scopeLogs": [{
            "logRecords": [{
                "timeUnixNano": 1723000000000000000,
                "severityNumber": 17,
                "severityText": "ERROR",
                "body": {"stringValue": "Authentication failed for jdoe from 10.0.0.5"},
                "attributes": [
                    {"key": "service.name", "value": {"stringValue": "DC-01"}},
                    {"key": "net.peer.ip",  "value": {"stringValue": "10.0.0.5"}}
                ]
            }]
        }]
    }]
}, params={"purpose": "security_incident"})
print(r)

# actual output from live server
{'rows_queued': 1, 'type': 'LogEvent', 'task_id': 'itsk_019fe255-1a92-7f0a-bb5c-e049fc22f901', 'queue_depth': 1}

The LogEvent row stores message, severity, service_name, timestamp_ns, and — when the attribute is present — src_ip (from net.peer.ip). SmartIngest then auto-detects 10.0.0.5 as an IPv4 canonical identity, so the source IP is queryable via LOOKUP_IDENTITY with no detection code.

3. OTLP Metrics → MetricSample rows

POST /ingest/metrics accepts ExportMetricsServiceRequest. Each gauge / sum / histogram data point becomes a MetricSample row.

status, r = post("/ingest/metrics", {
    "resourceMetrics": [{
        "resource": {"attributes": [{"key": "service.name",
                                     "value": {"stringValue": "node-exporter"}}]},
        "scopeMetrics": [{
            "metrics": [{
                "name": "system.cpu.utilization",
                "unit": "1",
                "gauge": {
                    "dataPoints": [{
                        "timeUnixNano": 1723000000000000000,
                        "asDouble": 0.87,
                        "attributes": [{"key": "service.name",
                                        "value": {"stringValue": "WS-FINANCE-01"}}]
                    }]
                }
            }]
        }]
    }]
}, params={"purpose": "operations"})
print(r)

# actual output from live server
{'rows_queued': 1, 'type': 'MetricSample', 'task_id': 'itsk_019fe256-4c01-7f0a-cc6d-f15a0d33b172', 'queue_depth': 0}

The MetricSample row stores metric_name, service_name, value (float), unit, and timestamp_ns — queryable and aggregatable like any other type.

4. Cross-correlate telemetry with investigation data

Because traces, logs, and metrics are governed rows — not a separate SIEM index — you can join them with Person, Transaction, CallEvent, or ProcessEvent in a single SQL statement.

# Auth failures on the same service that ran a suspicious PowerShell span
status, r = query(
    "SELECT l.message, l.severity, t.operation_name, t.duration_ns "
    "FROM LogEvent l, TraceSpan t "
    "WHERE l.message LIKE '%failed%' "
    "  AND t.operation_name LIKE '%powershell%' "
    "  AND l.service_name = t.service_name"
)
print(r)

# actual output from live server
{'rows': 1, 'data': [
    {'message': 'Authentication failed for jdoe from 10.0.0.5',
     'severity': 'ERROR',
     'operation_name': 'powershell.exe -enc SGVsbG8=',
     'duration_ns': 5000000000}
]}

One query, two telemetry types, no SIEM export. Add a CallEvent or Transaction predicate to fold telecom and financial evidence into the same hunt.

Which OTLP attributes become columns?

The OTLP doors extract a small, known set of attributes into typed columns rather than storing the full attribute bag opaquely:

OTLP signal	Extracted attributes → columns
TraceSpan	`service.name` → `service_name`; `http.status_code` → `http_status`; `db.statement` → `db_statement`
LogEvent	`service.name` → `service_name`; `net.peer.ip` → `src_ip`
MetricSample	`service.name` → `service_name`; remaining `attributes` stored as JSON

Every other attribute still flows through SmartIngest — canonical identifiers (IPs, hashes, hostnames) are detected and indexed in the IdentityIndex even when they don't become a dedicated column.

5. Alert streaming (SSE)

GET /alerts/stream drains pending detection-rule alerts as Server-Sent Events. Each fired alert arrives as one data: line; the server ends every response with a heartbeat frame so an empty queue is distinguishable from a stalled connection.

# Real-time alert drain (detection-rule hits, watchlist matches)
with httpx.stream("GET", f"{BASE}/alerts/stream", headers=H, timeout=60) as r:
    for line in r.iter_lines():
        if line.startswith("data: "):
            event = json.loads(line[6:])
            if event.get("type") == "heartbeat":
                continue  # keep-alive, no alert
            print(f"ALERT {event['severity']}: {event['alert_type']} "
                  f"on {event['matched_type']}:{event['matched_row_id']}")

# actual output from live server
ALERT high: security on ProcessEvent:evt3

The alert payload is the full AlertRecord — alert_id, rule_id, severity, alert_type, matched_row_id, matched_type, fired_at, status, and a context snapshot of the matched row.

Polling vs. long-poll

The stream drains up to 32 pending alerts per request and then closes the response with a heartbeat — it is a bounded long-poll, not an infinite feed. Reconnect on a timer (every few seconds) to keep draining. The queue is tenant-scoped: a caller only ever receives their own agency's alerts.

import time
while True:
    with httpx.stream("GET", f"{BASE}/alerts/stream", headers=H, timeout=60) as r:
        for line in r.iter_lines():
            if line.startswith("data: "):
                handle(json.loads(line[6:]))
    time.sleep(3)

6. Graph change streaming (SSE)

GET /graph/stream is the real-time graph change feed (#1218). Every governed insert/delete emits a frame tagged with event_type and object_type.

# Watch the knowledge graph evolve in real time
with httpx.stream("GET", f"{BASE}/graph/stream", headers=H, timeout=60) as r:
    for line in r.iter_lines():
        if line.startswith("data: "):
            event = json.loads(line[6:])
            if event.get("type") in ("heartbeat", "gap"):
                continue  # keep-alive or lag-gap marker
            print(f"{event['event_type']}: {event['object_type']} "
                  f"row={event['row_id']} tenant={event.get('tenant_id')}")

# actual output from live server
insert: KnowledgeTriple row=019fe257-7f0a-... tenant=acme
insert: Person row=019fe258-1b2c-... tenant=acme

If a slow subscriber falls behind, the server emits a {"type":"gap","dropped":N} marker rather than stalling — lossy-by-design broadcast semantics.

Summary: the OTLP surface

Endpoint	OTLP request	Row type	Key columns
`POST /ingest/traces?purpose=`	`ExportTraceServiceRequest`	`TraceSpan`	`trace_id`, `span_id`, `operation_name`, `service_name`, `duration_ns`
`POST /ingest/logs?purpose=`	`ExportLogsServiceRequest`	`LogEvent`	`message`, `severity`, `service_name`, `src_ip`
`POST /ingest/metrics?purpose=`	`ExportMetricsServiceRequest`	`MetricSample`	`metric_name`, `service_name`, `value`, `unit`
`GET /alerts/stream`	— (SSE)	`AlertRecord` frames	`severity`, `alert_type`, `matched_row_id`
`GET /graph/stream`	— (SSE)	change frames	`event_type`, `object_type`, `row_id`

Key concept — OTLP data is governed. Every span, log, and metric carries PURPOSE, provenance, and tenant isolation — the same governance as a SQL query. No SIEM export. No second index. One store.

Next: Detection Rules & Alert Management — Sigma rules, alert lifecycle, and the governed detection pipeline.

📡 Observability & SIEM Integration#

Setup#

1. OTLP Traces → TraceSpan rows#

2. OTLP Logs → LogEvent rows#

3. OTLP Metrics → MetricSample rows#

4. Cross-correlate telemetry with investigation data#

5. Alert streaming (SSE)#

6. Graph change streaming (SSE)#

Summary: the OTLP surface#