Improve local backend (#1762)

Co-authored-by: 6543 <6543@obermui.de>

pipeline/backend/local/local.go

@@ -31,11 +31,6 @@ import (
 	"github.com/woodpecker-ci/woodpecker/shared/constant"
 )
 
-const (
-	workingSubDir = "work"
-	homeSubDir    = "home"
-)
-
 // notAllowedEnvVarOverwrites are all env vars that can not be overwritten by step config
 var notAllowedEnvVarOverwrites = []string{
 	"CI_NETRC_MACHINE",
@@ -46,16 +41,23 @@ var notAllowedEnvVarOverwrites = []string{
 	"SHELL",
 }
 
+type workflowState struct {
+	stepCMDs     map[string]*exec.Cmd
+	baseDir      string
+	homeDir      string
+	workspaceDir string
+}
+
 type local struct {
-	// TODO: make cmd a cmd list to iterate over, the hard part is to have a common ReadCloser
-	cmd             *exec.Cmd
-	output          io.ReadCloser
-	workflowBaseDir string
+	workflows map[string]*workflowState
+	output    io.ReadCloser
 }
 
 // New returns a new local Engine.
 func New() types.Engine {
-	return &local{}
+	return &local{
+		workflows: make(map[string]*workflowState),
+	}
 }
 
 func (e *local) Name() string {
@@ -67,25 +69,52 @@ func (e *local) IsAvailable(context.Context) bool {
 }
 
 func (e *local) Load(context.Context) error {
-	dir, err := os.MkdirTemp("", "woodpecker-local-*")
-	if err != nil {
-		return err
-	}
-	e.workflowBaseDir = dir
+	// TODO: download plugin-git binary if not exist
 
-	if err := os.Mkdir(filepath.Join(e.workflowBaseDir, workingSubDir), 0o700); err != nil {
-		return err
-	}
-	return os.Mkdir(filepath.Join(e.workflowBaseDir, homeSubDir), 0o700)
+	return nil
 }
 
 // Setup the pipeline environment.
-func (e *local) Setup(_ context.Context, _ *types.Config) error {
+func (e *local) Setup(_ context.Context, c *types.Config) error {
+	baseDir, err := os.MkdirTemp("", "woodpecker-local-*")
+	if err != nil {
+		return err
+	}
+
+	state := &workflowState{
+		stepCMDs:     make(map[string]*exec.Cmd),
+		baseDir:      baseDir,
+		workspaceDir: filepath.Join(baseDir, "workspace"),
+		homeDir:      filepath.Join(baseDir, "home"),
+	}
+
+	if err := os.Mkdir(state.homeDir, 0o700); err != nil {
+		return err
+	}
+
+	if err := os.Mkdir(state.workspaceDir, 0o700); err != nil {
+		return err
+	}
+
+	// TODO: copy plugin-git binary to homeDir and set PATH
+
+	workflowID, err := e.getWorkflowIDFromConfig(c)
+	if err != nil {
+		return err
+	}
+
+	e.workflows[workflowID] = state
+
 	return nil
 }
 
 // Exec the pipeline step.
 func (e *local) Exec(ctx context.Context, step *types.Step) error {
+	state, err := e.getWorkflowStateFromStep(step)
+	if err != nil {
+		return err
+	}
+
 	// Get environment variables
 	env := os.Environ()
 	for a, b := range step.Environment {
@@ -96,14 +125,13 @@ func (e *local) Exec(ctx context.Context, step *types.Step) error {
 	}
 
 	// Set HOME
-	env = append(env, "HOME="+filepath.Join(e.workflowBaseDir, homeSubDir))
+	env = append(env, "HOME="+state.homeDir)
 
 	var command []string
 	if step.Image == constant.DefaultCloneImage {
 		// Default clone step
 		// TODO: use tmp HOME and insert netrc and delete it after clone
-		// TODO: download plugin-git binary if not exist
-		env = append(env, "CI_WORKSPACE="+filepath.Join(e.workflowBaseDir, workingSubDir, step.Environment["CI_REPO"]))
+		env = append(env, "CI_WORKSPACE="+state.workspaceDir)
 		command = append(command, "plugin-git")
 	} else {
 		// Use "image name" as run command
@@ -122,30 +150,33 @@ func (e *local) Exec(ctx context.Context, step *types.Step) error {
 	}
 
 	// Prepare command
-	e.cmd = exec.CommandContext(ctx, command[0], command[1:]...)
-	e.cmd.Env = env
+	cmd := exec.CommandContext(ctx, command[0], command[1:]...)
+	cmd.Env = env
+	cmd.Dir = state.workspaceDir
 
-	// Prepare working directory
-	if step.Image == constant.DefaultCloneImage {
-		e.cmd.Dir = filepath.Join(e.workflowBaseDir, workingSubDir, step.Environment["CI_REPO_OWNER"])
-	} else {
-		e.cmd.Dir = filepath.Join(e.workflowBaseDir, workingSubDir, step.Environment["CI_REPO"])
-	}
-	err := os.MkdirAll(e.cmd.Dir, 0o700)
-	if err != nil {
-		return err
-	}
 	// Get output and redirect Stderr to Stdout
-	e.output, _ = e.cmd.StdoutPipe()
-	e.cmd.Stderr = e.cmd.Stdout
+	e.output, _ = cmd.StdoutPipe()
+	cmd.Stderr = cmd.Stdout
 
-	return e.cmd.Start()
+	state.stepCMDs[step.Name] = cmd
+
+	return cmd.Start()
 }
 
 // Wait for the pipeline step to complete and returns
 // the completion results.
-func (e *local) Wait(context.Context, *types.Step) (*types.State, error) {
-	err := e.cmd.Wait()
+func (e *local) Wait(_ context.Context, step *types.Step) (*types.State, error) {
+	state, err := e.getWorkflowStateFromStep(step)
+	if err != nil {
+		return nil, err
+	}
+
+	cmd, ok := state.stepCMDs[step.Name]
+	if !ok {
+		return nil, fmt.Errorf("step cmd %s not found", step.Name)
+	}
+
+	err = cmd.Wait()
 	ExitCode := 0
 
 	var execExitError *exec.ExitError
@@ -167,6 +198,69 @@ func (e *local) Tail(context.Context, *types.Step) (io.ReadCloser, error) {
 }
 
 // Destroy the pipeline environment.
-func (e *local) Destroy(context.Context, *types.Config) error {
-	return os.RemoveAll(e.workflowBaseDir)
+func (e *local) Destroy(_ context.Context, c *types.Config) error {
+	state, err := e.getWorkflowStateFromConfig(c)
+	if err != nil {
+		return err
+	}
+
+	err = os.RemoveAll(state.baseDir)
+	if err != nil {
+		return err
+	}
+
+	workflowID, err := e.getWorkflowIDFromConfig(c)
+	if err != nil {
+		return err
+	}
+
+	delete(e.workflows, workflowID)
+
+	return err
+}
+
+func (e *local) getWorkflowIDFromStep(step *types.Step) (string, error) {
+	prefix := strings.Split(step.Name, "_stage_")
+	if len(prefix) < 2 {
+		return "", fmt.Errorf("invalid step name %s", step.Name)
+	}
+
+	return prefix[0], nil
+}
+
+func (e *local) getWorkflowIDFromConfig(c *types.Config) (string, error) {
+	if len(c.Volumes) < 1 {
+		return "", fmt.Errorf("no volumes found in config")
+	}
+
+	prefix := strings.Replace(c.Volumes[0].Name, "_default", "", 1)
+	return prefix, nil
+}
+
+func (e *local) getWorkflowStateFromConfig(c *types.Config) (*workflowState, error) {
+	workflowID, err := e.getWorkflowIDFromConfig(c)
+	if err != nil {
+		return nil, err
+	}
+
+	state, ok := e.workflows[workflowID]
+	if !ok {
+		return nil, fmt.Errorf("workflow %s not found", workflowID)
+	}
+
+	return state, nil
+}
+
+func (e *local) getWorkflowStateFromStep(step *types.Step) (*workflowState, error) {
+	workflowID, err := e.getWorkflowIDFromStep(step)
+	if err != nil {
+		return nil, err
+	}
+
+	state, ok := e.workflows[workflowID]
+	if !ok {
+		return nil, fmt.Errorf("workflow %s not found", workflowID)
+	}
+
+	return state, nil
 }

pipeline/backend/types/engine.go

@@ -8,8 +8,13 @@ import (
 // Engine defines a container orchestration backend and is used
 // to create and manage container resources.
 type Engine interface {
+	// Name returns the name of the backend.
 	Name() string
+
+	// Check if the backend is available.
 	IsAvailable(context.Context) bool
+
+	// Load the backend engine.
 	Load(context.Context) error
 
 	// Setup the pipeline environment.