Add txid to workflows (#72)

* add txid's to workflow submission labels
cello-proj · Aug 20, 2021 · 374912a · 374912a
1 parent 5097279
commit 374912a
Show file tree

Hide file tree

Showing 5 changed files with 12 additions and 6 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,9 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [0.7.0] - 2021-08-20
 ### Changed
 * config.listFrameworks() are now sorted (helps avoid flaky tests).
+* Transaction ID label added for workflow submissions.
 
 ### Fixed
 * config.listFrameworks() was returning additional empty items.

diff --git a/service/handlers.go b/service/handlers.go
@@ -318,8 +318,10 @@ func (h handler) createWorkflowFromRequest(_ context.Context, w http.ResponseWri
 	level.Debug(l).Log("message", "creating workflow parameters")
 	parameters := workflow.NewParameters(environmentVariablesString, executeCommand, executeContainerImageURI, cwr.TargetName, cwr.ProjectName, cwr.Parameters, credentialsToken)
 
+	workflowLabels := map[string]string{txIDHeader: r.Header.Get(txIDHeader)}
+
 	level.Debug(l).Log("message", "creating workflow")
-	workflowName, err := h.argo.Submit(h.argoCtx, workflowFrom, parameters)
+	workflowName, err := h.argo.Submit(h.argoCtx, workflowFrom, parameters, workflowLabels)
 	if err != nil {
 		level.Error(l).Log("message", "error creating workflow", "error", err)
 		h.errorResponse(w, "error creating workflow", http.StatusInternalServerError)

diff --git a/service/handlers_test.go b/service/handlers_test.go
@@ -90,7 +90,7 @@ func (m mockWorkflowSvc) List(ctx context.Context) ([]string, error) {
 	return []string{"project1-target1-abcde", "project2-target2-12345"}, nil
 }
 
-func (m mockWorkflowSvc) Submit(ctx context.Context, from string, parameters map[string]string) (string, error) {
+func (m mockWorkflowSvc) Submit(ctx context.Context, from string, parameters map[string]string, labels map[string]string) (string, error) {
 	return "success", nil
 }
 
@@ -757,6 +757,7 @@ func executeRequest(method string, url string, body *bytes.Buffer, asAdmin bool)
 		logger:                 log.NewNopLogger(),
 		newCredentialsProvider: newMockProvider,
 		argo:                   mockWorkflowSvc{},
+		argoCtx:                context.Background(),
 		config:                 config,
 		gitClient:              newMockGitClient(),
 		env: env.Vars{

diff --git a/service/internal/workflow/workflow.go b/service/internal/workflow/workflow.go
@@ -11,6 +11,7 @@ import (
 	argoWorkflowAPIClient "github.com/argoproj/argo-workflows/v3/pkg/apiclient/workflow"
 	argoWorkflowAPISpec "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 )
 
 const mainContainer = "main"
@@ -21,7 +22,7 @@ type Workflow interface {
 	Logs(ctx context.Context, workflowName string) (*Logs, error)
 	LogStream(ctx context.Context, workflowName string, data http.ResponseWriter) error
 	Status(ctx context.Context, workflowName string) (*Status, error)
-	Submit(ctx context.Context, from string, parameters map[string]string) (string, error)
+	Submit(ctx context.Context, from string, parameters map[string]string, labels map[string]string) (string, error)
 }
 
 // NewArgoWorkflow creates an Argo workflow.
@@ -166,7 +167,7 @@ func (a ArgoWorkflow) LogStream(ctx context.Context, workflowName string, w http
 }
 
 // Submit submits a workflow execution.
-func (a ArgoWorkflow) Submit(ctx context.Context, from string, parameters map[string]string) (string, error) {
+func (a ArgoWorkflow) Submit(ctx context.Context, from string, parameters map[string]string, workflowLabels map[string]string) (string, error) {
 	parts := strings.SplitN(from, "/", 2)
 	for _, part := range parts {
 		if part == "" {
@@ -191,6 +192,7 @@ func (a ArgoWorkflow) Submit(ctx context.Context, from string, parameters map[st
 		SubmitOptions: &argoWorkflowAPISpec.SubmitOpts{
 			GenerateName: generateNamePrefix,
 			Parameters:   parameterStrings,
+			Labels:       labels.FormatLabels(workflowLabels),
 		},
 	})
 

diff --git a/service/internal/workflow/workflow_test.go b/service/internal/workflow/workflow_test.go
@@ -147,7 +147,7 @@ func TestArgoSubmit(t *testing.T) {
 				"namespace",
 			)
 
-			workflow, err := argoWf.Submit(context.Background(), "test/test", map[string]string{"param": "value"})
+			workflow, err := argoWf.Submit(context.Background(), "test/test", map[string]string{"param": "value"}, map[string]string{"X-B3-TraceId": "test-txid"})
 			if err != nil {
 				if tt.errResult != nil && tt.errResult.Error() != err.Error() {
 					t.Errorf("\nwant: %v\n got: %v", tt.errResult, err)