Add one Task queue and worker per AM Pipeline (#633)

Author: DanielCosme

internal/batch/service.go

@@ -143,7 +143,8 @@ func (s *batchImpl) InitProcessingWorkflow(ctx context.Context, req *collection.
 	req.ValidationConfig = validation.Config{}
 	// req.MetadataConfig = metadata.Config{}
 	tr := noop.NewTracerProvider().Tracer("")
-	err := collection.InitProcessingWorkflow(ctx, tr, s.cc, s.taskQueue, req)
+
+	err := collection.InitProcessingWorkflow(ctx, tr, s.cc, req)
 	if err != nil {
 		s.logger.Error(err, "Error initializing processing workflow.")
 	}

internal/batch/workflow.go

@@ -60,11 +60,6 @@ func NewBatchActivity(batchsvc Service) *BatchActivity {
 }
 
 func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput) error {
-	pipelines := []string{}
-	if params.PipelineName != "" {
-		pipelines = append(pipelines, params.PipelineName)
-	}
-
 	if params.Depth < 0 {
 		params.Depth = 0
 	}
@@ -97,7 +92,7 @@ func (a *BatchActivity) Execute(ctx context.Context, params BatchWorkflowInput)
 			BatchDir:           filepath.Dir(path),
 			Key:                entry.Name(),
 			IsDir:              entry.IsDir(),
-			PipelineNames:      pipelines,
+			PipelineName:       params.PipelineName,
 			ProcessingConfig:   params.ProcessingConfig,
 			CompletedDir:       params.CompletedDir,
 			RetentionPeriod:    params.RetentionPeriod,

internal/batch/workflow_test.go

@@ -37,14 +37,14 @@ func TestBatchActivityStartsProcessingWorkflows(t *testing.T) {
 		BatchDir:         batchPath,
 		Key:              "transfer1",
 		IsDir:            true,
-		PipelineNames:    []string{"am"},
+		PipelineName:     "am",
 		ProcessingConfig: "automated",
 	})
 	serviceMock.EXPECT().InitProcessingWorkflow(ctx, &collection.ProcessingWorkflowRequest{
 		BatchDir:         batchPath,
 		Key:              "transfer2",
 		IsDir:            true,
-		PipelineNames:    []string{"am"},
+		PipelineName:     "am",
 		ProcessingConfig: "automated",
 	})
 

internal/collection/goa.go

@@ -267,7 +267,7 @@ func (w *goaWrapper) Retry(ctx context.Context, payload *goacollection.RetryPayl
 	req.WorkflowID = *goacol.WorkflowID
 	req.CollectionID = goacol.ID
 	tr := noop.NewTracerProvider().Tracer("")
-	if err := InitProcessingWorkflow(ctx, tr, w.cc, w.taskQueue, req); err != nil {
+	if err := InitProcessingWorkflow(ctx, tr, w.cc, req); err != nil {
 		return fmt.Errorf("error starting the new workflow instance: %w", err)
 	}
 

internal/collection/workflow.go

@@ -12,6 +12,7 @@ import (
 	temporalsdk_client "go.temporal.io/sdk/client"
 
 	"github.com/artefactual-labs/enduro/internal/metadata"
+	"github.com/artefactual-labs/enduro/internal/pipeline"
 	"github.com/artefactual-labs/enduro/internal/validation"
 )
 
@@ -28,10 +29,7 @@ type ProcessingWorkflowRequest struct {
 	// Name of the watcher that received this blob.
 	WatcherName string
 
-	// Pipelines that are available for processing. The workflow will choose one
-	// (randomly picked for now). If the slice is empty, it will be
-	// automatically populated from the list of all configured pipelines.
-	PipelineNames []string
+	PipelineName string
 
 	// Period of time to schedule the deletion of the original blob from the
 	// watched data source. nil means no deletion.
@@ -72,7 +70,7 @@ type ProcessingWorkflowRequest struct {
 	MetadataConfig metadata.Config
 }
 
-func InitProcessingWorkflow(ctx context.Context, tr trace.Tracer, c temporalsdk_client.Client, taskQueue string, req *ProcessingWorkflowRequest) error {
+func InitProcessingWorkflow(ctx context.Context, tr trace.Tracer, c temporalsdk_client.Client, req *ProcessingWorkflowRequest) error {
 	_, span := tr.Start(ctx, "InitProcessingWorkflow")
 	defer span.End()
 
@@ -89,7 +87,7 @@ func InitProcessingWorkflow(ctx context.Context, tr trace.Tracer, c temporalsdk_
 
 	opts := temporalsdk_client.StartWorkflowOptions{
 		ID:                    req.WorkflowID,
-		TaskQueue:             taskQueue,
+		TaskQueue:             pipeline.TaskQueueName(req.PipelineName),
 		WorkflowIDReusePolicy: temporalsdk_api_enums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
 	}
 	_, err := c.ExecuteWorkflow(ctx, opts, ProcessingWorkflowName, req)

internal/pipeline/pipeline.go

@@ -57,32 +57,38 @@ type Pipeline struct {
 	status          string
 	statusUpdatedAt time.Time
 	statusLock      sync.RWMutex
+	TaskQueue       string
 }
 
 func NewPipeline(logger logr.Logger, config Config) (*Pipeline, error) {
 	config.TransferDir = expandPath(config.TransferDir)
 	config.ProcessingDir = expandPath(config.ProcessingDir)
 
 	p := &Pipeline{
-		logger: logger,
-		sem:    semaphore.NewWeighted(int64(config.Capacity)),
-		config: &config,
-		client: httpClient(),
+		logger:    logger,
+		sem:       semaphore.NewWeighted(int64(config.Capacity)),
+		config:    &config,
+		client:    httpClient(),
+		TaskQueue: TaskQueueName(config.Name),
 	}
 
 	if config.ID != "" {
 		p.ID = config.ID
 	}
 
-	// init() enriches our record by retrieving the UUID but we still return
-	// the the object in case of errors.
+	// init() enriches our record by retrieving the UUID, but we still return
+	// the object in case of errors.
 	if err := p.init(); err != nil {
 		return p, err
 	}
 
 	return p, nil
 }
 
+func TaskQueueName(name string) string {
+	return fmt.Sprintf("pipeline-%s", name)
+}
+
 // init connects with the pipeline to retrieve its identifier, unless one has
 // already been defined via configuration.
 func (p *Pipeline) init() error {

internal/workflow/processing.go

@@ -286,27 +286,7 @@ func (w *ProcessingWorkflow) Execute(ctx temporalsdk_workflow.Context, req *coll
 		}
 	}
 
-	// Randomly choose the pipeline from the list of names provided. If the
-	// list is empty then choose one from the list of all configured pipelines.
-	{
-		var pick string
-		if err := temporalsdk_workflow.SideEffect(ctx, func(ctx temporalsdk_workflow.Context) interface{} {
-			names := req.PipelineNames
-			if len(names) < 1 {
-				names = w.pipelineRegistry.Names()
-				if len(names) < 1 {
-					return ""
-				}
-			}
-			src := rand.NewSource(time.Now().UnixNano())
-			rnd := rand.New(src) // #nosec G404 -- not security sensitive.
-			return names[rnd.Intn(len(names))]
-		}).Get(&pick); err != nil {
-			return err
-		}
-
-		tinfo.PipelineName = pick
-	}
+	tinfo.PipelineName = req.PipelineName
 
 	// Load pipeline configuration and hooks.
 	{
@@ -706,3 +686,18 @@ func (w *ProcessingWorkflow) transfer(sessCtx temporalsdk_workflow.Context, tinf
 
 	return nil
 }
+
+// RandomPipeline will randomly choose the pipeline from the list of names provided. If the
+// list is empty then choose one from the list of all configured pipelines.
+func RandomPipeline(pipelineNames []string, registry *pipeline.Registry) string {
+	names := pipelineNames
+	if len(names) < 1 {
+		names = registry.Names()
+		if len(names) < 1 {
+			return ""
+		}
+	}
+	src := rand.NewSource(time.Now().UnixNano())
+	rnd := rand.New(src) // #nosec G404 -- not security sensitive.
+	return names[rnd.Intn(len(names))]
+}

internal/workflow/processing_test.go

@@ -71,7 +71,7 @@ func (s *ProcessingWorkflowTestSuite) TestParseErrorIsIgnored() {
 	s.env.ExecuteWorkflow(s.workflow.Execute, &collection.ProcessingWorkflowRequest{
 		CollectionID:     0,
 		WatcherName:      "watcher",
-		PipelineNames:    []string{"pipeline"},
+		PipelineName:     "pipeline",
 		RetentionPeriod:  &retentionPeriod,
 		StripTopLevelDir: true,
 		Key:              "key",
@@ -107,7 +107,7 @@ func (s *ProcessingWorkflowTestSuite) TestParseError() {
 	s.env.ExecuteWorkflow(s.workflow.Execute, &collection.ProcessingWorkflowRequest{
 		CollectionID:     0,
 		WatcherName:      "watcher",
-		PipelineNames:    []string{"pipeline"},
+		PipelineName:     "pipeline",
 		RetentionPeriod:  &retentionPeriod,
 		StripTopLevelDir: true,
 		Key:              "key",

main.go

@@ -218,10 +218,18 @@ func main() {
 								attribute.String("key", event.Key),
 								attribute.Bool("dir", event.IsDir),
 							)
-							logger.V(1).Info("Starting new workflow", "watcher", event.WatcherName, "bucket", event.Bucket, "key", event.Key, "dir", event.IsDir)
+							pipelineName := workflow.RandomPipeline(event.PipelineName, pipelineRegistry)
+							logger.V(1).Info(
+								"Starting new workflow",
+								"watcher", event.WatcherName,
+								"bucket", event.Bucket,
+								"key", event.Key,
+								"dir", event.IsDir,
+								"pipeline", pipelineName,
+							)
 							req := collection.ProcessingWorkflowRequest{
 								WatcherName:        event.WatcherName,
-								PipelineNames:      event.PipelineName,
+								PipelineName:       pipelineName,
 								RetentionPeriod:    event.RetentionPeriod,
 								CompletedDir:       event.CompletedDir,
 								StripTopLevelDir:   event.StripTopLevelDir,
@@ -233,7 +241,8 @@ func main() {
 								ValidationConfig:   config.Validation,
 								MetadataConfig:     config.Metadata,
 							}
-							if err := collection.InitProcessingWorkflow(ctx, tracer, temporalClient, config.Temporal.TaskQueue, &req); err != nil {
+
+							if err := collection.InitProcessingWorkflow(ctx, tracer, temporalClient, &req); err != nil {
 								logger.Error(err, "Error initializing processing workflow.")
 							}
 							span.End()
@@ -247,62 +256,36 @@ func main() {
 		}
 	}
 
-	// Workflow and activity worker.
-	{
-		h := hooks.NewHooks(config.Hooks)
-
-		done := make(chan struct{})
-		w := temporalsdk_worker.New(temporalClient, config.Temporal.TaskQueue, temporalsdk_worker.Options{
-			EnableSessionWorker:                    true,
-			MaxConcurrentSessionExecutionSize:      config.Worker.MaxConcurrentSessionExecutionSize,
-			MaxConcurrentWorkflowTaskExecutionSize: config.Worker.MaxConcurrentWorkflowsExecutionsSize,
-			MaxHeartbeatThrottleInterval:           config.Worker.HeartbeatThrottleInterval,
-			DefaultHeartbeatThrottleInterval:       config.Worker.HeartbeatThrottleInterval,
-		})
-		if err != nil {
-			logger.Error(err, "Error creating Temporal worker.")
-			os.Exit(1)
+	for _, p := range pipelineRegistry.List() {
+		if p.Status(ctx) == "active" {
+			logger.Info("Creating worker and task queue", "Pipeline", p.TaskQueue)
+		} else {
+			logger.Error(errors.New("Pipeline not active"), "Cannot create worker and task queue for inactive pipeline")
+			continue
 		}
-
-		w.RegisterWorkflowWithOptions(workflow.NewProcessingWorkflow(h, colsvc, pipelineRegistry, logger, config.Workflow).Execute, temporalsdk_workflow.RegisterOptions{Name: collection.ProcessingWorkflowName})
-		w.RegisterActivityWithOptions(activities.NewAcquirePipelineActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.AcquirePipelineActivityName})
-		w.RegisterActivityWithOptions(activities.NewDownloadActivity(h, pipelineRegistry, wsvc).Execute, temporalsdk_activity.RegisterOptions{Name: activities.DownloadActivityName})
-		w.RegisterActivityWithOptions(archive.NewExtractActivity(config.ExtractActivity).Execute, temporalsdk_activity.RegisterOptions{Name: archive.ExtractActivityName})
-		w.RegisterActivityWithOptions(activities.NewBundleActivity().Execute, temporalsdk_activity.RegisterOptions{Name: activities.BundleActivityName})
-		w.RegisterActivityWithOptions(activities.NewValidateTransferActivity().Execute, temporalsdk_activity.RegisterOptions{Name: activities.ValidateTransferActivityName})
-		w.RegisterActivityWithOptions(activities.NewTransferActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.TransferActivityName})
-		w.RegisterActivityWithOptions(activities.NewPollTransferActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.PollTransferActivityName})
-		w.RegisterActivityWithOptions(activities.NewPollIngestActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.PollIngestActivityName})
-		w.RegisterActivityWithOptions(activities.NewCleanUpActivity().Execute, temporalsdk_activity.RegisterOptions{Name: activities.CleanUpActivityName})
-		w.RegisterActivityWithOptions(activities.NewHidePackageActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.HidePackageActivityName})
-		w.RegisterActivityWithOptions(activities.NewDeleteOriginalActivity(wsvc).Execute, temporalsdk_activity.RegisterOptions{Name: activities.DeleteOriginalActivityName})
-		w.RegisterActivityWithOptions(activities.NewDisposeOriginalActivity(wsvc).Execute, temporalsdk_activity.RegisterOptions{Name: activities.DisposeOriginalActivityName})
-		w.RegisterActivityWithOptions(activities.NewPopulateMetadataActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.PopulateMetadataActivityName})
-
-		w.RegisterActivityWithOptions(workflow.NewAsyncCompletionActivity(colsvc).Execute, temporalsdk_activity.RegisterOptions{Name: workflow.AsyncCompletionActivityName})
-		w.RegisterActivityWithOptions(nha_activities.NewUpdateHARIActivity(h).Execute, temporalsdk_activity.RegisterOptions{Name: nha_activities.UpdateHARIActivityName})
-		w.RegisterActivityWithOptions(nha_activities.NewUpdateProductionSystemActivity(h).Execute, temporalsdk_activity.RegisterOptions{Name: nha_activities.UpdateProductionSystemActivityName})
-
-		w.RegisterWorkflowWithOptions(collection.BulkWorkflow, temporalsdk_workflow.RegisterOptions{Name: collection.BulkWorkflowName})
-		w.RegisterActivityWithOptions(collection.NewBulkActivity(colsvc).Execute, temporalsdk_activity.RegisterOptions{Name: collection.BulkActivityName})
-
-		w.RegisterWorkflowWithOptions(batch.BatchWorkflow, temporalsdk_workflow.RegisterOptions{Name: batch.BatchWorkflowName})
-		w.RegisterActivityWithOptions(batch.NewBatchActivity(batchsvc).Execute, temporalsdk_activity.RegisterOptions{Name: batch.BatchActivityName})
-
-		g.Add(
-			func() error {
-				if err := w.Start(); err != nil {
-					return err
-				}
-				<-done
-				return nil
-			},
-			func(err error) {
-				w.Stop()
-				close(done)
-			},
+		registerWorker(
+			p.TaskQueue,
+			temporalClient,
+			pipelineRegistry,
+			config,
+			colsvc,
+			wsvc,
+			batchsvc,
+			logger,
+			&g,
 		)
 	}
+	registerWorker(
+		config.Temporal.TaskQueue,
+		temporalClient,
+		pipelineRegistry,
+		config,
+		colsvc,
+		wsvc,
+		batchsvc,
+		logger,
+		&g,
+	)
 
 	// Observability server.
 	{
@@ -506,3 +489,66 @@ func initTracerProvider(ctx context.Context, logger logr.Logger, cfg TelemetryCo
 
 	return tp, shutdown, nil
 }
+
+func registerWorker(
+	taskQueue string,
+	temporalClient temporalsdk_client.Client,
+	pipelineRegistry *pipeline.Registry,
+	config configuration,
+	colsvc collection.Service,
+	wsvc watcher.Service,
+	batchsvc batch.Service,
+	logger logr.Logger,
+	g *run.Group,
+) {
+	// Workflow and activity worker.
+	h := hooks.NewHooks(config.Hooks)
+
+	done := make(chan struct{})
+	w := temporalsdk_worker.New(temporalClient, taskQueue, temporalsdk_worker.Options{
+		EnableSessionWorker:                    true,
+		MaxConcurrentSessionExecutionSize:      config.Worker.MaxConcurrentSessionExecutionSize,
+		MaxConcurrentWorkflowTaskExecutionSize: config.Worker.MaxConcurrentWorkflowsExecutionsSize,
+		MaxHeartbeatThrottleInterval:           config.Worker.HeartbeatThrottleInterval,
+		DefaultHeartbeatThrottleInterval:       config.Worker.HeartbeatThrottleInterval,
+	})
+
+	w.RegisterWorkflowWithOptions(workflow.NewProcessingWorkflow(h, colsvc, pipelineRegistry, logger, config.Workflow).Execute, temporalsdk_workflow.RegisterOptions{Name: collection.ProcessingWorkflowName})
+	w.RegisterActivityWithOptions(activities.NewAcquirePipelineActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.AcquirePipelineActivityName})
+	w.RegisterActivityWithOptions(activities.NewDownloadActivity(h, pipelineRegistry, wsvc).Execute, temporalsdk_activity.RegisterOptions{Name: activities.DownloadActivityName})
+	w.RegisterActivityWithOptions(archive.NewExtractActivity(config.ExtractActivity).Execute, temporalsdk_activity.RegisterOptions{Name: archive.ExtractActivityName})
+	w.RegisterActivityWithOptions(activities.NewBundleActivity().Execute, temporalsdk_activity.RegisterOptions{Name: activities.BundleActivityName})
+	w.RegisterActivityWithOptions(activities.NewValidateTransferActivity().Execute, temporalsdk_activity.RegisterOptions{Name: activities.ValidateTransferActivityName})
+	w.RegisterActivityWithOptions(activities.NewTransferActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.TransferActivityName})
+	w.RegisterActivityWithOptions(activities.NewPollTransferActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.PollTransferActivityName})
+	w.RegisterActivityWithOptions(activities.NewPollIngestActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.PollIngestActivityName})
+	w.RegisterActivityWithOptions(activities.NewCleanUpActivity().Execute, temporalsdk_activity.RegisterOptions{Name: activities.CleanUpActivityName})
+	w.RegisterActivityWithOptions(activities.NewHidePackageActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.HidePackageActivityName})
+	w.RegisterActivityWithOptions(activities.NewDeleteOriginalActivity(wsvc).Execute, temporalsdk_activity.RegisterOptions{Name: activities.DeleteOriginalActivityName})
+	w.RegisterActivityWithOptions(activities.NewDisposeOriginalActivity(wsvc).Execute, temporalsdk_activity.RegisterOptions{Name: activities.DisposeOriginalActivityName})
+	w.RegisterActivityWithOptions(activities.NewPopulateMetadataActivity(pipelineRegistry).Execute, temporalsdk_activity.RegisterOptions{Name: activities.PopulateMetadataActivityName})
+
+	w.RegisterActivityWithOptions(workflow.NewAsyncCompletionActivity(colsvc).Execute, temporalsdk_activity.RegisterOptions{Name: workflow.AsyncCompletionActivityName})
+	w.RegisterActivityWithOptions(nha_activities.NewUpdateHARIActivity(h).Execute, temporalsdk_activity.RegisterOptions{Name: nha_activities.UpdateHARIActivityName})
+	w.RegisterActivityWithOptions(nha_activities.NewUpdateProductionSystemActivity(h).Execute, temporalsdk_activity.RegisterOptions{Name: nha_activities.UpdateProductionSystemActivityName})
+
+	w.RegisterWorkflowWithOptions(collection.BulkWorkflow, temporalsdk_workflow.RegisterOptions{Name: collection.BulkWorkflowName})
+	w.RegisterActivityWithOptions(collection.NewBulkActivity(colsvc).Execute, temporalsdk_activity.RegisterOptions{Name: collection.BulkActivityName})
+
+	w.RegisterWorkflowWithOptions(batch.BatchWorkflow, temporalsdk_workflow.RegisterOptions{Name: batch.BatchWorkflowName})
+	w.RegisterActivityWithOptions(batch.NewBatchActivity(batchsvc).Execute, temporalsdk_activity.RegisterOptions{Name: batch.BatchActivityName})
+
+	g.Add(
+		func() error {
+			if err := w.Start(); err != nil {
+				return err
+			}
+			<-done
+			return nil
+		},
+		func(err error) {
+			w.Stop()
+			close(done)
+		},
+	)
+}