add support of env variables to exec/run/local jobs and refactor tests

cli/config_test.go

@@ -257,12 +257,12 @@ func (s *SuiteConfig) TestLabelsConfig(c *C) {
 				"some": {
 					requiredLabel: "true",
 					serviceLabel:  "true",
-					labelPrefix + "." + jobRun + ".job1.schedule": "schedule1",
-					labelPrefix + "." + jobRun + ".job1.command":  "command1",
-					labelPrefix + "." + jobRun + ".job1.env":   "KEY1=value1",
-					labelPrefix + "." + jobRun + ".job2.schedule": "schedule2",
-					labelPrefix + "." + jobRun + ".job2.command":  "command2",
-					labelPrefix + "." + jobRun + ".job2.env":   `["KEY1=value1", "KEY2=value2"]`,
+					labelPrefix + "." + jobRun + ".job1.schedule":    "schedule1",
+					labelPrefix + "." + jobRun + ".job1.command":     "command1",
+					labelPrefix + "." + jobRun + ".job1.environment": "KEY1=value1",
+					labelPrefix + "." + jobRun + ".job2.schedule":    "schedule2",
+					labelPrefix + "." + jobRun + ".job2.command":     "command2",
+					labelPrefix + "." + jobRun + ".job2.environment": `["KEY1=value1", "KEY2=value2"]`,
 				},
 			},
 			ExpectedConfig: Config{
@@ -271,14 +271,14 @@ func (s *SuiteConfig) TestLabelsConfig(c *C) {
 						Schedule: "schedule1",
 						Command:  "command1",
 					},
-						Env: []string{"KEY1=value1"},
+						Environment: []string{"KEY1=value1"},
 					},
 					},
 					"job2": {RunJob: core.RunJob{BareJob: core.BareJob{
 						Schedule: "schedule2",
 						Command:  "command2",
 					},
-						Env: []string{"KEY1=value1", "KEY2=value2"},
+						Environment: []string{"KEY1=value1", "KEY2=value2"},
 					},
 					},
 				},

cli/docker-labels.go

@@ -161,7 +161,7 @@ func setJobParam(params map[string]interface{}, paramName, paramVal string) {
 			params[paramName] = arr
 			return
 		}
