diff --git a/.github/workflows/ci-build.yaml b/.github/workflows/ci-build.yaml
index 1b9503dff7ec..d5622e1811d3 100644
--- a/.github/workflows/ci-build.yaml
+++ b/.github/workflows/ci-build.yaml
@@ -88,28 +88,18 @@ jobs:
       matrix:
         include:
           - test: test-plugins
-            containerRuntimeExecutor: emissary
             profile: plugins
           - test: test-functional
-            containerRuntimeExecutor: emissary
             profile: minimal
           - test: test-api
-            containerRuntimeExecutor: emissary
             profile: mysql
           - test: test-cli
-            containerRuntimeExecutor: emissary
             profile: mysql
           - test: test-cron
-            containerRuntimeExecutor: emissary
             profile: minimal
           - test: test-examples
-            containerRuntimeExecutor: emissary
             profile: minimal
           - test: test-executor
-            containerRuntimeExecutor: emissary
-            profile: minimal
-          - test: test-executor
-            containerRuntimeExecutor: pns
             profile: minimal
     steps:
       - uses: actions/checkout@v3
@@ -144,11 +134,11 @@ jobs:
           echo '127.0.0.1 minio'    | sudo tee -a /etc/hosts
           echo '127.0.0.1 postgres' | sudo tee -a /etc/hosts
           echo '127.0.0.1 mysql'    | sudo tee -a /etc/hosts
-      - run: make install PROFILE=${{matrix.profile}} E2E_EXECUTOR=${{matrix.containerRuntimeExecutor}} STATIC_FILES=false
+      - run: make install PROFILE=${{matrix.profile}} STATIC_FILES=false
       - run: make controller $(go env GOPATH)/bin/goreman STATIC_FILES=false
       - run: make cli STATIC_FILES=false
         if: ${{matrix.test == 'test-api' || matrix.test == 'test-cli'}}
-      - run: make start PROFILE=${{matrix.profile}} E2E_EXECUTOR=${{matrix.containerRuntimeExecutor}} AUTH_MODE=client STATIC_FILES=false LOG_LEVEL=info API=${{matrix.test == 'test-api' || matrix.test == 'test-cli'}} UI=false > /tmp/argo.log 2>&1 &
+      - run: make start PROFILE=${{matrix.profile}} AUTH_MODE=client STATIC_FILES=false LOG_LEVEL=info API=${{matrix.test == 'test-api' || matrix.test == 'test-cli'}} UI=false > /tmp/argo.log 2>&1 &
       - run: make wait
         timeout-minutes: 4
       - run: make ${{matrix.test}} E2E_TIMEOUT=1m STATIC_FILES=false
diff --git a/Dockerfile b/Dockerfile
index 8119ae0a9099..f3d953f1ff90 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -45,7 +45,7 @@ ARG DOCKER_CHANNEL
 ARG DOCKER_VERSION
 ARG KUBECTL_VERSION
 
-RUN apk --no-cache add curl procps git tar libcap jq
+RUN apk --no-cache add curl git tar jq
 
 COPY hack/arch.sh hack/os.sh /bin/
 
@@ -87,7 +87,6 @@ RUN cat .dockerignore >> .gitignore
 RUN git status --porcelain | cut -c4- | xargs git update-index --skip-worktree
 
 RUN --mount=type=cache,target=/root/.cache/go-build make dist/argoexec
-RUN setcap CAP_SYS_PTRACE,CAP_SYS_CHROOT+ei dist/argoexec
 
 ####################################################################################################
 
diff --git a/Makefile b/Makefile
index 404a489cf784..6f3bb75ed5ad 100644
--- a/Makefile
+++ b/Makefile
@@ -100,7 +100,6 @@ endif
 ARGOEXEC_PKGS    := $(shell echo cmd/argoexec            && go list -f '{{ join .Deps "\n" }}' ./cmd/argoexec/            | grep 'argoproj/argo-workflows/v3/' | cut -c 39-)
 CLI_PKGS         := $(shell echo cmd/argo                && go list -f '{{ join .Deps "\n" }}' ./cmd/argo/                | grep 'argoproj/argo-workflows/v3/' | cut -c 39-)
 CONTROLLER_PKGS  := $(shell echo cmd/workflow-controller && go list -f '{{ join .Deps "\n" }}' ./cmd/workflow-controller/ | grep 'argoproj/argo-workflows/v3/' | cut -c 39-)
-E2E_EXECUTOR ?= emissary
 TYPES := $(shell find pkg/apis/workflow/v1alpha1 -type f -name '*.go' -not -name openapi_generated.go -not -name '*generated*' -not -name '*test.go')
 CRDS := $(shell find manifests/base/crds -type f -name 'argoproj.io_*.yaml')
 SWAGGER_FILES := pkg/apiclient/_.primary.swagger.json \
@@ -409,13 +408,8 @@ test: server/static/files.go dist/argosay
 install: githooks
 	kubectl get ns $(KUBE_NAMESPACE) || kubectl create ns $(KUBE_NAMESPACE)
 	kubectl config set-context --current --namespace=$(KUBE_NAMESPACE)
-	@echo "installing PROFILE=$(PROFILE), E2E_EXECUTOR=$(E2E_EXECUTOR)"
+	@echo "installing PROFILE=$(PROFILE)"
 	kubectl kustomize --load-restrictor=LoadRestrictionsNone test/e2e/manifests/$(PROFILE) | sed 's|quay.io/argoproj/|$(IMAGE_NAMESPACE)/|' | sed 's/namespace: argo/namespace: $(KUBE_NAMESPACE)/' | kubectl -n $(KUBE_NAMESPACE) apply --prune -l app.kubernetes.io/part-of=argo -f -
-ifneq ($(E2E_EXECUTOR),emissary)
-	# only change the executor from the default it we need to
-	kubectl patch cm/workflow-controller-configmap -p "{\"data\": {\"containerRuntimeExecutor\": \"$(E2E_EXECUTOR)\"}}"
-	kubectl apply -f manifests/quick-start/base/executor/$(E2E_EXECUTOR)
-endif
 ifeq ($(PROFILE),stress)
 	kubectl -n $(KUBE_NAMESPACE) apply -f test/stress/massive-workflow.yaml
 endif
