set max length for names

woodpecker-ci · pat-s · Jan 5, 2025 · Jan 5, 2025 · Jan 5, 2025 · Jan 5, 2025
commit 5a0c941d34c905e6a8a93cd25f21800e9ea9feb6
diff --git a/pipeline/backend/kubernetes/utils.go b/pipeline/backend/kubernetes/utils.go
@@ -34,16 +34,21 @@
 func dnsName(i string) (string, error) {
 	res := strings.ToLower(strings.ReplaceAll(i, "_", "-"))
 
-	// Check for invalid characters (dnsDisallowedCharacters)
 	invalidChars := dnsDisallowedCharacters.FindAllString(res, -1)
 	if len(invalidChars) > 0 {
 		return "", fmt.Errorf("name is not a valid kubernetes DNS name: found invalid characters '%v'", strings.Join(invalidChars, ""))
 	}
 
-	// Check if the entire string matches the dnsPattern
 	if !dnsPattern.MatchString(res) {
 		return "", fmt.Errorf("name is not a valid kubernetes DNS name")
 	}
+
+	// k8s pod name limit
+	const maxPodNameLength = 253
+	if len(res) > maxPodNameLength {
+		res = res[:maxPodNameLength]
+	}
+
 	return res, nil
 }
 

diff --git a/agent/runner.go b/agent/runner.go

 	repoName := extractRepositoryName(workflow.Config)       // hack
 	pipelineNumber := extractPipelineNumber(workflow.Config) // hack
 	pipelineName := extractPipelineName(workflow.Config)     // hack

 	r.counter.Add(
 		workflow.ID,
 			"workflow_id":     workflow.ID,
 			"repo":            repoName,
 			"pipeline_number": pipelineNumber,
 			"pipeline_name":   pipelineName,
 		}),
 	).Run(runnerCtx)

 	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"]
 }
diff --git a/pipeline/backend/docker/docker.go b/pipeline/backend/docker/docker.go
 	}, nil
 }

 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,
 	return err
 }

 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-%s", workflowName, step.Name)

 	config := e.toConfig(step, options)
 	hostConfig := toHostConfig(step, &e.config)
 	containerName := toContainerName(step, workflowName)

 	// create pull options with encoded authorization credentials.
 	pullOpts := image.PullOptions{}
 	return e.client.ContainerStart(ctx, containerName, container.StartOptions{})
 }

 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, workflowName)

 	wait, errC := e.client.ContainerWait(ctx, containerName, "")
 	select {
 	}, nil
 }

 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, workflowName), container.LogsOptions{
 		Follow:     true,
 		ShowStdout: true,
 		ShowStderr: true,
 	return rc, nil
 }

 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, workflowName)

 	if err := e.client.ContainerKill(ctx, containerName, "9"); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
 		return err
 	return nil
 }

 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, 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)
 			}
diff --git a/pipeline/backend/dummy/dummy.go b/pipeline/backend/dummy/dummy.go
 		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
