feat: Jestream configurable DiscardNew policies (#3151)

Signed-off-by: Taleb Zeghmi <talebz@zillowgroup.com>
argoproj · Jul 12, 2024 · a6b45f1 · a6b45f1
1 parent 0e8ef63
commit a6b45f1
Show file tree

Hide file tree

Showing 11 changed files with 55 additions and 9 deletions.
diff --git a/api/event-bus.html b/api/event-bus.html
diff --git a/api/event-bus.md b/api/event-bus.md
diff --git a/api/jsonschema/schema.json b/api/jsonschema/schema.json
@@ -497,7 +497,7 @@
           "type": "array"
         },
         "streamConfig": {
-          "description": "Optional configuration for the streams to be created in this JetStream service, if specified, it will be merged with the default configuration in controller-config. It accepts a YAML format configuration, available fields include, \"maxBytes\", \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m).",
+          "description": "Optional configuration for the streams to be created in this JetStream service, if specified, it will be merged with the default configuration in controller-config. It accepts a YAML format configuration, available fields include, \"maxBytes\", \"maxMsgs\", \"maxAge\" (e.g. 72h), \"replicas\" (1, 3, 5), \"duplicates\" (e.g. 5m), \"retention\" (e.g. 0: Limits (default), 1: Interest, 2: WorkQueue), \"Discard\" (e.g. 0: DiscardOld (default), 1: DiscardNew).",
           "type": "string"
         },
         "tolerations": {

diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
diff --git a/eventbus/jetstream/base/jetstream.go b/eventbus/jetstream/base/jetstream.go
@@ -127,11 +127,25 @@ func (stream *Jetstream) CreateStream(conn *JetstreamConnection) error {
 		return err
 	}
 
+	v.SetDefault("retention", 0) // Limits
+	v.SetDefault("discard", 0)   // DiscardOld
+
+	retentionPolicy, err := intToRetentionPolicy(v.GetInt("retention"))
+	if err != nil {
+		stream.Logger.Errorf("invalid retention policy: %s, error: %v", retentionPolicy, err)
+		return err
+	}
+
+	discardPolicy, err := intToDiscardPolicy(v.GetInt("discard"))
+	if err != nil {
+		stream.Logger.Errorf("invalid discard policy: %s, error: %v", discardPolicy, err)
+		return err
+	}
 	streamConfig := nats.StreamConfig{
 		Name:       common.JetStreamStreamName,
 		Subjects:   []string{common.JetStreamStreamName + ".*.*"},
-		Retention:  nats.LimitsPolicy,
-		Discard:    nats.DiscardOld,
+		Retention:  retentionPolicy,
+		Discard:    discardPolicy,
 		MaxMsgs:    v.GetInt64("maxMsgs"),
 		MaxAge:     v.GetDuration("maxAge"),
 		MaxBytes:   v.GetInt64("maxBytes"),
@@ -158,3 +172,18 @@ func (stream *Jetstream) CreateStream(conn *JetstreamConnection) error {
 	stream.Logger.Infof("Created Jetstream stream '%s' for connection %+v", common.JetStreamStreamName, conn)
 	return nil
 }
+
+func intToRetentionPolicy(i int) (nats.RetentionPolicy, error) {
+	if i < 0 || i > int(nats.WorkQueuePolicy) {
+		// Handle invalid value, return a default value or panic
+		return -1, fmt.Errorf("invalid int for RetentionPolicy: %d", i)
+	}
+	return nats.RetentionPolicy(i), nil
+}
+
+func intToDiscardPolicy(i int) (nats.DiscardPolicy, error) {
+	if i < 0 || i > int(nats.DiscardNew) {
+		return -1, fmt.Errorf("invalid int for DiscardPolicy: %d", i)
+	}
+	return nats.DiscardPolicy(i), nil
+}
diff --git a/manifests/base/controller-manager/controller-config.yaml b/manifests/base/controller-manager/controller-config.yaml
@@ -26,6 +26,10 @@ data:
           maxBytes: -1
           replicas: 3
           duplicates: 300s
+          # 0: Limits, 1: Interest, 2: WorkQueue
+          retention: 0
+          # 0: DiscardOld, 1: DiscardNew
+          discard: 0
         versions:
         - version: latest
           natsImage: nats:2.10.10

diff --git a/manifests/install.yaml b/manifests/install.yaml
@@ -330,6 +330,10 @@ data:
           maxBytes: -1
           replicas: 3
           duplicates: 300s
+          # 0: Limits, 1: Interest, 2: WorkQueue
+          retention: 0
+          # 0: DiscardOld, 1: DiscardNew
+          discard: 0
         versions:
         - version: latest
           natsImage: nats:2.10.10

diff --git a/manifests/namespace-install.yaml b/manifests/namespace-install.yaml
@@ -250,6 +250,10 @@ data:
           maxBytes: -1
           replicas: 3
           duplicates: 300s
+          # 0: Limits, 1: Interest, 2: WorkQueue
+          retention: 0
+          # 0: DiscardOld, 1: DiscardNew
+          discard: 0
         versions:
         - version: latest
           natsImage: nats:2.10.10

diff --git a/pkg/apis/eventbus/v1alpha1/generated.proto b/pkg/apis/eventbus/v1alpha1/generated.proto
diff --git a/pkg/apis/eventbus/v1alpha1/jetstream_eventbus.go b/pkg/apis/eventbus/v1alpha1/jetstream_eventbus.go
@@ -79,7 +79,8 @@ type JetStreamBus struct {
 	// +optional
 	StartArgs []string `json:"startArgs,omitempty" protobuf:"bytes,17,rep,name=startArgs"`
 	// Optional configuration for the streams to be created in this JetStream service, if specified, it will be merged with the default configuration in controller-config.
-	// It accepts a YAML format configuration, available fields include, "maxBytes", "maxMsgs", "maxAge" (e.g. 72h), "replicas" (1, 3, 5), "duplicates" (e.g. 5m).
+	// It accepts a YAML format configuration, available fields include, "maxBytes", "maxMsgs", "maxAge" (e.g. 72h), "replicas" (1, 3, 5), "duplicates" (e.g. 5m),
+	// "retention" (e.g. 0: Limits (default), 1: Interest, 2: WorkQueue), "Discard" (e.g. 0: DiscardOld (default), 1: DiscardNew).
 	// +optional
 	StreamConfig *string `json:"streamConfig,omitempty" protobuf:"bytes,18,opt,name=streamConfig"`
 	// Maximum number of bytes in a message payload, 0 means unlimited. Defaults to 1MB

diff --git a/pkg/apis/eventbus/v1alpha1/openapi_generated.go b/pkg/apis/eventbus/v1alpha1/openapi_generated.go