feat(controller): Use different container runtime executors for each workflow. Close #4254

Makefile

@@ -385,7 +385,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/argoproj\//$(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/containerRuntimeExecutor: docker/containerRuntimeExecutor: $(E2E_EXECUTOR)/' | kubectl -n $(KUBE_NAMESPACE) apply --prune -l app.kubernetes.io/part-of=argo -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

config/config.go

@@ -44,6 +44,8 @@ type Config struct {
 	// ContainerRuntimeExecutor specifies the container runtime interface to use, default is docker
 	ContainerRuntimeExecutor string `json:"containerRuntimeExecutor,omitempty"`
 
+	ContainerRuntimeExecutors ContainerRuntimeExecutors `json:"containerRuntimeExecutors,omitempty"`
+
 	// KubeletPort is needed when using the kubelet containerRuntimeExecutor, default to 10250
 	KubeletPort int `json:"kubeletPort,omitempty"`
 
@@ -105,6 +107,17 @@ type Config struct {
 	InitialDelay metav1.Duration `json:"initialDelay,omitempty"`
 }
 
+func (c Config) GetContainerRuntimeExecutor(labels labels.Labels) (string, error) {
+	name, err := c.ContainerRuntimeExecutors.Select(labels)
+	if err != nil {
+		return "", err
+	}
+	if name != "" {
+		return name, nil
+	}
+	return c.ContainerRuntimeExecutor, nil
+}
+
 // PodSpecLogStrategy contains the configuration for logging the pod spec in controller log for debugging purpose
 type PodSpecLogStrategy struct {
 	FailedPod bool `json:"failedPod,omitempty"`

config/config_test.go

@@ -4,6 +4,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/utils/pointer"
 )
 
@@ -71,3 +73,33 @@ func TestDatabaseConfig(t *testing.T) {
 	assert.Equal(t, "my-host", DatabaseConfig{Host: "my-host"}.GetHostname())
 	assert.Equal(t, "my-host:1234", DatabaseConfig{Host: "my-host", Port: 1234}.GetHostname())
 }
+
+func TestContainerRuntimeExecutor(t *testing.T) {
+	t.Run("Default", func(t *testing.T) {
+		c := Config{ContainerRuntimeExecutor: "foo"}
+		executor, err := c.GetContainerRuntimeExecutor(labels.Set{})
+		assert.NoError(t, err)
+		assert.Equal(t, "foo", executor)
+	})
+	t.Run("Error", func(t *testing.T) {
+		c := Config{ContainerRuntimeExecutor: "foo", ContainerRuntimeExecutors: ContainerRuntimeExecutors{
+			{Name: "bar", Selector: metav1.LabelSelector{
+				MatchLabels: map[string]string{"!": "!"},
+			}},
+		}}
+		executor, err := c.GetContainerRuntimeExecutor(labels.Set{})
+		assert.Error(t, err)
+		assert.Equal(t, "foo", executor)
+	})
+	t.Run("NoError", func(t *testing.T) {
+		c := Config{ContainerRuntimeExecutor: "foo", ContainerRuntimeExecutors: ContainerRuntimeExecutors{
+			{Name: "bar", Selector: metav1.LabelSelector{
+				MatchLabels: map[string]string{"baz": "qux"},
+			}},
+		}}
+		executor, err := c.GetContainerRuntimeExecutor(labels.Set(map[string]string{"baz": "qux"}))
+		assert.NoError(t, err)
+		assert.Equal(t, "bar", executor)
+	})
+
+}

config/container_runtime_executors.go

@@ -0,0 +1,36 @@
+package config
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+type ContainerRuntimeExecutors []ContainerRuntimeExecutor
+
+// select the correct executor to use
+// this may return an empty string of there is not executor found
+func (e ContainerRuntimeExecutors) Select(labels labels.Labels) (string, error) {
+	for _, c := range e {
+		ok, err := c.Matches(labels)
+		if err != nil {
+			return "", err
+		}
+		if ok {
+			return c.Name, nil
+		}
+	}
+	return "", nil
+}
+
+type ContainerRuntimeExecutor struct {
+	Name     string               `json:"name"`
+	Selector metav1.LabelSelector `json:"selector"`
+}
+
+func (e ContainerRuntimeExecutor) Matches(labels labels.Labels) (bool, error) {
+	x, err := metav1.LabelSelectorAsSelector(&e.Selector)
+	if err != nil {
+		return false, err
+	}
+	return x.Matches(labels), nil
+}

config/container_runtime_executors_test.go

@@ -0,0 +1,44 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+func TestContainerRuntimeExecutors(t *testing.T) {
+	t.Run("Empty", func(t *testing.T) {
+		x := ContainerRuntimeExecutors{}
+		e, err := x.Select(labels.Set{})
+		assert.NoError(t, err)
+		assert.Empty(t, e)
+	})
+	t.Run("Select", func(t *testing.T) {
+		x := ContainerRuntimeExecutors{
+			{
+				Name: "foo",
+				Selector: metav1.LabelSelector{
+					MatchLabels: map[string]string{"bar": ""},
+				},
+			},
+		}
+		e, err := x.Select(labels.Set(map[string]string{"bar": ""}))
+		assert.NoError(t, err)
+		assert.Equal(t, "foo", e)
+	})
+	t.Run("Error", func(t *testing.T) {
+		x := ContainerRuntimeExecutors{
+			{
+				Name: "foo",
+				Selector: metav1.LabelSelector{
+					MatchLabels: map[string]string{"!": "!"},
+				},
+			},
+		}
+		_, err := x.Select(labels.Set(map[string]string{"bar": ""}))
+		assert.Error(t, err)
+	})
+
+}