diff --git a/pipeline/backend/kubernetes/kubernetes.go b/pipeline/backend/kubernetes/kubernetes.go
 }

 // SetupWorkflow sets up the pipeline environment.
 func (e *kube) SetupWorkflow(ctx context.Context, conf *types.Config, taskUUID, workflowName string) error {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("Setting up Kubernetes primitives")

 	_, err := startVolume(ctx, e, conf.Volume.Name)
 	for _, stage := range conf.Stages {
 		for _, step := range stage.Steps {
 			if step.Type == types.StepTypeService {
 				svc, err := startService(ctx, e, step, workflowName)
 				if err != nil {
 					return err
 				}
 }

 // StartStep starts the pipeline step.
 func (e *kube) StartStep(ctx context.Context, step *types.Step, taskUUID, workflowName string) error {
 	options, err := parseBackendOptions(step)
 	if err != nil {
 		log.Error().Err(err).Msg("could not parse backend options")
 	}

 	if needsRegistrySecret(step) {
 		err = startRegistrySecret(ctx, e, step, workflowName)
 		if err != nil {
 			return err
 		}
 	}

 	log.Trace().Str("taskUUID", taskUUID).Msgf("starting step: %s", step.Name)
 	_, err = startPod(ctx, e, step, options, workflowName)
 	return err
 }

 // WaitStep waits for the pipeline step to complete and returns
 // the completion results.
 func (e *kube) WaitStep(ctx context.Context, step *types.Step, taskUUID, workflowName string) (*types.State, error) {
 	podName, err := stepToPodName(step, workflowName)
 	if err != nil {
 		return nil, err
 	}
 }

 // TailStep tails the pipeline step logs.
 func (e *kube) TailStep(ctx context.Context, step *types.Step, taskUUID, workflowName string) (io.ReadCloser, error) {
 	podName, err := stepToPodName(step, workflowName)
 	if err != nil {
 		return nil, err
 	}
 	return rc, nil
 }

 func (e *kube) DestroyStep(ctx context.Context, step *types.Step, taskUUID, workflowName string) error {
 	var errs []error
 	log.Trace().Str("taskUUID", taskUUID).Msgf("Stopping step: %s", step.Name)
 	if needsRegistrySecret(step) {
 		err := stopRegistrySecret(ctx, e, step, defaultDeleteOptions, workflowName)
 		if err != nil {
 			errs = append(errs, err)
 		}
 	}

 	err := stopPod(ctx, e, step, defaultDeleteOptions, workflowName)
 	if err != nil {
 		errs = append(errs, err)
 	}
 }

 // DestroyWorkflow destroys the pipeline environment.
 func (e *kube) DestroyWorkflow(ctx context.Context, conf *types.Config, taskUUID, workflowName string) error {
 	log.Trace().Str("taskUUID", taskUUID).Msg("deleting Kubernetes primitives")

 	for _, stage := range conf.Stages {
 		for _, step := range stage.Steps {
 			err := stopPod(ctx, e, step, defaultDeleteOptions, workflowName)
 			if err != nil {
 				return err
 			}

 			if step.Type == types.StepTypeService {
 				err := stopService(ctx, e, step, defaultDeleteOptions, workflowName)
 				if err != nil {
 					return err
 				}
diff --git a/pipeline/backend/kubernetes/pod.go b/pipeline/backend/kubernetes/pod.go
 		maps.Copy(labels, config.PodLabels)
 	}
 	if step.Type == types.StepTypeService {
 		labels[ServiceLabel], _ = serviceName(step, workflowName)
 	}
 	labels[StepLabel], err = stepLabel(step)
 	if err != nil {
 	return ev
 }

 func startPod(ctx context.Context, engine *kube, step *types.Step, options BackendOptions, workflowName string) (*v1.Pod, error) {
 	podName, err := stepToPodName(step, workflowName)
 	if err != nil {
 		return nil, err
 	}
 	engineConfig := engine.getConfig()
 	pod, err := mkPod(step, engineConfig, podName, engine.goos, options, workflowName)
 	if err != nil {
 		return nil, err
 	}
 	return engine.client.CoreV1().Pods(engineConfig.Namespace).Create(ctx, pod, meta_v1.CreateOptions{})
 }

 func stopPod(ctx context.Context, engine *kube, step *types.Step, deleteOpts meta_v1.DeleteOptions, workflowName string) error {
 	podName, err := stepToPodName(step, workflowName)
 	if err != nil {
 		return err
 	}
diff --git a/pipeline/backend/kubernetes/secrets.go b/pipeline/backend/kubernetes/secrets.go
 	labels := make(map[string]string)

 	if step.Type == types.StepTypeService {
 		labels[ServiceLabel], _ = serviceName(step, workflowName)
 	}
 	labels[StepLabel], err = stepLabel(step)
 	if err != nil {
 	return labels, nil
 }

 func startRegistrySecret(ctx context.Context, engine *kube, step *types.Step, workflowName string) error {
 	secret, err := mkRegistrySecret(step, engine.config, workflowName)
 	if err != nil {
 		return err
 	}
 	return nil
 }

 func stopRegistrySecret(ctx context.Context, engine *kube, step *types.Step, deleteOpts meta_v1.DeleteOptions, workflowName string) error {
 	name, err := registrySecretName(step, workflowName)
 	if err != nil {
 		return err
 	}
diff --git a/pipeline/backend/kubernetes/service.go b/pipeline/backend/kubernetes/service.go
 	}
 }

 func startService(ctx context.Context, engine *kube, step *types.Step, workflowName string) (*v1.Service, error) {
 	engineConfig := engine.getConfig()
 	svc, err := mkService(step, engineConfig, workflowName)
 	if err != nil {
 		return nil, err
 	}
 	return engine.client.CoreV1().Services(engineConfig.Namespace).Create(ctx, svc, meta_v1.CreateOptions{})
 }

 func stopService(ctx context.Context, engine *kube, step *types.Step, deleteOpts meta_v1.DeleteOptions, workflowName string) error {
 	svcName, err := serviceName(step, workflowName)
 	if err != nil {
 		return err
 	}
diff --git a/pipeline/backend/local/local.go b/pipeline/backend/local/local.go
 }

 // SetupWorkflow the pipeline environment.
 func (e *local) SetupWorkflow(_ context.Context, _ *types.Config, taskUUID, workflowName string) error {
 	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msg("create workflow environment")

 	baseDir, err := os.MkdirTemp(e.tempDir, "woodpecker-local-*")
 	if err != nil {
 }

 // StartStep the pipeline step.
 func (e *local) StartStep(ctx context.Context, step *types.Step, taskUUID, workflowName string) error {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)

 	state, err := e.getState(taskUUID)
 	if err != nil {

 // WaitStep for the pipeline step to complete and returns
 // the completion results.
 func (e *local) WaitStep(_ context.Context, step *types.Step, taskUUID, workflowName string) (*types.State, error) {
 	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)

 	state, err := e.getState(taskUUID)
 	if err != nil {
 }

 // TailStep the pipeline step logs.
 func (e *local) TailStep(_ context.Context, step *types.Step, taskUUID, workflowName string) (io.ReadCloser, error) {
 	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msgf("tail logs of step %s", step.Name)
 	return e.output, nil
 }

 func (e *local) DestroyStep(_ context.Context, _ *types.Step, _, _ string) error {
 	// WaitStep already waits for the command to finish, so there is nothing to do here.
 	return nil
 }

 // DestroyWorkflow the pipeline environment.
 func (e *local) DestroyWorkflow(_ context.Context, _ *types.Config, taskUUID, workflowName string) error {
 	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msg("delete workflow environment")

 	state, err := e.getState(taskUUID)
 	if err != nil {
diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go
 		if ctx.Err() != nil {
 			ctx = GetShutdownCtx()
 		}
 		if err := r.engine.DestroyWorkflow(ctx, r.spec, r.taskUUID, r.Description["pipeline_name"]); err != nil {
 			logger.Error().Err(err).Msg("could not destroy engine")
 		}
 	}()

 	r.started = time.Now().Unix()
 	if err := r.engine.SetupWorkflow(runnerCtx, r.spec, r.taskUUID, r.Description["pipeline_name"]); err != nil {
 		return err
 	}

 		select {
 		case <-r.ctx.Done():
 			return ErrCancel
 		case err := <-r.execAll(stage.Steps, r.Description["pipeline_name"]):
 			if err != nil {
 				r.err = err
 			}
 }

 // Executes a set of parallel steps.
 func (r *Runtime) execAll(steps []*backend.Step, workflowName string) <-chan error {
 	var g errgroup.Group
 	done := make(chan error)
 	logger := r.MakeLogger()
 				Str("step", step.Name).
 				Msg("executing")

 			processState, err := r.exec(step, workflowName)

 			logger.Debug().
 				Str("step", step.Name).
 }

 // Executes the step and returns the state and error.
 func (r *Runtime) exec(step *backend.Step, workflowName string) (*backend.State, error) {
 	if err := r.engine.StartStep(r.ctx, step, r.taskUUID, workflowName); err != nil {
 		return nil, err
 	}

 	var wg sync.WaitGroup
 	if r.logger != nil {
 		rc, err := r.engine.TailStep(r.ctx, step, r.taskUUID, workflowName)
 		if err != nil {
 			return nil, err
 		}
 	// We wait until all data was logged. (Needed for some backends like local as WaitStep kills the log stream)
 	wg.Wait()

 	waitState, err := r.engine.WaitStep(r.ctx, step, r.taskUUID, workflowName)
 	if err != nil {
 		if errors.Is(err, context.Canceled) {
 			return waitState, ErrCancel
 		return nil, err
 	}

 	if err := r.engine.DestroyStep(r.ctx, step, r.taskUUID, workflowName); err != nil {
 		return nil, err
 	}

	repoName := extractRepositoryName(workflow.Config) // hack
	pipelineNumber := extractPipelineNumber(workflow.Config) // hack
	pipelineName := extractPipelineName(workflow.Config) // hack
Check warning on line 75 in agent/runner.go Codecov / codecov/patch agent/runner.go#L75 `Added line #L75 was not covered by tests`

	r.counter.Add(
	workflow.ID,
	"workflow_id": workflow.ID,
	"repo": repoName,
	"pipeline_number": pipelineNumber,
	"pipeline_name": pipelineName,
Check warning on line 153 in agent/runner.go Codecov / codecov/patch agent/runner.go#L153 `Added line #L153 was not covered by tests`
	}),
	).Run(runnerCtx)

	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"]
Check warning on line 204 in agent/runner.go Codecov / codecov/patch agent/runner.go#L203-L204 `Added lines #L203 - L204 were not covered by tests`
	}
	}, nil
	}

	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")
Check warning on line 145 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L144-L145 `Added lines #L144 - L145 were not covered by tests`

	_, err := e.client.VolumeCreate(ctx, volume.CreateOptions{
	Name: conf.Volume.Name,
	return err
	}

	func (e docker) StartStep(ctx context.Context, step backend.Step, taskUUID, workflowName string) error {
Check warning on line 166 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L166 `Added line #L166 was not covered by tests`
	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-%s", workflowName, step.Name)
Check warning on line 172 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L172 `Added line #L172 was not covered by tests`

	config := e.toConfig(step, options)
	hostConfig := toHostConfig(step, &e.config)
	containerName := toContainerName(step, workflowName)
Check warning on line 176 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L176 `Added line #L176 was not covered by tests`

	// create pull options with encoded authorization credentials.
	pullOpts := image.PullOptions{}
	return e.client.ContainerStart(ctx, containerName, container.StartOptions{})
	}

	func (e docker) WaitStep(ctx context.Context, step backend.Step, taskUUID, workflowName string) (*backend.State, error) {
Check warning on line 249 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L249 `Added line #L249 was not covered by tests`
	log.Trace().Str("taskUUID", taskUUID).Msgf("wait for step %s", step.Name)

	containerName := toContainerName(step, workflowName)
Check warning on line 252 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L252 `Added line #L252 was not covered by tests`

	wait, errC := e.client.ContainerWait(ctx, containerName, "")
	select {
	}, nil
	}

	func (e docker) TailStep(ctx context.Context, step backend.Step, taskUUID, workflowName string) (io.ReadCloser, error) {
Check warning on line 272 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L272 `Added line #L272 was not covered by tests`
	log.Trace().Str("taskUUID", taskUUID).Msgf("tail logs of step %s", step.Name)

	logs, err := e.client.ContainerLogs(ctx, toContainerName(step, workflowName), container.LogsOptions{
Check warning on line 275 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L275 `Added line #L275 was not covered by tests`
	Follow: true,
	ShowStdout: true,
	ShowStderr: true,
	return rc, nil
	}

	func (e docker) DestroyStep(ctx context.Context, step backend.Step, taskUUID, workflowName string) error {
Check warning on line 296 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L296 `Added line #L296 was not covered by tests`
	log.Trace().Str("taskUUID", taskUUID).Msgf("stop step %s", step.Name)

	containerName := toContainerName(step, workflowName)
Check warning on line 299 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L299 `Added line #L299 was not covered by tests`

	if err := e.client.ContainerKill(ctx, containerName, "9"); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
	return err
	return nil
	}

	func (e docker) DestroyWorkflow(ctx context.Context, conf backend.Config, taskUUID, workflowName string) error {
Check warning on line 312 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L312 `Added line #L312 was not covered by tests`
	log.Trace().Str("taskUUID", taskUUID).Msgf("delete workflow environment")

	for _, stage := range conf.Stages {
	for _, step := range stage.Steps {
	containerName := toContainerName(step, workflowName)
Check warning on line 317 in pipeline/backend/docker/docker.go Codecov / codecov/patch pipeline/backend/docker/docker.go#L317 `Added line #L317 was not covered by tests`
	if err := e.client.ContainerKill(ctx, containerName, "9"); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
	log.Error().Err(err).Msgf("could not kill container '%s'", step.Name)
	}
	return fmt.Errorf("expected fail to setup workflow")
	}
	if workflowName == "" {
	return fmt.Errorf("expected fail to setup workflow")
	}
Check warning on line 87 in pipeline/backend/dummy/dummy.go Codecov / codecov/patch pipeline/backend/dummy/dummy.go#L86-L87 `Added lines #L86 - L87 were not covered by tests`
	log.Trace().Str("taskUUID", taskUUID).Msg("create workflow environment")
	e.kv.Store("task_"+taskUUID, "setup")
	return nil
	}

	// SetupWorkflow sets up the pipeline environment.
	func (e kube) SetupWorkflow(ctx context.Context, conf types.Config, taskUUID, workflowName string) error {
Check warning on line 191 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L191 `Added line #L191 was not covered by tests`
	log.Trace().Str("taskUUID", taskUUID).Msgf("Setting up Kubernetes primitives")

	_, err := startVolume(ctx, e, conf.Volume.Name)
	for _, stage := range conf.Stages {
	for _, step := range stage.Steps {
	if step.Type == types.StepTypeService {
	svc, err := startService(ctx, e, step, workflowName)
Check warning on line 203 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L203 `Added line #L203 was not covered by tests`
	if err != nil {
	return err
	}
	}

	// StartStep starts the pipeline step.
	func (e kube) StartStep(ctx context.Context, step types.Step, taskUUID, workflowName string) error {
Check warning on line 223 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L223 `Added line #L223 was not covered by tests`
	options, err := parseBackendOptions(step)
	if err != nil {
	log.Error().Err(err).Msg("could not parse backend options")
	}

	if needsRegistrySecret(step) {
	err = startRegistrySecret(ctx, e, step, workflowName)
Check warning on line 230 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L230 `Added line #L230 was not covered by tests`
	if err != nil {
	return err
	}
	}

	log.Trace().Str("taskUUID", taskUUID).Msgf("starting step: %s", step.Name)
	_, err = startPod(ctx, e, step, options, workflowName)
Check warning on line 237 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L237 `Added line #L237 was not covered by tests`
	return err
	}

	// WaitStep waits for the pipeline step to complete and returns
	// the completion results.
	func (e kube) WaitStep(ctx context.Context, step types.Step, taskUUID, workflowName string) (*types.State, error) {
	podName, err := stepToPodName(step, workflowName)
Check warning on line 244 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L243-L244 `Added lines #L243 - L244 were not covered by tests`
	if err != nil {
	return nil, err
	}
	}

	// TailStep tails the pipeline step logs.
	func (e kube) TailStep(ctx context.Context, step types.Step, taskUUID, workflowName string) (io.ReadCloser, error) {
	podName, err := stepToPodName(step, workflowName)
Check warning on line 320 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L319-L320 `Added lines #L319 - L320 were not covered by tests`
	if err != nil {
	return nil, err
	}
	return rc, nil
	}

	func (e kube) DestroyStep(ctx context.Context, step types.Step, taskUUID, workflowName string) error {
Check warning on line 391 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L391 `Added line #L391 was not covered by tests`
	var errs []error
	log.Trace().Str("taskUUID", taskUUID).Msgf("Stopping step: %s", step.Name)
	if needsRegistrySecret(step) {
	err := stopRegistrySecret(ctx, e, step, defaultDeleteOptions, workflowName)
Check warning on line 395 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L395 `Added line #L395 was not covered by tests`
	if err != nil {
	errs = append(errs, err)
	}
	}

	err := stopPod(ctx, e, step, defaultDeleteOptions, workflowName)
Check warning on line 401 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L401 `Added line #L401 was not covered by tests`
	if err != nil {
	errs = append(errs, err)
	}
	}

	// DestroyWorkflow destroys the pipeline environment.
	func (e kube) DestroyWorkflow(ctx context.Context, conf types.Config, taskUUID, workflowName string) error {
Check warning on line 409 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L409 `Added line #L409 was not covered by tests`
	log.Trace().Str("taskUUID", taskUUID).Msg("deleting Kubernetes primitives")

	for _, stage := range conf.Stages {
	for _, step := range stage.Steps {
	err := stopPod(ctx, e, step, defaultDeleteOptions, workflowName)
Check warning on line 414 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L414 `Added line #L414 was not covered by tests`
	if err != nil {
	return err
	}

	if step.Type == types.StepTypeService {
	err := stopService(ctx, e, step, defaultDeleteOptions, workflowName)
Check warning on line 420 in pipeline/backend/kubernetes/kubernetes.go Codecov / codecov/patch pipeline/backend/kubernetes/kubernetes.go#L420 `Added line #L420 was not covered by tests`
	if err != nil {
	return err
	}
	maps.Copy(labels, config.PodLabels)
	}
	if step.Type == types.StepTypeService {
	labels[ServiceLabel], _ = serviceName(step, workflowName)
Check warning on line 116 in pipeline/backend/kubernetes/pod.go Codecov / codecov/patch pipeline/backend/kubernetes/pod.go#L116 `Added line #L116 was not covered by tests`
	}
	labels[StepLabel], err = stepLabel(step)
	if err != nil {
	return ev
	}

	func startPod(ctx context.Context, engine kube, step types.Step, options BackendOptions, workflowName string) (*v1.Pod, error) {
	podName, err := stepToPodName(step, workflowName)
Check warning on line 528 in pipeline/backend/kubernetes/pod.go Codecov / codecov/patch pipeline/backend/kubernetes/pod.go#L527-L528 `Added lines #L527 - L528 were not covered by tests`
	if err != nil {
	return nil, err
	}
	engineConfig := engine.getConfig()
	pod, err := mkPod(step, engineConfig, podName, engine.goos, options, workflowName)
Check warning on line 533 in pipeline/backend/kubernetes/pod.go Codecov / codecov/patch pipeline/backend/kubernetes/pod.go#L533 `Added line #L533 was not covered by tests`
	if err != nil {
	return nil, err
	}
	return engine.client.CoreV1().Pods(engineConfig.Namespace).Create(ctx, pod, meta_v1.CreateOptions{})
	}

	func stopPod(ctx context.Context, engine kube, step types.Step, deleteOpts meta_v1.DeleteOptions, workflowName string) error {
	podName, err := stepToPodName(step, workflowName)
Check warning on line 543 in pipeline/backend/kubernetes/pod.go Codecov / codecov/patch pipeline/backend/kubernetes/pod.go#L542-L543 `Added lines #L542 - L543 were not covered by tests`
	if err != nil {
	return err
	}
	labels := make(map[string]string)

	if step.Type == types.StepTypeService {
	labels[ServiceLabel], _ = serviceName(step, workflowName)
Check warning on line 259 in pipeline/backend/kubernetes/secrets.go Codecov / codecov/patch pipeline/backend/kubernetes/secrets.go#L259 `Added line #L259 was not covered by tests`
	}
	labels[StepLabel], err = stepLabel(step)
	if err != nil {
	return labels, nil
	}

	func startRegistrySecret(ctx context.Context, engine kube, step types.Step, workflowName string) error {
	secret, err := mkRegistrySecret(step, engine.config, workflowName)
Check warning on line 270 in pipeline/backend/kubernetes/secrets.go Codecov / codecov/patch pipeline/backend/kubernetes/secrets.go#L269-L270 `Added lines #L269 - L270 were not covered by tests`
	if err != nil {
	return err
	}
	return nil
	}

	func stopRegistrySecret(ctx context.Context, engine kube, step types.Step, deleteOpts meta_v1.DeleteOptions, workflowName string) error {
	name, err := registrySecretName(step, workflowName)
Check warning on line 283 in pipeline/backend/kubernetes/secrets.go Codecov / codecov/patch pipeline/backend/kubernetes/secrets.go#L282-L283 `Added lines #L282 - L283 were not covered by tests`
	if err != nil {
	return err
	}
	}
	}

	func startService(ctx context.Context, engine kube, step types.Step, workflowName string) (*v1.Service, error) {
Check warning on line 80 in pipeline/backend/kubernetes/service.go Codecov / codecov/patch pipeline/backend/kubernetes/service.go#L80 `Added line #L80 was not covered by tests`
	engineConfig := engine.getConfig()
	svc, err := mkService(step, engineConfig, workflowName)
Check warning on line 82 in pipeline/backend/kubernetes/service.go Codecov / codecov/patch pipeline/backend/kubernetes/service.go#L82 `Added line #L82 was not covered by tests`
	if err != nil {
	return nil, err
	}
	return engine.client.CoreV1().Services(engineConfig.Namespace).Create(ctx, svc, meta_v1.CreateOptions{})
	}

	func stopService(ctx context.Context, engine kube, step types.Step, deleteOpts meta_v1.DeleteOptions, workflowName string) error {
	svcName, err := serviceName(step, workflowName)
Check warning on line 92 in pipeline/backend/kubernetes/service.go Codecov / codecov/patch pipeline/backend/kubernetes/service.go#L91-L92 `Added lines #L91 - L92 were not covered by tests`
	if err != nil {
	return err
	}
	}

	// SetupWorkflow the pipeline environment.
	func (e local) SetupWorkflow(_ context.Context, _ types.Config, taskUUID, workflowName string) error {
	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msg("create workflow environment")
Check warning on line 94 in pipeline/backend/local/local.go Codecov / codecov/patch pipeline/backend/local/local.go#L93-L94 `Added lines #L93 - L94 were not covered by tests`

	baseDir, err := os.MkdirTemp(e.tempDir, "woodpecker-local-*")
	if err != nil {
	}

	// StartStep the pipeline step.
	func (e local) StartStep(ctx context.Context, step types.Step, taskUUID, workflowName string) error {
	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)
Check warning on line 123 in pipeline/backend/local/local.go Codecov / codecov/patch pipeline/backend/local/local.go#L122-L123 `Added lines #L122 - L123 were not covered by tests`

	state, err := e.getState(taskUUID)
	if err != nil {

	// WaitStep for the pipeline step to complete and returns
	// the completion results.
	func (e local) WaitStep(_ context.Context, step types.Step, taskUUID, workflowName string) (*types.State, error) {
	log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s-%s", workflowName, step.Name)
Check warning on line 208 in pipeline/backend/local/local.go Codecov / codecov/patch pipeline/backend/local/local.go#L207-L208 `Added lines #L207 - L208 were not covered by tests`

	state, err := e.getState(taskUUID)
	if err != nil {
	}

	// TailStep the pipeline step logs.
	func (e local) TailStep(_ context.Context, step types.Step, taskUUID, workflowName string) (io.ReadCloser, error) {
	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msgf("tail logs of step %s", step.Name)
Check warning on line 238 in pipeline/backend/local/local.go Codecov / codecov/patch pipeline/backend/local/local.go#L237-L238 `Added lines #L237 - L238 were not covered by tests`
	return e.output, nil
	}

	func (e local) DestroyStep(_ context.Context, _ types.Step, _, _ string) error {
Check warning on line 242 in pipeline/backend/local/local.go Codecov / codecov/patch pipeline/backend/local/local.go#L242 `Added line #L242 was not covered by tests`
	// WaitStep already waits for the command to finish, so there is nothing to do here.
	return nil
	}

	// DestroyWorkflow the pipeline environment.
	func (e local) DestroyWorkflow(_ context.Context, _ types.Config, taskUUID, workflowName string) error {
	log.Trace().Str("taskUUID", taskUUID).Str("workflowName", workflowName).Msg("delete workflow environment")
Check warning on line 249 in pipeline/backend/local/local.go Codecov / codecov/patch pipeline/backend/local/local.go#L248-L249 `Added lines #L248 - L249 were not covered by tests`

	state, err := e.getState(taskUUID)
	if err != nil {
	if ctx.Err() != nil {
	ctx = GetShutdownCtx()
	}
	if err := r.engine.DestroyWorkflow(ctx, r.spec, r.taskUUID, r.Description["pipeline_name"]); err != nil {
Check warning on line 112 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L112 `Added line #L112 was not covered by tests`
	logger.Error().Err(err).Msg("could not destroy engine")
	}
	}()

	r.started = time.Now().Unix()
	if err := r.engine.SetupWorkflow(runnerCtx, r.spec, r.taskUUID, r.Description["pipeline_name"]); err != nil {
Check warning on line 118 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L118 `Added line #L118 was not covered by tests`
	return err
	}

	select {
	case <-r.ctx.Done():
	return ErrCancel
	case err := <-r.execAll(stage.Steps, r.Description["pipeline_name"]):
Check warning on line 126 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L126 `Added line #L126 was not covered by tests`
	if err != nil {
	r.err = err
	}
	}

	// Executes a set of parallel steps.
	func (r Runtime) execAll(steps []backend.Step, workflowName string) <-chan error {
Check warning on line 166 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L166 `Added line #L166 was not covered by tests`
	var g errgroup.Group
	done := make(chan error)
	logger := r.MakeLogger()
	Str("step", step.Name).
	Msg("executing")

	processState, err := r.exec(step, workflowName)
Check warning on line 209 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L209 `Added line #L209 was not covered by tests`

	logger.Debug().
	Str("step", step.Name).
	}

	// Executes the step and returns the state and error.
	func (r Runtime) exec(step backend.Step, workflowName string) (*backend.State, error) {
	if err := r.engine.StartStep(r.ctx, step, r.taskUUID, workflowName); err != nil {
Check warning on line 233 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L232-L233 `Added lines #L232 - L233 were not covered by tests`
	return nil, err
	}

	var wg sync.WaitGroup
	if r.logger != nil {
	rc, err := r.engine.TailStep(r.ctx, step, r.taskUUID, workflowName)
Check warning on line 239 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L239 `Added line #L239 was not covered by tests`
	if err != nil {
	return nil, err
	}
	// We wait until all data was logged. (Needed for some backends like local as WaitStep kills the log stream)
	wg.Wait()

	waitState, err := r.engine.WaitStep(r.ctx, step, r.taskUUID, workflowName)
Check warning on line 264 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L264 `Added line #L264 was not covered by tests`
	if err != nil {
	if errors.Is(err, context.Canceled) {
	return waitState, ErrCancel
	return nil, err
	}

	if err := r.engine.DestroyStep(r.ctx, step, r.taskUUID, workflowName); err != nil {
Check warning on line 272 in pipeline/pipeline.go Codecov / codecov/patch pipeline/pipeline.go#L272 `Added line #L272 was not covered by tests`
	return nil, err
	}