Collector Scheduler

The APScheduler component that orchestrates periodic polling of external sources.

---

Overview

Property	Value
Library	apscheduler 3.10+
Scheduler Type	BackgroundScheduler
Job Store	Memory (in-process)
Timezone	UTC
Misfire Grace Time	APScheduler default (not configured)
Coalesce	APScheduler default
Max Instances	1 per job

---

Job Configuration

Defined in 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: {}                # No polling - event-driven

Each job: 1. Reads cursor from Redis (polling:cursor:{source}) 2. Calls source-specific collector 3. On Kafka ACK → advances cursor 4. Logs success/failure with structured logging

---

Scheduler Lifecycle

sequenceDiagram
  participant MAIN as scheduler.py
  participant SCH as BackgroundScheduler
  participant JOB as Collector Job
  participant REDIS as Redis
  participant KAFKA as Kafka

  MAIN->>SCH: scheduler.start()
  SCH->>SCH: Load jobs from sources.yaml
  loop Every interval
    SCH->>JOB: Execute collect()
    JOB->>REDIS: GET polling:cursor:source
    JOB->>JOB: Fetch new data since cursor
    JOB->>KAFKA: Produce batch (acks=all)
    KAFKA-->>JOB: ACK
    JOB->>REDIS: SET polling:cursor:source = new_cursor
  end

---

Code Structure

# scheduler.py
from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler(timezone="UTC")

def add_polling_jobs():
    for source_name, config in SOURCES_CONFIG.items():
        if "poll_interval_seconds" in config:
            interval = config["poll_interval_seconds"]
            scheduler.add_job(
                collectors[source_name].run_once,
                "interval",
                seconds=interval,
                id=f"poll_{source_name}",
                name=f"Poll {source_name}",
            )

if __name__ == "__main__":
    add_polling_jobs()
    scheduler.start()
    # Also start FastAPI webhook receiver in same process
    uvicorn.run(webhook_app, host="0.0.0.0", port=8080)

---

Error Handling & Retries

Scenario	Behavior
Source API timeout	tenacity retry (exponential backoff, max 3)
Source API 5xx	Retry with backoff
Source API 401/403	Log error, do not advance cursor, alert
Kafka produce fail	Retry (max 5, linger.ms=20), then log to DLQ
Scheduler crash	Jobs not running → cursor unchanged → catches up on restart

---

Monitoring

Health Check

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

Job Status via API

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

Redis Cursors

# View all cursors
docker compose exec redis redis-cli KEYS "polling:cursor:*"

# View specific cursor
docker compose exec redis redis-cli GET "polling:cursor:jira"
# "2026-01-15T10:00:00Z"

---

Logs & Debugging

# Follow scheduler logs
docker compose logs -f collectors | grep -E "poll_|scheduler"

# Key log events
# polling_started  - Job started
# polling_completed - Success, cursor advanced
# polling_failed   - Error (check details)
# webhook_received - GitLab webhook

# Structured log example
{
  "timestamp": "2026-01-15T10:00:00Z",
  "level": "info",
  "event": "polling_completed",
  "source": "jira",
  "duration_ms": 1250,
  "events_collected": 42,
  "cursor_advanced": true
}

---

Tuning

Parameter	Default	Tuning Guidance
poll_interval_seconds	300-600s	Lower for critical sources; respect API rate limits
misfire_grace_time	APScheduler default	Increase if network latency high
coalesce	APScheduler default	Keep True to avoid burst after downtime
max_instances	1	Never >1 (prevents duplicate polling)

---

Adding a New Source

1. Create collector class in services/collectors/{name}_collector.py
2. Register in scheduler.py → collectors dict
3. Add config to config/sources.yaml
4. Add env vars to .env.example
5. Implement collect() method returning list[dict] (CanonicalEvent-ready)
6. Test: docker compose restart collectors && docker compose logs -f collectors