feat: Controller changes to support Serving as a builtin source #2357

BulkBeing · 2025-01-23T07:22:12Z

Tested the validation with below pipeline:

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: simple-pipeline
spec:
  vertices:
    - name: serving-in
      containerTemplate:
        imagePullPolicy: Never
      scale:
        min: 1
      source:
        serving:
          service: true
          msgIDHeaderKey: "X-Numaflow-Id"
          store:
            url: "redis://redis:6379"

    - name: compute-sum
      udf:
        container:
          # compute the sum
          image: quay.io/numaio/numaflow-go/reduce-sum:stable
        groupBy:
          window:
            fixed:
              length: 60s
          keyed: true
          storage:
            emptyDir: {}
      partitions: 2

    - name: serve-sink
      containerTemplate:
        imagePullPolicy: Never
      scale:
        min: 1
      sink:
        udsink:
          container:
            image: servesink:0.1
            imagePullPolicy: Never
            env:
              - name: NUMAFLOW_CALLBACK_URL_KEY
                value: "X-Numaflow-Callback-Url"
              - name: NUMAFLOW_MSG_ID_HEADER_KEY
                value: "X-Numaflow-Id"

  edges:
    - from: serving-in
      to: compute-sum
    - from: compute-sum
      to: serve-sink

➜  kubectl apply -f simple-pipeline-test.yaml
Error from server (BadRequest): error when creating "simple-pipeline-test.yaml": admission webhook "webhook.numaflow.numaproj.io" denied the request: pipeline has a Serving source "serving-in" and a reduce vertex "compute-sum". Reduce is not supported with Serving source

Signed-off-by: Sreekanth <[email protected]>

yhl25 · 2025-01-27T04:08:24Z

Add validation checks to not support reduce when serving source is configured? Also enable callback when serving source is configured so that the users can avoid setting the callback annotation?

codecov · 2025-01-27T23:45:25Z

Codecov Report

Attention: Patch coverage is 77.77778% with 18 lines in your changes missing coverage. Please review.

Project coverage is 69.68%. Comparing base (084be89) to head (5e7089a).
Report is 3 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/serving/src/config.rs	56.41%	17 Missing ⚠️
pkg/apis/numaflow/v1alpha1/vertex_types.go	96.77%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2357      +/-   ##
==========================================
- Coverage   69.70%   69.68%   -0.02%     
==========================================
  Files         361      361              
  Lines       49934    49984      +50     
==========================================
+ Hits        34805    34830      +25     
- Misses      14053    14084      +31     
+ Partials     1076     1070       -6

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

…ecified Signed-off-by: Sreekanth <[email protected]>

Signed-off-by: Sreekanth <[email protected]>

Makefile

whynowy · 2025-01-30T18:37:03Z

pkg/apis/numaflow/v1alpha1/vertex_types.go

