feat: enable map in monovertex (numaproj#2994)

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>
Signed-off-by: Yashash H L <yashashhl25@gmail.com>
Signed-off-by: Vigith Maurice <vigith@gmail.com>
Co-authored-by: Yashash H L <yashashhl25@gmail.com>
Co-authored-by: Vigith Maurice <vigith@gmail.com>

Author: 3 people

api/json-schema/schema.json

@@ -21263,6 +21263,9 @@
           },
           "type": "array"
         },
+        "udf": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.UDF"
+        },
         "updateStrategy": {
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.UpdateStrategy",
           "description": "The strategy to use to replace existing pods with new ones."

api/openapi-spec/swagger.json

@@ -21259,6 +21259,9 @@
             "$ref": "#/definitions/io.k8s.api.core.v1.Toleration"
           }
         },
+        "udf": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.UDF"
+        },
         "updateStrategy": {
           "description": "The strategy to use to replace existing pods with new ones.",
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.UpdateStrategy"

config/base/crds/full/numaflow.numaproj.io_monovertices.yaml

@@ -6049,6 +6049,21 @@ Refer to the Kubernetes API documentation for the fields of the
 
 <td>
 
+<code>udf</code></br> <em> <a href="#numaflow.numaproj.io/v1alpha1.UDF">
+UDF </a> </em>
+</td>
+
+<td>
+
+<em>(Optional)</em>
+</td>
+
+</tr>
+
+<tr>
+
+<td>
+
 <code>AbstractPodTemplate</code></br> <em>
 <a href="#numaflow.numaproj.io/v1alpha1.AbstractPodTemplate">
 AbstractPodTemplate </a> </em>
@@ -6546,6 +6561,21 @@ Description
 
 <td>
 
+<code>udf</code></br> <em> <a href="#numaflow.numaproj.io/v1alpha1.UDF">
+UDF </a> </em>
+</td>
+
+<td>
+
+<em>(Optional)</em>
+</td>
+
+</tr>
+
+<tr>
+
+<td>
+
 <code>AbstractPodTemplate</code></br> <em>
 <a href="#numaflow.numaproj.io/v1alpha1.AbstractPodTemplate">
 AbstractPodTemplate </a> </em>
@@ -13142,7 +13172,8 @@ UDF
 <p>
 
 (<em>Appears on:</em>
-<a href="#numaflow.numaproj.io/v1alpha1.AbstractVertex">AbstractVertex</a>)
+<a href="#numaflow.numaproj.io/v1alpha1.AbstractVertex">AbstractVertex</a>,
+<a href="#numaflow.numaproj.io/v1alpha1.MonoVertexSpec">MonoVertexSpec</a>)
 </p>
 
 <p>

config/install.yaml

@@ -482,38 +482,40 @@ type MonoVertexSpec struct {
 	Source   *Source `json:"source,omitempty" protobuf:"bytes,2,opt,name=source"`
 	Sink     *Sink   `json:"sink,omitempty" protobuf:"bytes,3,opt,name=sink"`
 	// +optional
-	AbstractPodTemplate `json:",inline" protobuf:"bytes,4,opt,name=abstractPodTemplate"`
+	UDF *UDF `json:"udf,omitempty" protobuf:"bytes,4,opt,name=udf"`
+	// +optional
+	AbstractPodTemplate `json:",inline" protobuf:"bytes,5,opt,name=abstractPodTemplate"`
 	// Container template for the main numa container.
 	// +optional
-	ContainerTemplate *ContainerTemplate `json:"containerTemplate,omitempty" protobuf:"bytes,5,opt,name=containerTemplate"`
+	ContainerTemplate *ContainerTemplate `json:"containerTemplate,omitempty" protobuf:"bytes,6,opt,name=containerTemplate"`
 	// +optional
 	// +patchStrategy=merge
 	// +patchMergeKey=name
-	Volumes []corev1.Volume `json:"volumes,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,6,rep,name=volumes"`
+	Volumes []corev1.Volume `json:"volumes,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,7,rep,name=volumes"`
 	// Limits define the limitations such as read batch size for the mono vertex.
 	// +optional
-	Limits *MonoVertexLimits `json:"limits,omitempty" protobuf:"bytes,7,opt,name=limits"`
+	Limits *MonoVertexLimits `json:"limits,omitempty" protobuf:"bytes,8,opt,name=limits"`
 	// Settings for autoscaling
 	// +optional
-	Scale Scale `json:"scale,omitempty" protobuf:"bytes,8,opt,name=scale"`
+	Scale Scale `json:"scale,omitempty" protobuf:"bytes,9,opt,name=scale"`
 	// List of customized init containers belonging to the pod.
 	// More info: https://kubernetes.io/docs/concepts/workloads/pods/init-containers/
 	// +optional
-	InitContainers []corev1.Container `json:"initContainers,omitempty" protobuf:"bytes,9,rep,name=initContainers"`
+	InitContainers []corev1.Container `json:"initContainers,omitempty" protobuf:"bytes,10,rep,name=initContainers"`
 	// List of customized sidecar containers belonging to the pod.
 	// +optional
-	Sidecars []corev1.Container `json:"sidecars,omitempty" protobuf:"bytes,10,rep,name=sidecars"`
+	Sidecars []corev1.Container `json:"sidecars,omitempty" protobuf:"bytes,11,rep,name=sidecars"`
 	// Template for the daemon service deployment.
 	// +optional
-	DaemonTemplate *DaemonTemplate `json:"daemonTemplate,omitempty" protobuf:"bytes,11,opt,name=daemonTemplate"`
+	DaemonTemplate *DaemonTemplate `json:"daemonTemplate,omitempty" protobuf:"bytes,12,opt,name=daemonTemplate"`
 	// The strategy to use to replace existing pods with new ones.
 	// +kubebuilder:default={"type": "RollingUpdate", "rollingUpdate": {"maxUnavailable": "25%"}}
 	// +optional
-	UpdateStrategy UpdateStrategy `json:"updateStrategy,omitempty" protobuf:"bytes,12,opt,name=updateStrategy"`
+	UpdateStrategy UpdateStrategy `json:"updateStrategy,omitempty" protobuf:"bytes,13,opt,name=updateStrategy"`
 	// Lifecycle defines the Lifecycle properties of a MonoVertex
 	// +kubebuilder:default={"desiredPhase": Running}
 	// +optional
-	Lifecycle MonoVertexLifecycle `json:"lifecycle,omitempty" protobuf:"bytes,13,opt,name=lifecycle"`
+	Lifecycle MonoVertexLifecycle `json:"lifecycle,omitempty" protobuf:"bytes,14,opt,name=lifecycle"`
 }
 
 func (mvspec MonoVertexSpec) DeepCopyWithoutReplicas() MonoVertexSpec {
@@ -537,6 +539,9 @@ func (mvspec MonoVertexSpec) buildContainers(req getContainerReq) ([]corev1.Cont
 	if mvspec.Source.UDSource != nil { // Only support UDSource for now.
 		sidecarContainers = append(sidecarContainers, mvspec.Source.getUDSourceContainer(req))
 	}
+	if mvspec.UDF != nil {
+		sidecarContainers = append(sidecarContainers, mvspec.UDF.getUDFContainer(req))
+	}
 	if mvspec.Source.UDTransformer != nil {
 		sidecarContainers = append(sidecarContainers, mvspec.Source.getUDTransformerContainer(req))
 	}