diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index 67f9712675..ee70b718dd 100644 --- a/pkg/durableemitter/durable_emitter.go +++ b/pkg/durableemitter/durable_emitter.go @@ -780,19 +780,11 @@ func (d *DurableEmitter) expiryLoop() { } } -func (d *DurableEmitter) queueStatsNearExpiryLead() time.Duration { - lead := 5 * time.Minute - if d.cfg.Metrics != nil && d.cfg.Metrics.NearExpiryLead > 0 { - lead = d.cfg.Metrics.NearExpiryLead - } - return lead -} - func (d *DurableEmitter) metricsLoop() { mc := d.cfg.Metrics poll := mc.PollInterval if poll <= 0 { - poll = 500 * time.Millisecond + poll = 10 * time.Second } ctx, cancel := d.stopCh.NewCtx() @@ -806,7 +798,7 @@ func (d *DurableEmitter) metricsLoop() { return case <-ticker.C: if obs, ok := d.store.(DurableQueueObserver); ok { - st, err := obs.ObserveDurableQueue(ctx, d.cfg.EventTTL, d.queueStatsNearExpiryLead()) + st, err := obs.ObserveDurableQueue(ctx, d.cfg.EventTTL) if err != nil { d.eng.Debugw("DurableEmitter: queue observe failed; keeping last depth", "error", err) } else { diff --git a/pkg/durableemitter/durable_emitter_metrics.go b/pkg/durableemitter/durable_emitter_metrics.go index 0b86d2e214..9852e3123c 100644 --- a/pkg/durableemitter/durable_emitter_metrics.go +++ b/pkg/durableemitter/durable_emitter_metrics.go @@ -20,8 +20,6 @@ import ( type DurableEmitterMetricsConfig struct { // PollInterval is how often queue and optional process gauges refresh. Zero = 10s. PollInterval time.Duration - // NearExpiryLead is the window before EventTTL used for queue.near_ttl (DLQ pressure proxy). Zero = 5m. - NearExpiryLead time.Duration // MaxQueuePayloadBytes, if > 0, records capacity_usage_ratio = queue_payload_bytes / max. MaxQueuePayloadBytes int64 } @@ -64,7 +62,7 @@ type durableEmitterMetrics struct { queueDepth metric.Int64Gauge queuePayloadBytes metric.Int64Gauge queueOldestAgeSec metric.Float64Gauge - queueNearTTL metric.Int64Gauge + queueTTLBudgetSec metric.Int64Gauge queueCapacityRatio metric.Float64Gauge procHeapInuse metric.Int64Gauge procHeapSys metric.Int64Gauge @@ -230,10 +228,10 @@ func newDurableEmitterMetrics(meter metric.Meter) (*durableEmitterMetrics, error ); err != nil { return nil, err } - if m.queueNearTTL, err = meter.Int64Gauge( - "durable_emitter.queue.near_ttl", - metric.WithUnit("{row}"), - metric.WithDescription("Rows within near-expiry window of EventTTL (DLQ pressure proxy; no separate DLQ table)"), + if m.queueTTLBudgetSec, err = meter.Int64Gauge( + "durable_emitter.queue.ttl_budget_seconds", + metric.WithUnit("s"), + metric.WithDescription("Seconds of TTL headroom for the oldest pending event (EventTTL - oldest age); low/negative → DLQ/expiry pressure. Alert engine decides what 'near' means"), ); err != nil { return nil, err } @@ -309,7 +307,7 @@ func (m *durableEmitterMetrics) recordStoreOp(ctx context.Context, op string, el } // recordQueueStats records the DB-derived queue statistics (payload bytes, -// oldest pending age, near-TTL count) from an already-observed snapshot. The +// oldest pending age, TTL budget) from an already-observed snapshot. The // queue depth gauge itself is recorded separately by DurableEmitter from the // same snapshot's authoritative TotalRows count. func (m *durableEmitterMetrics) recordQueueStats(ctx context.Context, st DurableQueueStats, maxBytes int64) { @@ -322,7 +320,7 @@ func (m *durableEmitterMetrics) recordQueueStats(ctx context.Context, st Durable } else { m.queueOldestAgeSec.Record(ctx, st.OldestPendingAge.Seconds()) } - m.queueNearTTL.Record(ctx, st.NearTTLCount) + m.queueTTLBudgetSec.Record(ctx, int64(st.TTLBudget/time.Second)) if maxBytes > 0 { m.queueCapacityRatio.Record(ctx, float64(st.PayloadBytes)/float64(maxBytes)) } diff --git a/pkg/durableemitter/durable_emitter_test.go b/pkg/durableemitter/durable_emitter_test.go index 0806422d81..6481a3c5cc 100644 --- a/pkg/durableemitter/durable_emitter_test.go +++ b/pkg/durableemitter/durable_emitter_test.go @@ -789,6 +789,35 @@ func TestDurableEmitter_MetricsRegistersEmitSuccess(t *testing.T) { assert.True(t, found, "expected durable_emitter.emit.success in exported metrics") } +func TestDurableEmitter_MetricsRegistersQueueTTLBudget(t *testing.T) { + meter, reader := newTestMeter(t) + + store := NewMemDurableEventStore() + be := newTestBatchEmitter() + cfg := DefaultConfig() + cfg.RetransmitInterval = time.Hour + cfg.Metrics = &DurableEmitterMetricsConfig{PollInterval: 25 * time.Millisecond} + + em, err := NewDurableEmitter(store, be, true, cfg, logger.Test(t), meter) + require.NoError(t, err) + servicetest.Run(t, em) + ctx := t.Context() + + require.Eventually(t, func() bool { + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(ctx, &rm)) + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == "durable_emitter.queue.ttl_budget_seconds" { + _, ok := m.Data.(metricdata.Gauge[int64]) + return ok + } + } + } + return false + }, 2*time.Second, 10*time.Millisecond, "expected durable_emitter.queue.ttl_budget_seconds Int64 gauge in exported metrics") +} + func counterSumByPhase(t *testing.T, rm metricdata.ResourceMetrics, name, phase string) int64 { t.Helper() var total int64 @@ -1460,12 +1489,13 @@ func (m *MemDurableEventStore) Len() int { } // ObserveDurableQueue implements DurableQueueObserver. -func (m *MemDurableEventStore) ObserveDurableQueue(_ context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error) { +func (m *MemDurableEventStore) ObserveDurableQueue(_ context.Context, eventTTL time.Duration) (DurableQueueStats, error) { m.mu.Lock() defer m.mu.Unlock() now := time.Now() var st DurableQueueStats st.TotalRows = int64(len(m.events)) + st.TTLBudget = eventTTL if len(m.events) == 0 { return st, nil } @@ -1478,14 +1508,8 @@ func (m *MemDurableEventStore) ObserveDurableQueue(_ context.Context, eventTTL, oldest = e.CreatedAt first = false } - age := now.Sub(e.CreatedAt) - if eventTTL > 0 && nearExpiryLead > 0 && nearExpiryLead < eventTTL { - threshold := eventTTL - nearExpiryLead - if age >= threshold && age < eventTTL { - st.NearTTLCount++ - } - } } st.OldestPendingAge = now.Sub(oldest) + st.TTLBudget = eventTTL - st.OldestPendingAge return st, nil } diff --git a/pkg/durableemitter/observable_store.go b/pkg/durableemitter/observable_store.go index 1773896fc7..628c8267e2 100644 --- a/pkg/durableemitter/observable_store.go +++ b/pkg/durableemitter/observable_store.go @@ -26,18 +26,20 @@ type DurableQueueStats struct { TotalRows int64 PayloadBytes int64 OldestPendingAge time.Duration // 0 if the queue is empty - // NearTTLCount is the number of rows within nearExpiryLead of EventTTL (still - // pending, not yet removed by expiry). Serves as a DLQ-pressure proxy; there is - // no separate dead-letter table in the default design. - NearTTLCount int64 + // TTLBudget is the remaining time before the oldest still-pending event hits + // EventTTL (i.e. EventTTL - OldestPendingAge). It is the headroom of the event + // closest to expiry and serves as a DLQ-pressure proxy; there is no separate + // dead-letter table in the default design. It goes negative when the expiry + // loop is behind. Equals EventTTL when the queue is empty. + TTLBudget time.Duration } // DurableQueueObserver is optionally implemented by DurableEventStore implementations // so DurableEmitter can export queue depth and age gauges when metrics are enabled. type DurableQueueObserver interface { - // ObserveDurableQueue returns live queue statistics. eventTTL and nearExpiryLead - // match Config (nearExpiryLead should be << eventTTL). - ObserveDurableQueue(ctx context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error) + // ObserveDurableQueue returns live queue statistics. eventTTL matches Config and + // is used to derive the remaining TTL budget of the oldest pending event. + ObserveDurableQueue(ctx context.Context, eventTTL time.Duration) (DurableQueueStats, error) } // BatchInserter is optionally implemented by DurableEventStore implementations @@ -125,12 +127,12 @@ func (s *metricsInstrumentedStore) DeleteExpired(ctx context.Context, ttl time.D return n, err } -func (s *metricsInstrumentedStore) ObserveDurableQueue(ctx context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error) { +func (s *metricsInstrumentedStore) ObserveDurableQueue(ctx context.Context, eventTTL time.Duration) (DurableQueueStats, error) { o, ok := s.inner.(DurableQueueObserver) if !ok { return DurableQueueStats{}, errors.New("inner DurableEventStore does not implement DurableQueueObserver") } - return o.ObserveDurableQueue(ctx, eventTTL, nearExpiryLead) + return o.ObserveDurableQueue(ctx, eventTTL) } func (s *metricsInstrumentedStore) InsertBatch(ctx context.Context, payloads [][]byte) ([]int64, error) { diff --git a/pkg/durableemitter/store.go b/pkg/durableemitter/store.go index 6f4e63b7d1..34255f5baf 100644 --- a/pkg/durableemitter/store.go +++ b/pkg/durableemitter/store.go @@ -165,7 +165,7 @@ type chipDurableQueueAgg struct { // cnt/payload_sum/min_created describe only the undelivered backlog, while total // is the authoritative physical row count (incl. delivered-but-not-purged rows) // used as the queue-depth gauge. -func (s *PgDurableEventStore) ObserveDurableQueue(ctx context.Context, eventTTL, nearExpiryLead time.Duration) (DurableQueueStats, error) { +func (s *PgDurableEventStore) ObserveDurableQueue(ctx context.Context, eventTTL time.Duration) (DurableQueueStats, error) { const qAgg = ` SELECT count(*) FILTER (WHERE delivered_at IS NULL)::bigint AS cnt, @@ -182,21 +182,13 @@ FROM ` + chipDurableEventsTable st.Depth = row.Cnt st.TotalRows = row.Total st.PayloadBytes = row.PayloadSum + // TTLBudget is the headroom of the oldest pending event before it hits EventTTL. + // With no backlog the whole budget is available; the "near expiry" threshold is + // applied downstream by the alert engine, not here. + st.TTLBudget = eventTTL if row.MinCreated != nil { st.OldestPendingAge = time.Since(*row.MinCreated) - } - if eventTTL > 0 && nearExpiryLead > 0 && nearExpiryLead < eventTTL { - ttlSec := int64(eventTTL.Round(time.Second) / time.Second) - leadSec := int64(nearExpiryLead.Round(time.Second) / time.Second) - const qNear = ` -SELECT count(*)::bigint -FROM ` + chipDurableEventsTable + ` -WHERE delivered_at IS NULL - AND created_at >= now() - ($1::bigint * interval '1 second') - AND created_at < now() - (($1::bigint - $2::bigint) * interval '1 second')` - if err := s.ds.GetContext(ctx, &st.NearTTLCount, qNear, ttlSec, leadSec); err != nil { - return DurableQueueStats{}, fmt.Errorf("durable queue near-ttl: %w", err) - } + st.TTLBudget = eventTTL - st.OldestPendingAge } return st, nil }