feat(processor): processor destination processing pipeline

Author: atzoum

app/apphandlers/embeddedAppHandler.go

@@ -289,6 +289,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 	if config.GetBoolVar(false, "Processor.DestinationIsolation.enabled") {
 		procRWHandle := jobsdb.NewForReadWrite(
 			"proc",
+			jobsdb.WithMultiConsumer(),
 			jobsdb.WithClearDB(options.ClearDB),
 			jobsdb.WithDSLimit(a.config.procDSLimit),
 			jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
@@ -299,7 +300,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
 			jobsdb.WithNumPartitions(partitionCount),
 		)
 		defer procRWHandle.Close()
-		procRWDB = jobsdb.NewPendingEventsJobsDB(procRWHandle, pendingEventsRegistry)
+		procRWDB = jobsdb.NewPendingEventsJobsDB(procRWHandle, pendingEventsRegistry, jobsdb.WithConsumerAsDestinationID())
 	}
 
 	var schemaForwarder schema_forwarder.Forwarder

app/apphandlers/processorAppHandler.go

@@ -273,6 +273,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 	if config.GetBoolVar(false, "Processor.DestinationIsolation.enabled") {
 		procRWHandle := jobsdb.NewForReadWrite(
 			"proc",
+			jobsdb.WithMultiConsumer(),
 			jobsdb.WithClearDB(options.ClearDB),
 			jobsdb.WithDSLimit(a.config.procDSLimit),
 			jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
@@ -283,7 +284,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
 			jobsdb.WithNumPartitions(partitionCount),
 		)
 		defer procRWHandle.Close()
-		procRWDB = jobsdb.NewPendingEventsJobsDB(procRWHandle, pendingEventsRegistry)
+		procRWDB = jobsdb.NewPendingEventsJobsDB(procRWHandle, pendingEventsRegistry, jobsdb.WithConsumerAsDestinationID())
 	}
 
 	var schemaForwarder schema_forwarder.Forwarder

processor/partition_worker.go processor/gw_partition_worker.goprocessor/partition_worker.go renamed to processor/gw_partition_worker.go

@@ -19,39 +19,39 @@ import (
 	"github.com/rudderlabs/rudder-server/utils/tracing"
 )
 
