feat: Fallback Sink

pkg/apis/proto/sink/v1/sink.proto

@@ -35,15 +35,24 @@ message ReadyResponse {
   bool ready = 1;
 }
 
+/*
+  * Status is the status of the response.
+ */
+enum Status {
+  SUCCESS = 0;
+  FAILURE = 1;
+  FALLBACK = 2;
+}
+
 /**
  * SinkResponse is the individual response of each message written to the sink.
  */
 message SinkResponse {
   message Result {
     // id is the ID of the message, can be used to uniquely identify the message.
     string id = 1;
-    // success denotes the status of persisting to disk. if set to false, it means writing to sink for the message failed.
-    bool success = 2;
+    // status denotes the status of persisting to sink. It can be SUCCESS, FAILURE, or FALLBACK.
+    Status status = 2;
     // err_msg is the error message, set it if success is set to false.
     string err_msg = 3;
   }

pkg/sinker/examples/log/Makefile

@@ -10,7 +10,7 @@ image-push: build
 	docker buildx build -t "quay.io/numaio/numaflow-go/sink-log:${TAG}" --platform linux/amd64,linux/arm64 --target log . --push
 
 .PHONY: image
-imageimage: build
+image: build
 	docker build -t "quay.io/numaio/numaflow-go/sink-log:${TAG}" --target log .
 	@if [ "$(PUSH)" = "true" ]; then docker push "quay.io/numaio/numaflow-go/sink-log:${TAG}"; fi
 

pkg/sinker/examples/log/main.go

@@ -20,6 +20,8 @@ func (l *logSink) Sink(ctx context.Context, datumStreamCh <-chan sinksdk.Datum)
 		fmt.Println("User Defined Sink:", string(d.Value()))
 		id := d.ID()
 		result = result.Append(sinksdk.ResponseOK(id))
+		// if we are not able to write to sink and if we have a fallback sink configured
+		// we can use sinksdk.ResponseFallback(id)) to write the message to fallback sink
 	}
 	return result
 }

pkg/sinker/options.go

@@ -1,5 +1,7 @@
 package sinker
 
+import "os"
+
 type options struct {
 	sockAddr           string
 	maxMessageSize     int
@@ -10,10 +12,19 @@ type options struct {
 type Option func(*options)
 
 func defaultOptions() *options {
+	defaultPath := serverInfoFilePath
+	defaultAddress := address
+
+	// If the container type is fallback sink, then use the fallback sink address and path.
+	if os.Getenv(EnvUDContainerType) == UDContainerFallbackSink {
+		defaultPath = fbServerInfoFilePath
+		defaultAddress = fbAddress
+	}
+
 	return &options{
-		sockAddr:           address,
+		sockAddr:           defaultAddress,
 		maxMessageSize:     defaultMaxMessageSize,
-		serverInfoFilePath: serverInfoFilePath,
+		serverInfoFilePath: defaultPath,
 	}
 }
 

pkg/sinker/service.go

@@ -12,10 +12,14 @@ import (
 )
 
 const (
-	uds                   = "unix"
-	defaultMaxMessageSize = 1024 * 1024 * 64 // 64MB
-	address               = "/var/run/numaflow/sink.sock"
-	serverInfoFilePath    = "/var/run/numaflow/sinker-server-info"
+	uds                     = "unix"
+	defaultMaxMessageSize   = 1024 * 1024 * 64 // 64MB
+	address                 = "/var/run/numaflow/sink.sock"
+	fbAddress               = "/var/run/numaflow/fb-sink.sock"
+	serverInfoFilePath      = "/var/run/numaflow/sinker-server-info"
+	fbServerInfoFilePath    = "/var/run/numaflow/fb-sinker-server-info"
+	EnvUDContainerType      = "NUMAFLOW_UD_CONTAINER_TYPE"
+	UDContainerFallbackSink = "fb-udsink"
 )
 
 // handlerDatum implements the Datum interface and is used in the sink functions.
@@ -78,11 +82,23 @@ func (fs *Service) SinkFn(stream sinkpb.Sink_SinkFnServer) error {
 		defer wg.Done()
 		messages := fs.Sinker.Sink(ctx, datumStreamCh)
 		for _, msg := range messages {
-			resultList = append(resultList, &sinkpb.SinkResponse_Result{
-				Id:      msg.ID,
-				Success: msg.Success,
-				ErrMsg:  msg.Err,
-			})
+			if msg.Fallback {
+				resultList = append(resultList, &sinkpb.SinkResponse_Result{
+					Id:     msg.ID,
+					Status: sinkpb.Status_FALLBACK,
+				})
+			} else if msg.Success {
+				resultList = append(resultList, &sinkpb.SinkResponse_Result{
+					Id:     msg.ID,
+					Status: sinkpb.Status_SUCCESS,
+				})
+			} else {
+				resultList = append(resultList, &sinkpb.SinkResponse_Result{
+					Id:     msg.ID,
+					Status: sinkpb.Status_FAILURE,
+					ErrMsg: msg.Err,
+				})
+			}
 		}
 	}()