extract pipeline_name from metadata to agent and add to container names

agent/runner.go

@@ -72,6 +72,7 @@ func (r *Runner) Run(runnerCtx, shutdownCtx context.Context) error { //nolint:co
 
 	repoName := extractRepositoryName(workflow.Config)       // hack
 	pipelineNumber := extractPipelineNumber(workflow.Config) // hack
+	pipelineName := extractPipelineName(workflow.Config)     // hack
 
 	r.counter.Add(
 		workflow.ID,
@@ -149,6 +150,7 @@ func (r *Runner) Run(runnerCtx, shutdownCtx context.Context) error { //nolint:co
 			"workflow_id":     workflow.ID,
 			"repo":            repoName,
 			"pipeline_number": pipelineNumber,
+			"pipeline_name":   pipelineName,
 		}),
 	).Run(runnerCtx)
 
@@ -197,3 +199,7 @@ func extractRepositoryName(config *backend.Config) string {
 func extractPipelineNumber(config *backend.Config) string {
 	return config.Stages[0].Steps[0].Environment["CI_PIPELINE_NUMBER"]
 }
+
+func extractPipelineName(config *backend.Config) string {
+	return config.Stages[0].Steps[0].Environment["CI_WORKFLOW_NAME"]
+}

pipeline/backend/docker/convert.go

@@ -69,8 +69,8 @@ func (e *docker) toConfig(step *types.Step, options BackendOptions) *container.C
 	return config
 }
 