diff --git a/cmd/argoexec/commands/root.go b/cmd/argoexec/commands/root.go
index 504766b24773..cd33bcb1cd11 100644
--- a/cmd/argoexec/commands/root.go
+++ b/cmd/argoexec/commands/root.go
@@ -24,7 +24,6 @@ import (
 	"github.com/argoproj/argo-workflows/v3/workflow/common"
 	"github.com/argoproj/argo-workflows/v3/workflow/executor"
 	"github.com/argoproj/argo-workflows/v3/workflow/executor/emissary"
-	"github.com/argoproj/argo-workflows/v3/workflow/executor/pns"
 )
 
 const (
@@ -76,11 +75,10 @@ func NewRootCommand() *cobra.Command {
 
 func initExecutor() *executor.WorkflowExecutor {
 	version := argo.GetVersion()
-	executorType := os.Getenv(common.EnvVarContainerRuntimeExecutor)
-	log.WithFields(log.Fields{"version": version.Version, "executorType": executorType}).Info("Starting Workflow Executor")
+	log.WithFields(log.Fields{"version": version.Version}).Info("Starting Workflow Executor")
 	config, err := clientConfig.ClientConfig()
 	checkErr(err)
-	config = restclient.AddUserAgent(config, fmt.Sprintf("argo-workflows/%s argo-executor/%s", version.Version, executorType))
+	config = restclient.AddUserAgent(config, fmt.Sprintf("argo-workflows/%s argo-executor", version.Version))
 
 	logs.AddK8SLogTransportWrapper(config) // lets log all request as we should typically do < 5 per pod, so this is will show up problems
 
@@ -108,14 +106,7 @@ func initExecutor() *executor.WorkflowExecutor {
 	annotationPatchTickDuration, _ := time.ParseDuration(os.Getenv(common.EnvVarProgressPatchTickDuration))
 	progressFileTickDuration, _ := time.ParseDuration(os.Getenv(common.EnvVarProgressFileTickDuration))
 
-	var cre executor.ContainerRuntimeExecutor
-	log.Infof("Creating a %s executor", executorType)
-	switch executorType {
-	case common.ContainerRuntimeExecutorPNS:
-		cre, err = pns.NewPNSExecutor(clientset, podName, namespace)
-	default:
-		cre, err = emissary.New()
-	}
+	cre, err := emissary.New()
 	checkErr(err)
 
 	wfExecutor := executor.NewExecutor(
diff --git a/config/config.go b/config/config.go
index 5fe7fa14942c..76f75fe5e192 100644
--- a/config/config.go
+++ b/config/config.go
@@ -32,11 +32,6 @@ type Config struct {
 	// KubeConfig specifies a kube config file for the wait & init containers
 	KubeConfig *KubeConfig `json:"kubeConfig,omitempty"`
 
-	// ContainerRuntimeExecutor specifies the container runtime interface to use, default is emissary
-	ContainerRuntimeExecutor string `json:"containerRuntimeExecutor,omitempty"`
-
-	ContainerRuntimeExecutors ContainerRuntimeExecutors `json:"containerRuntimeExecutors,omitempty"`
-
 	// ArtifactRepository contains the default location of an artifact repository for container artifacts
 	ArtifactRepository wfv1.ArtifactRepository `json:"artifactRepository,omitempty"`
 
@@ -110,17 +105,6 @@ type Config struct {
 	SSO SSOConfig `json:"sso,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
-}
-
 func (c Config) GetExecutor() *apiv1.Container {
 	if c.Executor != nil {
 		return c.Executor
diff --git a/config/config_test.go b/config/config_test.go
index b8198a90876a..bd3f257ac15b 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -4,39 +4,9 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 )
 
 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{"!": "!"},
-			}},
-		}}
-		_, err := c.GetContainerRuntimeExecutor(labels.Set{})
-		assert.Error(t, err)
-	})
-	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)
-	})
-}
diff --git a/config/controller_test.go b/config/controller_test.go
index d1ce72b5f611..bbad50ec3306 100644
--- a/config/controller_test.go
+++ b/config/controller_test.go
@@ -13,16 +13,9 @@ func Test_parseConfigMap(t *testing.T) {
 		err := parseConfigMap(&apiv1.ConfigMap{}, c)
 		assert.NoError(t, err)
 	})
-	t.Run("Config", func(t *testing.T) {
-		c := &Config{}
-		err := parseConfigMap(&apiv1.ConfigMap{Data: map[string]string{"config": "containerRuntimeExecutor: pns"}}, c)
-		if assert.NoError(t, err) {
-			assert.Equal(t, "pns", c.ContainerRuntimeExecutor)
-		}
-	})
 	t.Run("Complex", func(t *testing.T) {
 		c := &Config{}
-		err := parseConfigMap(&apiv1.ConfigMap{Data: map[string]string{"containerRuntimeExecutor": "pns", "artifactRepository": `    archiveLogs: true
+		err := parseConfigMap(&apiv1.ConfigMap{Data: map[string]string{"artifactRepository": `    archiveLogs: true
     s3:
       bucket: my-bucket
       endpoint: minio:9000
@@ -34,7 +27,6 @@ func Test_parseConfigMap(t *testing.T) {
         name: my-minio-cred
         key: secretkey`}}, c)
 		if assert.NoError(t, err) {
-			assert.Equal(t, "pns", c.ContainerRuntimeExecutor)
 			assert.NotEmpty(t, c.ArtifactRepository)
 		}
 	})
diff --git a/docs/environment-variables.md b/docs/environment-variables.md
index 2ba3a62f864b..95ccb5fc2b9f 100644
--- a/docs/environment-variables.md
+++ b/docs/environment-variables.md
@@ -117,14 +117,10 @@ spec:
 
 | Name | Type | Default | Description |
 |------|------|---------|-------------|
-| `ARGO_CONTAINER_RUNTIME_EXECUTOR` | `string` | `"docker"` | The name of the container runtime executor. |
-| `ARGO_KUBELET_PORT` | `int` | `10250` | The port to the Kubelet API. |
-| `ARGO_KUBELET_INSECURE` | `bool` | `false` | Whether to disable the TLS verification. |
 | `EXECUTOR_RETRY_BACKOFF_DURATION` | `time.Duration` | `1s` | The retry backoff duration when the workflow executor performs retries. |
 | `EXECUTOR_RETRY_BACKOFF_FACTOR` | `float` | `1.6` | The retry backoff factor when the workflow executor performs retries. |
 | `EXECUTOR_RETRY_BACKOFF_JITTER` | `float` | `0.5` | The retry backoff jitter when the workflow executor performs retries. |
 | `EXECUTOR_RETRY_BACKOFF_STEPS` | `int` | `5` | The retry backoff steps when the workflow executor performs retries. |
-| `PNS_PRIVILEGED` | `bool` | `false` | Whether to always set privileged on for PNS when PNS executor is used. |
 | `REMOVE_LOCAL_ART_PATH` | `bool` | `false` | Whether to remove local artifacts. |
 | `RESOURCE_STATE_CHECK_INTERVAL` | `time.Duration` | `5s` | The time interval between resource status checks against the specified success and failure conditions. |
 | `WAIT_CONTAINER_STATUS_CHECK_INTERVAL` | `time.Duration` | `5s` | The time interval for wait container to check whether the containers have completed. |
diff --git a/docs/running-locally.md b/docs/running-locally.md
index 441f5b826957..e3b14a5ab667 100644
--- a/docs/running-locally.md
+++ b/docs/running-locally.md
@@ -90,7 +90,7 @@ make argoexec-image
 4. Run the profile in a terminal window
 
     ```shell
-    make start PROFILE=minimal E2E_EXECUTOR=emissary AUTH_MODE=client STATIC_FILES=false LOG_LEVEL=info API=true UI=false
+    make start PROFILE=minimal AUTH_MODE=client STATIC_FILES=false LOG_LEVEL=info API=true UI=false
     ```
 
 5. Run the test in your IDE
diff --git a/docs/workflow-controller-configmap.yaml b/docs/workflow-controller-configmap.yaml
index 01fc661f6a74..fdf50f154a3f 100644
--- a/docs/workflow-controller-configmap.yaml
+++ b/docs/workflow-controller-configmap.yaml
@@ -146,6 +146,7 @@ data:
   # Specifies the container runtime interface to use (default: emissary)
   # must be one of: docker, kubelet, k8sapi, pns, emissary
   # It has lower precedence than either `--container-runtime-executor` and `containerRuntimeExecutors`.
+  # (removed in v3.4)
   containerRuntimeExecutor: emissary
 
   # Specifies the executor to use.
@@ -158,6 +159,7 @@ data:
   #
   # The list is in order of precedence; the first matching executor is used.
   # This has precedence over `containerRuntimeExecutor`.
+  # (removed in v3.4)
   containerRuntimeExecutors: |
     - name: emissary
       selector:
@@ -173,9 +175,11 @@ data:
   dockerSockPath: /var/someplace/else/docker.sock
 
   # kubelet port when using kubelet executor (default: 10250) (kubelet executor will be deprecated use emissary instead)
+  # (removed in v3.4)
   kubeletPort: 10250
 
   # disable the TLS verification of the kubelet executor (default: false)
+  # (removed in v3.4)
   kubeletInsecure: false
 
   # The command/args for each image, needed when the command is not specified and the emissary executor is used.
diff --git a/docs/workflow-executors.md b/docs/workflow-executors.md
index 6b42ad4d495e..a686791ee7da 100644
--- a/docs/workflow-executors.md
+++ b/docs/workflow-executors.md
@@ -48,7 +48,7 @@ The emissary will exit with code 64 if it fails. This may indicate a bug in the
 
 ## Docker (docker) 
 
-⚠️Deprecated. 
+⚠️Deprecated. Removed in v3.4.
 
 **default in <= v3.2**
 
@@ -67,7 +67,7 @@ The emissary will exit with code 64 if it fails. This may indicate a bug in the
 
 ## Kubelet (kubelet)
 
-⚠️Deprecated.
+⚠️Deprecated. Removed in v3.4.
 
 * Secure
     * No `privileged` access
@@ -84,7 +84,7 @@ The emissary will exit with code 64 if it fails. This may indicate a bug in the
 
 ## Kubernetes API (k8sapi)
 
-⚠️Deprecated.
+⚠️Deprecated. Removed in v3.4.
 
 * Reliability:
     * Works on GKE Autopilot
@@ -103,6 +103,8 @@ The emissary will exit with code 64 if it fails. This may indicate a bug in the
 
 ## Process Namespace Sharing (pns)
 
+⚠️Deprecated. Removed in v3.4.
+
 * More secure:
     * No `privileged` access
     * cannot escape the privileges of the pod's service account
diff --git a/examples/selected-executor-workflow.yaml b/examples/selected-executor-workflow.yaml
index 68321e195f2f..dd39bfb83124 100644
--- a/examples/selected-executor-workflow.yaml
+++ b/examples/selected-executor-workflow.yaml
@@ -3,8 +3,6 @@ kind: Workflow
 metadata:
   generateName: selected-executor-
   labels:
-    # run this workflow as a part of our test suite
-    workflows.argoproj.io/test: "true"
     # use the pns executor, rather than the default (typically emissary)
     workflows.argoproj.io/container-runtime-executor: pns
   annotations:
@@ -17,7 +15,7 @@ metadata:
       e.g. have a certain labels use certain executors.
 
     # this workflow will only run on workflows version v3.0.0
-    workflows.argoproj.io/version: ">= 3.0.0"
+    workflows.argoproj.io/version: ">= 3.0.0 < 3.4.0"
 spec:
   entrypoint: main
   templates:
diff --git a/go.mod b/go.mod
index c507ee5580ef..13ba4a98be46 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,6 @@ require (
 	github.com/imkira/go-interpol v1.1.0 // indirect
 	github.com/klauspost/pgzip v1.2.5
 	github.com/minio/minio-go/v7 v7.0.23
-	github.com/mitchellh/go-ps v0.0.0-20190716172923-621e5597135b
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/common v0.33.0
diff --git a/go.sum b/go.sum
index bd7a80bc4d17..9ece5fbcc0de 100644
--- a/go.sum
+++ b/go.sum
@@ -913,8 +913,6 @@ github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HK
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
-github.com/mitchellh/go-ps v0.0.0-20190716172923-621e5597135b h1:9+ke9YJ9KGWw5ANXK6ozjoK47uI3uNbXv4YVINBnGm8=
-github.com/mitchellh/go-ps v0.0.0-20190716172923-621e5597135b/go.mod h1:r1VsdOzOPt1ZSrGZWFoNhsAedKnEd6r9Np1+5blZCWk=
 github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
 github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7/go.mod h1:ZXFpozHsX6DPmq2I0TCekCxypsnAUbP2oI0UX1GXzOo=
 github.com/mitchellh/go-wordwrap v1.0.0/go.mod h1:ZXFpozHsX6DPmq2I0TCekCxypsnAUbP2oI0UX1GXzOo=
diff --git a/hack/test-examples.sh b/hack/test-examples.sh
index 62cb4acad5b5..9cc95faecbe1 100755
--- a/hack/test-examples.sh
+++ b/hack/test-examples.sh
@@ -3,8 +3,6 @@ set -eu -o pipefail
 
 # Load the configmaps that contains the parameter values used for certain examples.
 kubectl apply -f examples/configmaps/simple-parameters-configmap.yaml
-# Needed for examples/selected-executor-workflow.yaml.
-kubectl apply -f manifests/quick-start/base/executor/pns/executor-role.yaml
 
 echo "Checking for banned images..."
 grep -lR 'workflows.argoproj.io/test' examples/*  | while read f ; do
diff --git a/manifests/quick-start-minimal.yaml b/manifests/quick-start-minimal.yaml
index 10abc9f97017..56527eb5ce96 100644
--- a/manifests/quick-start-minimal.yaml
+++ b/manifests/quick-start-minimal.yaml
@@ -1164,15 +1164,6 @@ data:
       secretKeySecret:
         name: my-minio-cred
         key: secretkey
-  containerRuntimeExecutors: |
-    - name: emissary
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: emissary
-    - name: pns
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: pns
   executor: |
     resources:
       requests:
diff --git a/manifests/quick-start-mysql.yaml b/manifests/quick-start-mysql.yaml
index 76aedac03111..02760a05622c 100644
--- a/manifests/quick-start-mysql.yaml
+++ b/manifests/quick-start-mysql.yaml
@@ -1164,15 +1164,6 @@ data:
       secretKeySecret:
         name: my-minio-cred
         key: secretkey
-  containerRuntimeExecutors: |
-    - name: emissary
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: emissary
-    - name: pns
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: pns
   executor: |
     resources:
       requests:
diff --git a/manifests/quick-start-postgres.yaml b/manifests/quick-start-postgres.yaml
index aeb0fe509ca3..6e236bde991b 100644
--- a/manifests/quick-start-postgres.yaml
+++ b/manifests/quick-start-postgres.yaml
@@ -1164,15 +1164,6 @@ data:
       secretKeySecret:
         name: my-minio-cred
         key: secretkey
-  containerRuntimeExecutors: |
-    - name: emissary
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: emissary
-    - name: pns
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: pns
   executor: |
     resources:
       requests:
diff --git a/manifests/quick-start/base/executor/docker/executor-role.yaml b/manifests/quick-start/base/executor/docker/executor-role.yaml
index 3ee284a9c0d7..fd4349b59161 100644
--- a/manifests/quick-start/base/executor/docker/executor-role.yaml
+++ b/manifests/quick-start/base/executor/docker/executor-role.yaml
@@ -7,6 +7,7 @@ metadata:
       Recommended minimum permissions for the `docker` executor.
 
       This executor is superseded by  the `emmisary` executor, so we do not recommend you use it anymore.
+    workflows.argoproj.io/version: "< 3.4.0"
 rules:
   - apiGroups:
       - argoproj.io
diff --git a/manifests/quick-start/base/executor/k8sapi/executor-role.yaml b/manifests/quick-start/base/executor/k8sapi/executor-role.yaml
index a7c9e2ac4f04..7685a1445b74 100644
--- a/manifests/quick-start/base/executor/k8sapi/executor-role.yaml
+++ b/manifests/quick-start/base/executor/k8sapi/executor-role.yaml
@@ -7,6 +7,7 @@ metadata:
       Recommended minimum permissions for `k8siapi` executor.
 
       This executor is superseded by  the `emmisary` executor, so we do not recommend you use it anymore.
+    workflows.argoproj.io/version: "< 3.4.0"
 rules:
   - apiGroups:
       - argoproj.io
diff --git a/manifests/quick-start/base/executor/kubelet/executor-role.yaml b/manifests/quick-start/base/executor/kubelet/executor-role.yaml
index 94ada0b5279e..341b298c3d9c 100644
--- a/manifests/quick-start/base/executor/kubelet/executor-role.yaml
+++ b/manifests/quick-start/base/executor/kubelet/executor-role.yaml
@@ -7,6 +7,7 @@ metadata:
       Recommended minimum permissions for `kubelet` executor.
 
       This executor is superseded by  the `emmisary` executor, so we do not recommend you use it anymore.
+    workflows.argoproj.io/version: "< 3.4.0"
 rules:
   - apiGroups:
       - argoproj.io
diff --git a/manifests/quick-start/base/executor/kubelet/kubelet-executor-clusterrole.yaml b/manifests/quick-start/base/executor/kubelet/kubelet-executor-clusterrole.yaml
index 6f79bc1ac337..914730194696 100644
--- a/manifests/quick-start/base/executor/kubelet/kubelet-executor-clusterrole.yaml
+++ b/manifests/quick-start/base/executor/kubelet/kubelet-executor-clusterrole.yaml
@@ -2,6 +2,8 @@ apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
   name: kubelet-executor
+  annotations:
+    workflows.argoproj.io/version: "< 3.4.0"
 rules:
   # This allows the kubelet executor.
   - apiGroups:
diff --git a/manifests/quick-start/base/executor/kubelet/kubelet-executor-default-clusterrolebinding.yaml b/manifests/quick-start/base/executor/kubelet/kubelet-executor-default-clusterrolebinding.yaml
index f0aff8e6c480..e754dab2487c 100644
--- a/manifests/quick-start/base/executor/kubelet/kubelet-executor-default-clusterrolebinding.yaml
+++ b/manifests/quick-start/base/executor/kubelet/kubelet-executor-default-clusterrolebinding.yaml
@@ -2,6 +2,8 @@ apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: kubelet-executor-default
+  annotations:
+    workflows.argoproj.io/version: "< 3.4.0"
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: ClusterRole
diff --git a/manifests/quick-start/base/executor/pns/executor-role.yaml b/manifests/quick-start/base/executor/pns/executor-role.yaml
index 27d506bc5b00..ea5d7a4d1e13 100644
--- a/manifests/quick-start/base/executor/pns/executor-role.yaml
+++ b/manifests/quick-start/base/executor/pns/executor-role.yaml
@@ -5,6 +5,7 @@ metadata:
   annotations:
     workflows.argoproj.io/description: |
       Recomended minimum permissions for `pns` executor.
+    workflows.argoproj.io/version: "< 3.4.0"
 rules:
   - apiGroups:
       - argoproj.io
diff --git a/manifests/quick-start/base/overlays/workflow-controller-configmap.yaml b/manifests/quick-start/base/overlays/workflow-controller-configmap.yaml
index 51c7faefdb8b..ca5fec342c46 100644
--- a/manifests/quick-start/base/overlays/workflow-controller-configmap.yaml
+++ b/manifests/quick-start/base/overlays/workflow-controller-configmap.yaml
@@ -5,15 +5,6 @@ data:
       requests:
         cpu: 10m
         memory: 64Mi
-  containerRuntimeExecutors: |
-    - name: emissary
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: emissary
-    - name: pns
-      selector:
-        matchLabels:
-          workflows.argoproj.io/container-runtime-executor: pns
   images: |
     argoproj/argosay:v1:
       command: [cowsay]
diff --git a/workflow/common/common.go b/workflow/common/common.go
index d5c2b89d4c54..6413da48f51d 100644
--- a/workflow/common/common.go
+++ b/workflow/common/common.go
@@ -125,14 +125,6 @@ const (
 	EnvVarIncludeScriptOutput = "ARGO_INCLUDE_SCRIPT_OUTPUT"
 	// EnvVarTemplate is the template
 	EnvVarTemplate = "ARGO_TEMPLATE"
-	// EnvVarContainerRuntimeExecutor contains the name of the container runtime executor to use.
-	EnvVarContainerRuntimeExecutor = "ARGO_CONTAINER_RUNTIME_EXECUTOR"
-	// EnvVarDownwardAPINodeIP is the envvar used to get the `status.hostIP`
-	EnvVarDownwardAPINodeIP = "ARGO_KUBELET_HOST"
-	// EnvVarKubeletPort is used to configure the kubelet api port
-	EnvVarKubeletPort = "ARGO_KUBELET_PORT"
-	// EnvVarKubeletInsecure is used to disable the TLS verification
-	EnvVarKubeletInsecure = "ARGO_KUBELET_INSECURE"
 	// EnvVarArgoTrace is used enable tracing statements in Argo components
 	EnvVarArgoTrace = "ARGO_TRACE"
 	// EnvVarProgressPatchTickDuration sets the tick duration for patching pod annotations upon progress changes.
@@ -150,12 +142,6 @@ const (
 	// EnvAgentPatchRate is the rate that the Argo Agent will patch the Workflow TaskSet
 	EnvAgentPatchRate = "ARGO_AGENT_PATCH_RATE"
 
-	// ContainerRuntimeExecutorPNS indicates to use process namespace sharing as the container runtime executor
-	ContainerRuntimeExecutorPNS = "pns"
-
-	// ContainerRuntimeExecutorEmissary indicates to use emissary container runtime executor
-	ContainerRuntimeExecutorEmissary = "emissary"
-
 	// Variables that are added to the scope during template execution and can be referenced using {{}} syntax
 
 	// GlobalVarWorkflowName is a global workflow variable referencing the workflow's metadata.name field
diff --git a/workflow/controller/container_set_template.go b/workflow/controller/container_set_template.go
index 89665c5339bd..05cf079cda8b 100644
--- a/workflow/controller/container_set_template.go
+++ b/workflow/controller/container_set_template.go
@@ -5,7 +5,6 @@ import (
 	"fmt"
 
 	wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
-	"github.com/argoproj/argo-workflows/v3/workflow/common"
 )
 
 func (woc *wfOperationCtx) executeContainerSet(ctx context.Context, nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
@@ -13,11 +12,6 @@ func (woc *wfOperationCtx) executeContainerSet(ctx context.Context, nodeName str
 	if node == nil {
 		node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypePod, templateScope, tmpl, orgTmpl, opts.boundaryID, wfv1.NodePending)
 	}
-
-	if woc.getContainerRuntimeExecutor() != common.ContainerRuntimeExecutorEmissary && tmpl.HasSequencedContainers() {
-		woc.markNodePhase(nodeName, wfv1.NodeFailed, fmt.Sprintf("template has sequenced containers, so you must use the emissary executor rather than %q, learn more: https://argoproj.github.io/argo-workflows/workflow-executors/#emissary-emissary", woc.getContainerRuntimeExecutor()))
-		return woc.wf.GetNodeByName(nodeName), nil
-	}
 	includeScriptOutput, err := woc.includeScriptOutput(nodeName, opts.boundaryID)
 	if err != nil {
 		return node, err
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 439a31772d46..a8b1cad3e750 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -1090,20 +1090,6 @@ func (wfc *WorkflowController) GetManagedNamespace() string {
 	return wfc.Config.Namespace
 }
 
-func (wfc *WorkflowController) GetContainerRuntimeExecutor(labels labels.Labels) string {
-	executor, err := wfc.Config.GetContainerRuntimeExecutor(labels)
-	if err != nil {
-		log.WithError(err).Info("failed to determine container runtime executor")
-	}
-	if executor != "" {
-		return executor
-	}
-	if wfc.containerRuntimeExecutor != "" {
-		return wfc.containerRuntimeExecutor
-	}
-	return common.ContainerRuntimeExecutorEmissary
-}
-
 func (wfc *WorkflowController) getMetricsServerConfig() (metrics.ServerConfig, metrics.ServerConfig) {
 	// Metrics config
 	path := wfc.Config.MetricsConfig.Path
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 6f64fce2b0a4..e74a5051101f 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -26,7 +26,6 @@ import (
 	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	"k8s.io/client-go/tools/cache"
@@ -510,10 +509,6 @@ func (woc *wfOperationCtx) updateWorkflowMetadata() error {
 	return nil
 }
 
-func (woc *wfOperationCtx) getContainerRuntimeExecutor() string {
-	return woc.controller.GetContainerRuntimeExecutor(labels.Set(woc.wf.Labels))
-}
-
 func (woc *wfOperationCtx) getWorkflowDeadline() *time.Time {
 	if woc.execWf.Spec.ActiveDeadlineSeconds == nil {
 		return nil
@@ -3442,7 +3437,7 @@ func (woc *wfOperationCtx) setExecWorkflow(ctx context.Context) error {
 
 	// Perform one-time workflow validation
 	if woc.wf.Status.Phase == wfv1.WorkflowUnknown {
-		validateOpts := validate.ValidateOpts{ContainerRuntimeExecutor: woc.getContainerRuntimeExecutor()}
+		validateOpts := validate.ValidateOpts{}
 		wftmplGetter := templateresolution.WrapWorkflowTemplateInterface(woc.controller.wfclientset.ArgoprojV1alpha1().WorkflowTemplates(woc.wf.Namespace))
 		cwftmplGetter := templateresolution.WrapClusterWorkflowTemplateInterface(woc.controller.wfclientset.ArgoprojV1alpha1().ClusterWorkflowTemplates())
 
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index e7d1c8b84a96..591cc031dbef 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -21,7 +21,6 @@ import (
 	apierr "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes/fake"
@@ -7505,34 +7504,6 @@ func TestBuildRetryStrategyLocalScope(t *testing.T) {
 	assert.Equal(t, "6", localScope[common.LocalVarRetriesLastDuration])
 }
 
-func TestGetContainerRuntimeExecutor(t *testing.T) {
-	cancel, controller := newController()
-	defer cancel()
-	t.Run("Default", func(t *testing.T) {
-		executor := controller.GetContainerRuntimeExecutor(labels.Set{})
-		assert.Equal(t, common.ContainerRuntimeExecutorEmissary, executor)
-	})
-	controller.Config.ContainerRuntimeExecutor = "pns"
-	controller.Config.ContainerRuntimeExecutors = config.ContainerRuntimeExecutors{
-		{
-			Name: "emissary",
-			Selector: metav1.LabelSelector{
-				MatchLabels: map[string]string{
-					"workflows.argoproj.io/container-runtime-executor": "emissary",
-				},
-			},
-		},
-	}
-	t.Run("Configuration", func(t *testing.T) {
-		executor := controller.GetContainerRuntimeExecutor(labels.Set{})
-		assert.Equal(t, common.ContainerRuntimeExecutorPNS, executor)
-	})
-	t.Run("Labels", func(t *testing.T) {
-		executor := controller.GetContainerRuntimeExecutor(labels.Set{"workflows.argoproj.io/container-runtime-executor": "emissary"})
-		assert.Equal(t, common.ContainerRuntimeExecutorEmissary, executor)
-	})
-}
-
 var exitHandlerWithRetryNodeParam = `
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow
diff --git a/workflow/controller/workflowpod.go b/workflow/controller/workflowpod.go
index 3a588d6df5f3..09e5d941b997 100644
--- a/workflow/controller/workflowpod.go
+++ b/workflow/controller/workflowpod.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"os"
 	"path/filepath"
 	"strconv"
 	"time"
@@ -14,7 +13,6 @@ import (
 	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
-	"k8s.io/utils/pointer"
 
 	"github.com/argoproj/argo-workflows/v3/config"
 	"github.com/argoproj/argo-workflows/v3/errors"
@@ -196,9 +194,6 @@ func (woc *wfOperationCtx) createWorkflowPod(ctx context.Context, nodeName strin
 	if woc.controller.Config.InstanceID != "" {
 		pod.ObjectMeta.Labels[common.LabelKeyControllerInstanceID] = woc.controller.Config.InstanceID
 	}
-	if woc.getContainerRuntimeExecutor() == common.ContainerRuntimeExecutorPNS {
-		pod.Spec.ShareProcessNamespace = pointer.BoolPtr(true)
-	}
 
 	woc.addArchiveLocation(tmpl)
 
@@ -233,10 +228,8 @@ func (woc *wfOperationCtx) createWorkflowPod(ctx context.Context, nodeName strin
 
 	// Add init container only if it needs input artifacts. This is also true for
 	// script templates (which needs to populate the script)
-	if len(tmpl.Inputs.Artifacts) > 0 || tmpl.GetType() == wfv1.TemplateTypeScript || woc.getContainerRuntimeExecutor() == common.ContainerRuntimeExecutorEmissary {
-		initCtr := woc.newInitContainer(tmpl)
-		pod.Spec.InitContainers = []apiv1.Container{initCtr}
-	}
+	initCtr := woc.newInitContainer(tmpl)
+	pod.Spec.InitContainers = []apiv1.Container{initCtr}
 
 	addSchedulingConstraints(pod, wfSpec, tmpl)
 	woc.addMetadata(pod, tmpl)
@@ -376,7 +369,7 @@ func (woc *wfOperationCtx) createWorkflowPod(ctx context.Context, nodeName strin
 	}
 
 	for i, c := range pod.Spec.Containers {
-		if woc.getContainerRuntimeExecutor() == common.ContainerRuntimeExecutorEmissary && c.Name != common.WaitContainerName {
+		if c.Name != common.WaitContainerName {
 			// https://kubernetes.io/docs/tasks/inject-data-application/define-command-argument-container/#notes
 			if len(c.Command) == 0 {
 				x := woc.getImage(c.Image)
@@ -515,25 +508,6 @@ func (woc *wfOperationCtx) newInitContainer(tmpl *wfv1.Template) apiv1.Container
 func (woc *wfOperationCtx) newWaitContainer(tmpl *wfv1.Template) *apiv1.Container {
 	ctr := woc.newExecContainer(common.WaitContainerName, tmpl)
 	ctr.Command = []string{"argoexec", "wait", "--loglevel", getExecutorLogLevel()}
-	switch woc.getContainerRuntimeExecutor() {
-	case common.ContainerRuntimeExecutorPNS:
-		ctr.SecurityContext = &apiv1.SecurityContext{
-			Capabilities: &apiv1.Capabilities{
-				Add: []apiv1.Capability{
-					// necessary to access main's root filesystem when run with a different user id
-					apiv1.Capability("SYS_PTRACE"),
-					apiv1.Capability("SYS_CHROOT"),
-				},
-			},
-		}
-		// PNS_PRIVILEGED allows you to always set privileged on for PNS, this seems to be needed for certain systems
-		// https://github.com/argoproj/argo-workflows/issues/1256
-		if hasPrivilegedContainers(tmpl) || os.Getenv("PNS_PRIVILEGED") == "true" {
-			// if the main or sidecar is privileged, the wait sidecar must also run privileged,
-			// in order to SIGTERM/SIGKILL the pid
-			ctr.SecurityContext.Privileged = pointer.BoolPtr(true)
-		}
-	}
 	return ctr
 }
 
@@ -541,26 +515,6 @@ 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) {
-		return true
-	}
-	for _, side := range tmpl.Sidecars {
-		if containerIsPrivileged(&side.Container) {
-			return true
-		}
-	}
-	return false
-}
-
-func containerIsPrivileged(ctr *apiv1.Container) bool {
-	if ctr != nil && ctr.SecurityContext != nil && ctr.SecurityContext.Privileged != nil && *ctr.SecurityContext.Privileged {
-		return true
-	}
-	return false
-}
-
 func (woc *wfOperationCtx) createEnvVars() []apiv1.EnvVar {
 	execEnvVars := []apiv1.EnvVar{
 		{
@@ -581,10 +535,6 @@ func (woc *wfOperationCtx) createEnvVars() []apiv1.EnvVar {
 				},
 			},
 		},
-		{
-			Name:  common.EnvVarContainerRuntimeExecutor,
-			Value: woc.getContainerRuntimeExecutor(),
-		},
 		// This flag was introduced in Go 15 and will be removed in Go 16.
 		// x509: cannot validate certificate for ... because it doesn't contain any IP SANs
 		// https://github.com/argoproj/argo-workflows/issues/5563 - Upgrade to Go 16
diff --git a/workflow/controller/workflowpod_test.go b/workflow/controller/workflowpod_test.go
index 28338e350e18..05e960187192 100644
--- a/workflow/controller/workflowpod_test.go
+++ b/workflow/controller/workflowpod_test.go
@@ -653,20 +653,17 @@ func Test_createWorkflowPod_containerName(t *testing.T) {
 func Test_createWorkflowPod_emissary(t *testing.T) {
 	t.Run("NoCommand", func(t *testing.T) {
 		woc := newWoc()
-		woc.controller.containerRuntimeExecutor = common.ContainerRuntimeExecutorEmissary
 		_, err := woc.createWorkflowPod(context.Background(), "", []apiv1.Container{{}}, &wfv1.Template{Name: "my-tmpl"}, &createWorkflowPodOpts{})
 		assert.EqualError(t, err, "container \"main\" in template \"my-tmpl\", does not have the command specified: when using the emissary executor you must either explicitly specify the command, or list the image's command in the index: https://argoproj.github.io/argo-workflows/workflow-executors/#emissary-emissary")
 	})
 	t.Run("CommandNoArgs", func(t *testing.T) {
 		woc := newWoc()
-		woc.controller.containerRuntimeExecutor = common.ContainerRuntimeExecutorEmissary
 		pod, err := woc.createWorkflowPod(context.Background(), "", []apiv1.Container{{Command: []string{"foo"}}}, &wfv1.Template{}, &createWorkflowPodOpts{})
 		assert.NoError(t, err)
 		assert.Equal(t, []string{"/var/run/argo/argoexec", "emissary", "--", "foo"}, pod.Spec.Containers[1].Command)
 	})
 	t.Run("NoCommandWithImageIndex", func(t *testing.T) {
 		woc := newWoc()
-		woc.controller.containerRuntimeExecutor = common.ContainerRuntimeExecutorEmissary
 		pod, err := woc.createWorkflowPod(context.Background(), "", []apiv1.Container{{Image: "my-image"}}, &wfv1.Template{}, &createWorkflowPodOpts{})
 		if assert.NoError(t, err) {
 			assert.Equal(t, []string{"/var/run/argo/argoexec", "emissary", "--", "my-cmd"}, pod.Spec.Containers[1].Command)
@@ -675,7 +672,6 @@ func Test_createWorkflowPod_emissary(t *testing.T) {
 	})
 	t.Run("NoCommandWithArgsWithImageIndex", func(t *testing.T) {
 		woc := newWoc()
-		woc.controller.containerRuntimeExecutor = common.ContainerRuntimeExecutorEmissary
 		pod, err := woc.createWorkflowPod(context.Background(), "", []apiv1.Container{{Image: "my-image", Args: []string{"foo"}}}, &wfv1.Template{}, &createWorkflowPodOpts{})
 		if assert.NoError(t, err) {
 			assert.Equal(t, []string{"/var/run/argo/argoexec", "emissary", "--", "my-cmd"}, pod.Spec.Containers[1].Command)
@@ -684,7 +680,6 @@ func Test_createWorkflowPod_emissary(t *testing.T) {
 	})
 	t.Run("CommandFromPodSpecPatch", func(t *testing.T) {
 		woc := newWoc()
-		woc.controller.containerRuntimeExecutor = common.ContainerRuntimeExecutorEmissary
 		podSpec := &apiv1.PodSpec{}
 		podSpec.Containers = []apiv1.Container{{
 			Name:    "main",
@@ -721,7 +716,6 @@ func TestVolumeAndVolumeMounts(t *testing.T) {
 		woc := newWoc()
 		woc.volumes = volumes
 		woc.execWf.Spec.Templates[0].Container.VolumeMounts = volumeMounts
-		woc.controller.Config.ContainerRuntimeExecutor = common.ContainerRuntimeExecutorEmissary
 
 		tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 		assert.NoError(t, err)
@@ -787,7 +781,6 @@ func TestVolumesPodSubstitution(t *testing.T) {
 	woc.volumes = volumes
 	woc.execWf.Spec.Templates[0].Container.VolumeMounts = volumeMounts
 	woc.execWf.Spec.Templates[0].Inputs.Parameters = inputParameters
-	woc.controller.Config.ContainerRuntimeExecutor = common.ContainerRuntimeExecutorEmissary
 
 	tmplCtx, err := woc.createTemplateContext(wfv1.ResourceScopeLocal, "")
 	assert.NoError(t, err)
diff --git a/workflow/executor/k8sapi/client.go b/workflow/executor/k8sapi/client.go
deleted file mode 100644
index d7190bfb03ad..000000000000
--- a/workflow/executor/k8sapi/client.go
+++ /dev/null
@@ -1,160 +0,0 @@
-package k8sapi
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"strings"
-	"syscall"
-	"time"
-
-	corev1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
-	restclient "k8s.io/client-go/rest"
-
-	"github.com/argoproj/argo-workflows/v3/errors"
-	errorsutil "github.com/argoproj/argo-workflows/v3/util/errors"
-	waitutil "github.com/argoproj/argo-workflows/v3/util/wait"
-	"github.com/argoproj/argo-workflows/v3/workflow/common"
-	execcommon "github.com/argoproj/argo-workflows/v3/workflow/executor/common"
-)
-
-type k8sAPIClient struct {
-	clientset kubernetes.Interface
-	config    *restclient.Config
-	podName   string
-	namespace string
-}
-
-var _ execcommon.KubernetesClientInterface = &k8sAPIClient{}
-
-func newK8sAPIClient(clientset kubernetes.Interface, config *restclient.Config, podName, namespace string) *k8sAPIClient {
-	return &k8sAPIClient{
-		clientset: clientset,
-		config:    config,
-		podName:   podName,
-		namespace: namespace,
-	}
-}
-
-func errWithHelp(err error) error {
-	return fmt.Errorf("unable to get pods, you can check https://argoproj.github.io/argo-workflows/faq/: %w", err)
-}
-
-func isErrUnknownGetPods(err error) bool {
-	return strings.Contains(err.Error(), "unknown (get pods)")
-}
-
-func (c *k8sAPIClient) CreateArchive(ctx context.Context, containerName, sourcePath string) (*bytes.Buffer, error) {
-	command := []string{"tar", "cf", "-", sourcePath}
-	exec, err := common.ExecPodContainer(c.config, c.namespace, c.podName, containerName, true, false, command...)
-	if err != nil {
-		return nil, err
-	}
-	stdOut, _, err := common.GetExecutorOutput(exec)
-	if err != nil {
-		return nil, err
-	}
-	return stdOut, nil
-}
-
-func (c *k8sAPIClient) getLogsAsStream(ctx context.Context, containerName string) (io.ReadCloser, error) {
-	return c.clientset.CoreV1().Pods(c.namespace).
-		GetLogs(c.podName, &corev1.PodLogOptions{Container: containerName, SinceTime: &metav1.Time{}}).Stream(ctx)
-}
-
-var backoffOver30s = wait.Backoff{
-	Duration: 1 * time.Second,
-	Steps:    7,
-	Factor:   2,
-}
-
-func (c *k8sAPIClient) getPod(ctx context.Context) (*corev1.Pod, error) {
-	var pod *corev1.Pod
-	err := waitutil.Backoff(backoffOver30s, func() (bool, error) {
-		var err error
-		pod, err = c.clientset.CoreV1().Pods(c.namespace).Get(ctx, c.podName, metav1.GetOptions{})
-		if err != nil && isErrUnknownGetPods(err) {
-			return !errorsutil.IsTransientErr(err), errWithHelp(err)
-		}
-		return !errorsutil.IsTransientErr(err), err
-	})
-	return pod, err
-}
-
-func (c *k8sAPIClient) GetContainerStatus(ctx context.Context, containerName string) (*corev1.Pod, *corev1.ContainerStatus, error) {
-	pod, containerStatuses, err := c.GetContainerStatuses(ctx)
-	if err != nil {
-		return nil, nil, err
-	}
-	for _, s := range containerStatuses {
-		if s.Name != containerName {
-			continue
-		}
-		return pod, &s, nil
-	}
-	return nil, nil, errors.New(errors.CodeNotFound, fmt.Sprintf("container %q is not found in the pod %s", containerName, c.podName))
-}
-
-func (c *k8sAPIClient) GetContainerStatuses(ctx context.Context) (*corev1.Pod, []corev1.ContainerStatus, error) {
-	pod, err := c.getPod(ctx)
-	if err != nil {
-		return nil, nil, err
-	}
-	return pod, pod.Status.ContainerStatuses, nil
-}
-
-func (c *k8sAPIClient) KillContainer(pod *corev1.Pod, container *corev1.ContainerStatus, sig syscall.Signal) error {
-	command := []string{"/bin/sh", "-c", fmt.Sprintf("kill -%d 1", sig)}
-	exec, err := common.ExecPodContainer(c.config, c.namespace, c.podName, container.Name, true, true, command...)
-	if err != nil {
-		return err
-	}
-	_, _, err = common.GetExecutorOutput(exec)
-	return err
-}
-
-func (c *k8sAPIClient) killGracefully(ctx context.Context, containerNames []string, terminationGracePeriodDuration time.Duration) error {
-	return execcommon.KillGracefully(ctx, c, containerNames, terminationGracePeriodDuration)
-}
-
-func (c *k8sAPIClient) until(ctx context.Context, f func(pod *corev1.Pod) bool) error {
-	podInterface := c.clientset.CoreV1().Pods(c.namespace)
-	for {
-		done, err := func() (bool, error) {
-			w, err := podInterface.Watch(ctx, metav1.ListOptions{FieldSelector: "metadata.name=" + c.podName})
-			if err != nil {
-				if isErrUnknownGetPods(err) {
-					return true, errWithHelp(err)
-				}
-				return true, fmt.Errorf("failed to establish pod watch: %w", err)
-			}
-			defer w.Stop()
-			for {
-				select {
-				case <-ctx.Done():
-					return true, ctx.Err()
-				case event, open := <-w.ResultChan():
-					if !open {
-						return false, fmt.Errorf("channel not open")
-					}
-					pod, ok := event.Object.(*corev1.Pod)
-					if !ok {
-						return true, apierrors.FromObject(event.Object)
-					}
-					done := f(pod)
-					if done {
-						return true, nil
-					}
-				}
-			}
-		}()
-		if done {
-			return err
-		}
-	}
-}
diff --git a/workflow/executor/k8sapi/k8sapi.go b/workflow/executor/k8sapi/k8sapi.go
deleted file mode 100644
index 0873efa9b874..000000000000
--- a/workflow/executor/k8sapi/k8sapi.go
+++ /dev/null
@@ -1,72 +0,0 @@
-package k8sapi
-
-import (
-	"context"
-	"io"
-	"time"
-
-	log "github.com/sirupsen/logrus"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/client-go/kubernetes"
-	restclient "k8s.io/client-go/rest"
-
-	"github.com/argoproj/argo-workflows/v3/errors"
-	"github.com/argoproj/argo-workflows/v3/workflow/common"
-	execcommon "github.com/argoproj/argo-workflows/v3/workflow/executor/common"
-)
-
-type K8sAPIExecutor struct {
-	client *k8sAPIClient
-}
-
-func NewK8sAPIExecutor(clientset kubernetes.Interface, config *restclient.Config, podName, namespace string) *K8sAPIExecutor {
-	client := newK8sAPIClient(clientset, config, podName, namespace)
-	return &K8sAPIExecutor{
-		client: client,
-	}
-}
-
-func (k *K8sAPIExecutor) GetFileContents(containerName string, sourcePath string) (string, error) {
-	return "", errors.Errorf(errors.CodeNotImplemented, "GetFileContents() is not implemented in the k8sapi executor.")
-}
-
-func (k *K8sAPIExecutor) CopyFile(containerName string, sourcePath string, destPath string, compressionLevel int) error {
-	return errors.Errorf(errors.CodeNotImplemented, "CopyFile() is not implemented in the k8sapi executor.")
-}
-
-func (k *K8sAPIExecutor) GetOutputStream(ctx context.Context, containerName string, combinedOutput bool) (io.ReadCloser, error) {
-	log.Infof("Getting output of %s", containerName)
-	if !combinedOutput {
-		log.Warn("non combined output unsupported")
-	}
-	return k.client.getLogsAsStream(ctx, containerName)
-}
-
-// Wait for the container to complete
-func (k *K8sAPIExecutor) Wait(ctx context.Context, containerNames []string) error {
-	return k.Until(ctx, func(pod *corev1.Pod) bool {
-		return execcommon.AllTerminated(pod.Status.ContainerStatuses, containerNames) || pod.Status.Reason == common.ErrDeadlineExceeded
-	})
-}
-
-func (k *K8sAPIExecutor) Until(ctx context.Context, f func(pod *corev1.Pod) bool) error {
-	return k.client.until(ctx, f)
-}
-
-// Kill kills a list of containers first with a SIGTERM then with a SIGKILL after a grace period
-func (k *K8sAPIExecutor) Kill(ctx context.Context, containerNames []string, terminationGracePeriodDuration time.Duration) error {
-	log.Infof("Killing containers %v", containerNames)
-	return k.client.killGracefully(ctx, containerNames, terminationGracePeriodDuration)
-}
-
-func (k *K8sAPIExecutor) ListContainerNames(ctx context.Context) ([]string, error) {
-	pod, err := k.client.getPod(ctx)
-	if err != nil {
-		return nil, err
-	}
-	var containerNames []string
-	for _, c := range pod.Status.ContainerStatuses {
-		containerNames = append(containerNames, c.Name)
-	}
-	return containerNames, nil
-}
diff --git a/workflow/executor/k8sapi/k8sapi_test.go b/workflow/executor/k8sapi/k8sapi_test.go
deleted file mode 100644
index 33f669499723..000000000000
--- a/workflow/executor/k8sapi/k8sapi_test.go
+++ /dev/null
@@ -1,19 +0,0 @@
-package k8sapi
-
-import (
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func Test_backoffOver30s(t *testing.T) {
-	x := backoffOver30s
-	assert.Equal(t, 1*time.Second, x.Step())
-	assert.Equal(t, 2*time.Second, x.Step())
-	assert.Equal(t, 4*time.Second, x.Step())
-	assert.Equal(t, 8*time.Second, x.Step())
-	assert.Equal(t, 16*time.Second, x.Step())
-	assert.Equal(t, 32*time.Second, x.Step())
-	assert.Equal(t, 64*time.Second, x.Step())
-}
diff --git a/workflow/executor/pns/pns.go b/workflow/executor/pns/pns.go
deleted file mode 100644
index 8e6dbe200b4b..000000000000
--- a/workflow/executor/pns/pns.go
+++ /dev/null
@@ -1,362 +0,0 @@
-package pns
-
-import (
-	"bufio"
-	"context"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"path/filepath"
-	"strings"
-	"sync"
-	"syscall"
-	"time"
-
-	executil "github.com/argoproj/pkg/exec"
-	gops "github.com/mitchellh/go-ps"
-	log "github.com/sirupsen/logrus"
-	"k8s.io/client-go/kubernetes"
-
-	"github.com/argoproj/argo-workflows/v3/errors"
-	"github.com/argoproj/argo-workflows/v3/util/archive"
-	"github.com/argoproj/argo-workflows/v3/workflow/common"
-	"github.com/argoproj/argo-workflows/v3/workflow/executor/k8sapi"
-	osspecific "github.com/argoproj/argo-workflows/v3/workflow/executor/os-specific"
-)
-
-const anonymousPIDPrefix = "pid/"
-
-type PNSExecutor struct {
-	*k8sapi.K8sAPIExecutor
-	podName   string
-	namespace string
-
-	// mu for `containerNameToPID``
-	mu sync.RWMutex
-
-	// mu for `pidFileHandles``
-	pmu sync.RWMutex
-
-	containerNameToPID map[string]int
-
-	// pidFileHandles holds file handles to all root containers
-	pidFileHandles map[int]*os.File
-
-	// thisPID is the pid of this process
-	thisPID int
-	// rootFS holds a file descriptor to the root filesystem, allowing the executor to exit out of a chroot
-	rootFS *os.File
-}
-
-func NewPNSExecutor(clientset *kubernetes.Clientset, podName, namespace string) (*PNSExecutor, error) {
-	thisPID := os.Getpid()
-	log.Infof("Initialized PNS executor (namespace: %s, pod: %s, pid: %d)", namespace, podName, thisPID)
-	if thisPID == 1 {
-		return nil, errors.New(errors.CodeBadRequest, "process namespace sharing is not enabled on pod")
-	}
-	delegate := k8sapi.NewK8sAPIExecutor(clientset, nil, podName, namespace)
-	return &PNSExecutor{
-		K8sAPIExecutor:     delegate,
-		podName:            podName,
-		namespace:          namespace,
-		mu:                 sync.RWMutex{},
-		pmu:                sync.RWMutex{},
-		containerNameToPID: make(map[string]int),
-		pidFileHandles:     make(map[int]*os.File),
-		thisPID:            thisPID,
-	}, nil
-}
-
-func (p *PNSExecutor) GetFileContents(containerName string, sourcePath string) (string, error) {
-	err := p.enterChroot(containerName)
-	if err != nil {
-		return "", err
-	}
-	defer func() { _ = p.exitChroot() }()
-	out, err := ioutil.ReadFile(filepath.Clean(sourcePath))
-	if err != nil {
-		return "", err
-	}
-	return string(out), nil
-}
-
-// enterChroot enters chroot of the main container
-func (p *PNSExecutor) enterChroot(containerName string) error {
-	pid := p.getContainerPID(containerName)
-	if pid == 0 {
-		return fmt.Errorf("cannot enter chroot for container named %q: no PID known - maybe short running container", containerName)
-	}
-	p.pmu.RLock()
-	defer p.pmu.RUnlock()
-	if err := p.pidFileHandles[pid].Chdir(); err != nil {
-		return errors.InternalWrapErrorf(err, "failed to chdir to main filesystem: %v", err)
-	}
-	if err := osspecific.CallChroot(); err != nil {
-		return errors.InternalWrapErrorf(err, "failed to chroot to main filesystem: %v", err)
-	}
-	return nil
-}
-
-// exitChroot exits chroot
-func (p *PNSExecutor) exitChroot() error {
-	if err := p.rootFS.Chdir(); err != nil {
-		return errors.InternalWrapError(err)
-	}
-	err := osspecific.CallChroot()
-	if err != nil {
-		return errors.InternalWrapError(err)
-	}
-	return nil
-}
-
-// CopyFile copies a source file in a container to a local path
-func (p *PNSExecutor) CopyFile(containerName string, sourcePath string, destPath string, compressionLevel int) (err error) {
-	destFile, err := os.Create(destPath)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		// exit chroot and close the file. preserve the original error
-		deferErr := p.exitChroot()
-		if err == nil && deferErr != nil {
-			err = errors.InternalWrapError(deferErr)
-		}
-		deferErr = destFile.Close()
-		if err == nil && deferErr != nil {
-			err = errors.InternalWrapError(deferErr)
-		}
-	}()
-	w := bufio.NewWriter(destFile)
-	err = p.enterChroot(containerName)
-	if err != nil {
-		return err
-	}
-
-	err = archive.TarGzToWriter(sourcePath, compressionLevel, w)
-	return err
-}
-
-func (p *PNSExecutor) Wait(ctx context.Context, containerNames []string) error {
-	go p.pollRootProcesses(ctx, containerNames)
-
-	// Secure a filehandle on our own root. This is because we will chroot back and forth from
-	// the main container's filesystem, to our own.
-	rootFS, err := os.Open("/")
-	if err != nil {
-		return fmt.Errorf("failed to open my own root: %w", err)
-	}
-	p.rootFS = rootFS
-
-	/*
-		What is a "short running container" and "late starting container"?:
-
-		Short answer: any container that exits in <5s
-
-		Long answer:
-
-		Some containers are short running and we cannot determine their PIDs because they exit too quickly.
-		This loop allows 5s for `pollRootProcesses` find PIDs, so we define any container that exits <5s as short running
-
-		Unfortunately, we cannot assume that a container that did not appeared within the 5s has completed.
-		They may still be in `ContainerCreating` state - i.e. late starting.
-	*/
-	for i := 0; !p.haveContainerPIDs(containerNames) && i < 5; i++ {
-		time.Sleep(1 * time.Second)
-	}
-
-	return p.K8sAPIExecutor.Wait(ctx, containerNames)
-}
-
-// pollRootProcesses will poll /proc for root pids (pids without parents) in a tight loop, for the
-// purpose of securing an open file handle against /proc/<pid>/root as soon as possible.
-// It opens file handles on all root pids because at this point, we do not yet know which pid is the
-// "main" container.
-// Polling is necessary because it is not possible to use something like fsnotify against procfs.
-func (p *PNSExecutor) pollRootProcesses(ctx context.Context, containerNames []string) {
-	ctx, cancel := context.WithTimeout(ctx, time.Minute)
-	defer cancel()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			if err := p.secureRootFiles(); err != nil {
-				log.WithError(err).Warn("failed to secure root files")
-			}
-			if p.haveContainerPIDs(containerNames) {
-				return
-			}
-			time.Sleep(50 * time.Millisecond)
-		}
-	}
-}
-
-func (p *PNSExecutor) haveContainerPIDs(containerNames []string) bool {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	for _, n := range containerNames {
-		if p.containerNameToPID[n] == 0 {
-			return false
-		}
-	}
-	return true
-}
-
-// Kill a list of containers first with a SIGTERM then with a SIGKILL after a grace period
-func (p *PNSExecutor) Kill(ctx context.Context, containerNames []string, terminationGracePeriodDuration time.Duration) error {
-	var asyncErr error
-	wg := sync.WaitGroup{}
-	for _, containerName := range containerNames {
-		wg.Add(1)
-		go func(containerName string) {
-			err := p.killContainer(containerName, terminationGracePeriodDuration)
-			if err != nil && asyncErr != nil {
-				asyncErr = err
-			}
-			wg.Done()
-		}(containerName)
-	}
-	wg.Wait()
-	return asyncErr
-}
-
-func (p *PNSExecutor) ListContainerNames(ctx context.Context) ([]string, error) {
-	procs, err := gops.Processes()
-	if err != nil {
-		return nil, fmt.Errorf("failed to list processes: %w", err)
-	}
-	var containerNames []string
-	for _, proc := range procs {
-		if !p.isRootContainerProcess(proc) {
-			continue
-		}
-		n, err := containerNameForPID(proc.Pid())
-		if err != nil {
-			return nil, fmt.Errorf("failed to get container name for process %d: %w", proc.Pid(), err)
-		}
-		containerNames = append(containerNames, n)
-	}
-	return containerNames, nil
-}
-
-func (p *PNSExecutor) killContainer(containerName string, terminationGracePeriodDuration time.Duration) error {
-	pid := p.getContainerPID(containerName)
-	if pid == 0 {
-		log.Warnf("No PID for container named %q. Process assumed to have completed", containerName)
-		return nil
-	}
-	// On Unix systems, FindProcess always succeeds and returns a Process
-	// for the given pid, regardless of whether the process exists.
-	proc, _ := os.FindProcess(pid)
-	log.Infof("Sending SIGTERM to pid %d", pid)
-	err := proc.Signal(syscall.SIGTERM)
-	if err != nil {
-		log.Warnf("Failed to SIGTERM pid %d: %v", pid, err)
-	}
-	waitPIDOpts := executil.WaitPIDOpts{Timeout: terminationGracePeriodDuration}
-	err = executil.WaitPID(pid, waitPIDOpts)
-	if err == nil {
-		log.Infof("PID %d completed", pid)
-		return nil
-	}
-	if err != executil.ErrWaitPIDTimeout {
-		return err
-	}
-	log.Warnf("Timed out (%v) waiting for pid %d to complete after SIGTERM. Issuing SIGKILL", waitPIDOpts.Timeout, pid)
-	err = proc.Signal(syscall.SIGKILL)
-	if err != nil {
-		log.Warnf("Failed to SIGKILL pid %d: %v", pid, err)
-	}
-	return err
-}
-
-// returns the entries associated with the container id. Returns zero if it was unable
-// to be determined because no running root processes exist with that container ID
-func (p *PNSExecutor) getContainerPID(containerName string) int {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	if pid, ok := p.containerNameToPID[containerName]; ok {
-		return pid
-	}
-	procs, _ := gops.Processes()
-	for _, proc := range procs {
-		n, _ := containerNameForPID(proc.Pid())
-		if n == containerName {
-			return proc.Pid()
-		}
-	}
-	for n, pid := range p.containerNameToPID {
-		// the container can't be me, and it must be anonymous, otherwise we would have determined it
-		if pid != os.Getpid() && strings.HasPrefix(n, anonymousPIDPrefix) {
-			log.Infof("guessing container %q PID is %d", containerName, pid)
-			return pid
-		}
-	}
-	return 0
-}
-
-func containerNameForPID(pid int) (string, error) {
-	data, err := ioutil.ReadFile(fmt.Sprintf("/proc/%d/environ", pid))
-	if err != nil {
-		return "", err
-	}
-	prefix := common.EnvVarContainerName + "="
-	for _, l := range strings.Split(string(data), "\000") {
-		if strings.HasPrefix(l, prefix) {
-			return strings.TrimPrefix(l, prefix), nil
-		}
-	}
-	return fmt.Sprintf("%s%d", anonymousPIDPrefix, pid), nil // we give all a "container name", including a fake name for injected sidecars
-}
-
-func (p *PNSExecutor) secureRootFiles() error {
-	processes, err := gops.Processes()
-	if err != nil {
-		return err
-	}
-	for _, proc := range processes {
-		err = func() error {
-			pid := proc.Pid()
-			if !p.isRootContainerProcess(proc) {
-				return nil
-			}
-
-			fs, err := os.Open(fmt.Sprintf("/proc/%d/root", pid))
-			if err != nil {
-				return err
-			}
-
-			p.pmu.Lock()
-			defer p.pmu.Unlock()
-			if p.pidFileHandles[pid] != fs {
-				// the main container may have switched (e.g. gone from busybox to the user's container)
-				if prevInfo, ok := p.pidFileHandles[pid]; ok {
-					_ = prevInfo.Close()
-				}
-				p.pidFileHandles[pid] = fs
-				log.Infof("secured root for pid %d root: %s (%q)", pid, proc.Executable(), fs.Name())
-			}
-
-			containerName, err := containerNameForPID(pid)
-			if err != nil {
-				return err
-			}
-			if p.containerNameToPID[containerName] != pid {
-				p.mu.Lock()
-				defer p.mu.Unlock()
-				p.containerNameToPID[containerName] = pid
-				log.Infof("mapped container name %q to pid %d", containerName, pid)
-			}
-			return nil
-		}()
-		if err != nil {
-			log.WithError(err).Warnf("failed to secure root file handle for %d", proc.Pid())
-		}
-	}
-	return nil
-}
-
-func (p *PNSExecutor) isRootContainerProcess(proc gops.Process) bool {
-	// ignore the pause container, our own pid, and non-root processes
-	return proc.Pid() != 1 && proc.Pid() != p.thisPID && proc.PPid() == 0
-}
diff --git a/workflow/validate/validate.go b/workflow/validate/validate.go
index e92776aaeb08..b3a625fa3bc5 100644
--- a/workflow/validate/validate.go
+++ b/workflow/validate/validate.go
@@ -34,10 +34,6 @@ type ValidateOpts struct {
 	// input parameters to the workflow)
 	Lint bool
 
-	// ContainerRuntimeExecutor will trigger additional validation checks specific to different
-	// types of executors.
-	ContainerRuntimeExecutor string
-
 	// IgnoreEntrypoint indicates to skip/ignore the EntryPoint validation on workflow spec.
 	// Entrypoint is optional for WorkflowTemplate and ClusterWorkflowTemplate
 	IgnoreEntrypoint bool
@@ -405,10 +401,6 @@ func (ctx *templateValidationCtx) validateTemplate(tmpl *wfv1.Template, tmplCtx
 	if err != nil {
 		return err
 	}
-	err = ctx.validateBaseImageOutputs(newTmpl)
-	if err != nil {
-		return err
-	}
 	if newTmpl.ArchiveLocation != nil {
 		errPrefix := fmt.Sprintf("templates.%s.archiveLocation", newTmpl.Name)
 		err = validateArtifactLocation(errPrefix, *newTmpl.ArchiveLocation)
@@ -990,30 +982,6 @@ func validateOutputs(scope map[string]interface{}, globalParams map[string]strin
 	return nil
 }
 
-// validateBaseImageOutputs detects if the template contains an valid output from base image layer
-func (ctx *templateValidationCtx) validateBaseImageOutputs(tmpl *wfv1.Template) error {
-	// This validation is not applicable for DAG and Step Template types
-	if tmpl.GetType() == wfv1.TemplateTypeDAG || tmpl.GetType() == wfv1.TemplateTypeSteps {
-		return nil
-	}
-	switch ctx.ContainerRuntimeExecutor {
-	case common.ContainerRuntimeExecutorPNS:
-		// pns supports copying from the base image, but only if there is no volume mount underneath it
-		errMsg := "pns executor does not support outputs from base image layer with volume mounts. Use an emptyDir: https://argoproj.github.io/argo-workflows/empty-dir/"
-		for _, out := range tmpl.Outputs.Artifacts {
-			if common.FindOverlappingVolume(tmpl, out.Path) == nil {
-				// output is in the base image layer. need to verify there are no volume mounts under it
-				for _, volMnt := range tmpl.GetVolumeMounts() {
-					if strings.HasPrefix(volMnt.MountPath, out.Path+"/") {
-						return errors.Errorf(errors.CodeBadRequest, "templates.%s.outputs.artifacts.%s: %s", tmpl.Name, out.Name, errMsg)
-					}
-				}
-			}
-		}
-	}
-	return nil
-}
-
 // validateOutputParameter verifies that only one of valueFrom is defined in an output
 func validateOutputParameter(paramRef string, param *wfv1.Parameter) error {
 	if param.ValueFrom != nil && param.Value != nil {
diff --git a/workflow/validate/validate_test.go b/workflow/validate/validate_test.go
index f6f68e73b418..8691c239e353 100644
--- a/workflow/validate/validate_test.go
+++ b/workflow/validate/validate_test.go
@@ -38,11 +38,6 @@ func validate(yamlStr string) (*wfv1.Conditions, error) {
 	return ValidateWorkflow(wftmplGetter, cwftmplGetter, wf, ValidateOpts{})
 }
 
-func validateWithOptions(yamlStr string, opts ValidateOpts) (*wfv1.Conditions, error) {
-	wf := unmarshalWf(yamlStr)
-	return ValidateWorkflow(wftmplGetter, cwftmplGetter, wf, opts)
-}
-
 // validateWorkflowTemplate is a test helper to accept WorkflowTemplate YAML as a string and return
 // its validation result.
 func validateWorkflowTemplate(yamlStr string, opts ValidateOpts) error {
@@ -1496,122 +1491,6 @@ func TestCustomTemplatVariable(t *testing.T) {
 	assert.Equal(t, err, nil)
 }
 
-var baseImageOutputArtifact = `
-apiVersion: argoproj.io/v1alpha1
-kind: Workflow
-metadata:
-  generateName: base-image-out-art-
-spec:
-  entrypoint: base-image-out-art
-  templates:
-  - name: base-image-out-art
-    container:
-      image: alpine:latest
-      command: [echo, hello]
-    outputs:
-      artifacts:
-      - name: tmp
-        path: /tmp
-`
-
-var baseImageOutputParameter = `
-apiVersion: argoproj.io/v1alpha1
-kind: Workflow
-metadata:
-  generateName: base-image-out-art-
-spec:
-  entrypoint: base-image-out-art
-  templates:
-  - name: base-image-out-art
-    container:
-      image: alpine:latest
-      command: [echo, hello]
-    outputs:
-      parameters:
-      - name: tmp
-        valueFrom:
-          path: /tmp/file
-`
-
-var volumeMountOutputArtifact = `
-apiVersion: argoproj.io/v1alpha1
-kind: Workflow
-metadata:
-  generateName: base-image-out-art-
-spec:
-  entrypoint: base-image-out-art
-  volumes:
-  - name: workdir
-    emptyDir: {}
-  templates:
-  - name: base-image-out-art
-    container:
-      image: alpine:latest
-      command: [echo, hello]
-      volumeMounts:
-      - name: workdir
-        mountPath: /mnt/vol
-    outputs:
-      artifacts:
-      - name: workdir
-        path: /mnt/vol
-`
-
-var baseImageDirWithEmptyDirOutputArtifact = `
-apiVersion: argoproj.io/v1alpha1
-kind: Workflow
-metadata:
-  generateName: base-image-out-art-
-spec:
-  entrypoint: base-image-out-art
-  volumes:
-  - name: workdir
-    emptyDir: {}
-  templates:
-  - name: base-image-out-art
-    container:
-      image: alpine:latest
-      command: [echo, hello]
-      volumeMounts:
-      - name: workdir
-        mountPath: /mnt/vol
-    outputs:
-      artifacts:
-      - name: workdir
-        path: /mnt
-`
-
-// TestBaseImageOutputVerify verifies we error when we detect the condition when the container
-// runtime executor doesn't support output artifacts from a base image layer, and fails validation
-func TestBaseImageOutputVerify(t *testing.T) {
-	wfBaseOutArt := unmarshalWf(baseImageOutputArtifact)
-	wfBaseOutParam := unmarshalWf(baseImageOutputParameter)
-	wfEmptyDirOutArt := unmarshalWf(volumeMountOutputArtifact)
-	wfBaseWithEmptyDirOutArt := unmarshalWf(baseImageDirWithEmptyDirOutputArtifact)
-	var err error
-
-	for _, executor := range []string{common.ContainerRuntimeExecutorPNS, common.ContainerRuntimeExecutorEmissary, ""} {
-		switch executor {
-		case common.ContainerRuntimeExecutorPNS:
-			_, err = ValidateWorkflow(wftmplGetter, cwftmplGetter, wfBaseOutArt, ValidateOpts{ContainerRuntimeExecutor: executor})
-			assert.NoError(t, err)
-			_, err = ValidateWorkflow(wftmplGetter, cwftmplGetter, wfBaseOutParam, ValidateOpts{ContainerRuntimeExecutor: executor})
-			assert.NoError(t, err)
-			_, err = ValidateWorkflow(wftmplGetter, cwftmplGetter, wfBaseWithEmptyDirOutArt, ValidateOpts{ContainerRuntimeExecutor: executor})
-			assert.Error(t, err)
-		case common.ContainerRuntimeExecutorEmissary, "":
-			_, err = ValidateWorkflow(wftmplGetter, cwftmplGetter, wfBaseOutArt, ValidateOpts{ContainerRuntimeExecutor: executor})
-			assert.NoError(t, err)
-			_, err = ValidateWorkflow(wftmplGetter, cwftmplGetter, wfBaseOutParam, ValidateOpts{ContainerRuntimeExecutor: executor})
-			assert.NoError(t, err)
-			_, err = ValidateWorkflow(wftmplGetter, cwftmplGetter, wfBaseWithEmptyDirOutArt, ValidateOpts{ContainerRuntimeExecutor: executor})
-			assert.NoError(t, err)
-		}
-		_, err = ValidateWorkflow(wftmplGetter, cwftmplGetter, wfEmptyDirOutArt, ValidateOpts{ContainerRuntimeExecutor: executor})
-		assert.NoError(t, err)
-	}
-}
-
 var templateRefTarget = `
 apiVersion: argoproj.io/v1alpha1
 kind: WorkflowTemplate
@@ -2416,67 +2295,6 @@ func TestWorkflowWithWFTRefWithOverrideParam(t *testing.T) {
 	assert.NoError(t, err)
 }
 
-var dagAndStepLevelOutputArtifacts = `
-apiVersion: argoproj.io/v1alpha1
-kind: Workflow
-metadata:
-  generateName: dag-target-
-spec:
-  entrypoint: main
-  templates:
-  - name: main
-    outputs:
-      artifacts:
-        - name: artifact
-          from: "{{tasks.artifact-svn-retrieve.outputs.artifacts.artifact}}"
-    dag:
-      tasks:
-      - name: artifact-svn-retrieve
-        template: artifact-svn-retrieve
-      - name: step-tmpl
-        template: step
-
-  - name: step
-    outputs:
-      artifacts:
-        - name: artifact
-          from: "{{steps.artifact-svn-retrieve.outputs.artifacts.artifact}}"
-    steps:
-    - - name: artifact-svn-retrieve
-        template: artifact-svn-retrieve
-
-  - name: artifact-svn-retrieve
-    outputs:
-      artifacts:
-      - name: artifact
-        path: "/vol/hello_world.txt"
-    container:
-      image: docker/whalesay:latest
-      command: [sh, -c]
-      args: ["sleep 1; cowsay hello world | tee /vol/hello_world.txt"]
-      volumeMounts:
-      - name: vol
-        mountPath: "/vol"
-    volumes:
-    - name: vol
-      emptyDir: {}
-`
-
-func TestDagAndStepLevelOutputArtifactsForDiffExecutor(t *testing.T) {
-	t.Run("DefaultExecutor", func(t *testing.T) {
-		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: ""})
-		assert.NoError(t, err)
-	})
-	t.Run("EmissaryExecutor", func(t *testing.T) {
-		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: common.ContainerRuntimeExecutorEmissary})
-		assert.NoError(t, err)
-	})
-	t.Run("PNSExecutor", func(t *testing.T) {
-		_, err := validateWithOptions(dagAndStepLevelOutputArtifacts, ValidateOpts{ContainerRuntimeExecutor: common.ContainerRuntimeExecutorPNS})
-		assert.NoError(t, err)
-	})
-}
-
 var testWorkflowTemplateLabels = `
 apiVersion: argoproj.io/v1alpha1
 kind: WorkflowTemplate