Collectors Overview¶

The Collectors service is a single container running two roles in one process:

APScheduler — periodic polling of external APIs (Jira, Prometheus, Azure DevOps)
FastAPI Webhook Receiver — HTTP endpoints for push-based sources (GitLab)

Architecture¶

%%{init: {'theme':'default','themeVariables':{'primaryColor':'#1a73e8'}}}%%
graph TB
  subgraph CONTAINER["collectors :8080"]
    SCH[APScheduler<br/>BackgroundScheduler]
    WH[FastAPI App<br/>webhook_receiver.py]
    STATE[PollingState<br/>Redis]
    KPROD[Kafka Producer<br/>acks=all, LZ4]
  end

  subgraph SOURCES["External Sources"]
    JIR[Jira REST API]
    PROM[Prometheus HTTP API]
    AZDO[Azure DevOps WIQL]
    GIT[GitLab Webhook]
  end

  KAF[(Kafka<br/>raw_events<br/>raw_metrics)]

  SCH -->|Poll| JIR
  SCH -->|Poll| PROM
  SCH -->|Poll| AZDO
  GIT -->|Push| WH
  WH --> STATE
  SCH --> STATE
  JIR --> KPROD
  PROM --> KPROD
  AZDO --> KPROD
  WH --> KPROD
  KPROD --> KAF

Process Model¶

Aspect	Detail
Base Image	`python:3.11-slim` + `librdkafka-dev`
Entry Point	`python scheduler.py`
Web Server	Uvicorn (embedded in same process)
Scheduler	`APScheduler BackgroundScheduler`
Kafka Client	`confluent-kafka` (shared `libs/common/kafka_client.py`)
State Backend	Redis (`polling:cursor:{source}`)
Logging	`structlog` JSON
Health Check	`GET /health`

Configuration¶

`services/collectors/config/sources.yaml`¶

sources:
  jira:
    poll_interval_seconds: 600      # 10 minutes
  prometheus:
    poll_interval_seconds: 300      # 5 minutes
  azuredevops:
    poll_interval_seconds: 600      # 10 minutes
  gitlab_webhook: {}                # Event-driven, no polling

Environment Variables (`.env`)¶

Variable	Collector	Required
`JIRA_BASE_URL`	Jira	Yes
`JIRA_EMAIL`	Jira	Yes
`JIRA_API_TOKEN`	Jira	Yes
`PROMETHEUS_URL`	Prometheus	Yes
`PROMETHEUS_QUERY`	Prometheus	`up` (default)
`AZURE_DEVOPS_ORG`	Azure DevOps	Yes
`AZURE_DEVOPS_PROJECT`	Azure DevOps	Yes
`AZURE_DEVOPS_PAT`	Azure DevOps	Yes
`GITLAB_WEBHOOK_SECRET`	GitLab	Yes (HMAC)

Kafka Production¶

All collectors use the shared Kafka producer (libs/common/kafka_client.py):

# Configuration
bootstrap_servers = "kafka-broker:19092"
acks = "all"
retries = 5
linger_ms = 20
compression_type = "lz4"
value_serializer = orjson.dumps

Source	Topic	Key	Partitioning
Jira	`raw_events`	`source`	Hash
Prometheus	`raw_metrics`	`source`	Hash
Azure DevOps	`raw_events`	`source`	Hash
GitLab Webhook	`raw_events`	`source`	Hash

Polling State Management¶

Redis Key: polling:cursor:{source_name}

# State update happens ONLY after Kafka ACK
async def advance_cursor(source: str, new_cursor: str):
    await redis.set(f"polling:cursor:{source}", new_cursor)

Guarantees: - At-least-once delivery (Kafka acks=all) - Cursor advances only after successful produce - On restart: resumes from last acknowledged cursor - No duplicate prevention at collector level (pipeline handles dedup)

Webhook Receiver (GitLab)¶

Endpoint: POST /webhooks/gitlab

HMAC Verification:

# Header: X-Gitlab-Token
expected = hmac.new(
    secret.encode(),
    request.body,
    hashlib.sha256
).hexdigest()

Response Codes: | Code | Meaning | |------|---------| | 200 | Accepted, published to Kafka | | 400 | Invalid payload | | 401 | Invalid/missing HMAC | | 500 | Internal error (Kafka unavailable) |

Monitoring & Debugging¶

Health Check¶

curl http://localhost:8080/health
# {"status": "ok", "scheduler_running": true}

Collector Status via API¶

curl http://localhost:8000/api/monitoring/collectors
# {
#   "jira": {"last_poll": "2026-01-15T10:00:00Z", "status": "ok"},
#   "prometheus": {"last_poll": "2026-01-15T10:05:00Z", "status": "ok"},
#   ...
# }

View Polling Cursors¶

docker compose exec redis redis-cli KEYS "polling:cursor:*"
docker compose exec redis redis-cli GET "polling:cursor:jira"

Logs¶

docker compose logs -f collectors
# Look for:
# - "polling_completed" (success)
# - "polling_failed" (retry)
# - "webhook_received" (GitLab)

Scaling Considerations¶

Aspect	Current	Production Recommendation
Replicas	1	1 active + 1 standby (leader election)
Polling Intervals	5-10 min	Adjust per source criticality
Kafka Partitions	1	≥ collector replicas
Rate Limiting	tenacity retry	Add token bucket per source API

Note: APScheduler is not distributed. For HA, implement leader election (Redis lock) or use a dedicated scheduler (e.g., Celery Beat, Temporal).