feat: controller changes for adding user-defined serving store (#2421)

Signed-off-by: Vigith Maurice <[email protected]> Signed-off-by: Sidhant Kohli <[email protected]> Co-authored-by: Sidhant Kohli <[email protected]>
numaproj · Feb 23, 2025 · 0b8c9c5 · 0b8c9c5
1 parent 1dc2b31
commit 0b8c9c5
Show file tree

Hide file tree

Showing 33 changed files with 2,058 additions and 871 deletions.
diff --git a/api/json-schema/schema.json b/api/json-schema/schema.json
@@ -19582,6 +19582,10 @@
           "description": "ServiceAccountName applied to the pod",
           "type": "string"
         },
+        "servingStoreName": {
+          "description": "Names of the serving store used in this vertex.",
+          "type": "string"
+        },
         "sideInputs": {
           "description": "Names of the side inputs used in this vertex.",
           "items": {
@@ -21538,6 +21542,10 @@
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.PipelineLimits",
           "description": "Limits define the limitations such as buffer read batch size for all the vertices of a pipeline, they could be overridden by each vertex's settings"
         },
+        "servingStore": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.ServingStore",
+          "description": "ServingStore defines the Serving Store for this pipeline."
+        },
         "sideInputs": {
           "description": "SideInputs defines the Side Inputs of a pipeline.",
           "items": {
@@ -21932,32 +21940,26 @@
         "service": {
           "description": "Whether to create a ClusterIP Service",
           "type": "boolean"
-        },
-        "store": {
-          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.ServingStore",
-          "description": "Persistent store for the callbacks for serving and tracking"
         }
       },
       "required": [
-        "msgIDHeaderKey",
-        "store"
+        "msgIDHeaderKey"
       ],
       "type": "object"
     },
     "io.numaproj.numaflow.v1alpha1.ServingStore": {
-      "description": "ServingStore to track and store data and metadata for tracking and serving.",
+      "description": "ServingStore defines information of a Serving Store used in a pipeline",
       "properties": {
-        "ttl": {
-          "$ref": "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.Duration",
-          "description": "TTL for the data in the store and tracker"
+        "container": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.Container"
         },
-        "url": {
-          "description": "URL of the persistent store to write the callbacks",
+        "name": {
           "type": "string"
         }
       },
       "required": [
-        "url"
+        "name",
+        "container"
       ],
       "type": "object"
     },
@@ -22555,6 +22557,10 @@
           "description": "ServiceAccountName applied to the pod",
           "type": "string"
         },
+        "servingStoreName": {
+          "description": "Names of the serving store used in this vertex.",
+          "type": "string"
+        },
         "sideInputs": {
           "description": "Names of the side inputs used in this vertex.",
           "items": {
@@ -22984,6 +22990,9 @@
         "resources": {
           "$ref": "#/definitions/io.k8s.api.core.v1.ResourceRequirements"
         },
+        "servingStore": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.ServingStore"
+        },
         "volumeMounts": {
           "items": {
             "$ref": "#/definitions/io.k8s.api.core.v1.VolumeMount"
@@ -22998,7 +23007,8 @@
         "image",
         "volumeMounts",
         "resources",
-        "executeRustBinary"
+        "executeRustBinary",
+        "servingStore"
       ],
       "type": "object"
     }

diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
@@ -19590,6 +19590,10 @@
           "description": "ServiceAccountName applied to the pod",
           "type": "string"
         },
+        "servingStoreName": {
+          "description": "Names of the serving store used in this vertex.",
+          "type": "string"
+        },
         "sideInputs": {
           "description": "Names of the side inputs used in this vertex.",
           "type": "array",
@@ -21525,6 +21529,10 @@
           "description": "Limits define the limitations such as buffer read batch size for all the vertices of a pipeline, they could be overridden by each vertex's settings",
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.PipelineLimits"
         },
+        "servingStore": {
+          "description": "ServingStore defines the Serving Store for this pipeline.",
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.ServingStore"
+        },
         "sideInputs": {
           "description": "SideInputs defines the Side Inputs of a pipeline.",
           "type": "array",
@@ -21909,8 +21917,7 @@
       "description": "ServingSource is the HTTP endpoint for Numaflow.",
       "type": "object",
       "required": [
-        "msgIDHeaderKey",
-        "store"
+        "msgIDHeaderKey"
       ],
       "properties": {
         "auth": {
@@ -21923,26 +21930,21 @@
         "service": {
           "description": "Whether to create a ClusterIP Service",
           "type": "boolean"
-        },
-        "store": {
-          "description": "Persistent store for the callbacks for serving and tracking",
-          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.ServingStore"
         }
       }
     },
     "io.numaproj.numaflow.v1alpha1.ServingStore": {
-      "description": "ServingStore to track and store data and metadata for tracking and serving.",
+      "description": "ServingStore defines information of a Serving Store used in a pipeline",
       "type": "object",
       "required": [
-        "url"
+        "name",
+        "container"
       ],
       "properties": {
-        "ttl": {
-          "description": "TTL for the data in the store and tracker",
-          "$ref": "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.Duration"
+        "container": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.Container"
         },
-        "url": {
-          "description": "URL of the persistent store to write the callbacks",
+        "name": {
           "type": "string"
         }
       }
@@ -22537,6 +22539,10 @@
           "description": "ServiceAccountName applied to the pod",
           "type": "string"
         },
+        "servingStoreName": {
+          "description": "Names of the serving store used in this vertex.",
+          "type": "string"
+        },
         "sideInputs": {
           "description": "Names of the side inputs used in this vertex.",
           "type": "array",
@@ -22947,7 +22953,8 @@
         "image",
         "volumeMounts",
         "resources",
-        "executeRustBinary"
+        "executeRustBinary",
+        "servingStore"
       ],
       "properties": {
         "env": {
@@ -22971,6 +22978,9 @@
         "resources": {
           "$ref": "#/definitions/io.k8s.api.core.v1.ResourceRequirements"
         },
+        "servingStore": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.ServingStore"
+        },
         "volumeMounts": {
           "type": "array",
           "items": {

diff --git a/config/base/crds/full/numaflow.numaproj.io_monovertices.yaml b/config/base/crds/full/numaflow.numaproj.io_monovertices.yaml
@@ -5154,18 +5154,8 @@ spec:
                         type: string
                       service:
                         type: boolean
-                      store:
-                        properties:
-                          ttl:
-                            type: string
-                          url:
-                            type: string
-                        required:
-                        - url
-                        type: object
                     required:
                     - msgIDHeaderKey
-                    - store
                     type: object
                   transformer:
                     properties: