feat: Use gRPC bidirectional streaming for source transformer #2071

BulkBeing · 2024-09-18T12:47:24Z

Part of #2046

Related PRs:
numaproj/numaflow-go#147
numaproj/numaflow-rs#91

codecov · 2024-09-18T13:11:34Z

Codecov Report

Attention: Patch coverage is 72.25806% with 86 lines in your changes missing coverage. Please review.

Project coverage is 64.49%. Comparing base (ac7b33b) to head (0b9cede).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
pkg/sdkclient/sourcetransformer/client.go	38.23%	34 Missing and 8 partials ⚠️
rust/numaflow-core/src/transformer/user_defined.rs	81.45%	23 Missing ⚠️
pkg/isb/tracker/message_tracker.go	66.66%	8 Missing ⚠️
pkg/udf/rpc/grpc_batch_map.go	20.00%	4 Missing ⚠️
pkg/sources/transformer/grpc_transformer.go	94.82%	2 Missing and 1 partial ⚠️
pkg/sources/forward/applier/sourcetransformer.go	0.00%	2 Missing ⚠️
pkg/webhook/validator/validator.go	33.33%	1 Missing and 1 partial ⚠️
pkg/sources/source.go	0.00%	1 Missing ⚠️
pkg/udf/forward/forward.go	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2071      +/-   ##
==========================================
+ Coverage   64.44%   64.49%   +0.04%     
==========================================
  Files         324      324              
  Lines       30502    30579      +77     
==========================================
+ Hits        19658    19721      +63     
- Misses       9819     9820       +1     
- Partials     1025     1038      +13

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

vigith · 2024-09-27T00:38:29Z

pkg/sources/forward/applier/sourcetransformer.go

