fix: fix metrics

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## v0.0.31 (2021-06-04)
+
+ * [3dd39f1](https://github.com/argoproj/argo-workflows/commit/3dd39f16bf6d1104af3d2a3e4c23b98c22170639) fix(sidecar): updated to use fixed counters
+ * [6bef420](https://github.com/argoproj/argo-workflows/commit/6bef420f18718ca977365ff2ed0c46f213036ec6) fix: removed scrape annotations
+ * [3a7ab5c](https://github.com/argoproj/argo-workflows/commit/3a7ab5c238c351941fb6cbc7771da7479c80f607) ok
+
+### Contributors
+
+ * Alex Collins
+
 ## v0.0.30 (2021-06-03)
 
  * [c6763e5](https://github.com/argoproj/argo-workflows/commit/c6763e5b1f7f8112d0e7806b2344e78b15863810) feat(runner): Request Bearer token

api/v1alpha1/source_status.go

@@ -7,11 +7,22 @@ type SourceStatus struct {
 	Metrics     map[string]Metrics `json:"metrics,omitempty" protobuf:"bytes,4,rep,name=metrics"`
 }
 
-func (in *SourceStatus) AnyErrors() bool {
+func (in SourceStatus) GetTotal() uint64 {
+	var x uint64
 	for _, m := range in.Metrics {
-		if m.Errors > 0 {
-			return true
-		}
+		x += m.Total
 	}
-	return false
+	return x
+}
+
+func (in SourceStatus) GetErrors() uint64 {
+	var x uint64
+	for _, m := range in.Metrics {
+		x += m.Errors
+	}
+	return x
+}
+
+func (in SourceStatus) AnyErrors() bool {
+	return in.GetErrors() > 0
 }

api/v1alpha1/source_status_test.go

@@ -0,0 +1,21 @@
+package v1alpha1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSourceStatus_GetTotal(t *testing.T) {
+	assert.Equal(t, uint64(0), (SourceStatus{}).GetTotal())
+	assert.Equal(t, uint64(1), (SourceStatus{
+		Metrics: map[string]Metrics{"": {Total: 1}},
+	}).GetTotal())
+}
+
+func TestSourceStatus_GetErrors(t *testing.T) {
+	assert.Equal(t, uint64(0), (SourceStatus{}).GetErrors())
+	assert.Equal(t, uint64(1), (SourceStatus{
+		Metrics: map[string]Metrics{"": {Errors: 1}},
+	}).GetErrors())
+}

api/v1alpha1/source_statuses.go

@@ -23,6 +23,14 @@ 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()
+	}
+	return x
+}
+
 func (in SourceStatuses) IncErrors(name string, replica int, err error) {
 	x := in[name]
 	msg := err.Error()
@@ -42,11 +50,11 @@ func (in SourceStatuses) SetPending(name string, pending uint64) {
 	in[name] = x
 }
 
-func (in SourceStatuses) GetPending() int {
-	v := 0
+func (in SourceStatuses) GetPending() uint64 {
+	var v uint64
 	for _, s := range in {
 		if s.Pending != nil {
-			v += int(*s.Pending)
+			v += *s.Pending
 		}
 	}
 	return v
@@ -60,3 +68,7 @@ func (in SourceStatuses) AnyErrors() bool {
 	}
 	return false
 }
+
+func (in SourceStatuses) AnySunk() bool {
+	return in.GetTotal() > 0
+}

api/v1alpha1/source_statuses_test.go

@@ -101,9 +101,9 @@ func TestSourceStatuses_SetPending(t *testing.T) {
 }
 
 func TestSourceStatus_GetPending(t *testing.T) {
-	assert.Equal(t, 0, SourceStatuses{}.GetPending())
+	assert.Equal(t, uint64(0), SourceStatuses{}.GetPending())
 	v := uint64(1)
-	assert.Equal(t, 1, SourceStatuses{"0": {Pending: &v}}.GetPending())
+	assert.Equal(t, uint64(1), SourceStatuses{"0": {Pending: &v}}.GetPending())
 }
 
 func TestSourceStatus_AnyErrors(t *testing.T) {

api/v1alpha1/step_status.go

@@ -12,7 +12,7 @@ type StepStatus struct {
 	Selector       string         `json:"selector,omitempty" protobuf:"bytes,7,opt,name=selector"`
 	LastScaledAt   metav1.Time    `json:"lastScaledAt,omitempty" protobuf:"bytes,6,opt,name=lastScaledAt"`
 	SourceStatuses SourceStatuses `json:"sourceStatuses" protobuf:"bytes,3,rep,name=sourceStatuses"`
-	SinkStatues    SinkStatuses   `json:"sinkStatuses" protobuf:"bytes,4,rep,name=sinkStatuses"`
+	SinkStatues    SourceStatuses `json:"sinkStatuses" protobuf:"bytes,4,rep,name=sinkStatuses"`
 }
 
 func (m StepStatus) GetReplicas() int {

api/v1alpha1/step_types.go

@@ -46,7 +46,7 @@ func (in *Step) GetTargetReplicas(currentReplicas int, scalingDelay, peekDelay t
 	}
 
 	pending := in.Status.SourceStatuses.GetPending()
-	targetReplicas := in.Spec.CalculateReplicas(pending)
+	targetReplicas := in.Spec.CalculateReplicas(int(pending))
 
 	// do we need to peek? currentReplicas and targetReplicas must both be zero
 	if currentReplicas <= 0 && targetReplicas == 0 && time.Since(lastScaledAt) > peekDelay {