How camatgithub/kafka-time-lag-monitor works
Kafka Consumer Lag Monitor is a Python daemon that reports how stale consumer-group data is in seconds. Its workflow loads configuration, discovers Kafka groups and offsets, records history and group status in SQLite, then interpolates committed offsets against partition snapshots and writes JSON lag reports. Sampling also manages ONLINE/OFFLINE/RECOVERING cadence and idle-group retirement; housekeeping prunes retained history. The sampled source supports the principal call paths; idle retirement is included as a documented sampler capability, while its detailed wiring is not expanded.
Main entry points: main.py Daemon entry · sampler.py Sampling cycle · database.py Database operations · reporter.py Report generation
Components 17
Daemon runtime
- Daemon entry
src/main.py - Config loader
src/config.py
Kafka collection
- Sampling cycle
src/sampler.py - Group discovery
src/kafka_client.py - Offset collection
src/kafka_client.py - Group state machine
src/sampler.py - Idle retirement
src/sampler.py
State and storage
- Shared state
src/state_manager.py - Database operations
src/database.py - SQLite history
src/database.py - History maintenance
src/housekeeping.py
Lag reporting
- Report generation
src/reporter.py - Lag calculation
src/interpolation.py - JSON lag report
Other components
- Operator
- YAML config
- Kafka cluster
Connections 25
- Operator → Daemon entry: starts daemon
- Operator → YAML config: provides settings
- Daemon entry → Config loader: loads config
- YAML config → Config loader: reads YAML
- Daemon entry → Database operations: initializes store
- Daemon entry → Group discovery: checks connectivity
- Group discovery → Kafka cluster: lists groups
- Daemon entry → Sampling cycle: starts worker
- Daemon entry → Report generation: starts worker
- Daemon entry → History maintenance: starts worker
- Sampling cycle → Group discovery: discovers groups
- Sampling cycle → Offset collection: requests offsets
- Offset collection → Kafka cluster: fetches offsets
- Sampling cycle → Database operations: records samples
- Sampling cycle → Group state machine: evaluates status
- Sampling cycle → Idle retirement: handles idle groups
- Sampling cycle → Shared state: updates shared state
- Group state machine → Shared state: updates status
- Shared state → Database operations: persists status
- Database operations → SQLite history: reads and writes
- Report generation → Database operations: reads history
- Report generation → Shared state: reads status
- Report generation → Lag calculation: calculates lag
- Report generation → JSON lag report: writes report
- History maintenance → Database operations: prunes history