chore: jitter peek and jet stream de-dup (#454)

* chore: jitter peek and jet stream de-dup Signed-off-by: Derek Wang <[email protected]> * test case Signed-off-by: Derek Wang <[email protected]> * . Signed-off-by: Derek Wang <[email protected]>
argoproj-labs · Oct 15, 2021 · 0c3e48f · 0c3e48f
1 parent f81b33d
commit 0c3e48f
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 3 deletions.
diff --git a/manager/controllers/scaling/scaling.go b/manager/controllers/scaling/scaling.go
@@ -2,6 +2,7 @@ package scaling
 
 import (
 	"fmt"
+	"math/rand"
 	"time"
 
 	"github.com/antonmedv/expr"
@@ -100,6 +101,9 @@ func RequeueAfter(step dfv1.Step) (time.Duration, error) {
 	}); err != nil {
 		return 0, fmt.Errorf("failed to evaluate %q: %w", scale.ScalingDelay, err)
 	} else {
-		return scalingDelay, nil
+		s := rand.NewSource(time.Now().UnixNano())
+		r := rand.New(s)
+		ns := int(scalingDelay.Nanoseconds()) * (120 - r.Intn(40)) / 100
+		return time.Duration(ns), nil
 	}
 }
diff --git a/manager/controllers/scaling/scaling_test.go b/manager/controllers/scaling/scaling_test.go
@@ -90,6 +90,7 @@ func TestRequeueAfter(t *testing.T) {
 			},
 		})
 		assert.NoError(t, err)
-		assert.Equal(t, 4*time.Minute, requeueAfter)
+		assert.True(t, requeueAfter.Seconds() >= 4*time.Minute.Seconds()*0.8)
+		assert.True(t, requeueAfter.Seconds() <= 4*time.Minute.Seconds()*1.2)
 	})
 }
diff --git a/runner/sidecar/sink/jetstream/jetstream.go b/runner/sidecar/sink/jetstream/jetstream.go
@@ -42,7 +42,11 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, namespace,
 func (j jsSink) Sink(ctx context.Context, msg []byte) error {
 	span, _ := opentracing.StartSpanFromContext(ctx, fmt.Sprintf("jetstream-sink-%s", j.sinkName))
 	defer span.Finish()
-	if _, err := j.js.Publish(j.subject, msg); err != nil {
+	m, err := dfv1.MetaFromContext(ctx)
+	if err != nil {
+		return err
+	}
+	if _, err := j.js.Publish(j.subject, msg, nats.MsgId(m.ID)); err != nil {
 		return err
 	}
 	return nil