Fixes site_link_errors_total not counting errors (#2402)

Fixes #2378

Author: AryanP123

cmd/network-observer/internal/collector/metrics/metrics.go

@@ -109,21 +109,15 @@ type counterMetricByItem struct {
 func (c *counterMetricByItem) Ensure(id string, ct int) {
 	prev, ok := c.Items[id]
 	if !ok {
-		prev = ct
-		c.Items[id] = prev
-	}
-	if ct <= prev {
+		c.Items[id] = ct
 		return
 	}
-	c.Items[id] = ct
-	c.Counter.Add(float64(ct - prev))
-}
-
-func (m *counterMetricByItem) Remove(id string) {
-	if _, ok := m.Items[id]; !ok {
-		return
+	if ct > prev {
+		c.Counter.Add(float64(ct - prev))
 	}
-	delete(m.Items, id)
+	// When ct < prev the counter was reset (e.g. router restart). Accept
+	// the new value as the baseline without incrementing.
+	c.Items[id] = ct
 }
 
 type gaugeMetricByID struct {
@@ -411,15 +405,6 @@ func (a *Adaptor) Remove(record vanflow.Record) {
 			return
 		}
 		metric.Remove(record.ID)
-		linkErrorKey := siteLinkErrors{
-			SiteID: siteLink.SiteID,
-			Role:   siteLink.Role,
-		}
-		counter, ok := a.linkErrors[linkErrorKey]
-		if !ok {
-			return
-		}
-		counter.Remove(record.ID)
 	case vanflow.ListenerRecord:
 		site, wants, ok := a.listenerInfo(record)
 		if !ok {

cmd/network-observer/internal/collector/metrics/metrics_test.go

@@ -160,11 +160,14 @@ func TestSiteLinkErrorMetrics(t *testing.T) {
 	reg := prometheus.NewPedanticRegistry()
 	metrics := New(reg)
 	interRouter := ptrTo("inter-router")
+	statusUp := ptrTo("up")
 	statusDown := ptrTo("down")
 
 	metrics.Add(vanflow.RouterRecord{
 		BaseRecord: vanflow.NewBase("r01"), Parent: ptrTo("s01"), Mode: interRouter,
 	})
+
+	// Initial Add sets the baseline. Counter stays at zero.
 	metrics.Add(vanflow.LinkRecord{
 		BaseRecord: vanflow.NewBase("l01"),
 		Role:       interRouter,
@@ -174,6 +177,8 @@ func TestSiteLinkErrorMetrics(t *testing.T) {
 	})
 	assert.Equal(t, prom_testutil.CollectAndCount(reg, "skupper_site_link_errors_total"), 1)
 	assert.Equal(t, prom_testutil.ToFloat64(metrics.siteLinkErrors.WithLabelValues("s01", "inter-router")), 0.0)
+
+	// Update increments DownCount 12 -> 15: counter += 3.
 	metrics.Update(
 		vanflow.LinkRecord{
 			BaseRecord: vanflow.NewBase("l01"),
@@ -192,23 +197,51 @@ func TestSiteLinkErrorMetrics(t *testing.T) {
 	)
 	assert.Equal(t, prom_testutil.CollectAndCount(reg, "skupper_site_link_errors_total"), 1)
 	assert.Equal(t, prom_testutil.ToFloat64(metrics.siteLinkErrors.WithLabelValues("s01", "inter-router")), 3.0)
+
+	// Source reconnect with higher DownCount: the collector removes all
+	// records from the source and re-adds them. The re-added record
+	// carries the already-incremented DownCount. The counter must detect
+	// the delta across the Remove+Add boundary.
 	metrics.Remove(vanflow.LinkRecord{
 		BaseRecord: vanflow.NewBase("l01"),
 		Role:       interRouter,
 		Status:     statusDown,
 		Parent:     ptrTo("r01"),
 		DownCount:  ptrTo(uint64(15)),
 	})
-	// cannot decrease
-	assert.Equal(t, prom_testutil.CollectAndCount(reg, "skupper_site_link_errors_total"), 1)
+	// Counter must not decrease after removal.
 	assert.Equal(t, prom_testutil.ToFloat64(metrics.siteLinkErrors.WithLabelValues("s01", "inter-router")), 3.0)
+	metrics.Add(vanflow.LinkRecord{
+		BaseRecord: vanflow.NewBase("l01"),
+		Role:       interRouter,
+		Status:     statusUp,
+		Peer:       ptrTo("ap02"),
+		Parent:     ptrTo("r01"),
+		DownCount:  ptrTo(uint64(17)),
+	})
+	// Delta of 2 (17 - 15): counter = 5.
+	assert.Equal(t, prom_testutil.ToFloat64(metrics.siteLinkErrors.WithLabelValues("s01", "inter-router")), 5.0)
+
+	// Router restart resets DownCount to zero. Remove+Add with a lower
+	// DownCount resets the baseline without incrementing the counter.
+	metrics.Remove(vanflow.LinkRecord{
+		BaseRecord: vanflow.NewBase("l01"),
+		Role:       interRouter,
+		Status:     statusUp,
+		Peer:       ptrTo("ap02"),
+		Parent:     ptrTo("r01"),
+		DownCount:  ptrTo(uint64(17)),
+	})
 	metrics.Add(vanflow.LinkRecord{
 		BaseRecord: vanflow.NewBase("l01"),
 		Role:       interRouter,
 		Status:     statusDown,
 		Parent:     ptrTo("r01"),
 		DownCount:  ptrTo(uint64(0)),
 	})
+	assert.Equal(t, prom_testutil.ToFloat64(metrics.siteLinkErrors.WithLabelValues("s01", "inter-router")), 5.0)
+
+	// After reset, subsequent Updates still increment from the new baseline.
 	metrics.Update(
 		vanflow.LinkRecord{
 			BaseRecord: vanflow.NewBase("l01"),
@@ -225,9 +258,7 @@ func TestSiteLinkErrorMetrics(t *testing.T) {
 			DownCount:  ptrTo(uint64(1)),
 		},
 	)
-	// after delete, can still increment
-	assert.Equal(t, prom_testutil.CollectAndCount(reg, "skupper_site_link_errors_total"), 1)
-	assert.Equal(t, prom_testutil.ToFloat64(metrics.siteLinkErrors.WithLabelValues("s01", "inter-router")), 4.0)
+	assert.Equal(t, prom_testutil.ToFloat64(metrics.siteLinkErrors.WithLabelValues("s01", "inter-router")), 6.0)
 }
 
 func ptrTo[T any](s T) *T {