feat: Jestream configurable DiscardNew policies (#3151)

Signed-off-by: Taleb Zeghmi <talebz@zillowgroup.com>

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": {

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
+}

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

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

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

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