feat: snowpipe streaming

.github/workflows/tests.yaml

@@ -122,7 +122,7 @@ jobs:
           go-version-file: 'go.mod'
       - run: go version
       - run: go mod download # Not required, used to segregate module download vs test times
-      - run: make test exclude="/rudder-server/(jobsdb|integration_test|processor|regulation-worker|router|services|suppression-backup-service|warehouse)"
+      - run: FORCE_RUN_INTEGRATION_TESTS=true make test exclude="/rudder-server/(jobsdb|integration_test|processor|regulation-worker|router|services|suppression-backup-service|warehouse)"
       - name: Upload coverage report
         uses: actions/upload-artifact@v4
         with:
@@ -145,6 +145,7 @@ jobs:
           - integration_test/tracing
           - integration_test/backendconfigunavailability
           - integration_test/trackedusersreporting
+          - integration_test/snowpipestreaming
           - processor
           - regulation-worker
           - router
@@ -186,7 +187,8 @@ jobs:
           TEST_KAFKA_AZURE_EVENT_HUBS_CLOUD_CONNECTION_STRING: ${{ secrets.TEST_KAFKA_AZURE_EVENT_HUBS_CLOUD_CONNECTION_STRING }}
           TEST_S3_DATALAKE_CREDENTIALS: ${{ secrets.TEST_S3_DATALAKE_CREDENTIALS }}
           BIGQUERY_INTEGRATION_TEST_CREDENTIALS: ${{ secrets.BIGQUERY_INTEGRATION_TEST_CREDENTIALS }}
-        run: make test exclude="${{ matrix.exclude }}" package=${{ matrix.package }}
+          SNOWPIPE_STREAMING_KEYPAIR_UNENCRYPTED_INTEGRATION_TEST_CREDENTIALS: ${{ secrets.SNOWPIPE_STREAMING_KEYPAIR_UNENCRYPTED_INTEGRATION_TEST_CREDENTIALS }}
+        run: FORCE_RUN_INTEGRATION_TESTS=true make test exclude="${{ matrix.exclude }}" package=${{ matrix.package }}
       - name: Sanitize name for Artifact
         run: |
           name=$(echo -n "${{ matrix.package }}" | sed -e 's/[ \t:\/\\"<>|*?]/-/g' -e 's/--*/-/g')

gateway/handle.go

@@ -42,6 +42,7 @@ import (
 	sourcedebugger "github.com/rudderlabs/rudder-server/services/debugger/source"
 	"github.com/rudderlabs/rudder-server/services/rsources"
 	"github.com/rudderlabs/rudder-server/utils/misc"
+	"github.com/rudderlabs/rudder-server/utils/timeutil"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
 
@@ -488,7 +489,7 @@ func (gw *Handle) getJobDataFromRequest(req *webRequestT) (jobData *jobFromReq,
 			}
 			receivedAt, ok := userEvent.events[0]["receivedAt"].(string)
 			if !ok || !arctx.ReplaySource {
-				receivedAt = time.Now().Format(misc.RFC3339Milli)
+				receivedAt = timeutil.Now().Format(misc.RFC3339Milli)
 			}
 			singularEventBatch := SingularEventBatch{
 				Batch:      userEvent.events,

integration_test/snowpipestreaming/testdata/docker-compose.rudder-snowpipe-clients.yml

@@ -0,0 +1,11 @@
+version: "3.9"
+
+services:
+  rudder-snowpipe-clients:
+    image: "rudderstack/rudder-snowpipe-clients:develop"
+    ports:
+      - "9078"
+    healthcheck:
+      test: wget --no-verbose --tries=1 --spider http://localhost:9078/health || exit 1
+      interval: 1s
+      retries: 25

integration_test/snowpipestreaming/testdata/docker-compose.rudder-transformer.yml

@@ -0,0 +1,11 @@
+version: "3.9"
+
+services:
+  transformer:
+    image: "rudderstack/develop-rudder-transformer:fix.snowpipe-streaming-users"
+    ports:
+      - "9090:9090"
+    healthcheck:
+      test: wget --no-verbose --tries=1 --spider http://0.0.0.0:9090/health || exit 1
+      interval: 1s
+      retries: 25

processor/transformer/transformer.go

@@ -537,6 +537,9 @@ func (trans *handle) destTransformURL(destType string) string {
 			return destinationEndPoint + "?" + whSchemaVersionQueryParam
 		}
 	}
+	if destType == warehouseutils.SnowpipeStreaming {
+		return destinationEndPoint + "?" + fmt.Sprintf("whSchemaVersion=%s&whIDResolve=%v", trans.conf.GetString("Warehouse.schemaVersion", "v1"), warehouseutils.IDResolutionEnabled())
+	}
 	return destinationEndPoint
 }
 

router/batchrouter/asyncdestinationmanager/common/utils.go

@@ -3,7 +3,7 @@ package common
 import "slices"
 
 var (
-	asyncDestinations = []string{"MARKETO_BULK_UPLOAD", "BINGADS_AUDIENCE", "ELOQUA", "YANDEX_METRICA_OFFLINE_EVENTS", "BINGADS_OFFLINE_CONVERSIONS", "KLAVIYO_BULK_UPLOAD", "LYTICS_BULK_UPLOAD"}
+	asyncDestinations = []string{"MARKETO_BULK_UPLOAD", "BINGADS_AUDIENCE", "ELOQUA", "YANDEX_METRICA_OFFLINE_EVENTS", "BINGADS_OFFLINE_CONVERSIONS", "KLAVIYO_BULK_UPLOAD", "LYTICS_BULK_UPLOAD", "SNOWPIPE_STREAMING"}
 	sftpDestinations  = []string{"SFTP"}
 )
 

