fix: total/error calcs

api/v1alpha1/source_status.go

@@ -1,12 +1,21 @@
 package v1alpha1
 
+import "strconv"
+
 type SourceStatus struct {
 	LastMessage *Message           `json:"lastMessage,omitempty" protobuf:"bytes,2,opt,name=lastMessage"`
 	LastError   *Error             `json:"lastError,omitempty" protobuf:"bytes,5,opt,name=lastError"`
 	Pending     *uint64            `json:"pending,omitempty" protobuf:"varint,3,opt,name=pending"`
 	Metrics     map[string]Metrics `json:"metrics,omitempty" protobuf:"bytes,4,rep,name=metrics"`
 }
 
+func (m SourceStatus) GetMetrics(replica int) Metrics {
+	if x, ok := m.Metrics[strconv.Itoa(replica)]; ok {
+		return x
+	}
+	return Metrics{}
+}
+
 func (m SourceStatus) GetPending() uint64 {
 	if m.Pending != nil {
 		return *m.Pending
@@ -30,6 +39,10 @@ func (in SourceStatus) GetErrors() uint64 {
 	return x
 }
 
+func (in SourceStatus) AnySunk() bool {
+	return in.GetTotal() > 0
+}
+
 func (in SourceStatus) AnyErrors() bool {
 	return in.GetErrors() > 0
 }

api/v1alpha1/source_statuses.go

@@ -23,12 +23,11 @@ func (in SourceStatuses) Set(name string, replica int, msg string, rate resource
 	in[name] = x
 }
 
-func (in SourceStatuses) GetTotal() uint64 {
-	var x uint64
-	for _, s := range in {
-		x += s.GetTotal()
+func (in SourceStatuses) Get(name string) SourceStatus {
+	if x, ok := in[name]; ok {
+		return x
 	}
-	return x
+	return SourceStatus{}
 }
 
 func (in SourceStatuses) IncErrors(name string, replica int, err error) {
@@ -70,5 +69,10 @@ func (in SourceStatuses) AnyErrors() bool {
 }
 
 func (in SourceStatuses) AnySunk() bool {
-	return in.GetTotal() > 0
+	for _, s := range in {
+		if s.AnySunk() {
+			return true
+		}
+	}
+	return false
 }

runner/sidecar/sidecar.go

@@ -292,27 +292,27 @@ func connectSources(ctx context.Context, toMain func([]byte) error) error {
 		rateCounter := ratecounter.NewRateCounter(updateInterval)
 
 		labels := map[string]string{"sourceName": source.Name, "replica": strconv.Itoa(replica)}
-		if replica == 0 {
+		if replica == 0 { // only replica zero updates this value, so it the only replica that can be accurate
 			promauto.NewCounterFunc(prometheus.CounterOpts{
 				Name:        "pending",
 				Subsystem:   "sources",
 				Help:        "Pending messages",
 				ConstLabels: labels,
-			}, func() float64 { return float64(status.SourceStatuses[sourceName].GetPending()) })
+			}, func() float64 { return float64(status.SourceStatuses.Get(sourceName).GetPending()) })
 		}
 		promauto.NewCounterFunc(prometheus.CounterOpts{
 			Name:        "total",
 			Subsystem:   "sources",
 			Help:        "Total number of messages",
 			ConstLabels: labels,
-		}, func() float64 { return float64(status.SourceStatuses[sourceName].GetTotal()) })
+		}, func() float64 { return float64(status.SourceStatuses.Get(sourceName).GetMetrics(replica).Total) })
 
 		promauto.NewCounterFunc(prometheus.CounterOpts{
 			Subsystem:   "sources",
 			Name:        "errors",
 			Help:        "Total number of errors",
 			ConstLabels: labels,
-		}, func() float64 { return float64(status.SourceStatuses[sourceName].GetErrors()) })
+		}, func() float64 { return float64(status.SourceStatuses.Get(sourceName).GetMetrics(replica).Errors) })
 
 		f := func(msg []byte) error {
 			rateCounter.Incr(1)