-type partitionWorker struct {
+type gwPartitionWorker struct {
 	partition    string
-	pipelines    []*pipelineWorker
+	pipelines    []*gwPipelineWorker
 	logger       logger.Logger
 	stats        *processorStats
 	tracer       *tracing.Tracer
 	handle       workerHandle
 	statsFactory stats.Stats
 }
 
-// newPartitionWorker creates a new worker for the specified partition
-func newPartitionWorker(partition string, h workerHandle, t stats.Tracer, statsFactory stats.Stats) *partitionWorker {
-	w := &partitionWorker{
+// newGwPartitionWorker creates a new worker for the specified partition
+func newGwPartitionWorker(partition string, h workerHandle, t stats.Tracer, statsFactory stats.Stats) *gwPartitionWorker {
+	w := &gwPartitionWorker{
 		partition:    partition,
 		logger:       h.logger().Child(partition),
 		stats:        h.stats(),
-		tracer:       tracing.New(t, tracing.WithNamePrefix("partitionWorker")),
+		tracer:       tracing.New(t, tracing.WithNamePrefix("gwPartitionWorker")),
 		handle:       h,
 		statsFactory: statsFactory,
 	}
 	// Create workers for each pipeline
 	pipelinesPerPartition := h.config().pipelinesPerPartition
-	w.pipelines = make([]*pipelineWorker, pipelinesPerPartition)
+	w.pipelines = make([]*gwPipelineWorker, pipelinesPerPartition)
 	for i := range pipelinesPerPartition {
-		w.pipelines[i] = newPipelineWorker(i, partition, h, tracing.New(t, tracing.WithNamePrefix("pipelineWorker")))
+		w.pipelines[i] = newGwPipelineWorker(i, partition, h, tracing.New(t, tracing.WithNamePrefix("gwPipelineWorker")))
 	}
 
 	return w
 }
 
 // Work processes jobs for the specified partition
 // Returns true if work was done, false otherwise
-func (w *partitionWorker) Work() bool {
+func (w *gwPartitionWorker) Work() bool {
 	// If pipelining is disabled, use the legacy job handling path
 	if !w.handle.config().enablePipelining {
 		return w.handle.handlePendingGatewayJobs(w.partition)
@@ -106,24 +106,24 @@ func (w *partitionWorker) Work() bool {
 }
 
 // SleepDurations returns the min and max sleep durations for the worker
-func (w *partitionWorker) SleepDurations() (min, max time.Duration) {
+func (w *gwPartitionWorker) SleepDurations() (min, max time.Duration) {
 	return w.handle.config().readLoopSleep.Load(), w.handle.config().maxLoopSleep.Load()
 }
 
 // Stop stops the worker and waits until all its goroutines have stopped
-func (w *partitionWorker) Stop() {
+func (w *gwPartitionWorker) Stop() {
 	var wg sync.WaitGroup
 	for _, pipeline := range w.pipelines {
 		wg.Add(1)
-		go func(p *pipelineWorker) {
+		go func(p *gwPipelineWorker) {
 			defer wg.Done()
 			p.Stop()
 		}(pipeline)
 	}
 	wg.Wait() // Wait for all stop operations to complete
 }
 
-func (w *partitionWorker) sendToPreProcess(ctx context.Context, jobsByPipeline map[int][]*jobsdb.JobT) error {
+func (w *gwPartitionWorker) sendToPreProcess(ctx context.Context, jobsByPipeline map[int][]*jobsdb.JobT) error {
 	spanTags := stats.Tags{"partition": w.partition}
 	_, span := w.tracer.Trace(ctx, "Work.sendToPreProcess", tracing.WithTraceTags(spanTags))
 	defer span.End()

processor/partition_worker_test.go processor/gw_partition_worker_test.goprocessor/partition_worker_test.go renamed to processor/gw_partition_worker_test.go

@@ -21,7 +21,7 @@ import (
 	"github.com/rudderlabs/rudder-server/utils/workerpool"
 )
 
-func TestWorkerPool(t *testing.T) {
+func TestGwWorkerPool(t *testing.T) {
 	run := func(t *testing.T, pipelining, limitsReached, shouldProcessMultipleSubJobs bool) {
 		wh := &mockWorkerHandle{
 			pipelining: pipelining,
@@ -61,7 +61,7 @@ func TestWorkerPool(t *testing.T) {
 
 		// create a worker pool
 		wp := workerpool.New(poolCtx, func(partition string) workerpool.Worker {
-			return newPartitionWorker(partition, wh, stats.NOP.NewTracer(""), stats.NOP)
+			return newGwPartitionWorker(partition, wh, stats.NOP.NewTracer(""), stats.NOP)
 		}, logger.NOP)
 
 		// start pinging for work for 100 partitions
@@ -116,7 +116,7 @@ func TestWorkerPool(t *testing.T) {
 	})
 }
 
-func TestWorkerPoolIdle(t *testing.T) {
+func TestGwWorkerPoolIdle(t *testing.T) {
 	wh := &mockWorkerHandle{
 		pipelining: true,
 		log:        logger.NewLogger(),
@@ -138,7 +138,7 @@ func TestWorkerPoolIdle(t *testing.T) {
 	// create a worker pool
 	wp := workerpool.New(poolCtx,
 		func(partition string) workerpool.Worker {
-			return newPartitionWorker(partition, wh, stats.NOP.NewTracer(""), stats.NOP)
+			return newGwPartitionWorker(partition, wh, stats.NOP.NewTracer(""), stats.NOP)
 		},
 		logger.NOP,
 		workerpool.WithCleanupPeriod(200*time.Millisecond),

processor/pipeline_worker.go processor/gw_pipeline_worker.goprocessor/pipeline_worker.go renamed to processor/gw_pipeline_worker.go

@@ -15,9 +15,9 @@ import (
 	"github.com/rudderlabs/rudder-server/utils/tracing"
 )
 
-// newPipelineWorker new worker which manages a single pipeline of a partition
-func newPipelineWorker(index int, partition string, h workerHandle, t *tracing.Tracer) *pipelineWorker {
-	w := &pipelineWorker{
+// newGwPipelineWorker new worker which manages a single pipeline of a partition
+func newGwPipelineWorker(index int, partition string, h workerHandle, t *tracing.Tracer) *gwPipelineWorker {
+	w := &gwPipelineWorker{
 		index:     index,
 		handle:    h,
 		logger:    h.logger().Withn(logger.NewStringField("partition", partition)),
@@ -45,12 +45,12 @@ func newPipelineWorker(index int, partition string, h workerHandle, t *tracing.T
 	return w
 }
 
-// pipelineWorker performs all processing steps of a partition's pipeline:
+// gwPipelineWorker performs all processing steps of a partition's pipeline:
 //  1. preprocess
 //  2. preTransform
 //  3. transform
 //  4. store
-type pipelineWorker struct {
+type gwPipelineWorker struct {
 	index     int
 	partition string
 	handle    workerHandle
@@ -73,7 +73,7 @@ type pipelineWorker struct {
 }
 
 // start launches the various worker goroutines for the pipelined processing
-func (w *pipelineWorker) start() {
+func (w *gwPipelineWorker) start() {
 	// Setup context cancellation handler
 	w.lifecycle.wg.Add(1)
 	rruntime.Go(func() {
@@ -230,7 +230,7 @@ func (w *pipelineWorker) start() {
 }
 
 // Stop gracefully terminates the worker by canceling its context and waiting for goroutines to finish
-func (w *pipelineWorker) Stop() {
+func (w *gwPipelineWorker) Stop() {
 	w.lifecycle.cancel()
 	w.lifecycle.wg.Wait()
 }