feat(controller) Emissary executor. (argoproj#4925)

Signed-off-by: Alex Collins <[email protected]>

Makefile

@@ -377,7 +377,7 @@ endif
 
 # for local we have a faster target that prints to stdout, does not use json, and can cache because it has no coverage
 .PHONY: test
-test: server/static/files.go
+test: server/static/files.go dist/argosay
  env KUBECONFIG=/dev/null $(GOTEST) ./...
 
 .PHONY: install
@@ -411,6 +411,9 @@ endif
 test/e2e/images/argosay/v2/argosay: test/e2e/images/argosay/v2/main/argosay.go
  cd test/e2e/images/argosay/v2 && GOOS=linux CGO_ENABLED=0 go build -ldflags '-w -s' -o argosay ./main
 
+dist/argosay: test/e2e/images/argosay/v2/main/argosay.go
+ go build -ldflags '-w -s' -o dist/argosay ./test/e2e/images/argosay/v2/main
+
 .PHONY: test-images
 test-images:
  $(call docker_pull,argoproj/argosay:v1)

cmd/argo/commands/submit.go

@@ -247,7 +247,7 @@ func waitWatchOrLog(ctx context.Context, serviceClient workflowpkg.WorkflowServi
  if cliSubmitOpts.log {
  for _, workflow := range workflowNames {
  logWorkflow(ctx, serviceClient, namespace, workflow, "", &corev1.PodLogOptions{
- Container: "main",
+ Container: common.MainContainerName,
  Follow: true,
  Previous: false,
  })

cmd/argoexec/commands/emissary.go

@@ -0,0 +1,218 @@
+package commands
+
+import (
+ "compress/gzip"
+ "encoding/json"
+ "fmt"
+ "io"
+ "io/ioutil"
+ "os"
+ "os/exec"
+ "os/signal"
+ "path/filepath"
+ "strconv"
+ "syscall"
+ "time"
+
+ log "github.com/sirupsen/logrus"
+ "github.com/spf13/cobra"
+
+ wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+ "github.com/argoproj/argo-workflows/v3/util/archive"
+ "github.com/argoproj/argo-workflows/v3/workflow/common"
+ "github.com/argoproj/argo-workflows/v3/workflow/util/path"
+)
+
+var (
+ varRunArgo = "/var/run/argo"
+ containerName = os.Getenv(common.EnvVarContainerName)
+ includeScriptOutput = os.Getenv(common.EnvVarIncludeScriptOutput) == "true" // capture stdout/stderr
+ template = &wfv1.Template{}
+ logger = log.WithField("argo", true)
+)
+
+func NewEmissaryCommand() *cobra.Command {
+ return &cobra.Command{
+ Use: "emissary",
+ SilenceUsage: true, // this prevents confusing usage message being printed when we SIGTERM
+ RunE: func(cmd *cobra.Command, args []string) error {
+ exitCode := 64
+
+ defer func() {
+ err := ioutil.WriteFile(varRunArgo+"/ctr/"+containerName+"/exitcode", []byte(strconv.Itoa(exitCode)), 0600)
+ if err != nil {
+ logger.Error(fmt.Errorf("failed to write exit code: %w", err))
+ }
+ }()
+
+ // this also indicates we've started
+ if err := os.MkdirAll(varRunArgo+"/ctr/"+containerName, 0700); err != nil {
+ return fmt.Errorf("failed to create ctr directory: %w", err)
+ }
+
+ name, args := args[0], args[1:]
+
+ signals := make(chan os.Signal, 1)
+ defer close(signals)
+ signal.Notify(signals)
+ defer signal.Reset()
+ go func() {
+ for s := range signals {
+ if s != syscall.SIGCHLD {
+ _ = syscall.Kill(-os.Getpid(), s.(syscall.Signal))
+ }
+ }
+ }()
+
+ data, err := ioutil.ReadFile(varRunArgo + "/template")
+ if err != nil {
+ return fmt.Errorf("failed to read template: %w", err)
+ }
+
+ if err := json.Unmarshal(data, template); err != nil {
+ return fmt.Errorf("failed to unmarshal template: %w", err)
+ }
+
+ name, err = path.Search(name)
+ if err != nil {
+ return fmt.Errorf("failed to find name in PATH: %w", err)
+ }
+
+ command := exec.Command(name, args...)
+ command.Env = os.Environ()
+ command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+ command.Stdout = os.Stdout
+ command.Stderr = os.Stderr
+
+ // this may not be that important an optimisation, except for very long logs we don't want to capture
+ if includeScriptOutput {
+ logger.Info("capturing script output")
+ stdout, err := os.Create(varRunArgo + "/ctr/" + containerName + "/stdout")
+ if err != nil {
+ return fmt.Errorf("failed to open stdout: %w", err)
+ }
+ defer func() { _ = stdout.Close() }()
+ command.Stdout = io.MultiWriter(os.Stdout, stdout)
+
+ stderr, err := os.Create(varRunArgo + "/ctr/" + containerName + "/stderr")
+ if err != nil {
+ return fmt.Errorf("failed to open stderr: %w", err)
+ }
+ defer func() { _ = stderr.Close() }()
+ command.Stderr = io.MultiWriter(os.Stderr, stderr)
+ }
+
+ if err := command.Start(); err != nil {
+ return err
+ }
+
+ go func() {
+ for {
+ data, _ := ioutil.ReadFile(varRunArgo + "/ctr/" + containerName + "/signal")
+ _ = os.Remove(varRunArgo + "/ctr/" + containerName + "/signal")
+ s, _ := strconv.Atoi(string(data))
+ if s > 0 {
+ _ = syscall.Kill(command.Process.Pid, syscall.Signal(s))
+ }
+ time.Sleep(2 * time.Second)
+ }
+ }()
+
+ cmdErr := command.Wait()
+
+ if cmdErr == nil {
+ exitCode = 0
+ } else if exitError, ok := cmdErr.(*exec.ExitError); ok {
+ if exitError.ExitCode() >= 0 {
+ exitCode = exitError.ExitCode()
+ } else {
+ exitCode = 137 // SIGTERM
+ }
+ }
+
+ if containerName == common.MainContainerName {
+ for _, x := range template.Outputs.Parameters {
+ if x.ValueFrom != nil && x.ValueFrom.Path != "" {
+ if err := saveParameter(x.ValueFrom.Path); err != nil {
+ return err
+ }
+ }
+ }
+ for _, x := range template.Outputs.Artifacts {
+ if x.Path != "" {
+ if err := saveArtifact(x.Path); err != nil {
+ return err
+ }
+ }
+ }
+ } else {
+ logger.Info("not saving outputs - not main container")
+ }
+
+ return cmdErr // this is the error returned from cmd.Wait(), which maybe an exitError
+ },
+ }
+}
+
+func saveArtifact(srcPath string) error {
+ if common.FindOverlappingVolume(template, srcPath) != nil {
+ logger.Infof("no need to save artifact - on overlapping volume: %s", srcPath)
+ return nil
+ }
+ if _, err := os.Stat(srcPath); os.IsNotExist(err) { // might be optional, so we ignore
+ logger.WithError(err).Errorf("cannot save artifact %s", srcPath)
+ return nil
+ }
+ dstPath := varRunArgo + "/outputs/artifacts/" + srcPath + ".tgz"
+ logger.Infof("%s -> %s", srcPath, dstPath)
+ z := filepath.Dir(dstPath)
+ if err := os.MkdirAll(z, 0700); err != nil { // chmod rwx------
+ return fmt.Errorf("failed to create directory %s: %w", z, err)
+ }
+ dst, err := os.Create(dstPath)
+ if err != nil {
+ return fmt.Errorf("failed to create destination %s: %w", dstPath, err)
+ }
+ defer func() { _ = dst.Close() }()
+ if err = archive.TarGzToWriter(srcPath, gzip.DefaultCompression, dst); err != nil {
+ return fmt.Errorf("failed to tarball the output %s to %s: %w", srcPath, dstPath, err)
+ }
+ if err = dst.Close(); err != nil {
+ return fmt.Errorf("failed to close %s: %w", dstPath, err)
+ }
+ return nil
+}
+
+func saveParameter(srcPath string) error {
+ if common.FindOverlappingVolume(template, srcPath) != nil {
+ logger.Infof("no need to save parameter - on overlapping volume: %s", srcPath)
+ return nil
+ }
+ src, err := os.Open(srcPath)
+ if os.IsNotExist(err) { // might be optional, so we ignore
+ logger.WithError(err).Errorf("cannot save parameter %s", srcPath)
+ return nil
+ }
+ if err != nil {
+ return fmt.Errorf("failed to open %s: %w", srcPath, err)
+ }
+ defer func() { _ = src.Close() }()
+ dstPath := varRunArgo + "/outputs/parameters/" + srcPath
+ logger.Infof("%s -> %s", srcPath, dstPath)
+ z := filepath.Dir(dstPath)
+ if err := os.MkdirAll(z, 0700); err != nil { // chmod rwx------
+ return fmt.Errorf("failed to create directory %s: %w", z, err)
+ }
+ dst, err := os.Create(dstPath)
+ if err != nil {
+ return fmt.Errorf("failed to create %s: %w", srcPath, err)
+ }
+ defer func() { _ = dst.Close() }()
+ if _, err = io.Copy(dst, src); err != nil {
+ return fmt.Errorf("failed to copy %s to %s: %w", srcPath, dstPath, err)
+ }
+ if err = dst.Close(); err != nil {
+ return fmt.Errorf("failed to close %s: %w", dstPath, err)
+ }
+ return nil
+}

cmd/argoexec/commands/emissary_test.go

@@ -0,0 +1,119 @@
+package commands
+
+import (
+ "io/ioutil"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strconv"
+ "sync"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestEmissary(t *testing.T) {
+ tmp, err := ioutil.TempDir("", "")
+ assert.NoError(t, err)
+
+ varRunArgo = tmp
+ includeScriptOutput = true
+
+ wd, err := os.Getwd()
+ assert.NoError(t, err)
+
+ x := filepath.Join(wd, "../../../dist/argosay")
+
+ err = ioutil.WriteFile(varRunArgo+"/template", []byte(`{}`), 0600)
+ assert.NoError(t, err)
+
+ t.Run("Exit0", func(t *testing.T) {
+ err := run(x, []string{"exit"})
+ assert.NoError(t, err)
+ data, err := ioutil.ReadFile(varRunArgo + "/ctr/main/exitcode")
+ assert.NoError(t, err)
+ assert.Equal(t, "0", string(data))
+ })
+ t.Run("Exit1", func(t *testing.T) {
+ err := run(x, []string{"exit", "1"})
+ assert.Equal(t, 1, err.(*exec.ExitError).ExitCode())
+ data, err := ioutil.ReadFile(varRunArgo + "/ctr/main/exitcode")
+ assert.NoError(t, err)
+ assert.Equal(t, "1", string(data))
+ })
+ t.Run("Stdout", func(t *testing.T) {
+ err := run(x, []string{"echo", "hello", "/dev/stdout"})
+ assert.NoError(t, err)
+ data, err := ioutil.ReadFile(varRunArgo + "/ctr/main/stdout")
+ assert.NoError(t, err)
+ assert.Equal(t, "hello", string(data))
+ })
+ t.Run("Stderr", func(t *testing.T) {
+ err := run(x, []string{"echo", "hello", "/dev/stderr"})
+ assert.NoError(t, err)
+ data, err := ioutil.ReadFile(varRunArgo + "/ctr/main/stderr")
+ assert.NoError(t, err)
+ assert.Equal(t, "hello", string(data))
+ })
+ t.Run("Signal", func(t *testing.T) {
+ for signal, message := range map[syscall.Signal]string{
+ syscall.SIGTERM: "terminated",
+ syscall.SIGKILL: "killed",
+ } {
+ err := ioutil.WriteFile(varRunArgo+"/ctr/main/signal", []byte(strconv.Itoa(int(signal))), 0600)
+ assert.NoError(t, err)
+ var wg sync.WaitGroup
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ err := run(x, []string{"sleep", "5s"})
+ assert.EqualError(t, err, "signal: "+message)
+ }()
+ time.Sleep(time.Second)
+ }
+ })
+ t.Run("Artifact", func(t *testing.T) {
+ err = ioutil.WriteFile(varRunArgo+"/template", []byte(`
+{
+ "outputs": {
+ "artifacts": [
+ {"path": "/tmp/artifact"}
+ ]
+ }
+}
+`), 0600)
+ assert.NoError(t, err)
+ err := run(x, []string{"echo", "hello", "/tmp/artifact"})
+ assert.NoError(t, err)
+ data, err := ioutil.ReadFile(varRunArgo + "/outputs/artifacts/tmp/artifact.tgz")
+ assert.NoError(t, err)
+ assert.NotEmpty(t, string(data)) // data is tgz format
+ })
+ t.Run("Parameter", func(t *testing.T) {
+ err = ioutil.WriteFile(varRunArgo+"/template", []byte(`
+{
+ "outputs": {
+ "parameters": [
+ {
+ "valueFrom": {"path": "/tmp/parameter"}
+ }
+ ]
+ }
+}
+`), 0600)
+ assert.NoError(t, err)
+ err := run(x, []string{"echo", "hello", "/tmp/parameter"})
+ assert.NoError(t, err)
+ data, err := ioutil.ReadFile(varRunArgo + "/outputs/parameters/tmp/parameter")
+ assert.NoError(t, err)
+ assert.Equal(t, "hello", string(data))
+ })
+}
+
+func run(name string, args []string) error {
+ cmd := NewEmissaryCommand()
+ containerName = "main"
+ return cmd.RunE(cmd, append([]string{name}, args...))
+}

cmd/argoexec/commands/init.go

@@ -28,6 +28,10 @@ func loadArtifacts(ctx context.Context) error {
  defer wfExecutor.HandleError(ctx)
  defer stats.LogStats()
 
+ if err := wfExecutor.Init(); err != nil {
+ wfExecutor.AddError(err)
+ return err
+ }
  // Download input artifacts
  err := wfExecutor.StageFiles()
  if err != nil {