Prometheus Collector¶

Polls Prometheus via instant query (/api/v1/query) and publishes metric samples as raw_metrics to Kafka.

Configuration¶

Variable	Description	Default
`PROMETHEUS_URL`	Prometheus HTTP endpoint	`http://prometheus:9090`
`PROMETHEUS_QUERY`	PromQL instant query	`up`

Poll Interval: 300 seconds (5 min) — configurable in sources.yaml

Query Execution¶

# Instant query
url = f"{PROMETHEUS_URL}/api/v1/query"
params = {"query": PROMETHEUS_QUERY, "time": now()}
response = httpx.get(url, params=params, timeout=10.0)

Default Query: up — returns 1 for healthy targets, 0 for down.

Data Mapping¶

Each series in the result vector becomes a separate event:

Prometheus Field	Canonical Event Field
`metric.__name__`	`tags.prom_metric_name`
`metric.instance`	`host` + `tags.instance`
`metric.job`	`tags.job`
`metric.{label}`	`tags.{label}`
`value[1]` (sample)	`message` + `tags.value`
`value[0]` (timestamp)	`timestamp`

Fixed Fields: - source = prometheus - type = metric - environment = from label or production - severity = derived from metric value: - up == 0 → CRITICAL - up == 1 → INFO - Custom thresholds via query

Sample Output¶

Query: up

{
  "schema_version": "1.0",
  "timestamp": "2026-01-15T10:30:00Z",
  "source": "prometheus",
  "type": "metric",
  "severity": "CRITICAL",
  "message": "up{instance=\"api:8000\",job=\"api\"} = 0",
  "host": "api",
  "tags": {
    "instance": "api:8000",
    "job": "api",
    "value": "0",
    "prom_metric_name": "up"
  },
  "dedup_key": "sha256:..."
}

Query: rate(http_requests_total[5m])

{
  "source": "prometheus",
  "type": "metric",
  "severity": "INFO",
  "message": "rate(http_requests_total[5m]){handler=\"/health\",job=\"api\"} = 12.5",
  "tags": {
    "handler": "/health",
    "job": "api",
    "value": "12.5",
    "prom_metric_name": "http_requests_total"
  }
}

Use Cases¶

Query	Purpose	Severity Logic
`up`	Target health	0 = CRITICAL
`container_memory_usage_bytes / container_spec_memory_limit_bytes > 0.9`	Memory pressure	>0.9 = WARNING
`rate(http_requests_total{status=~"5.."}[5m]) > 0.1`	Error rate	>0.1 = ERROR
`node_filesystem_avail_bytes / node_filesystem_size_bytes < 0.1`	Disk space	<0.1 = CRITICAL

Custom Queries¶

Edit sources.yaml:

prometheus:
  poll_interval_seconds: 300
  query: |
    up
    OR
    container_memory_usage_bytes / container_spec_memory_limit_bytes > 0.9

Or set PROMETHEUS_QUERY env var for complex queries.

Troubleshooting¶

Symptom	Check
No events	Verify `PROMETHEUS_URL` reachable, query returns data
All `up=1`	Targets healthy; check Prometheus targets page
High cardinality	Avoid queries returning many series (use recording rules)
Query timeout	Increase timeout, simplify query

# Test query manually
curl "http://localhost:9090/api/v1/query?query=up"

# View collector logs
docker compose logs -f collectors | grep -i prometheus