@@ -25,13 +25,13 @@ import (
 // SourceTransformApplier applies the source transform on the read message and gives back a new message. Any UserError will be retried here, while
 // InternalErr can be returned and could be retried by the callee.
 type SourceTransformApplier interface {
-	ApplyTransform(ctx context.Context, message *isb.ReadMessage) ([]*isb.WriteMessage, error)
+	ApplyTransform(ctx context.Context, messages []*isb.ReadMessage) ([]isb.ReadWriteMessagePair, error)


could you please help me understand the need for this signature change?

Currently, we run the below func in goroutines and call ApplyTransform within it:

concurrentApplyTransformer(ctx context.Context, readMessagePair <-chan *isb.ReadWriteMessagePair)

It retrieves message from the readMessagePair channel, apply transformer, and store the WriteMessages in the same element.

With bidrectional streaming, the concurrency is moved to the server side. It might be difficult to have a function signature that takes a message and returns the output since we are not sure of the result order from the stream. Instead, we send all messages, wait for all output and aggregate the result in the new implementation.

vigith · 2024-09-27T00:41:36Z

pkg/sources/forward/data_forward.go

-func (df *DataForward) concurrentApplyTransformer(ctx context.Context, readMessagePair <-chan *isb.ReadWriteMessagePair) {
-	for message := range readMessagePair {
-		start := time.Now()
-		metrics.SourceTransformerReadMessagesCount.With(map[string]string{


don't we need this metrics?

vigith · 2024-09-27T00:59:55Z

pkg/sources/transformer/grpc_transformer.go

-		}
-		taggedMessages = append(taggedMessages, taggedMessage)
+	if !tracker.IsEmpty() {
+		return nil, errors.New("transform response for all requests were not received from UDF")


should we dump the tracker state at this point if you think it could help triaging later on?

vigith · 2024-09-27T01:04:16Z

rust/monovertex/src/config.rs

-    /// # Returns
-    /// A string representing the `OnFailureStrategy` enum variant.
-    fn to_string(&self) -> String {
+impl Display for OnFailureStrategy {


thank you :)

vigith · 2024-09-27T01:07:56Z

rust/monovertex/src/transformer.rs

-    pub(crate) async fn new(client: SourceTransformClient<Channel>) -> Result<Self> {
-        Ok(Self { client })
+    pub(crate) async fn new(mut client: SourceTransformClient<Channel>) -> Result<Self> {
+        let (read_tx, read_rx) = mpsc::channel(config().batch_size as usize);


NIT: referencing config directly can make testing difficult if you want to change the value for testing.

vigith · 2024-09-27T01:10:17Z

rust/monovertex/src/transformer.rs

+        let transform_fn_with_timeout = tokio::time::timeout(
+            Duration::from_secs(3),
+            client.source_transform_fn(Request::new(read_stream)),
+        );
+        let Ok(resp_stream) = transform_fn_with_timeout.await else {
+            return Err(Error::TransformerError(
+                "connection to transformer gRPC server timed out".to_string(),
+            ));
+        };


only 3 sec timeout? will this whole thing become flaky in a busy system?

Why do we need timeout?

Since we create the stream at the beginning and we don't create multiple streams, this should succeed immediately. I added the timeout because the tests were stuck when there was some issue with the server (server started,but there was a mismatch in the protobuf files on the client and the server). I will remove the timeouts.

vigith · 2024-09-27T01:12:57Z

rust/monovertex/src/transformer.rs

        }

-        Ok(Some(messages))
+        let sender_task = tokio::time::timeout(Duration::from_millis(2), sender);


please document your thought on why you did this approach.

whynowy · 2024-09-27T06:15:04Z

pkg/udf/rpc/tracker.go

@@ -6,68 +6,68 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 )

-// tracker is used to store a key value pair for string and *isb.ReadMessage
+// Tracker is used to store a key value pair for string and *isb.ReadMessage


Move this file to pkg/isb/ and rename to message_tracker.go, also rename the struck name?

whynowy · 2024-09-27T06:47:03Z

rust/monovertex/src/transformer.rs

+
+use crate::config::config;
+use crate::error::{Error, Result};
+use crate::message::{Message, Offset};


I suppose this Message will be moved to a separate package that will be shared by both MonoVertex and Pipeline.

whynowy · 2024-09-27T06:47:41Z

rust/monovertex/src/transformer.rs

+            };
+            for (i, result) in resp.results.into_iter().enumerate() {
+                if result.tags.iter().any(|x| x == DROP) {
+                    continue;


TODO: metrics

KeranYang · 2024-09-27T15:48:12Z

pkg/sdkclient/sourcetransformer/client.go

+	for {
+		select {
+		case <-ctx.Done():
+			return nil, fmt.Errorf("waiting for transformer gRPC server to be ready: %w", context.Cause(ctx))


Suggested change

return nil, fmt.Errorf("waiting for transformer gRPC server to be ready: %w", context.Cause(ctx))

return nil, fmt.Errorf("failed waiting for transformer gRPC server to be ready: %w", ctx.Err())

context.Cause(ctx) returns same error obj as ctx.Err(). let's change it to be consistent with other sdk clients.

KeranYang · 2024-09-27T15:48:27Z

pkg/sdkclient/sourcetransformer/client.go

+
+	c.stream, err = c.grpcClt.SourceTransformFn(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("creating a gRPC stream for source transform: %w", err)


Suggested change

return nil, fmt.Errorf("creating a gRPC stream for source transform: %w", err)

return nil, fmt.Errorf("failed creating a gRPC stream for source transform: %w", err)

I've seen somewhere (Go blog / talks), to not include "failed" in an error message, since error type itself indicates that it failed. If the error gets wrapped again on the caller's side, it could result in a final message of below pattern:

Failed to create user: Failed to create new account: Failed to INSERT new account record: database not connected

Compared to:

Failed to create user: creating new account: INSERT new account record: database not connected

KeranYang · 2024-09-27T15:49:44Z

pkg/sdkclient/sourcetransformer/client.go

 	return c, nil
 }
+func doHandshake(stream transformpb.SourceTransform_SourceTransformFnClient) error {


Suggested change

func doHandshake(stream transformpb.SourceTransform_SourceTransformFnClient) error {

func doHandshake(stream transformpb.SourceTransform_SourceTransformFnClient) error {

KeranYang · 2024-09-27T15:50:08Z

pkg/sdkclient/sourcetransformer/client.go

+		},
+	}
+	if err := stream.Send(handshakeReq); err != nil {
+		return fmt.Errorf("sending handshake request for source tansform: %w", err)


Suggested change

return fmt.Errorf("sending handshake request for source tansform: %w", err)

return fmt.Errorf("failed sending handshake request for source tansform: %w", err)

KeranYang · 2024-09-27T15:51:13Z

pkg/sdkclient/sourcetransformer/client.go

+
+	handshakeResp, err := stream.Recv()
+	if err != nil {
+		return fmt.Errorf("receiving handshake response from source transform stream: %w", err)


Suggested change

return fmt.Errorf("receiving handshake response from source transform stream: %w", err)

return fmt.Errorf("failed receiving handshake response from source transform stream: %w", err)

KeranYang · 2024-09-27T17:17:23Z

pkg/sources/transformer/grpc_transformer.go

-			Tags: result.Tags,
-		}
-		taggedMessages = append(taggedMessages, taggedMessage)
+	if !tracker.IsEmpty() {


with the current implementation, this if statement should never hold true, right? because we only break the loop when messageCount is 0.

KeranYang · 2024-09-27T17:19:02Z

pkg/udf/rpc/tracker.go

 		m:    make(map[string]*isb.ReadMessage),
 		lock: sync.RWMutex{},
 	}
 }

-// addRequest add a new entry for a given message to the tracker.
+// AddRequest add a new entry for a given message to the Tracker.


naming: please be concise. if we are saving messages, use AddMessage instead of Request.

KeranYang · 2024-09-27T17:29:26Z

pkg/udf/rpc/tracker.go

 // to indicate if it does not exist
-func (t *tracker) getRequest(id string) (*isb.ReadMessage, bool) {
+func (t *Tracker) GetRequest(id string) (*isb.ReadMessage, bool) {


can we simply return *isb.RM and use nil to indicate message doesn't exist?

KeranYang · 2024-09-27T17:35:42Z

pkg/udf/rpc/tracker.go

-	return &tracker{
+// NewTracker initializes a new instance of a Tracker
+func NewTracker() *Tracker {
+	return &Tracker{
 		m:    make(map[string]*isb.ReadMessage),
 		lock: sync.RWMutex{},


I have a fundamental question. With bidirectional stream, are we trying to improve performance? If so, is having this tracker with RWMutex making introducing extra io latency?

Mutex operations (unless there is too much contention) are much faster than network calls (nanoseconds vs micro/milli seconds). Unary calls has some overhead. As per the docs:

Use streaming RPCs when handling a long-lived logical flow of data from the client-to-server, server-to-client, or in both directions. Streams can avoid continuous RPC initiation, which includes connection load balancing at the client-side, starting a new HTTP/2 request at the transport layer, and invoking a user-defined method handler on the server side.

KeranYang · 2024-09-27T17:45:00Z

pkg/udf/rpc/tracker.go

-// removeRequest will remove the entry for a given id
-func (t *tracker) removeRequest(id string) {
+// RemoveRequest will remove the entry for a given id
+func (t *Tracker) RemoveRequest(id string) {


Remove can be merged with Get as one method.

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

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

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

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

chore: fixes and minor refactor

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

yhl25

LGTM!

BulkBeing force-pushed the source-transformer-streaming branch from 00b3445 to 1b9a06d Compare September 23, 2024 07:36

vigith assigned BulkBeing Sep 25, 2024

BulkBeing mentioned this pull request Sep 26, 2024

feat: Use bidirectional streaming for Source Transformer numaproj/numaflow-go#147

Merged

BulkBeing marked this pull request as ready for review September 26, 2024 15:26

BulkBeing requested review from whynowy and vigith as code owners September 26, 2024 15:26

BulkBeing mentioned this pull request Sep 26, 2024

feat: Bidirectional streaming for source transformer numaproj/numaflow-rs#91

Merged

vigith requested review from KeranYang and yhl25 September 27, 2024 00:39

vigith reviewed Sep 27, 2024

View reviewed changes

whynowy reviewed Sep 27, 2024

View reviewed changes

KeranYang reviewed Sep 27, 2024

View reviewed changes

BulkBeing added 2 commits September 30, 2024 12:44

Merge new changes for rust

a04f9cd

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

Merge changes from main

8c96ca7

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

BulkBeing force-pushed the source-transformer-streaming branch from 8cad234 to 8c96ca7 Compare September 30, 2024 08:26

BulkBeing and others added 9 commits October 1, 2024 10:08

Update Go and Rust sdks to latest master

9792e58

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

dummy testing

b8d0498

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

working - background goroutine

b71a6eb

Change SourceTransformFn interface

f80918f

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

fix duplicates

0cc303f

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

fix issues and refactor

35ad8e6

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

Merge pull request #1 from numaproj/testing-transformer

25b3022

chore: fixes and minor refactor

Return context cancellation error

0edc2e5

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

Fix unit test

0b9cede

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

yhl25 approved these changes Oct 2, 2024

View reviewed changes

yhl25 merged commit e69551b into numaproj:main Oct 2, 2024
25 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Use gRPC bidirectional streaming for source transformer #2071

feat: Use gRPC bidirectional streaming for source transformer #2071

BulkBeing commented Sep 18, 2024 •

edited

Loading

codecov bot commented Sep 18, 2024 •

edited

Loading

vigith Sep 27, 2024

BulkBeing Sep 29, 2024

vigith Sep 27, 2024

vigith Sep 27, 2024

vigith Sep 27, 2024

vigith Sep 27, 2024

vigith Sep 27, 2024

yhl25 Sep 27, 2024

BulkBeing Sep 28, 2024

vigith Sep 27, 2024

whynowy Sep 27, 2024

whynowy Sep 27, 2024

whynowy Sep 27, 2024

KeranYang Sep 27, 2024

KeranYang Sep 27, 2024

BulkBeing Sep 28, 2024

KeranYang Sep 27, 2024

KeranYang Sep 27, 2024

KeranYang Sep 27, 2024

KeranYang Sep 27, 2024

KeranYang Sep 27, 2024

KeranYang Sep 27, 2024

KeranYang Sep 27, 2024

BulkBeing Sep 28, 2024

KeranYang Sep 27, 2024

yhl25 left a comment

	return nil, fmt.Errorf("waiting for transformer gRPC server to be ready: %w", context.Cause(ctx))
	return nil, fmt.Errorf("failed waiting for transformer gRPC server to be ready: %w", ctx.Err())

	return nil, fmt.Errorf("creating a gRPC stream for source transform: %w", err)
	return nil, fmt.Errorf("failed creating a gRPC stream for source transform: %w", err)

	func doHandshake(stream transformpb.SourceTransform_SourceTransformFnClient) error {

	func doHandshake(stream transformpb.SourceTransform_SourceTransformFnClient) error {

	return fmt.Errorf("sending handshake request for source tansform: %w", err)
	return fmt.Errorf("failed sending handshake request for source tansform: %w", err)

	return fmt.Errorf("receiving handshake response from source transform stream: %w", err)
	return fmt.Errorf("failed receiving handshake response from source transform stream: %w", err)

feat: Use gRPC bidirectional streaming for source transformer #2071

feat: Use gRPC bidirectional streaming for source transformer #2071

Conversation

BulkBeing commented Sep 18, 2024 • edited Loading

codecov bot commented Sep 18, 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

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

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

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 left a comment

Choose a reason for hiding this comment

BulkBeing commented Sep 18, 2024 •

edited

Loading

codecov bot commented Sep 18, 2024 •

edited

Loading