fix: short-term fix for pending metric not being updated (#346)

argoproj-labs · Sep 18, 2021 · 657062a · 657062a
1 parent 7425aa5
commit 657062a
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 8 deletions.
diff --git a/runner/sidecar/sidecar.go b/runner/sidecar/sidecar.go
@@ -35,7 +35,6 @@ var (
 	ready               = false // we are ready to serve HTTP requests, also updates pod status condition
 	kubernetesInterface kubernetes.Interface
 	secretInterface     corev1.SecretInterface
-	prePatchHooks       []func(ctx context.Context) error // hooks to run before patching
 	replica             int
 	step                dfv1.Step // this is updated on start, and then periodically as we update the status
 	stepName            string

diff --git a/runner/sidecar/sources.go b/runner/sidecar/sources.go
@@ -19,6 +19,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 )
 
 func connectSources(ctx context.Context, process func(context.Context, []byte) error) error {
@@ -158,17 +159,16 @@ func connectSources(ctx context.Context, process func(context.Context, []byte) e
 			return sources[sourceName].Close()
 		})
 		if x, ok := sources[sourceName].(source.HasPending); ok && leadReplica() {
-			logger.Info("adding pre-patch hook", "source", sourceName)
-			prePatchHooks = append(prePatchHooks, func(ctx context.Context) error {
+			logger.Info("starting pending loop", "source", sourceName, "updateInterval", updateInterval.String())
+			go wait.JitterUntilWithContext(ctx, func(ctx context.Context) {
 				logger.Info("getting pending", "source", sourceName)
 				if pending, err := x.GetPending(ctx); err != nil {
-					return err
+					logger.Error(err, "failed to get pending", "source", sourceName)
 				} else {
 					logger.Info("got pending", "source", sourceName, "pending", pending)
 					pendingGauge.WithLabelValues(sourceName).Set(float64(pending))
 				}
-				return nil
-			})
+			}, updateInterval, 1.2, true)
 		}
 	}
 	return nil

diff --git a/test/kafka-stress/kafka_stress_test.go b/test/kafka-stress/kafka_stress_test.go
@@ -54,6 +54,7 @@ func TestKafkaSourceStress(t *testing.T) {
 	defer StartTPSReporter(t, "main", prefix, n)()
 
 	go PumpKafkaTopic(topic, n, prefix, Params.MessageSize)
+	WaitForPending()
 	WaitForTotalSunkMessages(n, Params.Timeout)
 }
 
@@ -97,5 +98,6 @@ func TestKafkaSinkStress(t *testing.T) {
 	defer StartTPSReporter(t, "main", prefix, n)()
 
 	go PumpKafkaTopic(topic, n, prefix, Params.MessageSize)
+	WaitForPending()
 	WaitForTotalSunkMessages(n, Params.Timeout)
 }
diff --git a/test/metric.go b/test/metric.go
@@ -14,8 +14,8 @@ import (
 	"github.com/prometheus/common/expfmt"
 )
 
-func WaitForNothingPending() {
-	ExpectMetric("pending", Eq(0))
+func WaitForPending() {
+	ExpectMetric("pending", Gt(0))
 }
 
 func WaitForTotalSourceMessages(v int) {

diff --git a/test/stan-stress/stan_stress_test.go b/test/stan-stress/stan_stress_test.go
@@ -53,6 +53,7 @@ func TestStanSourceStress(t *testing.T) {
 	defer StartTPSReporter(t, "main", prefix, n)()
 
 	go PumpSTANSubject(longSubject, n, prefix, Params.MessageSize)
+	WaitForPending()
 	WaitForTotalSunkMessages(n, Params.Timeout)
 }
 
@@ -91,5 +92,6 @@ func TestStanSinkStress(t *testing.T) {
 	defer StartTPSReporter(t, "main", prefix, n)()
 
 	go PumpSTANSubject(longSubject, n, prefix, Params.MessageSize)
+	WaitForPending()
 	WaitForTotalSunkMessages(n, Params.Timeout)
 }