Add redis metrics giving better view into redis

[nikola-jokic]

No description provided.

[greptile-apps]

Greptile Summary

This PR adds a new Redis stream metrics subsystem to the event ingester: a Scanner that pipelines XINFO STREAM + MEMORY USAGE calls across all Events:* keys, and a Prometheus Collector that periodically runs the scanner, caches an atomic snapshot, and gates metric emission behind Kubernetes or standalone leader election.

• P1 — CollectionInterval: 0 panics on startup: time.NewTicker is called unconditionally with c.config.CollectionInterval; if the field is absent from the YAML, Go's zero-value time.Duration(0) is passed and the process crashes with "non-positive interval for NewTicker". A guard or default (similar to the InitialCollectionDelayMax pattern already used two lines above) should be added before the NewTicker call.

Confidence Score: 4/5

Safe to merge after fixing the zero CollectionInterval panic; the feature is disabled by default so it only affects users who opt in without setting the interval.

Most concerns from prior review rounds are addressed. One new P1 remains: an unchecked zero CollectionInterval causes a hard panic. Feature defaults to disabled so production impact is low.

internal/eventingester/metrics/redis/collector.go (CollectionInterval zero guard), internal/eventingester/repository/scanner_test.go (Redis DB conflict)

Important Files Changed

Filename	Overview
internal/eventingester/metrics/redis/collector.go	New Prometheus collector with leader-gated snapshot caching; panics on zero CollectionInterval (P1), otherwise the reset-per-cycle and atomic snapshot approach is sound.
internal/eventingester/repository/scanner.go	New Redis SCAN + pipeline scanner; properly handles NOSTREAM/WRONGTYPE/redis.Nil per-key errors and the pipeline-level error guard, age computation is correct.
internal/eventingester/repository/types.go	Adds StreamInfo, RedisClient interface, and parseStreamKey; comment and implementation correctly document that jobSetId (not queue) may contain ':'.
internal/eventingester/ingester.go	Wires Redis metrics collector into an errgroup alongside the existing ingester pipeline; correctly conditionally creates separate Redis client for metrics, leader election setup looks correct.
internal/eventingester/repository/scanner_test.go	Good unit/integration tests for the scanner; uses DB 11 which conflicts with existing event_test.go and eventstore_test.go (P2 flakiness risk).
internal/eventingester/metrics/redis/collector_test.go	Comprehensive unit tests covering top-N, queue aggregation, histogram resets, leadership transitions, and concurrent collection; uses DB 12 (no conflict).
internal/common/constants/constants.go	Promotes local eventStreamPrefix constant "Events:" to a shared package-level constant; refactored correctly across eventstore.go, event_repository.go, and tests.

Sequence Diagram

sequenceDiagram
    participant App as ingester.Run
    participant LC as LeaderController
    participant C as Collector.Run (goroutine)
    participant S as Scanner.ScanAll
    participant R as Redis
    participant P as Prometheus Scraper

    App->>LC: leaderController.Run(ctx)
    App->>C: collector.Run(ctx)
    Note over C: initial jitter delay [0, initialDelayMax)
    loop Every CollectionInterval
        C->>LC: GetToken().Leader()?
        alt is leader
            C->>S: ScanAll(ctx)
            S->>R: SCAN Events:* (batched)
            S->>R: Pipeline: XINFO STREAM + MEMORY USAGE
            R-->>S: stream info + memory bytes
            S-->>C: []StreamInfo
            C->>C: resetMetricsForNewCycle()
            C->>C: update topN/queue gauges + histograms
            C->>C: collectSnapshot() state.Store(metrics)
        else not leader
            C->>C: ClearState()
        end
    end
    P->>C: Collect(ch)
    C->>LC: GetToken().Leader()?
    alt is leader
        C->>P: emit cached []prometheus.Metric from state
    else
        C-->>P: return (no metrics)
    end

Loading

Comments Outside Diff (1)

1. internal/eventingester/metrics/redis/collector.go, line 206 (link)

   time.NewTicker panics on zero CollectionInterval

   time.NewTicker(d) panics with "non-positive interval for NewTicker" if d <= 0. If collectionInterval is absent or zero in the YAML, the zero-value time.Duration(0) is passed here, causing a hard panic at startup whenever the feature is enabled. There is no validation or default guard for this field.

   Add a guard before calling NewTicker:

_{Reviews (29): Last reviewed commit: "format" | Re-trigger Greptile}

[JamesMurkin]

These buckets are too small

[JamesMurkin]

Discussed this

Make byte and events 30 buckets, and age 20 buckets

[JamesMurkin]

It'd be good to use a constant prefix - as we do everywhere else

Possibly it should be armada_event_redis although I'm not 100%

• I can envision we have more redis in future
• So either we'd need the name to distinguish the difference or we'd need another label (instance/purpose something like that)

[JamesMurkin]

Does this redis need metrics in its name? It is already under the metrics section

[nikola-jokic]

@JamesMurkin here is the initial delay as random interval up to a minute. I think this might be fine, but we can change that

[masipauskas]

notes:

• high level skim through, lgtm.
• I don't see how metrics from topN / queue histograms drop out for deleted queues, instead of being reset to 0.