fix: update local executor code to include outputs (#626)

Author: ecrupper

executor/local/build.go

@@ -237,6 +237,11 @@ func (c *client) AssembleBuild(ctx context.Context) error {
 	// output a new line for readability to stdout
 	fmt.Fprintln(c.stdout, "")
 
+	c.err = c.outputs.create(ctx, c.OutputCtn, (int64(60) * 30))
+	if c.err != nil {
+		return fmt.Errorf("unable to create outputs container: %w", c.err)
+	}
+
 	// assemble runtime build just before any containers execute
 	c.err = c.Runtime.AssembleBuild(ctx, c.pipeline)
 	if c.err != nil {
@@ -253,6 +258,15 @@ func (c *client) ExecBuild(ctx context.Context) error {
 	// https://pkg.go.dev/github.com/go-vela/worker/internal/build#Upload
 	defer func() { build.Upload(c.build, nil, c.err, nil) }()
 
+	// output maps for dynamic environment variables captured from volume
+	var opEnv, maskEnv map[string]string
+
+	// execute outputs container
+	c.err = c.outputs.exec(ctx, c.OutputCtn)
+	if c.err != nil {
+		return fmt.Errorf("unable to exec outputs container: %w", c.err)
+	}
+
 	// execute the services for the pipeline
 	for _, _service := range c.pipeline.Services {
 		// plan the service
@@ -294,6 +308,22 @@ func (c *client) ExecBuild(ctx context.Context) error {
 			return fmt.Errorf("unable to plan step: %w", c.err)
 		}
 
+		// poll outputs
+		opEnv, maskEnv, c.err = c.outputs.poll(ctx, c.OutputCtn)
+		if c.err != nil {
+			return fmt.Errorf("unable to exec outputs container: %w", c.err)
+		}
+
+		// merge env from outputs
+		//
+		//nolint:errcheck // only errors with empty environment input, which does not matter here
+		_step.MergeEnv(opEnv)
+
+		// merge env from masked outputs
+		//
+		//nolint:errcheck // only errors with empty environment input, which does not matter here
+		_step.MergeEnv(maskEnv)
+
 		// execute the step
 		c.err = c.ExecStep(ctx, _step)
 		if c.err != nil {
@@ -451,6 +481,12 @@ func (c *client) DestroyBuild(ctx context.Context) error {
 		}
 	}
 
+	// destroy output container
+	err = c.outputs.destroy(ctx, c.OutputCtn)
+	if err != nil {
+		fmt.Fprintln(c.stdout, "unable to destroy output container:", err)
+	}
+
 	// remove the runtime volume for the pipeline
 	err = c.Runtime.RemoveVolume(ctx, c.pipeline)
 	if err != nil {

executor/local/build_test.go

@@ -73,6 +73,7 @@ func TestLocal_CreateBuild(t *testing.T) {
 				WithBuild(_build),
 				WithPipeline(_pipeline),
 				WithRuntime(_runtime),
+				WithOutputCtn(testOutputsCtn()),
 			)
 			if err != nil {
 				t.Errorf("unable to create executor engine: %v", err)
@@ -152,6 +153,7 @@ func TestLocal_PlanBuild(t *testing.T) {
 				WithBuild(_build),
 				WithPipeline(_pipeline),
 				WithRuntime(_runtime),
+				WithOutputCtn(testOutputsCtn()),
 			)
 			if err != nil {
 				t.Errorf("unable to create executor engine: %v", err)
@@ -270,6 +272,7 @@ func TestLocal_AssembleBuild(t *testing.T) {
 				WithBuild(_build),
 				WithPipeline(_pipeline),
 				WithRuntime(_runtime),
+				WithOutputCtn(testOutputsCtn()),
 				withStreamRequests(streamRequests),
 			)
 			if err != nil {
@@ -374,6 +377,7 @@ func TestLocal_ExecBuild(t *testing.T) {
 				WithBuild(_build),
 				WithPipeline(_pipeline),
 				WithRuntime(_runtime),
+				WithOutputCtn(testOutputsCtn()),
 				withStreamRequests(streamRequests),
 			)
 			if err != nil {
@@ -570,6 +574,7 @@ func TestLocal_StreamBuild(t *testing.T) {
 				WithBuild(_build),
 				WithPipeline(_pipeline),
 				WithRuntime(_runtime),
+				WithOutputCtn(testOutputsCtn()),
 				withStreamRequests(streamRequests),
 			)
 			if err != nil {
@@ -691,6 +696,7 @@ func TestLocal_DestroyBuild(t *testing.T) {
 				WithBuild(_build),
 				WithPipeline(_pipeline),
 				WithRuntime(_runtime),
+				WithOutputCtn(testOutputsCtn()),
 			)
 			if err != nil {
 				t.Errorf("unable to create executor engine: %v", err)
@@ -718,3 +724,11 @@ func TestLocal_DestroyBuild(t *testing.T) {
 		})
 	}
 }
+
+func testOutputsCtn() *pipeline.Container {
+	return &pipeline.Container{
+		ID:          "outputs_test",
+		Environment: make(map[string]string),
+		Detach:      true,
+	}
+}

executor/local/local.go

@@ -17,10 +17,11 @@ import (
 type (
 	// client manages communication with the pipeline resources.
 	client struct {
-		Vela     *vela.Client
-		Runtime  runtime.Engine
-		Hostname string
-		Version  string
+		Vela      *vela.Client
+		Runtime   runtime.Engine
+		Hostname  string
+		Version   string
+		OutputCtn *pipeline.Container
 
 		// private fields
 		init           *pipeline.Container
@@ -31,11 +32,17 @@ type (
 		err            error
 		streamRequests chan message.StreamRequest
 
+		outputs *outputSvc
+
 		// internal field partially exported for tests
 		stdout           *os.File
 		mockStdoutReader *os.File
 	}
 
+	svc struct {
+		client *client
+	}
+
 	// MockedClient is for internal use to facilitate testing the local executor.
 	MockedClient interface {
 		MockStdout() *os.File
@@ -77,6 +84,8 @@ func New(opts ...Opt) (*client, error) {
 	// Add stdout by default
 	c.stdout = os.Stdout
 
+	c.outputs = &outputSvc{client: c}
+
 	// instantiate streamRequests channel (which may be overridden using withStreamRequests()).
 	c.streamRequests = make(chan message.StreamRequest)
 

executor/local/opts.go

@@ -120,6 +120,16 @@ func WithMockStdout(mock bool) Opt {
 	}
 }
 
+// WithOutputCtn sets the outputs container in the executor client for Linux.
+func WithOutputCtn(ctn *pipeline.Container) Opt {
+	return func(c *client) error {
+		// set the outputs container in the client
+		c.OutputCtn = ctn
+
+		return nil
+	}
+}
+
 // withStreamRequests sets the streamRequests channel in the executor client for Linux
 // (primarily used for tests).
 func withStreamRequests(s chan message.StreamRequest) Opt {

executor/local/outputs.go

@@ -0,0 +1,130 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package local
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"fmt"
+
+	envparse "github.com/hashicorp/go-envparse"
+	"github.com/sirupsen/logrus"
+
+	"github.com/go-vela/server/compiler/types/pipeline"
+)
+
+// outputSvc handles communication with the outputs container during the build.
+type outputSvc svc
+
+// create configures the outputs container for execution.
+func (o *outputSvc) create(ctx context.Context, ctn *pipeline.Container, timeout int64) error {
+	// exit if outputs container has not been configured
+	if len(ctn.Image) == 0 {
+		return nil
+	}
+
+	// Encode script content to Base64
+	script := base64.StdEncoding.EncodeToString(
+		[]byte(fmt.Sprintf("mkdir /vela/outputs\nsleep %d\n", timeout)),
+	)
+
+	// set the entrypoint for the ctn
+	ctn.Entrypoint = []string{"/bin/sh", "-c"}
+
+	// set the commands for the ctn
+	ctn.Commands = []string{"echo $VELA_BUILD_SCRIPT | base64 -d | /bin/sh -e"}
+
+	// set the environment variables for the ctn
+	ctn.Environment["HOME"] = "/root"
+	ctn.Environment["SHELL"] = "/bin/sh"
+	ctn.Environment["VELA_BUILD_SCRIPT"] = script
+
+	// setup the runtime container
+	err := o.client.Runtime.SetupContainer(ctx, ctn)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// destroy cleans up outputs container after execution.
+func (o *outputSvc) destroy(ctx context.Context, ctn *pipeline.Container) error {
+	// exit if outputs container has not been configured
+	if len(ctn.Image) == 0 {
+		return nil
+	}
+
+	// inspect the runtime container
+	err := o.client.Runtime.InspectContainer(ctx, ctn)
+	if err != nil {
+		return err
+	}
+
+	// remove the runtime container
+	err = o.client.Runtime.RemoveContainer(ctx, ctn)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// exec runs the outputs sidecar container for a pipeline.
+func (o *outputSvc) exec(ctx context.Context, _outputs *pipeline.Container) error {
+	// exit if outputs container has not been configured
+	if len(_outputs.Image) == 0 {
+		return nil
+	}
+
+	// run the runtime container
+	err := o.client.Runtime.RunContainer(ctx, _outputs, o.client.pipeline)
+	if err != nil {
+		return err
+	}
+
+	// inspect the runtime container
+	err = o.client.Runtime.InspectContainer(ctx, _outputs)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// poll tails the output for sidecar container.
+func (o *outputSvc) poll(ctx context.Context, ctn *pipeline.Container) (map[string]string, map[string]string, error) {
+	// exit if outputs container has not been configured
+	if len(ctn.Image) == 0 {
+		return nil, nil, nil
+	}
+
+	// grab outputs
+	outputBytes, err := o.client.Runtime.PollOutputsContainer(ctx, ctn, "/vela/outputs/.env")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	reader := bytes.NewReader(outputBytes)
+
+	outputMap, err := envparse.Parse(reader)
+	if err != nil {
+		logrus.Debugf("unable to parse output map: %v", err)
+	}
+
+	// grab masked outputs
+	maskedBytes, err := o.client.Runtime.PollOutputsContainer(ctx, ctn, "/vela/outputs/masked.env")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	reader = bytes.NewReader(maskedBytes)
+
+	maskMap, err := envparse.Parse(reader)
+	if err != nil {
+		logrus.Debugf("unable to parse masked output map: %v", err)
+	}
+
+	return outputMap, maskMap, nil
+}