feat: publish to callback endpoint for tracking #1753

yhl25 · 2024-06-06T08:36:10Z

No description provided.

Signed-off-by: Yashash H L <[email protected]>

whynowy · 2024-06-10T21:50:46Z

pkg/apis/numaflow/v1alpha1/pipeline_types.go

+	// Callback defines the callback configuration for the messages processed by the pipeline.
+	// can be used for tracking the message processing status.
+	// +optional
+	Callback Callback `json:"callback,omitempty" protobuf:"bytes,9,opt,name=callback"`


As we discussed, can we use annotation at this moment?

whynowy · 2024-06-10T21:51:07Z

pkg/apis/numaflow/v1alpha1/pipeline_types.go

+type Callback struct {
+	// Enabled indicates whether callback is enabled for the pipeline.
+	// +optional
+	Enabled bool `json:"enabled,omitempty" protobuf:"varint,1,opt,name=enabled"`


This is not needed.

pkg/udf/rpc/grpc_map.go

Signed-off-by: Yashash H L <[email protected]>

codecov · 2024-06-14T07:19:00Z

Codecov Report

Attention: Patch coverage is 65.23077% with 113 lines in your changes missing coverage. Please review.

Project coverage is 56.82%. Comparing base (f69d830) to head (928bdfc).

Files	Patch %	Lines
pkg/shared/callback/callback.go	68.42%	24 Missing and 12 partials ⚠️
pkg/isb/serde.go	51.11%	10 Missing and 12 partials ⚠️
pkg/sources/forward/data_forward.go	66.66%	8 Missing and 2 partials ⚠️
pkg/sources/http/http.go	0.00%	6 Missing ⚠️
pkg/udf/forward/options.go	0.00%	5 Missing ⚠️
pkg/sinks/forward/options.go	0.00%	4 Missing ⚠️
pkg/sources/forward/options.go	0.00%	4 Missing ⚠️
pkg/sources/transformer/grpc_transformer.go	60.00%	4 Missing ⚠️
pkg/udf/rpc/grpc_map.go	60.00%	4 Missing ⚠️
pkg/udf/rpc/grpc_mapstream.go	60.00%	4 Missing ⚠️
... and 7 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1753      +/-   ##
==========================================
+ Coverage   56.63%   56.82%   +0.19%     
==========================================
  Files         216      218       +2     
  Lines       17350    17581     +231     
==========================================
+ Hits         9826     9991     +165     
- Misses       6692     6735      +43     
- Partials      832      855      +23

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

Signed-off-by: Yashash H L <[email protected]>

whynowy · 2024-06-17T05:20:35Z

pkg/shared/callback/callback.go

+		return fmt.Errorf("received non-OK response status: %s", resp.Status)
+	}
+
+	_ = resp.Body.Close()


Use defer before line 239.

whynowy · 2024-06-17T05:23:36Z

pkg/shared/callback/callback.go

