feat(executor): Minimize the number of Kubernetes API requests made by executors

.github/workflows/ci-build.yaml

@@ -77,7 +77,7 @@ jobs:
       - name: Pre-pull images
         env:
           GOPATH: /home/runner/go
-        run: make pull-build-images test-images &
+        run: make test-images &
       - name: Create Kubeconfig
         run: |
           mkdir -p ~/.kube

Makefile

@@ -383,7 +383,7 @@ install: $(MANIFESTS) $(E2E_MANIFESTS) dist/kustomize
 	kubectl get ns $(KUBE_NAMESPACE) || kubectl create ns $(KUBE_NAMESPACE)
 	kubectl config set-context --current --namespace=$(KUBE_NAMESPACE)
 	@echo "installing PROFILE=$(PROFILE) VERSION=$(VERSION), E2E_EXECUTOR=$(E2E_EXECUTOR)"
-	dist/kustomize build --load_restrictor=none test/e2e/manifests/$(PROFILE) | sed 's/image: argoproj/image: $(IMAGE_NAMESPACE)/' | sed 's/:latest/:$(VERSION)/' | sed 's/pns/$(E2E_EXECUTOR)/' | kubectl -n $(KUBE_NAMESPACE) apply -f -
+	dist/kustomize build --load_restrictor=none test/e2e/manifests/$(PROFILE) | sed 's/argoproj\//$(IMAGE_NAMESPACE)\//' | sed 's/:latest/:$(VERSION)/' | sed 's/pns/$(E2E_EXECUTOR)/' | kubectl -n $(KUBE_NAMESPACE) apply -f -
 	kubectl -n $(KUBE_NAMESPACE) apply -f test/stress/massive-workflow.yaml
 	kubectl -n $(KUBE_NAMESPACE) rollout restart deploy workflow-controller
 	kubectl -n $(KUBE_NAMESPACE) rollout restart deploy argo-server
@@ -394,10 +394,6 @@ ifeq ($(RUN_MODE),kubernetes)
 	kubectl -n $(KUBE_NAMESPACE) scale deploy/argo-server --replicas 1
 endif
 
-.PHONY: pull-build-images
-pull-build-images:
-	./hack/pull-build-images.sh
-
 .PHONY: argosay
 argosay: test/e2e/images/argosay/v2/argosay
 	cd test/e2e/images/argosay/v2 && docker build . -t argoproj/argosay:v2

cmd/argoexec/commands/root.go

@@ -101,11 +101,11 @@ func initExecutor() *executor.WorkflowExecutor {
 	case common.ContainerRuntimeExecutorK8sAPI:
 		cre, err = k8sapi.NewK8sAPIExecutor(clientset, config, podName, namespace)
 	case common.ContainerRuntimeExecutorKubelet:
-		cre, err = kubelet.NewKubeletExecutor()
+		cre, err = kubelet.NewKubeletExecutor(namespace, podName)
 	case common.ContainerRuntimeExecutorPNS:
-		cre, err = pns.NewPNSExecutor(clientset, podName, namespace, tmpl.Outputs.HasOutputs())
+		cre, err = pns.NewPNSExecutor(clientset, podName, namespace)
 	default:
-		cre, err = docker.NewDockerExecutor()
+		cre, err = docker.NewDockerExecutor(namespace, podName)
 	}
 	checkErr(err)
 

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -621,6 +621,15 @@ func (tmpl *Template) HasPodSpecPatch() bool {
 	return tmpl.PodSpecPatch != ""
 }
 