-func toContainerName(step *types.Step) string {
-	return "wp_" + step.UUID[:5] + "-" + step.Name
+func toContainerName(step *types.Step, workflowName string) string {
+	return "wp_" + step.UUID[:5] + "-" + workflowName + "-" + step.Name
 }
 
 // returns a container host configuration.

pipeline/backend/docker/convert_test.go

@@ -125,8 +125,8 @@ var (
 )
 
 func TestToContainerName(t *testing.T) {
-	assert.EqualValues(t, "wp_f5182-hello", toContainerName(testCmdStep))
-	assert.EqualValues(t, "wp_d841e-lint", toContainerName(testPluginStep))
+	assert.EqualValues(t, "wp_f5182-workflowNameTest-hello", toContainerName(testCmdStep, "workflowNameTest"))
+	assert.EqualValues(t, "wp_d841e-workflowNameTest-lint", toContainerName(testPluginStep, "workflowNameTest"))
 }
 
 func TestStepToConfig(t *testing.T) {

pipeline/backend/docker/docker.go

@@ -141,8 +141,8 @@ func (e *docker) Load(ctx context.Context) (*backend.BackendInfo, error) {
 	}, nil
 }
 
-func (e *docker) SetupWorkflow(ctx context.Context, conf *backend.Config, taskUUID string) error {
-	log.Trace().Str("taskUUID", taskUUID).Msg("create workflow environment")
+func (e *docker) SetupWorkflow(ctx context.Context, conf *backend.Config, taskUUID, workflowName string) error {
+	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msg("create workflow environment")
 
 	_, err := e.client.VolumeCreate(ctx, volume.CreateOptions{
 		Name:   conf.Volume.Name,
@@ -163,17 +163,17 @@ func (e *docker) SetupWorkflow(ctx context.Context, conf *backend.Config, taskUU
 	return err
 }
 
-func (e *docker) StartStep(ctx context.Context, step *backend.Step, taskUUID string) error {
+func (e *docker) StartStep(ctx context.Context, step *backend.Step, taskUUID, workflowName string) error {
 	options, err := parseBackendOptions(step)
 	if err != nil {
 		log.Error().Err(err).Msg("could not parse backend options")
 	}
 
-	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s", step.Name)
+	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)
 
 	config := e.toConfig(step, options)
 	hostConfig := toHostConfig(step, &e.config)
-	containerName := toContainerName(step)
+	containerName := toContainerName(step, workflowName)
 
 	// create pull options with encoded authorization credentials.
 	pullOpts := image.PullOptions{}
@@ -246,10 +246,10 @@ func (e *docker) StartStep(ctx context.Context, step *backend.Step, taskUUID str
 	return e.client.ContainerStart(ctx, containerName, container.StartOptions{})
 }
 
-func (e *docker) WaitStep(ctx context.Context, step *backend.Step, taskUUID string) (*backend.State, error) {
+func (e *docker) WaitStep(ctx context.Context, step *backend.Step, taskUUID, workflowName string) (*backend.State, error) {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("wait for step %s", step.Name)
 
-	containerName := toContainerName(step)
+	containerName := toContainerName(step, workflowName)
 
 	wait, errC := e.client.ContainerWait(ctx, containerName, "")
 	select {
@@ -269,10 +269,10 @@ func (e *docker) WaitStep(ctx context.Context, step *backend.Step, taskUUID stri
 	}, nil
 }
 
-func (e *docker) TailStep(ctx context.Context, step *backend.Step, taskUUID string) (io.ReadCloser, error) {
+func (e *docker) TailStep(ctx context.Context, step *backend.Step, taskUUID, workflowName string) (io.ReadCloser, error) {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("tail logs of step %s", step.Name)
 
-	logs, err := e.client.ContainerLogs(ctx, toContainerName(step), container.LogsOptions{
+	logs, err := e.client.ContainerLogs(ctx, toContainerName(step, workflowName), container.LogsOptions{
 		Follow:     true,
 		ShowStdout: true,
 		ShowStderr: true,
@@ -293,10 +293,10 @@ func (e *docker) TailStep(ctx context.Context, step *backend.Step, taskUUID stri
 	return rc, nil
 }
 
-func (e *docker) DestroyStep(ctx context.Context, step *backend.Step, taskUUID string) error {
+func (e *docker) DestroyStep(ctx context.Context, step *backend.Step, taskUUID, workflowName string) error {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("stop step %s", step.Name)
 
-	containerName := toContainerName(step)
+	containerName := toContainerName(step, workflowName)
 
 	if err := e.client.ContainerKill(ctx, containerName, "9"); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
 		return err
@@ -309,12 +309,12 @@ func (e *docker) DestroyStep(ctx context.Context, step *backend.Step, taskUUID s
 	return nil
 }
 
-func (e *docker) DestroyWorkflow(ctx context.Context, conf *backend.Config, taskUUID string) error {
+func (e *docker) DestroyWorkflow(ctx context.Context, conf *backend.Config, taskUUID, workflowName string) error {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("delete workflow environment")
 
 	for _, stage := range conf.Stages {
 		for _, step := range stage.Steps {
-			containerName := toContainerName(step)
+			containerName := toContainerName(step, workflowName)
 			if err := e.client.ContainerKill(ctx, containerName, "9"); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
 				log.Error().Err(err).Msgf("could not kill container '%s'", step.Name)
 			}

pipeline/backend/dummy/dummy.go

@@ -78,17 +78,20 @@ func (e *dummy) Load(_ context.Context) (*backend.BackendInfo, error) {
 	}, nil
 }
 
-func (e *dummy) SetupWorkflow(_ context.Context, _ *backend.Config, taskUUID string) error {
+func (e *dummy) SetupWorkflow(_ context.Context, _ *backend.Config, taskUUID, workflowName string) error {
 	if taskUUID == WorkflowSetupFailUUID {
 		return fmt.Errorf("expected fail to setup workflow")
 	}
+	if workflowName == "" {
+		return fmt.Errorf("expected fail to setup workflow")
+	}
 	log.Trace().Str("taskUUID", taskUUID).Msg("create workflow environment")
 	e.kv.Store("task_"+taskUUID, "setup")
 	return nil
 }
 
-func (e *dummy) StartStep(_ context.Context, step *backend.Step, taskUUID string) error {
-	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s", step.Name)
+func (e *dummy) StartStep(_ context.Context, step *backend.Step, taskUUID, workflowName string) error {
+	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)
 
 	// internal state checks
 	_, exist := e.kv.Load("task_" + taskUUID)
@@ -114,8 +117,8 @@ func (e *dummy) StartStep(_ context.Context, step *backend.Step, taskUUID string
 	return nil
 }
 
-func (e *dummy) WaitStep(ctx context.Context, step *backend.Step, taskUUID string) (*backend.State, error) {
-	log.Trace().Str("taskUUID", taskUUID).Msgf("wait for step %s", step.Name)
+func (e *dummy) WaitStep(ctx context.Context, step *backend.Step, taskUUID, workflowName string) (*backend.State, error) {
+	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)
 
 	_, exist := e.kv.Load("task_" + taskUUID)
 	if !exist {
@@ -172,8 +175,8 @@ func (e *dummy) WaitStep(ctx context.Context, step *backend.Step, taskUUID strin
 	}, nil
 }
 
-func (e *dummy) TailStep(_ context.Context, step *backend.Step, taskUUID string) (io.ReadCloser, error) {
-	log.Trace().Str("taskUUID", taskUUID).Msgf("tail logs of step %s", step.Name)
+func (e *dummy) TailStep(_ context.Context, step *backend.Step, taskUUID, workflowName string) (io.ReadCloser, error) {
+	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)
 
 	_, exist := e.kv.Load("task_" + taskUUID)
 	if !exist {
@@ -196,8 +199,8 @@ func (e *dummy) TailStep(_ context.Context, step *backend.Step, taskUUID string)
 	return io.NopCloser(strings.NewReader(dummyExecStepOutput(step))), nil
 }
 
-func (e *dummy) DestroyStep(_ context.Context, step *backend.Step, taskUUID string) error {
-	log.Trace().Str("taskUUID", taskUUID).Msgf("stop step %s", step.Name)
+func (e *dummy) DestroyStep(_ context.Context, step *backend.Step, taskUUID, workflowName string) error {
+	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msgf("stop step %s", step.Name)
 
 	_, exist := e.kv.Load("task_" + taskUUID)
 	if !exist {
@@ -217,8 +220,8 @@ func (e *dummy) DestroyStep(_ context.Context, step *backend.Step, taskUUID stri
 	return nil
 }
 
-func (e *dummy) DestroyWorkflow(_ context.Context, _ *backend.Config, taskUUID string) error {
-	log.Trace().Str("taskUUID", taskUUID).Msgf("delete workflow environment")
+func (e *dummy) DestroyWorkflow(_ context.Context, _ *backend.Config, taskUUID, workflowName string) error {
+	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msgf("delete workflow environment")
 
 	_, exist := e.kv.Load("task_" + taskUUID)
 	if !exist {

pipeline/backend/dummy/dummy_test.go

@@ -34,22 +34,22 @@ func TestSmalPipelineDummyRun(t *testing.T) {
 	_, err := dummyEngine.Load(ctx)
 	assert.NoError(t, err)
 
-	assert.Error(t, dummyEngine.SetupWorkflow(ctx, nil, dummy.WorkflowSetupFailUUID))
+	assert.Error(t, dummyEngine.SetupWorkflow(ctx, nil, dummy.WorkflowSetupFailUUID, "workflowTestName"))
 
 	t.Run("expect fail of step func with non setup workflow", func(t *testing.T) {
 		step := &types.Step{Name: "step1", UUID: "SID_1"}
 		nonExistWorkflowID := "WID_NONE"
 
-		err := dummyEngine.StartStep(ctx, step, nonExistWorkflowID)
+		err := dummyEngine.StartStep(ctx, step, nonExistWorkflowID, "workflowTestName")
 		assert.Error(t, err)
 
-		_, err = dummyEngine.TailStep(ctx, step, nonExistWorkflowID)
+		_, err = dummyEngine.TailStep(ctx, step, nonExistWorkflowID, "workflowTestName")
 		assert.Error(t, err)
 
-		_, err = dummyEngine.WaitStep(ctx, step, nonExistWorkflowID)
+		_, err = dummyEngine.WaitStep(ctx, step, nonExistWorkflowID, "workflowTestName")
 		assert.Error(t, err)
 
-		err = dummyEngine.DestroyStep(ctx, step, nonExistWorkflowID)
+		err = dummyEngine.DestroyStep(ctx, step, nonExistWorkflowID, "workflowTestName")
 		assert.Error(t, err)
 	})
 
@@ -63,11 +63,11 @@ func TestSmalPipelineDummyRun(t *testing.T) {
 		}
 		workflowUUID := "WID_1"
 
-		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 
-		assert.NoError(t, dummyEngine.StartStep(ctx, step, workflowUUID))
+		assert.NoError(t, dummyEngine.StartStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		reader, err := dummyEngine.TailStep(ctx, step, workflowUUID)
+		reader, err := dummyEngine.TailStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.NoError(t, err)
 		log, err := io.ReadAll(reader)
 		assert.NoError(t, err)
@@ -81,14 +81,14 @@ echo nein
 ------------------
 `, string(log))
 
-		state, err := dummyEngine.WaitStep(ctx, step, workflowUUID)
+		state, err := dummyEngine.WaitStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.NoError(t, err)
 		assert.NoError(t, state.Error)
 		assert.EqualValues(t, 0, state.ExitCode)
 
-		assert.NoError(t, dummyEngine.DestroyStep(ctx, step, workflowUUID))
+		assert.NoError(t, dummyEngine.DestroyStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 	})
 
 	t.Run("step exec error", func(t *testing.T) {
@@ -100,21 +100,21 @@ echo nein
 		}
 		workflowUUID := "WID_1"
 
-		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 
-		assert.NoError(t, dummyEngine.StartStep(ctx, step, workflowUUID))
+		assert.NoError(t, dummyEngine.StartStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		_, err := dummyEngine.TailStep(ctx, step, workflowUUID)
+		_, err := dummyEngine.TailStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.NoError(t, err)
 
-		state, err := dummyEngine.WaitStep(ctx, step, workflowUUID)
+		state, err := dummyEngine.WaitStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.NoError(t, err)
 		assert.NoError(t, state.Error)
 		assert.EqualValues(t, 1, state.ExitCode)
 
-		assert.NoError(t, dummyEngine.DestroyStep(ctx, step, workflowUUID))
+		assert.NoError(t, dummyEngine.DestroyStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 	})
 
 	t.Run("step tail error", func(t *testing.T) {
@@ -125,19 +125,19 @@ echo nein
 		}
 		workflowUUID := "WID_1"
 
-		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 
-		assert.NoError(t, dummyEngine.StartStep(ctx, step, workflowUUID))
+		assert.NoError(t, dummyEngine.StartStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		_, err := dummyEngine.TailStep(ctx, step, workflowUUID)
+		_, err := dummyEngine.TailStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.Error(t, err)
 
-		_, err = dummyEngine.WaitStep(ctx, step, workflowUUID)
+		_, err = dummyEngine.WaitStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.NoError(t, err)
 
-		assert.NoError(t, dummyEngine.DestroyStep(ctx, step, workflowUUID))
+		assert.NoError(t, dummyEngine.DestroyStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 	})
 
 	t.Run("step start fail", func(t *testing.T) {
@@ -149,20 +149,20 @@ echo nein
 		}
 		workflowUUID := "WID_1"
 
-		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.SetupWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 
-		assert.Error(t, dummyEngine.StartStep(ctx, step, workflowUUID))
+		assert.Error(t, dummyEngine.StartStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		_, err := dummyEngine.TailStep(ctx, step, workflowUUID)
+		_, err := dummyEngine.TailStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.Error(t, err)
 
-		state, err := dummyEngine.WaitStep(ctx, step, workflowUUID)
+		state, err := dummyEngine.WaitStep(ctx, step, workflowUUID, "workflowTestName")
 		assert.Error(t, err)
 		assert.Error(t, state.Error)
 		assert.EqualValues(t, 0, state.ExitCode)
 
-		assert.Error(t, dummyEngine.DestroyStep(ctx, step, workflowUUID))
+		assert.Error(t, dummyEngine.DestroyStep(ctx, step, workflowUUID, "workflowTestName"))
 
-		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID))
+		assert.NoError(t, dummyEngine.DestroyWorkflow(ctx, nil, workflowUUID, "workflowTestName"))
 	})
 }