feat: configurable jetstream storage (#328)

Signed-off-by: jyu6 <[email protected]>
numaproj · Nov 18, 2022 · 049e5c6 · 049e5c6
1 parent ee5cd64
commit 049e5c6
Show file tree

Hide file tree

Showing 7 changed files with 35 additions and 3 deletions.
diff --git a/config/advanced-install/namespaced-controller-wo-crds.yaml b/config/advanced-install/namespaced-controller-wo-crds.yaml
@@ -136,6 +136,8 @@ data:
             maxMsgs: 2000000
             maxAge: 168h
             maxBytes: -1
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
             duplicates: 60s
           # The default consumer properties for the created streams
@@ -147,12 +149,16 @@ data:
             history: 5
             ttl: 3h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
           procBucket:
             maxValueSize: 0
             history: 1
             ttl: 72h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
         versions:
         - version: latest

diff --git a/config/base/controller-manager/numaflow-controller-config.yaml b/config/base/controller-manager/numaflow-controller-config.yaml
@@ -52,6 +52,8 @@ data:
             maxMsgs: 2000000
             maxAge: 168h
             maxBytes: -1
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
             duplicates: 60s
           # The default consumer properties for the created streams
@@ -63,12 +65,16 @@ data:
             history: 5
             ttl: 3h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
           procBucket:
             maxValueSize: 0
             history: 1
             ttl: 72h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
         versions:
         - version: latest

diff --git a/config/install.yaml b/config/install.yaml
@@ -10451,6 +10451,8 @@ data:
             maxMsgs: 2000000
             maxAge: 168h
             maxBytes: -1
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
             duplicates: 60s
           # The default consumer properties for the created streams
@@ -10462,12 +10464,16 @@ data:
             history: 5
             ttl: 3h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
           procBucket:
             maxValueSize: 0
             history: 1
             ttl: 72h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
         versions:
         - version: latest

diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
@@ -10362,6 +10362,8 @@ data:
             maxMsgs: 2000000
             maxAge: 168h
             maxBytes: -1
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
             duplicates: 60s
           # The default consumer properties for the created streams
@@ -10373,12 +10375,16 @@ data:
             history: 5
             ttl: 3h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
           procBucket:
             maxValueSize: 0
             history: 1
             ttl: 72h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
         versions:
         - version: latest

diff --git a/docs/inter-step-buffer-service.md b/docs/inter-step-buffer-service.md
@@ -111,6 +111,8 @@ bufferConfig: |
     maxMsgs: 50000
     maxAge: 168h
     maxBytes: -1
+    # 0: File, 1: Memory
+    storage: 0
     replicas: 3
     duplicates: 60s
   # The consumer properties for the created streams

diff --git a/docs/numaflow-controller-config.yaml b/docs/numaflow-controller-config.yaml
@@ -52,6 +52,8 @@ data:
             maxMsgs: 2000000
             maxAge: 168h
             maxBytes: -1
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
             duplicates: 60s
           # The default consumer properties for the created streams
@@ -63,12 +65,16 @@ data:
             history: 5
             ttl: 72h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
           procBucket:
             maxValueSize: 0
             history: 1
             ttl: 72h
             maxBytes: 0
+            # 0: File, 1: Memory
+            storage: 0
             replicas: 3
         versions:
         - version: latest

diff --git a/pkg/isbsvc/jetstream_service.go b/pkg/isbsvc/jetstream_service.go
@@ -100,7 +100,7 @@ func (jss *jetStreamSvc) CreateBuffers(ctx context.Context, buffers []dfv1.Buffe
 					MaxMsgs:    v.GetInt64("stream.maxMsgs"),
 					MaxAge:     v.GetDuration("stream.maxAge"),
 					MaxBytes:   v.GetInt64("stream.maxBytes"),
-					Storage:    nats.FileStorage,
+					Storage:    nats.StorageType(v.GetInt("stream.storage")),
 					Replicas:   v.GetInt("stream.replicas"),
 					Duplicates: v.GetDuration("stream.duplicates"), // No duplication in this period
 				}); err != nil {
@@ -133,7 +133,7 @@ func (jss *jetStreamSvc) CreateBuffers(ctx context.Context, buffers []dfv1.Buffe
 				History:      uint8(v.GetUint("otBucket.history")),
 				TTL:          v.GetDuration("otBucket.ttl"),
 				MaxBytes:     v.GetInt64("otBucket.maxBytes"),
-				Storage:      nats.FileStorage,
+				Storage:      nats.StorageType(v.GetInt("otBucket.storage")),
 				Replicas:     v.GetInt("otBucket.replicas"),
 				Placement:    nil,
 			}); err != nil {
@@ -153,7 +153,7 @@ func (jss *jetStreamSvc) CreateBuffers(ctx context.Context, buffers []dfv1.Buffe
 				History:      uint8(v.GetUint("procBucket.history")),
 				TTL:          v.GetDuration("procBucket.ttl"),
 				MaxBytes:     v.GetInt64("procBucket.maxBytes"),
-				Storage:      nats.FileStorage,
+				Storage:      nats.StorageType(v.GetInt("procBucket.storage")),
 				Replicas:     v.GetInt("procBucket.replicas"),
 				Placement:    nil,
 			}); err != nil {