router/batchrouter/asyncdestinationmanager/manager.go

@@ -16,6 +16,7 @@ import (
 	lyticsBulkUpload "github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/lytics_bulk_upload"
 	marketobulkupload "github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/marketo-bulk-upload"
 	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/sftp"
+	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/snowpipestreaming"
 	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/yandexmetrica"
 )
 
@@ -41,6 +42,8 @@ func newRegularManager(
 		return klaviyobulkupload.NewManager(logger, statsFactory, destination)
 	case "LYTICS_BULK_UPLOAD":
 		return lyticsBulkUpload.NewManager(logger, statsFactory, destination)
+	case "SNOWPIPE_STREAMING":
+		return snowpipestreaming.New(conf, logger, statsFactory, destination), nil
 	}
 	return nil, errors.New("invalid destination type")
 }

router/batchrouter/asyncdestinationmanager/snowpipestreaming/apiadapter.go

@@ -0,0 +1,112 @@
+package snowpipestreaming
+
+import (
+	"context"
+	"strconv"
+
+	"github.com/rudderlabs/rudder-go-kit/logger"
+	"github.com/rudderlabs/rudder-go-kit/stats"
+
+	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
+	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/snowpipestreaming/internal/model"
+)
+
+func newApiAdapter(
+	logger logger.Logger,
+	statsFactory stats.Stats,
+	api api,
+	destination *backendconfig.DestinationT,
+) api {
+	return &apiAdapter{
+		logger:       logger,
+		statsFactory: statsFactory,
+		destination:  destination,
+		api:          api,
+	}
+}
+
+func (a *apiAdapter) defaultTags() stats.Tags {
+	return stats.Tags{
+		"module":        "batch_router",
+		"workspaceId":   a.destination.WorkspaceID,
+		"destType":      a.destination.DestinationDefinition.Name,
+		"destinationId": a.destination.ID,
+	}
+}
+
+func (a *apiAdapter) CreateChannel(ctx context.Context, req *model.CreateChannelRequest) (*model.ChannelResponse, error) {
+	a.logger.Infon("Creating channel",
+		logger.NewStringField("rudderIdentifier", req.RudderIdentifier),
+		logger.NewStringField("partition", req.Partition),
+		logger.NewStringField("database", req.TableConfig.Database),
+		logger.NewStringField("namespace", req.TableConfig.Schema),
+		logger.NewStringField("table", req.TableConfig.Table),
+	)
+	tags := a.defaultTags()
+	tags["api"] = "create_channel"
+
+	responseTimeStat := a.statsFactory.NewTaggedStat("snowpipe_streaming_api_response_time", stats.TimerType, tags)
+	defer responseTimeStat.RecordDuration()()
+
+	resp, err := a.api.CreateChannel(ctx, req)
+	if err != nil {
+		tags["status"] = "false"
+		return nil, err
+	}
+	tags["status"] = strconv.FormatBool(resp.Success)
+	tags["code"] = resp.Code
+	return resp, nil
+}
+
+func (a *apiAdapter) DeleteChannel(ctx context.Context, channelID string, sync bool) error {
+	a.logger.Infon("Deleting channel",
+		logger.NewStringField("channelId", channelID),
+		logger.NewBoolField("sync", sync),
+	)
+	tags := a.defaultTags()
+	tags["api"] = "delete_channel"
+
+	responseTimeStat := a.statsFactory.NewTaggedStat("snowpipe_streaming_api_response_time", stats.TimerType, tags)
+	defer responseTimeStat.RecordDuration()()
+
+	err := a.api.DeleteChannel(ctx, channelID, sync)
+	if err != nil {
+		tags["status"] = "false"
+		return err
+	}
+	tags["status"] = "true"
+	return nil
+}
+
+func (a *apiAdapter) Insert(ctx context.Context, channelID string, insertRequest *model.InsertRequest) (*model.InsertResponse, error) {
+	tags := a.defaultTags()
+	tags["api"] = "insert"
+
+	responseTimeStat := a.statsFactory.NewTaggedStat("snowpipe_streaming_api_response_time", stats.TimerType, tags)
+	defer responseTimeStat.RecordDuration()()
+
+	resp, err := a.api.Insert(ctx, channelID, insertRequest)
+	if err != nil {
+		tags["status"] = "false"
+		return nil, err
+	}
+	tags["status"] = strconv.FormatBool(resp.Success)
+	tags["code"] = resp.Code
+	return resp, nil
+}
+
+func (a *apiAdapter) Status(ctx context.Context, channelID string) (*model.StatusResponse, error) {
+	tags := a.defaultTags()
+	tags["api"] = "status"
+
+	responseTimeStat := a.statsFactory.NewTaggedStat("snowpipe_streaming_api_response_time", stats.TimerType, tags)
+	defer responseTimeStat.RecordDuration()()
+
+	resp, err := a.api.Status(ctx, channelID)
+	if err != nil {
+		tags["status"] = "false"
+		return nil, err
+	}
+	tags["status"] = strconv.FormatBool(resp.Success)
+	return resp, nil
+}