JetStreamBus
|
(Optional)
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).
|
diff --git a/api/event-bus.md b/api/event-bus.md
index 1af65d8e46..00f622a537 100644
--- a/api/event-bus.md
+++ b/api/event-bus.md
@@ -622,7 +622,9 @@ 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).
+(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).
diff --git a/api/jsonschema/schema.json b/api/jsonschema/schema.json
index 42abf12d38..3e8ac84b1e 100644
--- 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
index ad0db5b788..e3e4fae04e 100644
--- a/api/openapi-spec/swagger.json
+++ b/api/openapi-spec/swagger.json
@@ -490,7 +490,7 @@
}
},
"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/eventbus/jetstream/base/jetstream.go b/eventbus/jetstream/base/jetstream.go
index 3087830fe1..6cd34cd169 100644
--- 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
index 119648be15..f7663eab70 100644
--- 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
index ad68a6ccf5..09499a1984 100644
--- 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
index 5c28b136a1..1649db8844 100644
--- 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
index d05cb3d67a..d786f3f951 100644
--- a/pkg/apis/eventbus/v1alpha1/generated.proto
+++ b/pkg/apis/eventbus/v1alpha1/generated.proto
@@ -191,7 +191,8 @@ message JetStreamBus {
repeated string startArgs = 17;
// 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
optional string streamConfig = 18;
diff --git a/pkg/apis/eventbus/v1alpha1/jetstream_eventbus.go b/pkg/apis/eventbus/v1alpha1/jetstream_eventbus.go
index 97cad48fe2..ddb261bb17 100644
--- 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
index e2427fb6d4..44465bc26c 100644
--- a/pkg/apis/eventbus/v1alpha1/openapi_generated.go
+++ b/pkg/apis/eventbus/v1alpha1/openapi_generated.go
@@ -443,7 +443,7 @@ func schema_pkg_apis_eventbus_v1alpha1_JetStreamBus(ref common.ReferenceCallback
},
"streamConfig": {
SchemaProps: spec.SchemaProps{
- 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{"string"},
Format: "",
},