+	for _, vtx := range req.PipelineSpec.Vertices {
+		if vtx.IsASource() && vtx.Source.Serving != nil {
+			commonEnvVars = append(commonEnvVars,
+				corev1.EnvVar{Name: EnvCallbackEnabled, Value: "true"},


It seems like we changed the logic from getting the information from annotations.

Are we expecting it's always true?

This PR removes the need for having an annotation to enable callbacks. Instead, if a serving source is used, the environment variable to enable callbacks will be set automatically. The EnvCallbackEnabled is checked to construct the callback http client, the callback only happens if the message contains the X-Numaflow-Callback-Url header.

pkg/apis/numaflow/v1alpha1/vertex_types.go

Signed-off-by: Sreekanth <[email protected]>

whynowy · 2025-01-31T07:09:55Z

pkg/apis/numaflow/v1alpha1/vertex_types.go

+		)
+
+		// if auth is configured, set the auth token in the environment
+		if servingSource.Auth != nil && servingSource.Auth.Token != nil {


This should have been mounted as a volume to the container, there's no need to add env vars to it.

whynowy · 2025-01-31T07:15:06Z

pkg/apis/numaflow/v1alpha1/vertex_types.go

@@ -343,13 +349,61 @@ func (v Vertex) GetPodSpec(req GetVertexPodSpecReq) (*corev1.PodSpec, error) {
 	}

 	if v.IsASource() && v.Spec.Source.Serving != nil {
-		servingContainer, err := v.getServingContainer(req)
+		servingSource := v.Spec.Source.Serving
+		servingSourceBytes, err := json.Marshal(servingSource)


There's already an env var EnvVertexObject, no need to do this.

… source spec as env variable Signed-off-by: Sreekanth <[email protected]>

kohlisid · 2025-01-31T16:59:57Z

pkg/reconciler/validator/pipeline_validate.go

+	}
+	if servingSource != nil {
+		for _, v := range pl.Spec.Vertices {
+			if v.UDF != nil && v.UDF.GroupBy != nil {


We have a utility function IsReduceUDF()

Signed-off-by: Sreekanth <[email protected]>

whynowy

@kohlisid - please review.

yhl25 · 2025-02-03T07:09:40Z

pkg/apis/numaflow/v1alpha1/vertex_types.go

+			// set the serving source stream name in the environment because the numa container will be reading from it
+			corev1.EnvVar{Name: EnvServingJetstreamStream, Value: req.ServingSourceStreamName},


we don't need js stream anymore right?

yhl25 · 2025-02-03T07:11:55Z

rust/serving/src/config.rs

+        #[derive(Deserialize)]
+        struct Source {
+            serving: Serving,
+        }
+        #[derive(Deserialize)]
+        struct Spec {
+            source: Source,
+        }
+
+        #[derive(Deserialize)]
+        struct VertexObject {
+            spec: Spec,
+        }


Is it possible to use the generated models?

Signed-off-by: Sreekanth <[email protected]>

BulkBeing added 3 commits January 23, 2025 12:49

Avoid adding second container when serving source is used

ab2ace0

Signed-off-by: Sreekanth <[email protected]>

Merge branch 'main' into serving-controller

fb3b365

Fix tests

0d464a2

Signed-off-by: Sreekanth <[email protected]>

Merge branch 'main' into serving-controller

940cf45

BulkBeing added 5 commits January 28, 2025 09:42

Set env vars to enable callback automatically if Serving source is sp…

a55bfb4

…ecified Signed-off-by: Sreekanth <[email protected]>

Validation to ensure reduce is not used with Serving source

ba51aac

Signed-off-by: Sreekanth <[email protected]>

Unit test for Monovertex changes

246ebd7

Signed-off-by: Sreekanth <[email protected]>

Merge branch 'main' into serving-controller

4c16244

Unit test for pipeline validation

0c16343

Signed-off-by: Sreekanth <[email protected]>

BulkBeing marked this pull request as ready for review January 30, 2025 10:20

BulkBeing requested review from whynowy and vigith as code owners January 30, 2025 10:20

vigith requested a review from yhl25 January 30, 2025 16:06

whynowy reviewed Jan 30, 2025

View reviewed changes

Remove setting env variable NUMAFLOW_CALLBACK_URL

96a1883

Signed-off-by: Sreekanth <[email protected]>

BulkBeing requested a review from whynowy January 31, 2025 05:18

Merge branch 'main' into serving-controller

5a54cb6

whynowy reviewed Jan 31, 2025

View reviewed changes

Parse vertex object spec env variable and remove the need for serving…

5114b4b

… source spec as env variable Signed-off-by: Sreekanth <[email protected]>

kohlisid reviewed Jan 31, 2025

View reviewed changes

BulkBeing added 2 commits February 3, 2025 09:05

Use secret volume instead of env variable for serving API auth token

84629cf

Signed-off-by: Sreekanth <[email protected]>

Fix unit tests

06fc738

Signed-off-by: Sreekanth <[email protected]>

BulkBeing requested review from kohlisid and whynowy February 3, 2025 04:20

whynowy approved these changes Feb 3, 2025

View reviewed changes

yhl25 reviewed Feb 3, 2025

View reviewed changes

BulkBeing added 2 commits February 3, 2025 20:25

Remove Serving specific jetstream stream env variable

ec8ccd0

Signed-off-by: Sreekanth <[email protected]>

Parse vertex object with model spec

5e7089a

Signed-off-by: Sreekanth <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Controller changes to support Serving as a builtin source #2357

feat: Controller changes to support Serving as a builtin source #2357

BulkBeing commented Jan 23, 2025 •

edited

Loading

yhl25 commented Jan 27, 2025 •

edited

Loading

codecov bot commented Jan 27, 2025 •

edited

Loading

whynowy Jan 30, 2025

BulkBeing Jan 31, 2025

whynowy Jan 31, 2025

whynowy Jan 31, 2025

kohlisid Jan 31, 2025

whynowy left a comment

yhl25 Feb 3, 2025

yhl25 Feb 3, 2025

		// set the serving source stream name in the environment because the numa container will be reading from it
		corev1.EnvVar{Name: EnvServingJetstreamStream, Value: req.ServingSourceStreamName},

feat: Controller changes to support Serving as a builtin source #2357

Are you sure you want to change the base?

feat: Controller changes to support Serving as a builtin source #2357

Conversation

BulkBeing commented Jan 23, 2025 • edited Loading

yhl25 commented Jan 27, 2025 • edited Loading

codecov bot commented Jan 27, 2025 • edited Loading

Codecov Report

whynowy Jan 30, 2025

Choose a reason for hiding this comment

BulkBeing Jan 31, 2025

Choose a reason for hiding this comment

whynowy Jan 31, 2025

Choose a reason for hiding this comment

whynowy Jan 31, 2025

Choose a reason for hiding this comment

kohlisid Jan 31, 2025

Choose a reason for hiding this comment

whynowy left a comment

Choose a reason for hiding this comment

yhl25 Feb 3, 2025

Choose a reason for hiding this comment

yhl25 Feb 3, 2025

Choose a reason for hiding this comment

BulkBeing commented Jan 23, 2025 •

edited

Loading

yhl25 commented Jan 27, 2025 •

edited

Loading

codecov bot commented Jan 27, 2025 •

edited

Loading