+func (tmpl *Template) GetSidecarNames() []string {
+	var containerNames []string
+	for _, s := range tmpl.Sidecars {
+		containerNames = append(containerNames, s.Name)
+	}
+	return containerNames
+
+}
+
 type Artifacts []Artifact
 
 func (a Artifacts) GetArtifactByName(name string) *Artifact {

pkg/apis/workflow/v1alpha1/workflow_types_test.go

@@ -610,3 +610,12 @@ func TestWorkflow_GetSemaphoreKeys(t *testing.T) {
 	assert.Contains(keys, "test/template")
 	assert.Contains(keys, "test/template1")
 }
+
+func TestTemplate_GetSidecarNames(t *testing.T) {
+	m := &Template{
+		Sidecars: []UserContainer{
+			{Container: corev1.Container{Name: "sidecar-0"}},
+		},
+	}
+	assert.ElementsMatch(t, []string{"sidecar-0"}, m.GetSidecarNames())
+}

test/e2e/signals_test.go

@@ -14,6 +14,9 @@ import (
 	"github.com/argoproj/argo-workflows/v3/test/e2e/fixtures"
 )
 
+// Tests the use of signals to kill containers.
+// argoproj/argosay:v2 does not contain sh, so you must use argoproj/argosay:v1.
+// Killing often requires SIGKILL, which is issued 30s after SIGTERM. So tests need longer (>30s) timeout.
 type SignalsSuite struct {
 	fixtures.E2ESuite
 }
@@ -37,10 +40,10 @@ func (s *SignalsSuite) TestStopBehavior() {
 		WaitForWorkflow(1 * time.Minute).
 		Then().
 		ExpectWorkflow(func(t *testing.T, m *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
-			assert.Equal(t, wfv1.WorkflowFailed, status.Phase)
+			assert.Contains(t, []wfv1.WorkflowPhase{wfv1.WorkflowFailed, wfv1.WorkflowError}, status.Phase)
 			nodeStatus := status.Nodes.FindByDisplayName("A")
 			if assert.NotNil(t, nodeStatus) {
-				assert.Equal(t, wfv1.NodeFailed, nodeStatus.Phase)
+				assert.Contains(t, []wfv1.NodePhase{wfv1.NodeFailed, wfv1.NodeError}, nodeStatus.Phase)
 			}
 			nodeStatus = status.Nodes.FindByDisplayName("A.onExit")
 			if assert.NotNil(t, nodeStatus) {
@@ -66,10 +69,10 @@ func (s *SignalsSuite) TestTerminateBehavior() {
 		WaitForWorkflow(1 * time.Minute).
 		Then().
 		ExpectWorkflow(func(t *testing.T, m *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
-			assert.Equal(t, wfv1.WorkflowFailed, status.Phase)
+			assert.Contains(t, []wfv1.WorkflowPhase{wfv1.WorkflowFailed, wfv1.WorkflowError}, status.Phase)
 			nodeStatus := status.Nodes.FindByDisplayName("A")
 			if assert.NotNil(t, nodeStatus) {
-				assert.Equal(t, wfv1.NodeFailed, nodeStatus.Phase)
+				assert.Contains(t, []wfv1.NodePhase{wfv1.NodeFailed, wfv1.NodeError}, nodeStatus.Phase)
 			}
 			nodeStatus = status.Nodes.FindByDisplayName("A.onExit")
 			assert.Nil(t, nodeStatus)
@@ -133,7 +136,7 @@ spec:
 		WaitForWorkflow(1 * time.Minute).
 		Then().
 		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
-			assert.Equal(t, wfv1.WorkflowFailed, status.Phase)
+			assert.Contains(t, []wfv1.WorkflowPhase{wfv1.WorkflowFailed, wfv1.WorkflowError}, status.Phase)
 			assert.Len(t, status.Nodes, 3)
 			node := status.Nodes.FindByDisplayName("retry-backoff-2(1)")
 			if assert.NotNil(t, node) {

test/e2e/testdata/sidecar-workflow.yaml

@@ -9,8 +9,12 @@ spec:
   templates:
     - name: main
       container:
-        image: argoproj/argosay:v2
+        image: argoproj/argosay:v1
+        args: [ sleep, "5s" ]
       sidecars:
         - name: sidecar-0
-          image: argoproj/argosay:v2
-          argso: [ sleep, "999" ]
+          image: argoproj/argosay:v1
+          args: [ sleep, "999s" ]
+        - name: sidecar-1
+          image: argoproj/argosay:v1
+          args: [ sleep, "999s" ]

test/stress/tool/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	_ "k8s.io/client-go/plugin/pkg/client/auth"
 	"k8s.io/client-go/tools/clientcmd"
 	"sigs.k8s.io/yaml"
 
@@ -45,7 +46,7 @@ spec:
 	}
 
 	ctx := context.Background()
-	for i := 0; i < 100; i++ {
+	for i := 0; i < 500; i++ {
 		_, err := w.Create(ctx, wf, metav1.CreateOptions{})
 		if err != nil {
 			panic(err)

workflow/common/common.go

@@ -105,6 +105,8 @@ const (
 
 	// EnvVarPodName contains the name of the pod (currently unused)
 	EnvVarPodName = "ARGO_POD_NAME"
+	// EnvVarContainerName container the container's name for the current pod
+	EnvVarContainerName = "ARGO_CONTAINER_NAME"
 	// EnvVarContainerRuntimeExecutor contains the name of the container runtime executor to use, empty is equal to "docker"
 	EnvVarContainerRuntimeExecutor = "ARGO_CONTAINER_RUNTIME_EXECUTOR"
 	// EnvVarDownwardAPINodeIP is the envvar used to get the `status.hostIP`

workflow/controller/operator.go

@@ -1272,110 +1272,64 @@ func inferFailedReason(pod *apiv1.Pod) (wfv1.NodePhase, string) {
 	// We only get one message to set for the overall node status.
 	// If multiple containers failed, in order of preference:
 	// init, main (annotated), main (exit code), wait, sidecars
-	for _, ctr := range pod.Status.InitContainerStatuses {
-		// Virtual Kubelet environment will not set the terminate on waiting container
-		// https://github.com/argoproj/argo-workflows/issues/3879
-		// https://github.com/virtual-kubelet/virtual-kubelet/blob/7f2a02291530d2df14905702e6d51500dd57640a/node/sync.go#L195-L208
-		if ctr.State.Waiting != nil {
-			return wfv1.NodeError, fmt.Sprintf("Pod failed before %s container starts", ctr.Name)
+	order := func(n string) int {
+		order, ok := map[string]int{
+			common.InitContainerName: 0,
+			common.MainContainerName: 1,
+			common.WaitContainerName: 2,
+		}[n]
+		if ok {
+			return order
 		}
-		if ctr.State.Terminated == nil {
-			// We should never get here
-			log.Warnf("Pod %s phase was Failed but %s did not have terminated state", pod.ObjectMeta.Name, ctr.Name)
-			continue
-		}
-		if ctr.State.Terminated.ExitCode == 0 {
-			continue
-		}
-		errMsg := "failed to load artifacts"
-		for _, msg := range []string{annotatedMsg, ctr.State.Terminated.Message} {
-			if msg != "" {
-				errMsg += ": " + msg
-				break
-			}
-		}
-		// NOTE: we consider artifact load issues as Error instead of Failed
-		return wfv1.NodeError, errMsg
+		return 3
 	}
-	failMessages := make(map[string]string)
-	for _, ctr := range pod.Status.ContainerStatuses {
+
+	ctrs := append(pod.Status.InitContainerStatuses, pod.Status.ContainerStatuses...)
+	sort.Slice(ctrs, func(i, j int) bool { return order(ctrs[i].Name) < order(ctrs[j].Name) })
+
+	for _, ctr := range ctrs {
+
 		// Virtual Kubelet environment will not set the terminate on waiting container
 		// https://github.com/argoproj/argo-workflows/issues/3879
 		// https://github.com/virtual-kubelet/virtual-kubelet/blob/7f2a02291530d2df14905702e6d51500dd57640a/node/sync.go#L195-L208
 
 		if ctr.State.Waiting != nil {
 			return wfv1.NodeError, fmt.Sprintf("Pod failed before %s container starts", ctr.Name)
 		}
-		if ctr.State.Terminated == nil {
+		t := ctr.State.Terminated
+		if t == nil {
 			// We should never get here
-			log.Warnf("Pod %s phase was Failed but %s did not have terminated state", pod.ObjectMeta.Name, ctr.Name)
-			continue
-		}
-		if ctr.State.Terminated.ExitCode == 0 {
-			continue
-		}
-		if ctr.State.Terminated.Message == "" && ctr.State.Terminated.Reason == "OOMKilled" {
-			failMessages[ctr.Name] = ctr.State.Terminated.Reason
-			continue
-		}
-		if ctr.Name == common.WaitContainerName {
-			errDetails := ""
-			for _, msg := range []string{annotatedMsg, ctr.State.Terminated.Message} {
-				if msg != "" {
-					errDetails = msg
-					break
-				}
-			}
-			if errDetails == "" {
-				// executor is expected to annotate a message to the pod upon any errors.
-				// If we failed to see the annotated message, it is likely the pod ran with
-				// insufficient privileges. Give a hint to that effect.
-				errDetails = fmt.Sprintf("verify serviceaccount %s:%s has necessary privileges", pod.ObjectMeta.Namespace, pod.Spec.ServiceAccountName)
-			}
-			errMsg := fmt.Sprintf("failed to save outputs: %s", errDetails)
-			failMessages[ctr.Name] = errMsg
+			log.Warnf("Pod %s phase was Failed but %s did not have terminated state", pod.Name, ctr.Name)
 			continue
 		}
-		if ctr.State.Terminated.Message != "" {
-			errMsg := ctr.State.Terminated.Message
-			if ctr.Name != common.MainContainerName {
-				errMsg = fmt.Sprintf("sidecar '%s' %s", ctr.Name, errMsg)
-			}
-			failMessages[ctr.Name] = errMsg
+		if t.ExitCode == 0 {
 			continue
 		}
-		errMsg := fmt.Sprintf("failed with exit code %d", ctr.State.Terminated.ExitCode)
-		if ctr.Name != common.MainContainerName {
-			if ctr.State.Terminated.ExitCode == 137 || ctr.State.Terminated.ExitCode == 143 {
+
+		msg := fmt.Sprintf("exit code %d: %s; %s; %s", t.ExitCode, t.Reason, t.Message, annotatedMsg)
+
+		switch ctr.Name {
+		case common.InitContainerName:
+			return wfv1.NodeError, msg
+		case common.MainContainerName:
+			return wfv1.NodeFailed, msg
+		case common.WaitContainerName:
+			// executor is expected to annotate a message to the pod upon any errors.
+			// If we failed to see the annotated message, it is likely the pod ran with
+			// insufficient privileges. Give a hint to that effect.
+			return wfv1.NodeError, fmt.Sprintf("%s; verify serviceaccount %s:%s has necessary privileges", msg, pod.Namespace, pod.Spec.ServiceAccountName)
+		default:
+			if t.ExitCode == 137 || t.ExitCode == 143 {
 				// if the sidecar was SIGKILL'd (exit code 137) assume it was because argoexec
 				// forcibly killed the container, which we ignore the error for.
 				// Java code 143 is a normal exit 128 + 15 https://github.com/elastic/elasticsearch/issues/31847
-				log.Infof("Ignoring %d exit code of sidecar '%s'", ctr.State.Terminated.ExitCode, ctr.Name)
-				continue
+				log.Infof("Ignoring %d exit code of container '%s'", t.ExitCode, ctr.Name)
+			} else {
+				return wfv1.NodeFailed, msg
 			}
-			errMsg = fmt.Sprintf("sidecar '%s' %s", ctr.Name, errMsg)
 		}
-		failMessages[ctr.Name] = errMsg
-	}
-	if failMsg, ok := failMessages[common.MainContainerName]; ok {
-		_, ok = failMessages[common.WaitContainerName]
-		isResourceTemplate := !ok
-		if isResourceTemplate && annotatedMsg != "" {
-			// For resource templates, we prefer the annotated message
-			// over the vanilla exit code 1 error
-			return wfv1.NodeFailed, annotatedMsg
-		}
-		return wfv1.NodeFailed, failMsg
-	}
-	if failMsg, ok := failMessages[common.WaitContainerName]; ok {
-		return wfv1.NodeError, failMsg
 	}
 
-	// If we get here, both the main and wait container succeeded. Iterate the fail messages to
-	// identify the sidecar which failed and return the message.
-	for _, failMsg := range failMessages {
-		return wfv1.NodeFailed, failMsg
-	}
 	// If we get here, we have detected that the main/wait containers succeed but the sidecar(s)
 	// were  SIGKILL'd. The executor may have had to forcefully terminate the sidecar (kill -9),
 	// resulting in a 137 exit code (which we had ignored earlier). If failMessages is empty, it

workflow/controller/operator_test.go

@@ -5439,7 +5439,7 @@ func TestPodFailureWithContainerOOM(t *testing.T) {
 		assert.NotNil(t, pod)
 		nodeStatus, msg := inferFailedReason(&pod)
 		assert.Equal(t, tt.phase, nodeStatus)
-		assert.Equal(t, msg, "OOMKilled")
+		assert.Contains(t, msg, "OOMKilled")
 	}
 }
 

workflow/controller/workflowpod.go

@@ -291,6 +291,11 @@ func (woc *wfOperationCtx) createWorkflowPod(ctx context.Context, nodeName strin
 	}
 	addOutputArtifactsVolumes(pod, tmpl)
 
+	for i, c := range pod.Spec.Containers {
+		c.Env = append(c.Env, apiv1.EnvVar{Name: common.EnvVarContainerName, Value: c.Name}) // used to identify the container name of the process
+		pod.Spec.Containers[i] = c
+	}
+
 	// Set the container template JSON in pod annotations, which executor examines for things like
 	// artifact location/path.
 	tmplBytes, err := json.Marshal(tmpl)
@@ -428,13 +433,13 @@ func substitutePodParams(pod *apiv1.Pod, globalParams common.Parameters, tmpl *w
 
 func (woc *wfOperationCtx) newInitContainer(tmpl *wfv1.Template) apiv1.Container {
 	ctr := woc.newExecContainer(common.InitContainerName, tmpl)
-	ctr.Command = []string{"argoexec", "init"}
+	ctr.Command = []string{"argoexec", "init", "--loglevel", getExecutorLogLevel()}
 	return *ctr
 }
 
 func (woc *wfOperationCtx) newWaitContainer(tmpl *wfv1.Template) (*apiv1.Container, error) {
 	ctr := woc.newExecContainer(common.WaitContainerName, tmpl)
-	ctr.Command = []string{"argoexec", "wait"}
+	ctr.Command = []string{"argoexec", "wait", "--loglevel", getExecutorLogLevel()}
 	switch woc.controller.GetContainerRuntimeExecutor() {
 	case common.ContainerRuntimeExecutorPNS:
 		ctr.SecurityContext = &apiv1.SecurityContext{
@@ -459,6 +464,10 @@ func (woc *wfOperationCtx) newWaitContainer(tmpl *wfv1.Template) (*apiv1.Contain
 	return ctr, nil
 }
 
+func getExecutorLogLevel() string {
+	return log.GetLevel().String()
+}
+
 // hasPrivilegedContainers tests if the main container or sidecars is privileged
 func hasPrivilegedContainers(tmpl *wfv1.Template) bool {
 	if containerIsPrivileged(tmpl.Container) {