# camatgithub/kafka-time-lag-monitor architecture

> Architecture diagram of the GitHub repository camatgithub/kafka-time-lag-monitor, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-09-25). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/camatgithub/kafka-time-lag-monitor
- Repository: https://github.com/camatgithub/kafka-time-lag-monitor
- Video tour (about a minute): https://gitdiagram.com/camatgithub/kafka-time-lag-monitor/video

## Overview

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: [`src/main.py`](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/main.py) (Daemon entry), [`src/sampler.py`](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/sampler.py) (Sampling cycle), [`src/database.py`](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/database.py) (Database operations), [`src/reporter.py`](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/reporter.py) (Report generation)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_runtime["Daemon runtime"]
  node_main["Daemon entry<br/>[main.py]"]
  node_config["Config loader<br/>[config.py]"]
end

subgraph group_collection["Kafka collection"]
  node_sampler["Sampling cycle<br/>[sampler.py]"]
  node_group_discovery["Group discovery<br/>[kafka_client.py]"]
  node_offset_collection["Offset collection<br/>[kafka_client.py]"]
  node_state_machine["Group state machine<br/>[sampler.py]"]
  node_idle_retirement["Idle retirement<br/>[sampler.py]"]
end

subgraph group_state["State and storage"]
  node_state_manager["Shared state<br/>[state_manager.py]"]
  node_database_api["Database operations<br/>[database.py]"]
  node_sqlite[("SQLite history<br/>[database.py]")]
  node_housekeeping["History maintenance<br/>[housekeeping.py]"]
end

subgraph group_reporting["Lag reporting"]
  node_reporter["Report generation<br/>[reporter.py]"]
  node_interpolation["Lag calculation<br/>[interpolation.py]"]
  node_json_output["JSON lag report"]
end

node_operator(("Operator"))
node_config_file["YAML config"]
node_kafka{{"Kafka cluster"}}

node_operator -->|"starts daemon"| node_main
node_operator -->|"provides settings"| node_config_file
node_main -->|"loads config"| node_config
node_config_file -->|"reads YAML"| node_config
node_main -->|"initializes store"| node_database_api
node_main -->|"checks connectivity"| node_group_discovery
node_group_discovery -->|"lists groups"| node_kafka
node_main -->|"starts worker"| node_sampler
node_main -->|"starts worker"| node_reporter
node_main -->|"starts worker"| node_housekeeping
node_sampler -->|"discovers groups"| node_group_discovery
node_sampler -->|"requests offsets"| node_offset_collection
node_offset_collection -->|"fetches offsets"| node_kafka
node_sampler -->|"records samples"| node_database_api
node_sampler -->|"evaluates status"| node_state_machine
node_sampler -->|"handles idle groups"| node_idle_retirement
node_sampler -->|"updates shared state"| node_state_manager
node_state_machine -->|"updates status"| node_state_manager
node_state_manager -->|"persists status"| node_database_api
node_database_api -->|"reads and writes"| node_sqlite
node_reporter -->|"reads history"| node_database_api
node_reporter -->|"reads status"| node_state_manager
node_reporter -->|"calculates lag"| node_interpolation
node_reporter -->|"writes report"| node_json_output
node_housekeeping -->|"prunes history"| node_database_api

click node_main "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/main.py"
click node_config "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/config.py"
click node_sampler "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/sampler.py"
click node_group_discovery "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/kafka_client.py"
click node_offset_collection "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/kafka_client.py"
click node_state_machine "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/sampler.py"
click node_idle_retirement "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/sampler.py"
click node_state_manager "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/state_manager.py"
click node_database_api "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/database.py"
click node_sqlite "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/database.py"
click node_housekeeping "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/housekeeping.py"
click node_reporter "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/reporter.py"
click node_interpolation "https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/interpolation.py"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_main,node_config toneBlue
class node_sampler,node_group_discovery,node_offset_collection,node_state_machine,node_idle_retirement toneAmber
class node_state_manager,node_database_api,node_sqlite,node_housekeeping toneMint
class node_reporter,node_interpolation,node_json_output toneRose
class node_operator,node_config_file,node_kafka toneIndigo
```

## Components

### Daemon runtime

- [Daemon entry](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/main.py): `src/main.py`
- [Config loader](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/config.py): `src/config.py`

### Kafka collection

- [Sampling cycle](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/sampler.py): `src/sampler.py`
- [Group discovery](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/kafka_client.py): `src/kafka_client.py`
- [Offset collection](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/kafka_client.py): `src/kafka_client.py`
- [Group state machine](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/sampler.py): `src/sampler.py`
- [Idle retirement](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/sampler.py): `src/sampler.py`

### State and storage

- [Shared state](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/state_manager.py): `src/state_manager.py`
- [Database operations](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/database.py): `src/database.py`
- [SQLite history](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/database.py): `src/database.py`
- [History maintenance](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/housekeeping.py): `src/housekeeping.py`

### Lag reporting

- [Report generation](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/reporter.py): `src/reporter.py`
- [Lag calculation](https://github.com/camatgithub/kafka-time-lag-monitor/blob/master/src/interpolation.py): `src/interpolation.py`
- **JSON lag report**

### Other components

- **Operator**
- **YAML config**
- **Kafka cluster**

## Connections

- 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