fix: group all the non-iterator activities under TriggerFromAPI condi…

…tion
instill-ai · Nov 15, 2024 · 40dde1e · 40dde1e
1 parent e0db95c
commit 40dde1e
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 20 deletions.
diff --git a/pkg/worker/minioactivity.go b/pkg/worker/minioactivity.go
@@ -70,7 +70,7 @@ func (w *worker) UploadInputsToMinioActivity(ctx context.Context, param *UploadI
 
 func (w *worker) UploadRecipeToMinioActivity(ctx context.Context, param *UploadRecipeToMinioActivityParam) error {
 	log := w.log.With(zap.String("PipelineTriggerUID", param.PipelineTriggerID))
-	log.Info("UploadReceiptToMinioActivity started")
+	log.Info("UploadRecipeToMinioActivity started")
 
 	wfm, err := w.memoryStore.GetWorkflowMemory(ctx, param.PipelineTriggerID)
 	if err != nil {
@@ -116,7 +116,7 @@ func (w *worker) UploadRecipeToMinioActivity(ctx context.Context, param *UploadR
 		return err
 	}
 
-	log.Info("UploadReceiptToMinioActivity finished")
+	log.Info("UploadRecipeToMinioActivity finished")
 	return nil
 }
 

diff --git a/pkg/worker/workflow.go b/pkg/worker/workflow.go
@@ -250,26 +250,23 @@ func (w *worker) TriggerPipelineWorkflow(ctx workflow.Context, param *TriggerPip
 		}).Get(ctx, nil); err != nil {
 			return err
 		}
-	}
 
-	err := workflow.ExecuteActivity(minioCtx, w.UploadInputsToMinioActivity, &UploadInputsToMinioActivityParam{
-		PipelineTriggerID: param.SystemVariables.PipelineTriggerID,
-		ExpiryRuleTag:     param.SystemVariables.ExpiryRuleTag,
-	}).Get(ctx, nil)
-	if err != nil {
-		logger.Error("Failed to upload pipeline run input", zap.Error(err))
-	}
+		err := workflow.ExecuteActivity(minioCtx, w.UploadInputsToMinioActivity, &UploadInputsToMinioActivityParam{
+			PipelineTriggerID: param.SystemVariables.PipelineTriggerID,
+			ExpiryRuleTag:     param.SystemVariables.ExpiryRuleTag,
+		}).Get(ctx, nil)
+		if err != nil {
+			logger.Error("Failed to upload pipeline run input", zap.Error(err))
+		}
 
-	err = workflow.ExecuteActivity(minioCtx, w.UploadRecipeToMinioActivity, &UploadRecipeToMinioActivityParam{
-		PipelineTriggerID: param.SystemVariables.PipelineTriggerID,
-		ExpiryRuleTag:     param.SystemVariables.ExpiryRuleTag,
-	}).Get(ctx, nil)
-	if err != nil {
-		logger.Error("Failed to upload pipeline run recipe", zap.Error(err))
-	}
+		err = workflow.ExecuteActivity(minioCtx, w.UploadRecipeToMinioActivity, &UploadRecipeToMinioActivityParam{
+			PipelineTriggerID: param.SystemVariables.PipelineTriggerID,
+			ExpiryRuleTag:     param.SystemVariables.ExpiryRuleTag,
+		}).Get(ctx, nil)
+		if err != nil {
+			logger.Error("Failed to upload pipeline run recipe", zap.Error(err))
+		}
 
-	// TODO group everything under condition
-	if param.TriggerFromAPI {
 		if err := workflow.ExecuteActivity(ctx, w.InitComponentsActivity, &InitComponentsActivityParam{
 			WorkflowID:      workflowID,
 			SystemVariables: param.SystemVariables,
@@ -283,7 +280,7 @@ func (w *worker) TriggerPipelineWorkflow(ctx workflow.Context, param *TriggerPip
 	}
 
 	dagData := &LoadDAGDataActivityResult{}
-	err = workflow.ExecuteActivity(ctx, w.LoadDAGDataActivity, workflowID).Get(ctx, dagData)
+	err := workflow.ExecuteActivity(ctx, w.LoadDAGDataActivity, workflowID).Get(ctx, dagData)
 	if err != nil {
 		logger.Error("Failed to load DAG", zap.Error(err))
 	}