diff --git a/harnesses/aggregator-head-lag/cmd/script/metrics.go b/harnesses/aggregator-head-lag/cmd/script/metrics.go index 25afea98c..2ff6cc59f 100644 --- a/harnesses/aggregator-head-lag/cmd/script/metrics.go +++ b/harnesses/aggregator-head-lag/cmd/script/metrics.go @@ -40,6 +40,9 @@ var ( headLagRaces *prometheus.CounterVec headLagFirstShare *prometheus.GaugeVec headLagRefMatches *prometheus.CounterVec + headLagPoolSeen *prometheus.CounterVec + headLagPoolEntries *prometheus.GaugeVec + headLagRacesVoided *prometheus.CounterVec refClockEntries prometheus.Gauge // Fast-trade latency (for comparison with Pulse V2) @@ -228,6 +231,40 @@ func init() { ) prometheus.MustRegister(headLagRefMatches) + // What the pool subscription actually sees, per chain. The race filter + // reads this set, and until now nothing published its rate, so a filter + // that passed everything and a pool that is genuinely busy produced the + // same observable. + headLagPoolSeen = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "head_lag_pool_observed_total", + Help: "Trades observed by the bench pool subscription, per chain.", + }, + []string{"chain"}, + ) + prometheus.MustRegister(headLagPoolSeen) + + headLagPoolEntries = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "head_lag_pool_trades_entries", + Help: "Live entries in the pool membership set (TTL bounded).", + }, + []string{}, + ) + prometheus.MustRegister(headLagPoolEntries) + + // Why a race was discarded. "off_pool" is the filter doing its job; + // "single_participant" is the pre-existing rule. A population that stays + // high with off_pool near zero means the filter is not biting. + headLagRacesVoided = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "head_lag_races_voided_total", + Help: "Races closed without being scored, by reason.", + }, + []string{"chain", "reason"}, + ) + prometheus.MustRegister(headLagRacesVoided) + // Per-transaction race (race.go): who reported the trade first, and // how often, our node reference included as aggregator="reference". headLagFirst = prometheus.NewCounterVec( @@ -516,6 +553,18 @@ func RecordHeadLagRefMiss(aggregator, chain, region string) { } // RecordRefClockSize publishes the reference window occupancy. +func RecordPoolObserved(chain string) { + headLagPoolSeen.WithLabelValues(chain).Inc() +} + +func RecordPoolSetSize(n int) { + headLagPoolEntries.WithLabelValues().Set(float64(n)) +} + +func RecordRaceVoided(chain, reason string) { + headLagRacesVoided.WithLabelValues(chain, reason).Inc() +} + func RecordRefClockSize(n int) { refClockEntries.Set(float64(n)) } // RecordBlockchainHead records the current blockchain head block number diff --git a/harnesses/aggregator-head-lag/cmd/script/race.go b/harnesses/aggregator-head-lag/cmd/script/race.go index 9218655ff..4e34e1257 100644 --- a/harnesses/aggregator-head-lag/cmd/script/race.go +++ b/harnesses/aggregator-head-lag/cmd/script/race.go @@ -195,6 +195,7 @@ func (b *raceBook) closeLocked(e *raceEntry, k string) { e.closed = true e.void = true e.closedAt = time.Now() + RecordRaceVoided(e.chain, "off_pool") return } // Our own node takes part when it saw the trade. It also validates @@ -214,6 +215,7 @@ func (b *raceBook) closeLocked(e *raceEntry, k string) { e.closed = true e.void = true e.closedAt = time.Now() + RecordRaceVoided(e.chain, "single_participant") return } providers := 0 diff --git a/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go b/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go index a293129bc..a1d2b051a 100644 --- a/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go @@ -178,6 +178,7 @@ func runReferenceMonitor(stopChan <-chan struct{}) { reference.sweep() poolTrades.sweep() RecordRefClockSize(reference.size()) + RecordPoolSetSize(poolTrades.size()) } } }() @@ -304,6 +305,7 @@ func refConnect(p HeadLagPool, url string, stopChan <-chan struct{}) error { } reference.observe(p.ChainName, r.Value.Signature, now) poolTrades.observe(p.ChainName, r.Value.Signature, now) + RecordPoolObserved(p.ChainName) continue } @@ -320,6 +322,7 @@ func refConnect(p HeadLagPool, url string, stopChan <-chan struct{}) error { } reference.observe(p.ChainName, r.TransactionHash, now) poolTrades.observe(p.ChainName, r.TransactionHash, now) + RecordPoolObserved(p.ChainName) } }