Don't produce *_lag_seconds when all metric's values are hardcoded to 0.

.envrc

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+
+# Export all:
+# - (should be) .gitignored
+# - (potentially) secret environment variables
+# - from dotenv-formatted files w/names starting w/`.env.`
+DOTENV_FILES="$(find . -maxdepth 1 -type f -name '.env.*' -and -not -name '.envrc')"
+for file in ${DOTENV_FILES}; do
+  dotenv "${file}"
+done
+export DOTENV_FILES
+
+# Load nix env for all the cool people
+use flake

.github/workflows/post-release.yml

@@ -41,7 +41,7 @@ jobs:
           sudo apt-get update
           sudo apt-get install -y cmake build-essential libssl-dev libsasl2-dev zlib1g-dev libzstd-dev liblz4-dev pkg-config libcurl4-openssl-dev
 
-      - uses: dtolnay/rust-toolchain@1.93
+      - uses: dtolnay/rust-toolchain@1.96
         with:
           targets: ${{ matrix.target }}
 

.github/workflows/release-plz-pr.yml

@@ -30,7 +30,7 @@ jobs:
           sudo apt-get install -y cmake build-essential libssl-dev libsasl2-dev zlib1g-dev libzstd-dev liblz4-dev pkg-config jq libcurl4-openssl-dev
 
       - name: Install Rust toolchain
-        uses: dtolnay/rust-toolchain@1.93
+        uses: dtolnay/rust-toolchain@1.96
 
       - name: Rust version
         id: rust_version

.github/workflows/release.yml

@@ -30,7 +30,7 @@ jobs:
           sudo apt-get install -y cmake build-essential libssl-dev libsasl2-dev zlib1g-dev libzstd-dev liblz4-dev pkg-config libcurl4-openssl-dev
 
       - name: Install Rust toolchain
-        uses: dtolnay/rust-toolchain@1.93
+        uses: dtolnay/rust-toolchain@1.96
 
       - name: Run release-plz
         uses: release-plz/action@v0.5

.gitignore

@@ -2,3 +2,6 @@
 .claude
 .idea/
 CLAUDE.md
+.direnv/
+result
+.env.*

Cargo.toml

@@ -11,7 +11,7 @@ documentation = "https://docs.rs/klag-exporter"
 readme = "README.md"
 keywords = ["kafka", "prometheus", "metrics", "lag", "exporter"]
 categories = ["development-tools::profiling"]
-exclude = ["test-stack/", "docs/", ".github/", ".claude/"]
+exclude = ["test-stack/", "docs/", ".github/", ".claude/", "flake.*", ".envrc"]
 
 [dependencies]
 tokio = { version = "1", features = ["full"] }

Dockerfile.dev

@@ -1,5 +1,5 @@
 # Build stage
-FROM rust:1.92-bookworm AS builder
+FROM rust:1.96-bookworm AS builder
 
 # Install build dependencies for rdkafka
 RUN apt-get update && apt-get install -y \

flake.nix

@@ -0,0 +1,78 @@
+{
+  inputs = {
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+    flake-utils.url = "github:numtide/flake-utils";
+
+    rust-overlay = {
+      url = "github:oxalica/rust-overlay";
+      inputs.nixpkgs.follows = "nixpkgs";
+    };
+  };
+
+  outputs =
+    inputs:
+    inputs.flake-utils.lib.eachDefaultSystem (
+      system:
+      let
+        pkgs = import inputs.nixpkgs {
+          inherit system;
+          overlays = [ (import inputs.rust-overlay) ];
+        };
+        inherit (pkgs) lib;
+        cargoToml = lib.fromTOML (lib.readFile ./Cargo.toml);
+
+        rustToolchain = pkgs.rust-bin.stable."1.93.0".default.override {
+          extensions = [
+            "clippy"
+            "rust-analyzer"
+            "rust-src"
+            "rustfmt"
+          ];
+        };
+
+        klag-exporter =
+          (pkgs.makeRustPlatform {
+            cargo = rustToolchain;
+            rustc = rustToolchain;
+          }).buildRustPackage
+            {
+              pname = cargoToml.package.name;
+              version = "${cargoToml.package.version}-${
+                if lib.hasAttr "revCount" inputs.self then
+                  "${lib.toString inputs.self.revCount}-${inputs.self.shortRev}"
+                else
+                  "gitDirty"
+              }";
+
+              src = lib.cleanSource ./.;
+              cargoLock.lockFile = ./Cargo.lock;
+              LIBCLANG_PATH = "${lib.makeLibraryPath [ pkgs.libclang ]}";
+
+              buildInputs = [
+                pkgs.openssl
+                pkgs.cyrus_sasl
+                pkgs.curl
+              ];
+              nativeBuildInputs = [
+                pkgs.pkg-config
+                pkgs.cmake
+              ];
+            };
+      in
+      {
+        checks = { inherit klag-exporter; };
+        devShells.default = pkgs.mkShell {
+          hardeningDisable = [ "fortify" ];
+          LIBCLANG_PATH = "${lib.makeLibraryPath [ pkgs.libclang ]}"; # for dev builds
+          inputsFrom = [ klag-exporter ];
+          packages = with pkgs; [
+            cargo-watch
+          ];
+        };
+        packages = {
+          inherit klag-exporter;
+          default = klag-exporter;
+        };
+      }
+    );
+}

