woodpecker/pipeline/pipeline.go

package pipeline

import (
	"context"
	"strings"
	"sync"
	"time"

	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
	"golang.org/x/sync/errgroup"

	backend "github.com/woodpecker-ci/woodpecker/pipeline/backend/types"
	"github.com/woodpecker-ci/woodpecker/pipeline/multipart"
)

type (
	// State defines the pipeline and process state.
	State struct {
		// Global state of the pipeline.
		Pipeline struct {
			// Pipeline time started
			Time int64 `json:"time"`
			// Current pipeline step
			Step *backend.Step `json:"step"`
			// Current pipeline error state
			Error error `json:"error"`
		}

		// Current process state.
		Process *backend.State
	}
)

// Runtime is a configuration runtime.
type Runtime struct {
	err     error
	spec    *backend.Config
	engine  backend.Engine
	started int64

	ctx    context.Context
	tracer Tracer
	logger Logger

	Description map[string]string // The runtime descriptors.
}

// New returns a new runtime using the specified runtime
// configuration and runtime engine.
func New(spec *backend.Config, opts ...Option) *Runtime {
	r := new(Runtime)
	r.Description = map[string]string{}
	r.spec = spec
	r.ctx = context.Background()
	for _, opts := range opts {
		opts(r)
	}
	return r
}

func (r *Runtime) MakeLogger() zerolog.Logger {
	logCtx := log.With()
	for key, val := range r.Description {
		logCtx = logCtx.Str(key, val)
	}
	return logCtx.Logger()
}

// Starts the execution of the pipeline and waits for it to complete
func (r *Runtime) Run() error {
	logger := r.MakeLogger()
	logger.Debug().Msgf("Executing %d stages, in order of:", len(r.spec.Stages))
	for _, stage := range r.spec.Stages {
		steps := []string{}
		for _, step := range stage.Steps {
			steps = append(steps, step.Name)
		}

		logger.Debug().
			Str("Stage", stage.Name).
			Str("Steps", strings.Join(steps, ",")).
			Msg("stage")
	}

	defer func() {
		if err := r.engine.Destroy(r.ctx, r.spec); err != nil {
			logger.Error().Err(err).Msg("could not destroy engine")
		}
	}()

	r.started = time.Now().Unix()
	if err := r.engine.Setup(r.ctx, r.spec); err != nil {
		return err
	}

	for _, stage := range r.spec.Stages {
		select {
		case <-r.ctx.Done():
			return ErrCancel
		case err := <-r.execAll(stage.Steps):
			if err != nil {
				r.err = err
			}
		}
	}

	return r.err
}

// Updates the current status of a step
func (r *Runtime) traceStep(processState *backend.State, err error, step *backend.Step) error {
	if r.tracer == nil {
		// no tracer nothing to trace :)
		return nil
	}

	if processState == nil {
		processState = new(backend.State)
		if err != nil {
			processState.Error = err
			processState.Exited = true
			processState.OOMKilled = false
			processState.ExitCode = 126 // command invoked cannot be executed.
		}
	}

	state := new(State)
	state.Pipeline.Time = r.started
	state.Pipeline.Step = step
	state.Process = processState // empty
	state.Pipeline.Error = r.err

	if traceErr := r.tracer.Trace(state); traceErr != nil {
		return traceErr
	}
	return err
}

// Executes a set of parallel steps
func (r *Runtime) execAll(steps []*backend.Step) <-chan error {
	var g errgroup.Group
	done := make(chan error)
	logger := r.MakeLogger()

	for _, step := range steps {
		// required since otherwise the loop variable
		// will be captured by the function. This will
		// recreate the step "variable"
		step := step
		g.Go(func() error {
			// Case the pipeline was already complete.
			logger.Debug().
				Str("Step", step.Name).
				Msg("Prepare")

			switch {
			case r.err != nil && !step.OnFailure:
				logger.Debug().
					Str("Step", step.Name).
					Err(r.err).
					Msgf("Skipped due to OnFailure=%t", step.OnFailure)
				return nil
			case r.err == nil && !step.OnSuccess:
				logger.Debug().
					Str("Step", step.Name).
					Msgf("Skipped due to OnSuccess=%t", step.OnSuccess)
				return nil
			}

			// Trace started.
			err := r.traceStep(nil, nil, step)
			if err != nil {
				return err
			}

			logger.Debug().
				Str("Step", step.Name).
				Msg("Executing")

			processState, err := r.exec(step)

			logger.Debug().
				Str("Step", step.Name).
				Msg("Complete")

			// if we got a nil process but an error state
			// then we need to log the internal error to the step.
			if r.logger != nil && err != nil && processState == nil {
				_ = r.logger.Log(step, multipart.New(strings.NewReader(
					"Backend engine error while running step: "+err.Error(),
				)))
			}

			// Return the error after tracing it.
			return r.traceStep(processState, err, step)
		})
	}

	go func() {
		done <- g.Wait()
		close(done)
	}()
	return done
}

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

	var wg sync.WaitGroup
	if r.logger != nil {
		rc, err := r.engine.Tail(r.ctx, step)
		if err != nil {
			return nil, err
		}

		wg.Add(1)
		go func() {
			defer wg.Done()
			logger := r.MakeLogger()

			if err := r.logger.Log(step, multipart.New(rc)); err != nil {
				logger.Error().Err(err).Msg("process logging failed")
			}
			_ = rc.Close()
		}()
	}

	// nothing else to do, this is a detached process.
	if step.Detached {
		return nil, nil
	}

	// Some pipeline backends, such as local, will close the pipe from Tail on Wait,
	// so first make sure all reading has finished.
	wg.Wait()
	waitState, err := r.engine.Wait(r.ctx, step)
	if err != nil {
		return nil, err
	}

	if waitState.OOMKilled {
		return waitState, &OomError{
			Name: step.Name,
			Code: waitState.ExitCode,
		}
	} else if waitState.ExitCode != 0 {
		return waitState, &ExitError{
			Name: step.Name,
			Code: waitState.ExitCode,
		}
	}

	return waitState, nil
}