fix(artifacts): Explicit archive strategy. Fixes #2140

Makefile

@@ -71,7 +71,7 @@ CONTROLLER_IMAGE_FILE  := dist/controller-image.$(VERSION)
 STATIC_BUILD          ?= true
 CI                    ?= false
 DB                    ?= postgres
-K3D                   := $(shell if [ `which kubectl` <> '' ] && [ "`kubectl config current-context`" = "k3s-default" ]; then echo true; else echo false; fi)
+K3D                   := $(shell if [ "`which kubectl`" != '' ] && [ "`kubectl config current-context`" = "k3s-default" ]; then echo true; else echo false; fi)
 # which components to start, useful if you want to disable them to debug
 COMPONENTS            := controller,argo-server
 LOG_LEVEL             := debug

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -1720,6 +1720,13 @@ func (args *Arguments) GetParameterByName(name string) *Parameter {
 	return nil
 }
 
+func (a *Artifact) GetArchive() *ArchiveStrategy {
+	if a == nil || a.Archive == nil {
+		return &ArchiveStrategy{}
+	}
+	return a.Archive
+}
+
 // GetTemplateByName retrieves a defined template by its name
 func (wf *Workflow) GetTemplateByName(name string) *Template {
 	for _, t := range wf.Spec.Templates {

pkg/apis/workflow/v1alpha1/workflow_types_test.go

@@ -38,6 +38,11 @@ func TestArtifactLocation_HasLocation(t *testing.T) {
 	assert.True(t, (&ArtifactLocation{GCS: &GCSArtifact{Key: "my-key", GCSBucket: GCSBucket{Bucket: "my-bucket"}}}).HasLocation())
 }
 
+func TestArtifact_GetArchive(t *testing.T) {
+	assert.NotNil(t, (&Artifact{}).GetArchive())
+	assert.Equal(t, &ArchiveStrategy{None: &NoneStrategy{}}, (&Artifact{Archive: &ArchiveStrategy{None: &NoneStrategy{}}}).GetArchive())
+}
+
 func TestNodes_FindByDisplayName(t *testing.T) {
 	assert.Nil(t, Nodes{}.FindByDisplayName(""))
 	assert.NotNil(t, Nodes{"": NodeStatus{DisplayName: "foo"}}.FindByDisplayName("foo"))

test/e2e/fixtures/when.go

@@ -120,6 +120,7 @@ func (w *When) waitForWorkflow(workflowName string, test func(wf *wfv1.Workflow)
 		time.Sleep(timeout)
 		timeoutCh <- true
 	}()
+	start := time.Now()
 	for {
 		select {
 		case event := <-watch.ResultChan():
@@ -128,7 +129,7 @@ func (w *When) waitForWorkflow(workflowName string, test func(wf *wfv1.Workflow)
 				logCtx.WithFields(log.Fields{"type": event.Type, "phase": wf.Status.Phase, "message": wf.Status.Message}).Info("...")
 				w.hydrateWorkflow(wf)
 				if test(wf) {
-					logCtx.Infof("Condition met")
+					logCtx.Infof("Condition met after %v", time.Since(start))
 					return w
 				}
 			} else {

test/e2e/functional_test.go

@@ -21,6 +21,18 @@ type FunctionalSuite struct {
 	fixtures.E2ESuite
 }
 
+func (s *FunctionalSuite) TestArchiveStrategies() {
+	s.Given().
+		Workflow(`@testdata/archive-strategies.yaml`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(30 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+}
+
 func (s *FunctionalSuite) TestContinueOnFail() {
 	s.Given().
 		Workflow(`

test/e2e/images/argosay/v2/main/argosay.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -25,6 +26,8 @@ func argosay(args ...string) error {
 		args = []string{"echo"}
 	}
 	switch args[0] {
+	case "assert_contains":
+		return assertContains(args[1:])
 	case "cat":
 		return cat(args[1:])
 	case "echo":
@@ -34,7 +37,24 @@ func argosay(args ...string) error {
 	case "sleep":
 		return sleep(args[1:])
 	}
-	return errors.New("usage: argosay [cat [file...]|echo [string] [file]|sleep duration|exit [code]]")
+	return errors.New("usage: argosay [assert_contains file string|cat [file...]|echo [string] [file]|sleep duration|exit [code]]")
+}
+
+func assertContains(args []string) error {
+	switch len(args) {
+	case 2:
+		filename := args[0]
+		substr := args[1]
+		data, err := ioutil.ReadFile(filename)
+		if err != nil {
+			return err
+		}
+		if !strings.Contains(string(data), substr) {
+			return fmt.Errorf(`expected "%s" to contain "%s", but was "%s"`, filename, substr, string(data))
+		}
+		return nil
+	}
+	return errors.New("usage: assert_contains data string")
 }
 
 func cat(args []string) error {

test/e2e/images/argosay/v2/main/argosay_test.go

@@ -11,6 +11,14 @@ func Test(t *testing.T) {
 		assert.NoError(t, argosay())
 		assert.Error(t, argosay("garbage"))
 	})
+	t.Run("assert_contains", func(t *testing.T) {
+		assert.NoError(t, argosay("echo", "foo", "/tmp/foo"))
+		assert.Error(t, argosay("assert_contains"))
+		assert.Error(t, argosay("assert_contains", "/tmp/foo"))
+		assert.Error(t, argosay("assert_contains", "/tmp/not-exists", "foo"))
+		assert.NoError(t, argosay("assert_contains", "/tmp/foo", "foo"))
+		assert.Error(t, argosay("assert_contains", "/tmp/foo", "bar"))
+	})
 	t.Run("echo", func(t *testing.T) {
 		assert.NoError(t, argosay("echo"))
 		assert.NoError(t, argosay("echo", "foo"))

test/e2e/testdata/archive-strategies.yaml

@@ -0,0 +1,106 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: archive-strategies
+  labels:
+    argo-e2e: true
+spec:
+  entrypoint: main
+  templates:
+    - name: main
+      dag:
+        tasks:
+          # use the default settings (auto-detect)
+          - name: generate-default
+            template: generate-default
+          - name: consume-default
+            template: consume-default
+            dependencies: [generate-default]
+            arguments:
+              artifacts:
+                - name: file-default
+                  from: "{{tasks.generate-default.outputs.artifacts.file-default}}"
+
+          # explicitly tar
+          - name: generate-tar
+            template: generate-tar
+          - name: consume-tar
+            template: consume-tar
+            dependencies: [generate-tar]
+            arguments:
+              artifacts:
+                - name: file-tar
+                  from: "{{tasks.generate-tar.outputs.artifacts.file-tar}}"
+
+          # explicitly none
+          - name: generate-none
+            template: generate-none
+          - name: consume-none
+            template: consume-none
+            dependencies: [generate-none]
+            arguments:
+              artifacts:
+                - name: file-none
+                  from: "{{tasks.generate-none.outputs.artifacts.file-none}}"
+
+    - name: generate-default
+      container:
+        image: argoproj/argosay:v2
+        args: [echo, hello, /tmp/file-default]
+      outputs:
+        artifacts:
+          - name: file-default
+            path: /tmp/file-default
+
+    - name: consume-default
+      inputs:
+        artifacts:
+          - name: file-default
+            path: /tmp/file-default
+      container:
+        image: argoproj/argosay:v2
+        args: [assert_contains, /tmp/file-default, hello]
+
+    - name: generate-tar
+      container:
+        image: argoproj/argosay:v2
+        args: [echo, hello, /tmp/file-tar]
+      outputs:
+        artifacts:
+          - name: file-tar
+            path: /tmp/file-tar
+            archive:
+              tar: {}
+
+    - name: consume-tar
+      inputs:
+        artifacts:
+          - name: file-tar
+            path: /tmp/file-tar
+            archive:
+              tar: {}
+      container:
+        image: argoproj/argosay:v2
+        args: [assert_contains, /tmp/file-tar, hello]
+
+    - name: generate-none
+      container:
+        image: argoproj/argosay:v2
+        args: [echo, hello, /tmp/file-none]
+      outputs:
+        artifacts:
+          - name: file-none
+            path: /tmp/file-none
+            archive:
+              none: {}
+
+    - name: consume-none
+      inputs:
+        artifacts:
+          - name: file-none
+            path: /tmp/file-none
+            archive:
+              none: {}
+      container:
+        image: argoproj/argosay:v2
+        args: [assert_contains, /tmp/file-none, hello]

workflow/executor/executor.go

@@ -133,7 +133,7 @@ func (we *WorkflowExecutor) LoadArtifacts() error {
 				log.Warnf("Ignoring optional artifact '%s' which was not supplied", art.Name)
 				continue
 			} else {
-				return errors.Errorf("", "required artifact %s not supplied", art.Name)
+				return errors.Errorf("required artifact %s not supplied", art.Name)
 			}
 		}
 		artDriver, err := we.InitDriver(&art)
@@ -171,11 +171,22 @@ func (we *WorkflowExecutor) LoadArtifacts() error {
 			return err
 		}
 
-		ok, err := isTarball(tempArtPath)
-		if err != nil {
-			return err
+		isTar := false
+		if art.GetArchive().None != nil {
+			// explicitly not a tar
+			isTar = false
+		} else if art.GetArchive().Tar != nil {
+			// explicitly a tar
+			isTar = true
+		} else {
+			// auto-detect
+			isTar, err = isTarball(tempArtPath)
+			if err != nil {
+				return err
+			}
 		}
-		if ok {
+
+		if isTar {
 			err = untar(tempArtPath, artPath)
 			_ = os.Remove(tempArtPath)
 		} else {
@@ -807,16 +818,19 @@ func (we *WorkflowExecutor) AddAnnotation(key, value string) error {
 
 // isTarball returns whether or not the file is a tarball
 func isTarball(filePath string) (bool, error) {
-	log.Info("Checking if the file is a tarball: ", filePath)
-
+	log.Infof("Detecting if %s is a tarball", filePath)
 	f, err := os.Open(filePath)
 	if err != nil {
-		return false, errors.InternalWrapError(err)
+		return false, err
 	}
 	defer f.Close()
-
-	tr := tar.NewReader(f)
-	_, err = tr.Next()
+	gzr, err := gzip.NewReader(f)
+	if err != nil {
+		return false, nil
+	}
+	defer gzr.Close()
+	tarr := tar.NewReader(gzr)
+	_, err = tarr.Next()
 	return err == nil, nil
 }
 

workflow/executor/executor_test.go

@@ -175,24 +175,23 @@ func TestDefaultParametersEmptyString(t *testing.T) {
 func TestIsTarball(t *testing.T) {
 	tests := []struct {
 		path      string
-		expectOK  bool
+		isTarball bool
 		expectErr bool
 	}{
-		{"testdata/istarball.tar", true, false},
-		{"testdata/istarball_small.csv", false, false},
-		{"testdata/istarball_small.csv.gz", false, false},
-		{"testdata/istarball_big.csv", false, false},
-		{"testdata/istarball_big.csv.gz", false, false},
-		{"testdata/istarball_notfound.csv", false, true},
+		{"testdata/file", false, false},
+		{"testdata/file.tar", false, false},
+		{"testdata/file.gz", false, false},
+		{"testdata/file.tar.gz", true, false},
+		{"testdata/not-found", false, true},
 	}
 
 	for _, test := range tests {
 		ok, err := isTarball(test.path)
 		if test.expectErr {
-			assert.Error(t, err)
+			assert.Error(t, err, test.path)
 		} else {
-			assert.NoError(t, err)
+			assert.NoError(t, err, test.path)
 		}
-		assert.Equal(t, test.expectOK, ok)
+		assert.Equal(t, test.isTarball, ok, test.path)
 	}
 }