-	case "env":
+	case "environment":
 		arr := []string{} // allow providing JSON arr of env keyvalues
 		if err := json.Unmarshal([]byte(paramVal), &arr); err == nil {
 			params[paramName] = arr

core/execjob.go

@@ -8,11 +8,14 @@ import (
 )
 
 type ExecJob struct {
-	BareJob   `mapstructure:",squash"`
-	Client    *docker.Client `json:"-"`
-	Container string
-	User      string `default:"root"`
-	TTY       bool   `default:"false"`
+	BareJob     `mapstructure:",squash"`
+	Client      *docker.Client `json:"-"`
+	Container   string
+	User        string `default:"root"`
+	TTY         bool   `default:"false"`
+	Environment []string
+
+	execID string
 }
 
 func NewExecJob(c *docker.Client) *ExecJob {
@@ -25,11 +28,27 @@ func (j *ExecJob) Run(ctx *Context) error {
 		return err
 	}
 
-	if err := j.startExec(ctx.Execution, exec); err != nil {
+	if exec != nil {
+		j.execID = exec.ID
+	}
+
+	if err := j.startExec(ctx.Execution); err != nil {
 		return err
 	}
 
-	return j.inspectExec(exec)
+	inspect, err := j.inspectExec()
+	if err != nil {
+		return err
+	}
+
+	switch inspect.ExitCode {
+	case 0:
+		return nil
+	case -1:
+		return ErrUnexpected
+	default:
+		return fmt.Errorf("error non-zero exit code: %d", inspect.ExitCode)
+	}
 }
 
 func (j *ExecJob) buildExec() (*docker.Exec, error) {
@@ -41,6 +60,7 @@ func (j *ExecJob) buildExec() (*docker.Exec, error) {
 		Cmd:          args.GetArgs(j.Command),
 		Container:    j.Container,
 		User:         j.User,
+		Env:          j.Environment,
 	})
 
 	if err != nil {
@@ -50,8 +70,8 @@ func (j *ExecJob) buildExec() (*docker.Exec, error) {
 	return exec, nil
 }
 
-func (j *ExecJob) startExec(e *Execution, exec *docker.Exec) error {
-	err := j.Client.StartExec(exec.ID, docker.StartExecOptions{
+func (j *ExecJob) startExec(e *Execution) error {
+	err := j.Client.StartExec(j.execID, docker.StartExecOptions{
 		Tty:          j.TTY,
 		OutputStream: e.OutputStream,
 		ErrorStream:  e.ErrorStream,
@@ -65,19 +85,12 @@ func (j *ExecJob) startExec(e *Execution, exec *docker.Exec) error {
 	return nil
 }
 
-func (j *ExecJob) inspectExec(exec *docker.Exec) error {
-	i, err := j.Client.InspectExec(exec.ID)
+func (j *ExecJob) inspectExec() (*docker.ExecInspect, error) {
+	i, err := j.Client.InspectExec(j.execID)
 
 	if err != nil {
-		return fmt.Errorf("error inspecting exec: %s", err)
+		return i, fmt.Errorf("error inspecting exec: %s", err)
 	}
 
-	switch i.ExitCode {
-	case 0:
-		return nil
-	case -1:
-		return ErrUnexpected
-	default:
-		return fmt.Errorf("error non-zero exit code: %d", i.ExitCode)
-	}
+	return i, nil
 }

core/execjob_test.go

@@ -3,8 +3,10 @@ package core
 import (
 	"archive/tar"
 	"bytes"
+	"encoding/json"
+	"net/http"
 
-	"github.com/fsouza/go-dockerclient"
+	docker "github.com/fsouza/go-dockerclient"
 	"github.com/fsouza/go-dockerclient/testing"
 	. "gopkg.in/check.v1"
 )
@@ -18,11 +20,33 @@ type SuiteExecJob struct {
 
 var _ = Suite(&SuiteExecJob{})
 
+// overwrite version handler, because
+// exec configuration Env is only supported in API#1.25 and above
+// https://github.com/fsouza/go-dockerclient/blob/0f57349a7248b9b35ad2193ffe70953d5893e2b8/testing/server.go#L1607
+func versionDockerHandler(w http.ResponseWriter, r *http.Request) {
+	envs := map[string]interface{}{
+		"Version":       "1.10.1",
+		"Os":            "linux",
+		"KernelVersion": "3.13.0-77-generic",
+		"GoVersion":     "go1.17.1",
+		"GitCommit":     "9e83765",
+		"Arch":          "amd64",
+		"ApiVersion":    "1.27",
+		"BuildTime":     "2015-12-01T07:09:13.444803460+00:00",
+		"Experimental":  false,
+	}
+	w.WriteHeader(http.StatusOK)
+	json.NewEncoder(w).Encode(envs)
+
+}
+
 func (s *SuiteExecJob) SetUpTest(c *C) {
 	var err error
 	s.server, err = testing.NewServer("127.0.0.1:0", nil, nil)
 	c.Assert(err, IsNil)
 
+	s.server.CustomHandler("/version", http.HandlerFunc(versionDockerHandler))
+
 	s.client, err = docker.NewClient(s.server.URL())
 	c.Assert(err, IsNil)
 
@@ -38,6 +62,7 @@ func (s *SuiteExecJob) TestRun(c *C) {
 	job := &ExecJob{Client: s.client}
 	job.Container = ContainerFixture
 	job.Command = `echo -a "foo bar"`
+	job.Environment = []string{"test_Key1=value1", "test_Key2=value2"}
 	job.User = "foo"
 	job.TTY = true
 
@@ -49,13 +74,15 @@ func (s *SuiteExecJob) TestRun(c *C) {
 
 	container, err := s.client.InspectContainer(ContainerFixture)
 	c.Assert(err, IsNil)
+	c.Assert(len(container.ExecIDs) > 0, Equals, true)
 
-	exec, err := s.client.InspectExec(container.ExecIDs[0])
+	exec, err := job.inspectExec()
 	c.Assert(err, IsNil)
 	c.Assert(exec.ProcessConfig.EntryPoint, Equals, "echo")
 	c.Assert(exec.ProcessConfig.Arguments, DeepEquals, []string{"-a", "foo bar"})
 	c.Assert(exec.ProcessConfig.User, Equals, "foo")
 	c.Assert(exec.ProcessConfig.Tty, Equals, true)
+	// no way to check for env :|
 }
 
 func (s *SuiteExecJob) buildContainer(c *C) {

core/localjob.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"os"
 	"os/exec"
 
 	"github.com/gobs/args"
@@ -37,7 +38,9 @@ func (j *LocalJob) buildCommand(ctx *Context) (*exec.Cmd, error) {
 		Args:   args,
 		Stdout: ctx.Execution.OutputStream,
 		Stderr: ctx.Execution.ErrorStream,
-		Env:    j.Environment,
-		Dir:    j.Dir,
+		// add custom env variables to the existing ones
+		// instead of overwriting them
+		Env: append(os.Environ(), j.Environment...),
+		Dir: j.Dir,
 	}, nil
 }

core/localjob_test.go

@@ -1,6 +1,8 @@
 package core
 
 import (
+	"strings"
+
 	"github.com/armon/circbuf"
 
 	. "gopkg.in/check.v1"
@@ -22,3 +24,29 @@ func (s *SuiteLocalJob) TestRun(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(b.String(), Equals, "foo bar\n")
 }
+
+func (s *SuiteLocalJob) TestEnvironment(c *C) {
+	job := &LocalJob{}
+	job.Command = `env`
+	env := []string{"test_Key1=value1", "test_Key2=value2"}
+	job.Environment = env
+
+	b, _ := circbuf.NewBuffer(1000)
+	e := NewExecution()
+	e.OutputStream = b
+
+	err := job.Run(&Context{Execution: e})
+	c.Assert(err, IsNil)
+
+	// check that expected keys are present in the system env
+	for _, expectedEnv := range env {
+		found := false
+		for _, systemEnv := range strings.Split(strings.TrimSuffix(b.String(), "\n"), "\n") {
+			if expectedEnv == systemEnv {
+				found = true
+				break
+			}
+		}
+		c.Assert(found, Equals, true)
+	}
+}

core/runjob.go

@@ -29,11 +29,13 @@ type RunJob struct {
 	Delete string `default:"true"`
 	Pull   string `default:"true"`
 
-	Image     string
-	Network   string
-	Container string
-	Volume    []string
-	Env       []string
+	Image       string
+	Network     string
+	Container   string
+	Volume      []string
+	Environment []string
+
+	containerID string
 }
 
 func NewRunJob(c *docker.Client) *RunJob {
@@ -92,18 +94,31 @@ func (j *RunJob) Run(ctx *Context) error {
 			return err
 		}
 	} else {
-		container, err = j.getContainer(j.Container)
+		container, err = j.Client.InspectContainer(j.Container)
 		if err != nil {
 			return err
 		}
 	}
 
+	if container != nil {
+		j.containerID = container.ID
+	}
+
+	// cleanup container if it is a created one
+	if j.Container == "" {
+		defer func() {
+			if delErr := j.deleteContainer(); delErr != nil {
+				ctx.Warn("failed to delete container: " + delErr.Error())
+			}
+		}()
+	}
+
 	startTime := time.Now()
-	if err := j.startContainer(ctx.Execution, container); err != nil {
+	if err := j.startContainer(); err != nil {
 		return err
 	}
 
-	err = j.watchContainer(container.ID)
+	err = j.watchContainer()
 	if err == ErrUnexpected {
 		return err
 	}
@@ -120,14 +135,6 @@ func (j *RunJob) Run(ctx *Context) error {
 		ctx.Warn("failed to fetch container logs: " + logsErr.Error())
 	}
 
-	if j.Container == "" {
-		defer func() {
-			if delErr := j.deleteContainer(container.ID); delErr != nil {
-				ctx.Warn("failed to delete container: " + delErr.Error())
-			}
-		}()
-	}
-
 	return err
 }
 
@@ -163,7 +170,7 @@ func (j *RunJob) buildContainer() (*docker.Container, error) {
 			Tty:          j.TTY,
 			Cmd:          args.GetArgs(j.Command),
 			User:         j.User,
-			Env:          j.Env,
+			Env:          j.Environment,
 		},
 		NetworkingConfig: &docker.NetworkingConfig{},
 		HostConfig: &docker.HostConfig{
@@ -193,12 +200,16 @@ func (j *RunJob) buildContainer() (*docker.Container, error) {
 	return c, nil
 }
 
-func (j *RunJob) startContainer(e *Execution, c *docker.Container) error {
-	return j.Client.StartContainer(c.ID, &docker.HostConfig{})
+func (j *RunJob) startContainer() error {
+	return j.Client.StartContainer(j.containerID, &docker.HostConfig{})
+}
+
+func (j *RunJob) stopContainer(timeout uint) error {
+	return j.Client.StopContainer(j.containerID, timeout)
 }
 
-func (j *RunJob) getContainer(id string) (*docker.Container, error) {
-	container, err := j.Client.InspectContainer(id)
+func (j *RunJob) getContainer() (*docker.Container, error) {
+	container, err := j.Client.InspectContainer(j.containerID)
 	if err != nil {
 		return nil, err
 	}
@@ -210,7 +221,7 @@ const (
 	maxProcessDuration = time.Hour * 24
 )
 
-func (j *RunJob) watchContainer(containerID string) error {
+func (j *RunJob) watchContainer() error {
 	var s docker.State
 	var r time.Duration
 	for {
@@ -221,7 +232,7 @@ func (j *RunJob) watchContainer(containerID string) error {
 			return ErrMaxTimeRunning
 		}
 
-		c, err := j.Client.InspectContainer(containerID)
+		c, err := j.Client.InspectContainer(j.containerID)
 		if err != nil {
 			return err
 		}
@@ -242,12 +253,12 @@ func (j *RunJob) watchContainer(containerID string) error {
 	}
 }
 
-func (j *RunJob) deleteContainer(containerID string) error {
+func (j *RunJob) deleteContainer() error {
 	if delete, _ := strconv.ParseBool(j.Delete); !delete {
 		return nil
 	}
 
 	return j.Client.RemoveContainer(docker.RemoveContainerOptions{
-		ID: containerID,
+		ID: j.containerID,
 	})
 }