From 17a0af3f814e467ffbbe275cad20f1b1fac2552d Mon Sep 17 00:00:00 2001 From: Tarcisio Ferraz Date: Fri, 10 Jul 2026 19:15:19 -0300 Subject: [PATCH 1/4] ttl budget metric --- pkg/durableemitter/durable_emitter.go | 10 +--------- pkg/durableemitter/durable_emitter_metrics.go | 16 +++++++-------- pkg/durableemitter/durable_emitter_test.go | 11 +++------- pkg/durableemitter/observable_store.go | 20 ++++++++++--------- pkg/durableemitter/store.go | 20 ++++++------------- 5 files changed, 28 insertions(+), 49 deletions(-) diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index 67f9712675..ac02d8d81d 100644 --- a/pkg/durableemitter/durable_emitter.go +++ b/pkg/durableemitter/durable_emitter.go @@ -780,14 +780,6 @@ 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 @@ -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..a8329407b6 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.Float64Gauge 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.Float64Gauge( + "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, st.TTLBudget.Seconds()) 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..9f3c090dff 100644 --- a/pkg/durableemitter/durable_emitter_test.go +++ b/pkg/durableemitter/durable_emitter_test.go @@ -1460,12 +1460,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 +1479,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 } From be35b1882f76bfb11b674f3db34d97b2603bc05d Mon Sep 17 00:00:00 2001 From: Tarcisio Ferraz Date: Mon, 13 Jul 2026 16:10:40 -0300 Subject: [PATCH 2/4] poll every 10s --- pkg/durableemitter/durable_emitter.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index ac02d8d81d..ee70b718dd 100644 --- a/pkg/durableemitter/durable_emitter.go +++ b/pkg/durableemitter/durable_emitter.go @@ -784,7 +784,7 @@ 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() From 7d81332c2c1339480c5f613d16ea55f29819c1ed Mon Sep 17 00:00:00 2001 From: Tarcisio Ferraz Date: Mon, 13 Jul 2026 16:11:16 -0300 Subject: [PATCH 3/4] add tests --- pkg/durableemitter/durable_emitter_test.go | 29 ++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/pkg/durableemitter/durable_emitter_test.go b/pkg/durableemitter/durable_emitter_test.go index 9f3c090dff..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 From 7bc5dca9e4d5bd58bf97f6ced5aaef6d57d33fa4 Mon Sep 17 00:00:00 2001 From: Tarcisio Ferraz Date: Mon, 13 Jul 2026 16:11:42 -0300 Subject: [PATCH 4/4] use int64 --- pkg/durableemitter/durable_emitter_metrics.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/durableemitter/durable_emitter_metrics.go b/pkg/durableemitter/durable_emitter_metrics.go index a8329407b6..9852e3123c 100644 --- a/pkg/durableemitter/durable_emitter_metrics.go +++ b/pkg/durableemitter/durable_emitter_metrics.go @@ -62,7 +62,7 @@ type durableEmitterMetrics struct { queueDepth metric.Int64Gauge queuePayloadBytes metric.Int64Gauge queueOldestAgeSec metric.Float64Gauge - queueTTLBudgetSec metric.Float64Gauge + queueTTLBudgetSec metric.Int64Gauge queueCapacityRatio metric.Float64Gauge procHeapInuse metric.Int64Gauge procHeapSys metric.Int64Gauge @@ -228,7 +228,7 @@ func newDurableEmitterMetrics(meter metric.Meter) (*durableEmitterMetrics, error ); err != nil { return nil, err } - if m.queueTTLBudgetSec, err = meter.Float64Gauge( + 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"), @@ -320,7 +320,7 @@ func (m *durableEmitterMetrics) recordQueueStats(ctx context.Context, st Durable } else { m.queueOldestAgeSec.Record(ctx, st.OldestPendingAge.Seconds()) } - m.queueTTLBudgetSec.Record(ctx, st.TTLBudget.Seconds()) + m.queueTTLBudgetSec.Record(ctx, int64(st.TTLBudget/time.Second)) if maxBytes > 0 { m.queueCapacityRatio.Record(ctx, float64(st.PayloadBytes)/float64(maxBytes)) }