+	})
+
+	if dOpts.callbackURL != "" {
+		client := &http.Client{


This has been existing in GetClient(), do not need it here?

we don't need it

whynowy · 2024-06-17T05:30:01Z

pkg/sources/source.go

@@ -74,7 +75,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 		healthCheckers           []metrics.HealthChecker
 		idleManager              wmb.IdleManager
 		pipelineName             = sp.VertexInstance.Vertex.Spec.PipelineName
-		vertexName               = sp.VertexInstance.Vertex.Name
+		vertexName               = sp.VertexInstance.Vertex.Spec.Name


Will existing pipelines get impacted with this fix?

pkg/isb/message.go

whynowy · 2024-06-17T05:40:03Z

pkg/isb/message.go

@@ -73,6 +74,23 @@ type Header struct {
 	Headers map[string]string
 }

+// MessageID is the message ID of the message which is used for exactly-once-semantics.
+type MessageID struct {


Will this change impact existing pipelines?

No, it should not impact.

Signed-off-by: Yashash H L <[email protected]>

Signed-off-by: Yashash H L <[email protected]> Co-authored-by: Derek Wang <[email protected]>

whynowy · 2024-06-18T20:13:08Z

pkg/sinks/sink.go

@@ -87,7 +92,7 @@ func (u *SinkProcessor) Start(ctx context.Context) error {
 		// create reader for each partition. Each partition is a group in redis
 		for index, bufferPartition := range u.VertexInstance.Vertex.OwnedBuffers() {
 			fromGroup := bufferPartition + "-group"
-			consumer := fmt.Sprintf("%s-%v", u.VertexInstance.Vertex.Name, u.VertexInstance.Replica)
+			consumer := fmt.Sprintf("%s-%v", vertexName, u.VertexInstance.Replica)


This is risky. vertex.spec.Name is not unique in one namespace, while vertex.Name is.

whynowy · 2024-06-18T20:19:10Z

pkg/udf/map_udf.go

-	enableMapUdfStream, err := u.VertexInstance.Vertex.MapUdfStreamEnabled()
-	if err != nil {
-		return fmt.Errorf("failed to parse UDF map streaming metadata, %w", err)
+	enableMapStreamStr := os.Getenv(dfv1.EnvMapStreaming)


Use pkg/shared/util.LookupEnvBoolOr()

Signed-off-by: Yashash H L <[email protected]>

Signed-off-by: Vigith Maurice <[email protected]>

Signed-off-by: Yashash H L <[email protected]>

Signed-off-by: Vigith Maurice <[email protected]>

vigith · 2024-06-18T15:06:46Z

examples/1-simple-pipeline.yaml

-          rpu: 5
-          duration: 1s
+        http: {}


don't we have to update the quickstart if we change the source?

This was added by mistake.

vigith · 2024-06-19T15:12:31Z

pkg/shared/callback/callback.go

+	}
+
+	// If the client is not in the cache, create a new one
+	client := &http.Client{


if the callback server (for this url) is restarted, will the http.Client auto-reconnect internally or will this client become invalid?

For the server restart case, If I am not wrong the client doesn't maintain a persistent connection to the server, so it simply creates a new connection when there is a post request, so ideally it should be fine. But still, the client caches the TCP connections. Let me run some tests, If I find any issues I will create a follow-up PR to fix it.

Signed-off-by: Vigith Maurice <[email protected]>

yhl25 and others added 9 commits May 9, 2024 10:15

callback

c666027

Signed-off-by: Yashash H L <[email protected]>

poc

86d5774

Signed-off-by: Yashash H L <[email protected]>

Merge branch 'main' of github.com:numaproj/numaflow into cb-poc

13fe80f

support for tracking

4bd3ad3

Signed-off-by: Yashash H L <[email protected]>

cb poc

98951b4

Signed-off-by: Yashash H L <[email protected]>

Merge branch 'main' of github.com:numaproj/numaflow into cb-poc

0efa750

Forward message headers in jetstream source (#1738)

62b1d04

lint

82f38c3

Signed-off-by: Yashash H L <[email protected]>

Merge branch 'main' into cb-poc

6249f98

whynowy reviewed Jun 10, 2024

View reviewed changes

yhl25 added 3 commits June 14, 2024 11:06

Merge branch 'main' of github.com:numaproj/numaflow into cb-poc

61dec39

fix tests

af58b3a

Signed-off-by: Yashash H L <[email protected]>

using annotations

d3cd083

Signed-off-by: Yashash H L <[email protected]>

yhl25 added 2 commits June 14, 2024 14:03

add unit tests

f3d7b3b

Signed-off-by: Yashash H L <[email protected]>

revert example spec

0b00a7d

Signed-off-by: Yashash H L <[email protected]>

whynowy reviewed Jun 17, 2024

View reviewed changes

yhl25 and others added 4 commits June 17, 2024 21:34

review comments

0cab01a

Signed-off-by: Yashash H L <[email protected]>

use env to store annotations for map stream and callback

5a0700a

Signed-off-by: Yashash H L <[email protected]>

Update pkg/isb/message.go

b776214

Signed-off-by: Yashash H L <[email protected]> Co-authored-by: Derek Wang <[email protected]>

Merge branch 'main' into cb-poc

cf7f2e1

vigith changed the title ~~feat: Support For Publishing Callbacks.~~ feat: publish to callback endpoint for tracking Jun 18, 2024

whynowy reviewed Jun 18, 2024

View reviewed changes

yhl25 added 3 commits June 19, 2024 09:50

review comments

a272d64

Signed-off-by: Yashash H L <[email protected]>

Merge branch 'cb-poc' of github.com:numaproj/numaflow into cb-poc

90af6a8

Merge branch 'main' of github.com:numaproj/numaflow into cb-poc

d539048

yhl25 marked this pull request as ready for review June 19, 2024 04:22

yhl25 requested a review from vigith as a code owner June 19, 2024 04:22

yhl25 requested a review from whynowy June 19, 2024 04:22

vigith and others added 5 commits June 18, 2024 22:45

chore: copyright

4584627

Signed-off-by: Vigith Maurice <[email protected]>

Merge remote-tracking branch 'refs/remotes/origin/cb-poc' into cb-poc

18d89cb

Signed-off-by: Vigith Maurice <[email protected]>

rephrase code comments

8f6f6d8

Signed-off-by: Yashash H L <[email protected]>

chore: minor refactor

9c9af21

Signed-off-by: Vigith Maurice <[email protected]>

chore: minor refactor

0d41583

Signed-off-by: Vigith Maurice <[email protected]>

vigith approved these changes Jun 19, 2024

View reviewed changes

whynowy approved these changes Jun 19, 2024

View reviewed changes

chore: revert example

928bdfc

Signed-off-by: Vigith Maurice <[email protected]>

yhl25 enabled auto-merge (squash) June 19, 2024 16:27

yhl25 merged commit 8da7c22 into main Jun 19, 2024
25 checks passed

yhl25 deleted the cb-poc branch June 19, 2024 16:42

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: publish to callback endpoint for tracking #1753

feat: publish to callback endpoint for tracking #1753

yhl25 commented Jun 6, 2024

whynowy Jun 10, 2024

whynowy Jun 10, 2024

codecov bot commented Jun 14, 2024 •

edited

Loading

whynowy Jun 17, 2024

whynowy Jun 17, 2024

yhl25 Jun 17, 2024

whynowy Jun 17, 2024

yhl25 Jun 17, 2024

whynowy Jun 17, 2024

yhl25 Jun 17, 2024 •

edited

Loading

whynowy Jun 18, 2024

whynowy Jun 18, 2024

vigith Jun 18, 2024

yhl25 Jun 19, 2024

vigith Jun 19, 2024

yhl25 Jun 19, 2024

feat: publish to callback endpoint for tracking #1753

feat: publish to callback endpoint for tracking #1753

Conversation

yhl25 commented Jun 6, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Jun 14, 2024 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yhl25 Jun 17, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Jun 14, 2024 •

edited

Loading

yhl25 Jun 17, 2024 •

edited

Loading