src/cluster/manager.rs

@@ -13,8 +13,8 @@ use std::time::{Duration, Instant};
 use tokio::sync::broadcast;
 use tracing::{debug, error, info, instrument, warn};
 
-/// Default timeout for a single collection cycle (should be less than poll_interval)
-const DEFAULT_COLLECTION_TIMEOUT: Duration = Duration::from_secs(60);
+/// Default timeout for a single collection cycle (should be less than `poll_interval`)
+const DEFAULT_COLLECTION_TIMEOUT: Duration = Duration::from_mins(1);
 
 pub struct ClusterManager {
     cluster_name: String,
@@ -34,7 +34,7 @@ pub struct ClusterManager {
 
 impl ClusterManager {
     pub fn new(
-        config: ClusterConfig,
+        config: &ClusterConfig,
         registry: Arc<MetricsRegistry>,
         exporter_config: &ExporterConfig,
     ) -> Result<Self> {
@@ -43,11 +43,11 @@ impl ClusterManager {
         let filters = config.compile_filters()?;
         let performance = exporter_config.performance.clone();
 
-        let client = Arc::new(KafkaClient::with_performance(&config, performance.clone())?);
+        let client = Arc::new(KafkaClient::with_performance(config, performance.clone())?);
         let offset_collector = OffsetCollector::with_performance(
             Arc::clone(&client),
             filters,
-            performance.clone(),
+            performance,
             exporter_config.granularity,
         );
 
@@ -59,7 +59,7 @@ impl ClusterManager {
                     // the Tier-3 resident-memory saving for rate-mode users:
                     // no BaseConsumer pool, no extra librdkafka clients.
                     let ts_consumer =
-                        TimestampConsumer::with_pool_size(&config, ts_cfg.max_concurrent_fetches)?;
+                        TimestampConsumer::with_pool_size(config, ts_cfg.max_concurrent_fetches)?;
                     TimestampSampler::new_message(ts_consumer, ts_cfg.cache_ttl)
                 }
                 crate::config::TimestampSamplingMode::Rate => {
@@ -101,7 +101,7 @@ impl ClusterManager {
             timestamp_sampler,
             registry,
             poll_interval: exporter_config.poll_interval,
-            max_backoff: Duration::from_secs(300),
+            max_backoff: Duration::from_mins(5),
             granularity: exporter_config.granularity,
             max_concurrent_fetches: exporter_config.timestamp_sampling.max_concurrent_fetches,
             cache_cleanup_interval: exporter_config.timestamp_sampling.cache_ttl * 2,
@@ -303,12 +303,13 @@ impl ClusterManager {
             now_ms,
             poll_time_ms,
             &snapshot.compacted_topics,
+            self.timestamp_sampler.is_some(),
         );
 
         // Update registry with granularity and custom labels
         self.registry.update_with_options(
             &self.cluster_name,
-            lag_metrics,
+            &lag_metrics,
             self.granularity,
             &self.cluster_labels,
         );
@@ -320,7 +321,7 @@ impl ClusterManager {
         debug!(
             cluster = %self.cluster_name,
             elapsed_ms = scrape_duration_ms,
-            timestamp_cache_size = self.timestamp_sampler.as_ref().map(|s| s.cache_size()).unwrap_or(0),
+            timestamp_cache_size = self.timestamp_sampler.as_ref().map_or(0, super::super::collector::timestamp_sampler::TimestampSampler::cache_size),
             "Collection cycle completed"
         );