Repository: linkerd/linkerd2
Pull request #11918 (2024-01-12 03:47:45)

build(deps): bump golang.org/x/tools from 0.16.1 to 0.17.0
Bumps [golang.org/x/tools](https://github.com/golang/tools) from 0.16.1 to 0.17.0.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/golang/tools/commit/0b1f1d4bc227cc2e610854f23e14696becb9e46c"><code>0b1f1d4</code></a> gopls/internal/lsp/cache: (re-)ensure clean shutdown</li>
<li><a href="https://github.com/golang/tools/commit/706525de51451bb409297f305f23ac473eb693ff"><code>706525d</code></a> gopls/internal/lsp/source/completion: support postfix completion (iferr,</li>
<li><a href="https://github.com/golang/tools/commit/581c0b357f883185d60780718e053e600005dac4"><code>581c0b3</code></a> gopls/internal/lsp/source: add receiver name to stubbed methods</li>
<li><a href="https://github.com/golang/tools/commit/c95fa0ff4c2370b6f4b78947fc45987c8d0d664a"><code>c95fa0f</code></a> gopls/internal/test: skip marker tests on darwin builders if -short</li>
<li><a href="https://github.com/golang/tools/commit/25a0e9d3e36e70c0dd9ab7302baa370bd20373b1"><code>25a0e9d</code></a> go.mod: update golang.org/x dependencies</li>
<li><a href="https://github.com/golang/tools/commit/c9c95f97fda18f56ab4c8c406e9f48a4cb3b3d7e"><code>c9c95f9</code></a> internal/refactor/inline: improve a confusing error message</li>
<li><a href="https://github.com/golang/tools/commit/ba8672b53f4b6840528df6fd534cb16f65278711"><code>ba8672b</code></a> go/analysis/passes/unusedresult: add functions from slices package</li>
<li><a href="https://github.com/golang/tools/commit/3e8a0a3b0845dc1af9def080412d0a11d11a10d6"><code>3e8a0a3</code></a> gopls/internal/lsp/cache: address additional comments from CL 553095</li>
<li><a href="https://github.com/golang/tools/commit/920d665b996049294bff24b972efc9f3d949a8ba"><code>920d665</code></a> gopls/internal/cmd: factor three loops in fix subcommand</li>
<li><a href="https://github.com/golang/tools/commit/782573673af31588817cb7e79a1baeca1570609c"><code>7825736</code></a> gopls/internal/lsp/cache: simplify critical errors</li>
<li>Additional commits viewable in <a href="https://github.com/golang/tools/compare/v0.16.1...v0.17.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=golang.org/x/tools&package-manager=go_modules&previous-version=0.16.1&new-version=0.17.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 41bf3aa42..2b4bc30c4 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -148,7 +148,7 @@ jobs:
     timeout-minutes: 5
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
-      - uses: tj-actions/changed-files@716b1e13042866565e00e85fd4ec490e186c4a2f
+      - uses: tj-actions/changed-files@62f4729b5df35e6e0e01265fa70a82ccaf196b4b
         id: changed
         with:
           files: |
@@ -287,7 +287,7 @@ jobs:
     timeout-minutes: 5
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
-      - uses: tj-actions/changed-files@716b1e13042866565e00e85fd4ec490e186c4a2f
+      - uses: tj-actions/changed-files@62f4729b5df35e6e0e01265fa70a82ccaf196b4b
         id: changed
         with:
           files: |
@@ -330,7 +330,7 @@ jobs:
     timeout-minutes: 5
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
-      - uses: tj-actions/changed-files@716b1e13042866565e00e85fd4ec490e186c4a2f
+      - uses: tj-actions/changed-files@62f4729b5df35e6e0e01265fa70a82ccaf196b4b
         id: changed
         with:
           files: |
diff --git a/CHANGES.md b/CHANGES.md
index 8c1833b89..a3a3bd278 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,5 +1,10 @@
 # Changes
 
+## edge-24.1.2
+
+This edge release incrementally improves support for ExternalWorkload resources
+throughout the control plane.
+
 ## edge-24.1.1
 
 This edge release introduces a number of different fixes and improvements. More
diff --git a/Cargo.lock b/Cargo.lock
index ec31c5562..fac92f84e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -615,9 +615,9 @@ dependencies = [
 
 [[package]]
 name = "h2"
-version = "0.3.22"
+version = "0.3.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d6250322ef6e60f93f9a2162799302cd6f68f79f6e5d85c8c16f14d1d958178"
+checksum = "bb2c4422095b67ee78da96fbb51a4cc413b3b25883c7717ff7ca1ab31022c9c9"
 dependencies = [
  "bytes",
  "fnv",
@@ -1297,6 +1297,7 @@ dependencies = [
  "k8s-openapi",
  "kube",
  "linkerd-policy-controller-core",
+ "linkerd-policy-controller-grpc",
  "linkerd-policy-controller-k8s-api",
  "linkerd2-proxy-api",
  "maplit",
@@ -1588,9 +1589,9 @@ dependencies = [
 
 [[package]]
 name = "pest_generator"
-version = "2.7.5"
+version = "2.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68bd1206e71118b5356dae5ddc61c8b11e28b09ef6a31acbd15ea48a28e0c227"
+checksum = "2a31940305ffc96863a735bef7c7994a00b325a7138fdbc5bda0f1a0476d3275"
 dependencies = [
  "pest",
  "pest_meta",
@@ -1601,9 +1602,9 @@ dependencies = [
 
 [[package]]
 name = "pest_meta"
-version = "2.7.5"
+version = "2.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c747191d4ad9e4a4ab9c8798f1e82a39affe7ef9648390b7e5548d18e099de6"
+checksum = "a7ff62f5259e53b78d1af898941cdcdccfae7385cf7d793a6e55de5d05bb4b7d"
 dependencies = [
  "once_cell",
  "pest",
@@ -2561,9 +2562,9 @@ dependencies = [
 
 [[package]]
 name = "typenum"
-version = "1.16.0"
+version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "497961ef93d974e23eb6f433eb5fe1b7930b659f06d12dec6fc44a8f554c0bba"
+checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825"
 
 [[package]]
 name = "ucd-trie"
diff --git a/charts/linkerd-control-plane/Chart.yaml b/charts/linkerd-control-plane/Chart.yaml
index 37f9a3a29..ad7d38dae 100644
--- a/charts/linkerd-control-plane/Chart.yaml
+++ b/charts/linkerd-control-plane/Chart.yaml
@@ -16,7 +16,7 @@ dependencies:
   - name: partials
     version: 0.1.0
     repository: file://../partials
-version: 1.18.4-edge
+version: 1.18.5-edge
 icon: https://linkerd.io/images/logo-only-200h.png
 maintainers:
   - name: Linkerd authors
diff --git a/charts/linkerd-control-plane/README.md b/charts/linkerd-control-plane/README.md
index aacee978a..760886326 100644
--- a/charts/linkerd-control-plane/README.md
+++ b/charts/linkerd-control-plane/README.md
@@ -3,7 +3,7 @@
 Linkerd gives you observability, reliability, and security
 for your microservices — with no code change required.
 
-![Version: 1.18.4-edge](https://img.shields.io/badge/Version-1.18.4--edge-informational?style=flat-square)
+![Version: 1.18.5-edge](https://img.shields.io/badge/Version-1.18.5--edge-informational?style=flat-square)
 ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square)
 ![AppVersion: edge-XX.X.X](https://img.shields.io/badge/AppVersion-edge--XX.X.X-informational?style=flat-square)
 
diff --git a/charts/linkerd-control-plane/templates/destination-rbac.yaml b/charts/linkerd-control-plane/templates/destination-rbac.yaml
index 7fa47eb30..d05ed0dd3 100644
--- a/charts/linkerd-control-plane/templates/destination-rbac.yaml
+++ b/charts/linkerd-control-plane/templates/destination-rbac.yaml
@@ -23,10 +23,16 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
   {{- if .Values.enableEndpointSlices }}
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
   {{- end }}
 ---
 kind: ClusterRoleBinding
@@ -243,6 +249,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
diff --git a/controller/api/destination/endpoint_translator.go b/controller/api/destination/endpoint_translator.go
index 4fc280d3a..88acf8078 100644
--- a/controller/api/destination/endpoint_translator.go
+++ b/controller/api/destination/endpoint_translator.go
@@ -9,9 +9,11 @@ import (
 	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
 	"github.com/linkerd/linkerd2-proxy-api/go/net"
 	"github.com/linkerd/linkerd2/controller/api/destination/watcher"
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
 	"github.com/linkerd/linkerd2/controller/k8s"
 	"github.com/linkerd/linkerd2/pkg/addr"
 	pkgK8s "github.com/linkerd/linkerd2/pkg/k8s"
+	"github.com/linkerd/linkerd2/pkg/util"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	logging "github.com/sirupsen/logrus"
@@ -379,7 +381,14 @@ func (et *endpointTranslator) sendClientAdd(set watcher.AddressSet) {
 			opaquePorts = watcher.GetAnnotatedOpaquePorts(address.Pod, et.defaultOpaquePorts)
 			wa, err = createWeightedAddr(address, opaquePorts, et.enableH2Upgrade, et.identityTrustDomain, et.controllerNS)
 			if err != nil {
-				et.log.Errorf("Failed to translate endpoints to weighted addr: %s", err)
+				et.log.Errorf("Failed to translate Pod endpoints to weighted addr: %s", err)
+				continue
+			}
+		} else if address.ExternalWorkload != nil {
+			opaquePorts = getAnnotatedOpaquePortsForExternalWorkload(address.ExternalWorkload, et.defaultOpaquePorts)
+			wa, err = createWeightedAddrForExternalWorkload(address, opaquePorts)
+			if err != nil {
+				et.log.Errorf("Failed to translate ExternalWorkload endpoints to weighted addr: %s", err)
 				continue
 			}
 		} else {
@@ -480,6 +489,74 @@ func toAddr(address watcher.Address) (*net.TcpAddress, error) {
 	}, nil
 }
 
+func createWeightedAddrForExternalWorkload(
+	address watcher.Address,
+	opaquePorts map[uint32]struct{},
+) (*pb.WeightedAddr, error) {
+	tcpAddr, err := toAddr(address)
+	if err != nil {
+		return nil, err
+	}
+
+	weightedAddr := pb.WeightedAddr{
+		Addr:         tcpAddr,
+		Weight:       defaultWeight,
+		MetricLabels: map[string]string{},
+	}
+
+	weightedAddr.MetricLabels = pkgK8s.GetExternalWorkloadLabels(address.OwnerKind, address.OwnerName, address.ExternalWorkload)
+	// If the address is not backed by an ExternalWorkload, there is no additional metadata
+	// to add.
+	if address.ExternalWorkload == nil {
+		return &weightedAddr, nil
+	}
+
+	weightedAddr.ProtocolHint = &pb.ProtocolHint{}
+	_, opaquePort := opaquePorts[address.Port]
+	// If address is set as opaque by a Server, or its port is set as
+	// opaque by annotation or default value, then set the hinted protocol to
+	// Opaque.
+	if address.OpaqueProtocol || opaquePort {
+		weightedAddr.ProtocolHint.Protocol = &pb.ProtocolHint_Opaque_{
+			Opaque: &pb.ProtocolHint_Opaque{},
+		}
+
+		port, err := getInboundPortFromExternalWorkload(&address.ExternalWorkload.Spec)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read inbound port: %w", err)
+		}
+		weightedAddr.ProtocolHint.OpaqueTransport = &pb.ProtocolHint_OpaqueTransport{
+			InboundPort: port,
+		}
+	} else {
+		weightedAddr.ProtocolHint.Protocol = &pb.ProtocolHint_H2_{
+			H2: &pb.ProtocolHint_H2{},
+		}
+	}
+
+	// we assume external workloads support only SPIRE identity
+	weightedAddr.TlsIdentity = &pb.TlsIdentity{
+		Strategy: &pb.TlsIdentity_UriLikeIdentity_{
+			UriLikeIdentity: &pb.TlsIdentity_UriLikeIdentity{
+				Uri: address.ExternalWorkload.Spec.MeshTls.Identity,
+			},
+		},
+		ServerName: &pb.TlsIdentity_DnsLikeIdentity{
+			Name: address.ExternalWorkload.Spec.MeshTls.ServerName,
+		},
+	}
+
+	weightedAddr.MetricLabels = pkgK8s.GetExternalWorkloadLabels(address.OwnerKind, address.OwnerName, address.ExternalWorkload)
+	// Set a zone label, even if it is empty (for consistency).
+	z := ""
+	if address.Zone != nil {
+		z = *address.Zone
+	}
+	weightedAddr.MetricLabels["zone"] = z
+
+	return &weightedAddr, nil
+}
+
 func createWeightedAddr(
 	address watcher.Address,
 	opaquePorts map[uint32]struct{},
@@ -558,12 +635,13 @@ func createWeightedAddr(
 		!isSkippedInboundPort {
 
 		id := fmt.Sprintf("%s.%s.serviceaccount.identity.%s.%s", sa, ns, controllerNSLabel, identityTrustDomain)
+		tlsId := &pb.TlsIdentity_DnsLikeIdentity{Name: id}
+
 		weightedAddr.TlsIdentity = &pb.TlsIdentity{
 			Strategy: &pb.TlsIdentity_DnsLikeIdentity_{
-				DnsLikeIdentity: &pb.TlsIdentity_DnsLikeIdentity{
-					Name: id,
-				},
+				DnsLikeIdentity: tlsId,
 			},
+			ServerName: tlsId,
 		}
 	}
 
@@ -609,3 +687,65 @@ func getInboundPort(podSpec *corev1.PodSpec) (uint32, error) {
 	}
 	return 0, fmt.Errorf("failed to find %s environment variable in any container for given pod spec", envInboundListenAddr)
 }
+
+// getInboundPortFromExternalWorkload gets the inbound port from the ExternalWorkload spec
+// variable.
+func getInboundPortFromExternalWorkload(ewSpec *ewv1alpha1.ExternalWorkloadSpec) (uint32, error) {
+	for _, p := range ewSpec.Ports {
+		if p.Name == pkgK8s.ProxyPortName {
+			return uint32(p.Port), nil
+		}
+	}
+
+	return 0, fmt.Errorf("failed to find %s port for given ExternalWorkloadSpec", pkgK8s.ProxyPortName)
+}
+
+// getAnnotatedOpaquePortsForExternalWorkload returns the opaque ports for the external workload given its
+// annotations, or the default opaque ports if it's not annotated
+func getAnnotatedOpaquePortsForExternalWorkload(ew *ewv1alpha1.ExternalWorkload, defaultPorts map[uint32]struct{}) map[uint32]struct{} {
+	if ew == nil {
+		return defaultPorts
+	}
+	annotation, ok := ew.Annotations[pkgK8s.ProxyOpaquePortsAnnotation]
+	if !ok {
+		return defaultPorts
+	}
+	opaquePorts := make(map[uint32]struct{})
+	if annotation != "" {
+		for _, pr := range parseExternalWorkloadOpaquePorts(annotation, ew) {
+			for _, port := range pr.Ports() {
+				opaquePorts[uint32(port)] = struct{}{}
+			}
+		}
+	}
+	return opaquePorts
+}
+
+func parseExternalWorkloadOpaquePorts(override string, ew *ewv1alpha1.ExternalWorkload) []util.PortRange {
+	portRanges := util.GetPortRanges(override)
+	var values []util.PortRange
+	for _, pr := range portRanges {
+		port, named := isNamedInExternalWorkload(pr, ew)
+		if named {
+			values = append(values, util.PortRange{UpperBound: int(port), LowerBound: int(port)})
+		} else {
+			pr, err := util.ParsePortRange(pr)
+			if err != nil {
+				logging.Warnf("Invalid port range [%v]: %s", pr, err)
+				continue
+			}
+			values = append(values, pr)
+		}
+	}
+	return values
+}
+
+func isNamedInExternalWorkload(pr string, ew *ewv1alpha1.ExternalWorkload) (int32, bool) {
+	for _, p := range ew.Spec.Ports {
+		if p.Name == pr {
+			return p.Port, true
+		}
+	}
+
+	return 0, false
+}
diff --git a/controller/api/destination/external-workload/endpoints_controller.go b/controller/api/destination/external-workload/endpoints_controller.go
new file mode 100644
index 000000000..9646a9bfd
--- /dev/null
+++ b/controller/api/destination/external-workload/endpoints_controller.go
@@ -0,0 +1,564 @@
+package externalworkload
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
+	"github.com/linkerd/linkerd2/controller/k8s"
+	logging "github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/client-go/util/workqueue"
+)
+
+const (
+	// Name of the lease resource the controller will use
+	leaseName = "linkerd-destination-endpoint-write"
+
+	// Duration of the lease
+	leaseDuration = 30 * time.Second
+
+	// Deadline for the leader to refresh its lease. Core controllers have a
+	// deadline of 10 seconds.
+	leaseRenewDeadline = 10 * time.Second
+
+	// Duration a leader elector should wait in between action re-tries.
+	// Core controllers have a value of 2 seconds.
+	leaseRetryPeriod = 2 * time.Second
+
+	// Max retries for a service to be reconciled
+	maxRetryBudget = 15
+)
+
+// EndpointsController reconciles service memberships for ExternalWorkload resources
+// by writing EndpointSlice objects for Services that select one or more
+// external endpoints.
+type EndpointsController struct {
+	k8sAPI   *k8s.API
+	log      *logging.Entry
+	queue    workqueue.RateLimitingInterface
+	stop     chan struct{}
+	isLeader atomic.Bool
+
+	lec leaderelection.LeaderElectionConfig
+	sync.RWMutex
+}
+
+// The EndpointsController code has been structured (and modified) based on the
+// core EndpointSlice controller. Copyright 2014 The Kubernetes Authors
+// https://github.com/kubernetes/kubernetes/blob/29fad383dab0dd7b7b563ec9eae10156616a6f34/pkg/controller/endpointslice/endpointslice_controller.go
+//
+// There are some fundamental differences between the core endpoints controller
+// and Linkerd's endpoints controller; for one, the churn rate is expected to be
+// much lower for a controller that reconciles ExternalWorkload resources.
+// Furthermore, the structure of the resource is different, statuses do not
+// contain as many conditions, and the lifecycle of an ExternalWorkload is
+// different to that of a Pod (e.g. a workload is long lived).
+//
+// NewEndpointsController creates a new controller. The controller must be
+// started with its `Start()` method.
+func NewEndpointsController(k8sAPI *k8s.API, hostname, controllerNs string, stopCh chan struct{}, exportQueueMetrics bool) (*EndpointsController, error) {
+	workQueueConfig := workqueue.RateLimitingQueueConfig{
+		Name: "endpoints_controller_workqueue",
+	}
+	if exportQueueMetrics {
+		workQueueConfig.MetricsProvider = newWorkQueueMetricsProvider()
+	}
+
+	ec := &EndpointsController{
+		k8sAPI: k8sAPI,
+		queue:  workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workQueueConfig),
+		stop:   stopCh,
+		log: logging.WithFields(logging.Fields{
+			"component": "external-endpoints-controller",
+		}),
+	}
+
+	_, err := k8sAPI.Svc().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    ec.updateService,
+		DeleteFunc: ec.updateService,
+		UpdateFunc: func(_, newObj interface{}) {
+			ec.updateService(newObj)
+		},
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = k8sAPI.ExtWorkload().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			ew, ok := obj.(*ewv1alpha1.ExternalWorkload)
+			if !ok {
+				ec.log.Errorf("couldn't get ExternalWorkload from object %#v", obj)
+				return
+			}
+
+			if len(ew.Spec.WorkloadIPs) == 0 {
+				ec.log.Debugf("ExternalWorkload %s/%s has no IP addresses", ew.Namespace, ew.Name)
+				return
+			}
+
+			if len(ew.Spec.Ports) == 0 {
+				ec.log.Debugf("ExternalWorkload %s/%s has no ports", ew.Namespace, ew.Name)
+				return
+			}
+
+			services, err := ec.getSvcMembership(ew)
+			if err != nil {
+				ec.log.Errorf("failed to get service membership for %s/%s: %v", ew.Namespace, ew.Name, err)
+				return
+			}
+
+			for _, svc := range services {
+				ec.queue.Add(svc)
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			ew, ok := obj.(*ewv1alpha1.ExternalWorkload)
+			if !ok {
+				tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					ec.log.Errorf("couldn't get object from tombstone %+v", obj)
+					return
+				}
+				ew, ok = tombstone.Obj.(*ewv1alpha1.ExternalWorkload)
+				if !ok {
+					ec.log.Errorf("tombstone contained object that is not an ExternalWorkload %+v", obj)
+					return
+				}
+			}
+
+			services, err := ec.getSvcMembership(ew)
+			if err != nil {
+				ec.log.Errorf("failed to get service membership for %s/%s: %v", ew.Namespace, ew.Name, err)
+				return
+			}
+
+			for _, svc := range services {
+				ec.queue.Add(svc)
+			}
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			old, ok := oldObj.(*ewv1alpha1.ExternalWorkload)
+			if !ok {
+				ec.log.Errorf("couldn't get ExternalWorkload from object %#v", oldObj)
+				return
+			}
+
+			updated, ok := newObj.(*ewv1alpha1.ExternalWorkload)
+			if !ok {
+				ec.log.Errorf("couldn't get ExternalWorkload from object %#v", newObj)
+				return
+			}
+
+			// Ignore resync updates. If nothing has changed in the object, then
+			// the update processing is redudant.
+			if old.ResourceVersion == updated.ResourceVersion {
+				ec.log.Tracef("skipping ExternalWorkload resync update event")
+				return
+			}
+
+			services, err := ec.servicesToUpdate(old, updated)
+			if err != nil {
+				ec.log.Errorf("failed to get list of services to update: %v", err)
+				return
+			}
+
+			for _, svc := range services {
+				ec.queue.Add(svc)
+			}
+		},
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	// Store configuration for leader elector client. The leader elector will
+	// accept three callbacks. When a lease is claimed, the elector will mark
+	// the manager as a 'leader'. When a lease is released, the elector will set
+	// the isLeader value back to false.
+	ec.lec = leaderelection.LeaderElectionConfig{
+		// When runtime context is cancelled, lock will be released. Implies any
+		// code guarded by the lease _must_ finish before cancelling.
+		ReleaseOnCancel: true,
+		Lock: &resourcelock.LeaseLock{
+			LeaseMeta: metav1.ObjectMeta{
+				Name:      leaseName,
+				Namespace: controllerNs,
+			},
+			Client: k8sAPI.Client.CoordinationV1(),
+			LockConfig: resourcelock.ResourceLockConfig{
+				Identity: hostname,
+			},
+		},
+		LeaseDuration: leaseDuration,
+		RenewDeadline: leaseRenewDeadline,
+		RetryPeriod:   leaseRetryPeriod,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(context.Context) {
+				ec.Lock()
+				defer ec.Unlock()
+				ec.isLeader.Store(true)
+			},
+			OnStoppedLeading: func() {
+				ec.Lock()
+				defer ec.Unlock()
+				ec.isLeader.Store(false)
+				ec.log.Infof("%s released lease", hostname)
+			},
+			OnNewLeader: func(identity string) {
+				if identity == hostname {
+					ec.log.Infof("%s acquired lease", hostname)
+				}
+			},
+		},
+	}
+
+	return ec, nil
+}
+
+// Start will run the endpoint manager's processing loop and leader elector.
+//
+// The function will spawn three background tasks; one to run the leader elector
+// client, one that will process updates applied by the informer
+// callbacks and one to handle shutdown signals and propagate them to all
+// components.
+//
+// Warning: Do not call Start() more than once
+func (ec *EndpointsController) Start() {
+	// Create a parent context that will be used by leader elector to gracefully
+	// shutdown.
+	//
+	// When cancelled (either through cancel function or by having its done
+	// channel closed), the leader elector will release the lease and stop its
+	// execution.
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		for {
+			// Block until a lease is acquired or a lease has been released
+			leaderelection.RunOrDie(ctx, ec.lec)
+			// If the context has been cancelled, exit the function, otherwise
+			// continue spinning.
+			select {
+			case <-ctx.Done():
+				ec.log.Trace("leader election client received shutdown signal")
+				return
+			default:
+			}
+		}
+	}()
+
+	// When a shutdown signal is received over the manager's stop channel, it is
+	// propagated to the elector through the context object and to the queue
+	// through its dedicated `Shutdown()` function.
+	go func() {
+		// Block until a shutdown signal arrives
+		<-ec.stop
+		// Do not drain the queue since we may not hold the lease.
+		ec.queue.ShutDown()
+		// Propagate shutdown to elector
+		cancel()
+		ec.log.Infof("received shutdown signal")
+	}()
+
+	// Start a background task to process updates.
+	go ec.processQueue()
+}
+
+// processQueue spins and pops elements off the queue. When the queue has
+// received a shutdown signal it exists.
+//
+// The queue uses locking internally so this function is thread safe and can
+// have many workers call it in parallel; workers will not process the same item
+// at the same time.
+func (ec *EndpointsController) processQueue() {
+	for {
+		item, quit := ec.queue.Get()
+		if quit {
+			ec.log.Trace("queue received shutdown signal")
+			return
+		}
+
+		key := item.(string)
+		err := ec.processUpdate(key)
+		ec.handleError(err, key)
+
+		// Tell the queue that we are done processing this key. This will
+		// unblock the key for other workers to process if executing in
+		// parallel, or if it needs to be re-queued because another update has
+		// been received.
+		ec.queue.Done(key)
+	}
+}
+
+// processUpdate will run a reconciliation function for a single Service object
+// that needs to have its EndpointSlice objects reconciled.
+//
+// TODO (matei): remove lint during impl of processUpdate. CI complains error is
+// always nil
+//
+//nolint:all
+func (ec *EndpointsController) processUpdate(update string) error {
+	// TODO (matei): reconciliation logic
+	ec.log.Infof("received %s", update)
+	return nil
+}
+
+// handleError will look at the result of the queue update processing step and
+// decide whether an update should be re-tried or marked as done.
+//
+// The queue operates with an error budget. When exceeded, the item is evicted
+// from the queue (and its retry history wiped). Otherwise, the item is enqueued
+// according to the queue's rate limiting algorithm.
+func (ec *EndpointsController) handleError(err error, key string) {
+	if err == nil {
+		// Wipe out rate limiting history for key when processing was successful.
+		// Next time this key is used, it will get its own fresh rate limiter
+		// error budget
+		ec.queue.Forget(key)
+		return
+	}
+
+	if ec.queue.NumRequeues(key) < maxRetryBudget {
+		ec.queue.AddRateLimited(key)
+		return
+	}
+
+	ec.queue.Forget(key)
+	ec.log.Errorf("dropped Service %s out of update queue: %v", key, err)
+}
+
+// === Callbacks ===
+
+// When a service update has been received (regardless of the event type, i.e.
+// can be Added, Modified, Deleted) send it to the endpoint controller for
+// processing.
+func (ec *EndpointsController) updateService(obj interface{}) {
+	// Use client-go's generic key function to make a key of the format
+	// <namespace>/<name>
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		ec.log.Infof("failed to get key for object %+v: %v", obj, err)
+	}
+
+	namespace, _, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		ec.log.Infof("failed to get namespace from key %s: %v", key, err)
+	}
+
+	// Skip processing 'core' services
+	if namespace == "kube-system" {
+		return
+	}
+
+	ec.queue.Add(key)
+}
+
+func isReady(ew *ewv1alpha1.ExternalWorkload) bool {
+	if len(ew.Status.Conditions) == 0 {
+		return false
+	}
+
+	// Loop through the conditions and look at each condition in turn starting
+	// from the top.
+	for i := range ew.Status.Conditions {
+		cond := ew.Status.Conditions[i]
+		// Stop once we find a 'Ready' condition. We expect a resource to only
+		// have one 'Ready' type condition.
+		if cond.Type == ewv1alpha1.WorkloadReady && cond.Status == ewv1alpha1.ConditionTrue {
+			return true
+		}
+	}
+
+	return false
+}
+
+// === Util ===
+
+// Check whether two label sets are matching
+func labelsChanged(old, updated *ewv1alpha1.ExternalWorkload) bool {
+	if len(old.Labels) != len(updated.Labels) {
+		return true
+	}
+
+	for upK, upV := range updated.Labels {
+		oldV, ok := old.Labels[upK]
+		if !ok || oldV != upV {
+			return true
+		}
+	}
+
+	return false
+}
+
+// specChanged will check whether two workload resource specs have changed
+//
+// Note: we are interested in changes to the ports, ips and readiness fields
+// since these are going to influence a change in a service's underlying
+// endpoint slice
+func specChanged(old, updated *ewv1alpha1.ExternalWorkload) bool {
+	if isReady(old) != isReady(updated) {
+		return true
+	}
+
+	if len(old.Spec.Ports) != len(updated.Spec.Ports) ||
+		len(old.Spec.WorkloadIPs) != len(updated.Spec.WorkloadIPs) {
+		return true
+	}
+
+	// Determine if the ports have changed between workload resources
+	portSet := make(map[int32]ewv1alpha1.PortSpec)
+	for _, ps := range updated.Spec.Ports {
+		portSet[ps.Port] = ps
+	}
+
+	for _, oldPs := range old.Spec.Ports {
+		// If the port number is present in the new workload but not the old
+		// one, then we have a diff and we return early
+		newPs, ok := portSet[oldPs.Port]
+		if !ok {
+			return true
+		}
+
+		// If the port is present in both workloads, we check to see if any of
+		// the port spec's values have changed, e.g. name or protocol
+		if newPs.Name != oldPs.Name || newPs.Protocol != oldPs.Protocol {
+			return true
+		}
+	}
+
+	// Determine if the ips have changed between workload resources. If an IP
+	// is documented for one workload but not the other, then we have a diff.
+	ipSet := make(map[string]struct{})
+	for _, addr := range updated.Spec.WorkloadIPs {
+		ipSet[addr.Ip] = struct{}{}
+	}
+
+	for _, addr := range old.Spec.WorkloadIPs {
+		if _, ok := ipSet[addr.Ip]; !ok {
+			return true
+		}
+	}
+
+	return false
+}
+
+func toSet(s []string) map[string]struct{} {
+	set := map[string]struct{}{}
+	for _, k := range s {
+		set[k] = struct{}{}
+	}
+	return set
+}
+
+// servicesToUpdate will look at an old and an updated external workload
+// resource and determine which services need to be reconciled. The outcome is
+// determined by what has changed in-between resources (selections, spec, or
+// both).
+func (ec *EndpointsController) servicesToUpdate(old, updated *ewv1alpha1.ExternalWorkload) ([]string, error) {
+	labelsChanged := labelsChanged(old, updated)
+	specChanged := specChanged(old, updated)
+	if !labelsChanged && !specChanged {
+		ec.log.Debugf("skipping update; nothing has changed between old rv %s and new rv %s", old.ResourceVersion, updated.ResourceVersion)
+		return nil, nil
+	}
+
+	newSelection, err := ec.getSvcMembership(updated)
+	if err != nil {
+		return nil, err
+
+	}
+
+	oldSelection, err := ec.getSvcMembership(old)
+	if err != nil {
+		return nil, err
+	}
+
+	result := map[string]struct{}{}
+	// Determine the list of services we need to update based on the difference
+	// between our old and updated workload.
+	//
+	// Service selections (i.e. services that select a workload through a label
+	// selector) may reference an old workload, a new workload, or both,
+	// depending on the workload's labels.
+	if labelsChanged && specChanged {
+		// When the selection has changed, and the workload has changed, all
+		// services need to be updated so we consider the union of selections.
+		result = toSet(append(newSelection, oldSelection...))
+	} else if specChanged {
+		// When the workload resource has changed, it is enough to consider
+		// either the oldSelection slice or the newSelection slice, since they
+		// are equal. We have the same set of services to update since no
+		// selection has been changed by the update.
+		return newSelection, nil
+	} else {
+		// When the selection has changed, then we need to consider only
+		// services that reference the old workload's labels, or the new
+		// workload's labels, but not both. Services that select both are
+		// unchanged since the workload has not changed.
+		newSelectionSet := toSet(newSelection)
+		oldSelectionSet := toSet(oldSelection)
+
+		// Determine selections that reference only the old workload resource
+		for _, oldSvc := range oldSelection {
+			if _, ok := newSelectionSet[oldSvc]; !ok {
+				result[oldSvc] = struct{}{}
+			}
+		}
+
+		// Determine selections that reference only the new workload resource
+		for _, newSvc := range newSelection {
+			if _, ok := oldSelectionSet[newSvc]; !ok {
+				result[newSvc] = struct{}{}
+			}
+		}
+	}
+
+	var resultSlice []string
+	for svc := range result {
+		resultSlice = append(resultSlice, svc)
+	}
+
+	return resultSlice, nil
+}
+
+// getSvcMembership accepts a pointer to an external workload resource and
+// returns a set of service keys (<namespace>/<name>). The set includes all
+// services local to the workload's namespace that match the workload.
+func (ec *EndpointsController) getSvcMembership(workload *ewv1alpha1.ExternalWorkload) ([]string, error) {
+	keys := []string{}
+	services, err := ec.k8sAPI.Svc().Lister().Services(workload.Namespace).List(labels.Everything())
+	if err != nil {
+		return keys, err
+	}
+
+	for _, svc := range services {
+		if svc.Spec.Selector == nil {
+			continue
+		}
+
+		// Taken from upstream k8s code, this checks whether a given object has
+		// a deleted state before returning a `namespace/name` key. This is
+		// important since we do not want to consider a service that has been
+		// deleted and is waiting for cache eviction
+		key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(svc)
+		if err != nil {
+			return []string{}, err
+		}
+
+		// Check if service selects our ExternalWorkload.
+		if labels.ValidatedSetSelector(svc.Spec.Selector).Matches(labels.Set(workload.Labels)) {
+			keys = append(keys, key)
+		}
+	}
+
+	return keys, nil
+}
diff --git a/controller/api/destination/external-workload/queue_metrics.go b/controller/api/destination/external-workload/queue_metrics.go
new file mode 100644
index 000000000..b5a1a051b
--- /dev/null
+++ b/controller/api/destination/external-workload/queue_metrics.go
@@ -0,0 +1,116 @@
+package externalworkload
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"k8s.io/client-go/util/workqueue"
+)
+
+// Code is functionally the same as usptream metrics provider. The difference is that
+// we rely on promauto instead of init() method for metrics registering as the logic
+// used to register the metrics in the upstream implementation uses k8s.io/component-base/metrics/legacyregistry
+// The latter does not work with our metrics registry and hence these metrics do not
+// end up exposed via our admin's server /metrics endpoint.
+//
+// https://github.com/kubernetes/component-base/blob/68f947b04ec3a353e63bbef2c1f935bb9ce0061d/metrics/prometheus/workqueue/metrics.go
+
+const (
+	WorkQueueSubsystem         = "workqueue"
+	DepthKey                   = "depth"
+	AddsKey                    = "adds_total"
+	QueueLatencyKey            = "queue_duration_seconds"
+	WorkDurationKey            = "work_duration_seconds"
+	UnfinishedWorkKey          = "unfinished_work_seconds"
+	LongestRunningProcessorKey = "longest_running_processor_seconds"
+	RetriesKey                 = "retries_total"
+)
+
+type queueMetricsProvider struct {
+	depth                   *prometheus.GaugeVec
+	adds                    *prometheus.CounterVec
+	latency                 *prometheus.HistogramVec
+	workDuration            *prometheus.HistogramVec
+	unfinished              *prometheus.GaugeVec
+	longestRunningProcessor *prometheus.GaugeVec
+	retries                 *prometheus.CounterVec
+}
+
+func newWorkQueueMetricsProvider() workqueue.MetricsProvider {
+	return &queueMetricsProvider{
+		depth: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Subsystem: WorkQueueSubsystem,
+			Name:      DepthKey,
+			Help:      "Current depth of workqueue",
+		}, []string{"name"}),
+
+		adds: promauto.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: WorkQueueSubsystem,
+			Name:      AddsKey,
+			Help:      "Total number of adds handled by workqueue",
+		}, []string{"name"}),
+
+		latency: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Subsystem: WorkQueueSubsystem,
+			Name:      QueueLatencyKey,
+			Help:      "How long in seconds an item stays in workqueue before being requested.",
+			Buckets:   prometheus.ExponentialBuckets(10e-9, 10, 10),
+		}, []string{"name"}),
+
+		workDuration: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Subsystem: WorkQueueSubsystem,
+			Name:      WorkDurationKey,
+			Help:      "How long in seconds processing an item from workqueue takes.",
+			Buckets:   prometheus.ExponentialBuckets(10e-9, 10, 10),
+		}, []string{"name"}),
+
+		unfinished: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Subsystem: WorkQueueSubsystem,
+			Name:      UnfinishedWorkKey,
+			Help: "How many seconds of work has done that " +
+				"is in progress and hasn't been observed by work_duration. Large " +
+				"values indicate stuck threads. One can deduce the number of stuck " +
+				"threads by observing the rate at which this increases.",
+		}, []string{"name"}),
+
+		longestRunningProcessor: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Subsystem: WorkQueueSubsystem,
+			Name:      LongestRunningProcessorKey,
+			Help: "How many seconds has the longest running " +
+				"processor for workqueue been running.",
+		}, []string{"name"}),
+
+		retries: promauto.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: WorkQueueSubsystem,
+			Name:      RetriesKey,
+			Help:      "Total number of retries handled by workqueue",
+		}, []string{"name"}),
+	}
+}
+
+func (p queueMetricsProvider) NewDepthMetric(name string) workqueue.GaugeMetric {
+	return p.depth.WithLabelValues(name)
+}
+
+func (p queueMetricsProvider) NewAddsMetric(name string) workqueue.CounterMetric {
+	return p.adds.WithLabelValues(name)
+}
+
+func (p queueMetricsProvider) NewLatencyMetric(name string) workqueue.HistogramMetric {
+	return p.latency.WithLabelValues(name)
+}
+
+func (p queueMetricsProvider) NewWorkDurationMetric(name string) workqueue.HistogramMetric {
+	return p.workDuration.WithLabelValues(name)
+}
+
+func (p queueMetricsProvider) NewUnfinishedWorkSecondsMetric(name string) workqueue.SettableGaugeMetric {
+	return p.unfinished.WithLabelValues(name)
+}
+
+func (p queueMetricsProvider) NewLongestRunningProcessorSecondsMetric(name string) workqueue.SettableGaugeMetric {
+	return p.longestRunningProcessor.WithLabelValues(name)
+}
+
+func (p queueMetricsProvider) NewRetriesMetric(name string) workqueue.CounterMetric {
+	return p.retries.WithLabelValues(name)
+}
diff --git a/controller/api/destination/watcher/endpoints_watcher.go b/controller/api/destination/watcher/endpoints_watcher.go
index 6fefdccc9..e127d59a5 100644
--- a/controller/api/destination/watcher/endpoints_watcher.go
+++ b/controller/api/destination/watcher/endpoints_watcher.go
@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
 	"github.com/linkerd/linkerd2/controller/gen/apis/server/v1beta2"
 	"github.com/linkerd/linkerd2/controller/k8s"
 	consts "github.com/linkerd/linkerd2/pkg/k8s"
@@ -36,6 +37,7 @@ const (
 )
 
 const endpointTargetRefPod = "Pod"
+const endpointTargetRefExternalWorkload = "ExternalWorkload"
 
 type (
 	// Address represents an individual port on a specific endpoint.
@@ -47,6 +49,7 @@ type (
 		IP                string
 		Port              Port
 		Pod               *corev1.Pod
+		ExternalWorkload  *ewv1alpha1.ExternalWorkload
 		OwnerName         string
 		OwnerKind         string
 		Identity          string
@@ -704,13 +707,8 @@ func (sp *servicePublisher) updateServer(server *v1beta2.Server, isAdd bool) {
 	sp.Lock()
 	defer sp.Unlock()
 
-	selector, err := metav1.LabelSelectorAsSelector(server.Spec.PodSelector)
-	if err != nil {
-		sp.log.Errorf("failed to create Selector: %s", err)
-		return
-	}
 	for _, pp := range sp.ports {
-		pp.updateServer(server, selector, isAdd)
+		pp.updateServer(server, isAdd)
 	}
 }
 
@@ -899,6 +897,7 @@ func (pp *portPublisher) endpointSliceToAddresses(es *discovery.EndpointSlice) A
 					pp.log.Errorf("failed to set address OpaqueProtocol: %s", err)
 					continue
 				}
+
 				address.Zone = endpoint.Zone
 				if endpoint.Hints != nil {
 					zones := make([]discovery.ForZone, len(endpoint.Hints.ForZones))
@@ -909,6 +908,32 @@ func (pp *portPublisher) endpointSliceToAddresses(es *discovery.EndpointSlice) A
 			}
 		}
 
+		if endpoint.TargetRef.Kind == endpointTargetRefExternalWorkload {
+			for _, IPAddr := range endpoint.Addresses {
+				address, id, err := pp.newExtRefAddress(resolvedPort, IPAddr, endpoint.TargetRef.Name, es.Namespace)
+				if err != nil {
+					pp.log.Errorf("Unable to create new address: %v", err)
+					continue
+				}
+
+				err = setToServerProtocolExternalWorkload(pp.k8sAPI, &address, resolvedPort)
+				if err != nil {
+					pp.log.Errorf("failed to set address OpaqueProtocol: %s", err)
+					continue
+				}
+
+				address.Zone = endpoint.Zone
+				if endpoint.Hints != nil {
+					zones := make([]discovery.ForZone, len(endpoint.Hints.ForZones))
+					copy(zones, endpoint.Hints.ForZones)
+					address.ForZones = zones
+				}
+
+				addresses[id] = address
+			}
+
+		}
+
 	}
 	return AddressSet{
 		Addresses:          addresses,
@@ -1050,6 +1075,33 @@ func (pp *portPublisher) newPodRefAddress(endpointPort Port, endpointIP, podName
 	return addr, id, nil
 }
 
+func (pp *portPublisher) newExtRefAddress(endpointPort Port, endpointIP, externalWorkloadName, externalWorkloadNamespace string) (Address, ExternalWorkloadID, error) {
+	id := ExternalWorkloadID{
+		Name:      externalWorkloadName,
+		Namespace: externalWorkloadNamespace,
+	}
+
+	ew, err := pp.k8sAPI.ExtWorkload().Lister().ExternalWorkloads(id.Namespace).Get(id.Name)
+	if err != nil {
+		return Address{}, ExternalWorkloadID{}, fmt.Errorf("unable to fetch ExternalWorkload %v: %w", id, err)
+	}
+
+	addr := Address{
+		IP:               endpointIP,
+		Port:             endpointPort,
+		ExternalWorkload: ew,
+	}
+
+	ownerRefs := ew.GetOwnerReferences()
+	if len(ownerRefs) == 1 {
+		parent := ownerRefs[0]
+		addr.OwnerName = parent.Name
+		addr.OwnerName = strings.ToLower(parent.Kind)
+	}
+
+	return addr, id, nil
+}
+
 func (pp *portPublisher) resolveESTargetPort(slicePorts []discovery.EndpointPort) Port {
 	if slicePorts == nil {
 		return undefinedEndpointPort
@@ -1180,12 +1232,21 @@ func (pp *portPublisher) unsubscribe(listener EndpointUpdateListener) {
 
 	pp.metrics.setSubscribers(len(pp.listeners))
 }
-
-func (pp *portPublisher) updateServer(server *v1beta2.Server, selector labels.Selector, isAdd bool) {
+func (pp *portPublisher) updateServer(server *v1beta2.Server, isAdd bool) {
 	updated := false
 	for id, address := range pp.addresses.Addresses {
-		if address.Pod != nil && selector.Matches(labels.Set(address.Pod.Labels)) {
-			var portMatch bool
+		portMatch := false
+		if address.Pod != nil {
+			selector, err := metav1.LabelSelectorAsSelector(server.Spec.PodSelector)
+			if err != nil {
+				pp.log.Errorf("failed to create Selector: %s", err)
+				return
+			}
+
+			if !selector.Matches(labels.Set(address.Pod.Labels)) {
+				continue
+			}
+
 			switch server.Spec.Port.Type {
 			case intstr.Int:
 				if server.Spec.Port.IntVal == int32(address.Port) {
@@ -1202,16 +1263,46 @@ func (pp *portPublisher) updateServer(server *v1beta2.Server, selector labels.Se
 			default:
 				continue
 			}
-			if portMatch {
-				if isAdd && server.Spec.ProxyProtocol == opaqueProtocol {
-					address.OpaqueProtocol = true
-				} else {
-					address.OpaqueProtocol = false
+
+		} else if address.ExternalWorkload != nil {
+			selector, err := metav1.LabelSelectorAsSelector(server.Spec.ExternalWorkloadSelector)
+			if err != nil {
+				pp.log.Errorf("failed to create Selector: %s", err)
+				return
+			}
+
+			if !selector.Matches(labels.Set(address.ExternalWorkload.Labels)) {
+				continue
+			}
+
+			switch server.Spec.Port.Type {
+			case intstr.Int:
+				if server.Spec.Port.IntVal == int32(address.Port) {
+					portMatch = true
 				}
-				if pp.addresses.Addresses[id].OpaqueProtocol != address.OpaqueProtocol {
-					pp.addresses.Addresses[id] = address
-					updated = true
+			case intstr.String:
+				for _, p := range address.ExternalWorkload.Spec.Ports {
+					if p.Port == int32(address.Port) && p.Name == server.Spec.Port.StrVal {
+						portMatch = true
+					}
 				}
+			default:
+				continue
+			}
+
+		} else {
+			continue
+		}
+
+		if portMatch {
+			if isAdd && server.Spec.ProxyProtocol == opaqueProtocol {
+				address.OpaqueProtocol = true
+			} else {
+				address.OpaqueProtocol = false
+			}
+			if pp.addresses.Addresses[id].OpaqueProtocol != address.OpaqueProtocol {
+				pp.addresses.Addresses[id] = address
+				updated = true
 			}
 		}
 	}
@@ -1406,6 +1497,47 @@ func SetToServerProtocol(k8sAPI *k8s.API, address *Address, port Port) error {
 	return nil
 }
 
+// setToServerProtocolExternalWorkload sets the address's OpaqueProtocol field based off any
+// Servers that select it and override the expected protocol for ExternalWorkloads.
+func setToServerProtocolExternalWorkload(k8sAPI *k8s.API, address *Address, port Port) error {
+	if address.ExternalWorkload == nil {
+		return fmt.Errorf("endpoint not backed by ExternalWorkload: %s:%d", address.IP, address.Port)
+	}
+	servers, err := k8sAPI.Srv().Lister().Servers("").List(labels.Everything())
+	if err != nil {
+		return fmt.Errorf("failed to list Servers: %w", err)
+	}
+	for _, server := range servers {
+		selector, err := metav1.LabelSelectorAsSelector(server.Spec.ExternalWorkloadSelector)
+		if err != nil {
+			return fmt.Errorf("failed to create Selector: %w", err)
+		}
+		if server.Spec.ProxyProtocol == opaqueProtocol && selector.Matches(labels.Set(address.ExternalWorkload.Labels)) {
+			var portMatch bool
+			switch server.Spec.Port.Type {
+			case intstr.Int:
+				if server.Spec.Port.IntVal == int32(port) {
+					portMatch = true
+				}
+			case intstr.String:
+				for _, p := range address.ExternalWorkload.Spec.Ports {
+					if p.Port == int32(port) && p.Name == server.Spec.Port.StrVal {
+						portMatch = true
+					}
+
+				}
+			default:
+				continue
+			}
+			if portMatch {
+				address.OpaqueProtocol = true
+				return nil
+			}
+		}
+	}
+	return nil
+}
+
 func latestUpdated(managedFields []metav1.ManagedFieldsEntry) time.Time {
 	var latest time.Time
 	for _, field := range managedFields {
diff --git a/controller/api/destination/watcher/k8s.go b/controller/api/destination/watcher/k8s.go
index 529b199d8..ef698921b 100644
--- a/controller/api/destination/watcher/k8s.go
+++ b/controller/api/destination/watcher/k8s.go
@@ -37,6 +37,8 @@ type (
 	PodID = ID
 	// ProfileID is the namespace-qualified name of a service profile.
 	ProfileID = ID
+	// PodID is the namespace-qualified name of an ExternalWorkload.
+	ExternalWorkloadID = ID
 
 	// Port is a numeric port.
 	Port      = uint32
diff --git a/controller/cmd/destination/main.go b/controller/cmd/destination/main.go
index 845c9e03c..4edddbde5 100644
--- a/controller/cmd/destination/main.go
+++ b/controller/cmd/destination/main.go
@@ -107,7 +107,7 @@ func Main(args []string) {
 			*kubeConfigPath,
 			true,
 			"local",
-			k8s.Endpoint, k8s.ES, k8s.Pod, k8s.Svc, k8s.SP, k8s.Job, k8s.Srv,
+			k8s.Endpoint, k8s.ES, k8s.Pod, k8s.Svc, k8s.SP, k8s.Job, k8s.Srv, k8s.ExtWorkload,
 		)
 	} else {
 		k8sAPI, err = k8s.InitializeAPI(
@@ -115,7 +115,7 @@ func Main(args []string) {
 			*kubeConfigPath,
 			true,
 			"local",
-			k8s.Endpoint, k8s.Pod, k8s.Svc, k8s.SP, k8s.Job, k8s.Srv,
+			k8s.Endpoint, k8s.Pod, k8s.Svc, k8s.SP, k8s.Job, k8s.Srv, k8s.ExtWorkload,
 		)
 	}
 	if err != nil {
diff --git a/go.mod b/go.mod
index af027bed1..e56283b2a 100644
--- a/go.mod
+++ b/go.mod
@@ -7,7 +7,7 @@ require (
 	github.com/briandowns/spinner v0.0.0-20190212173954-5cf08d0ac778
 	github.com/clarketm/json v1.15.7
 	github.com/emicklei/proto v1.13.0
-	github.com/evanphx/json-patch v5.7.0+incompatible
+	github.com/evanphx/json-patch v5.8.0+incompatible
 	github.com/fatih/color v1.16.0
 	github.com/fsnotify/fsnotify v1.7.0
 	github.com/go-openapi/spec v0.20.14
@@ -18,7 +18,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/imdario/mergo v0.3.16
 	github.com/julienschmidt/httprouter v1.3.0
-	github.com/linkerd/linkerd2-proxy-api v0.12.0
+	github.com/linkerd/linkerd2-proxy-api v0.13.0
 	github.com/mattn/go-isatty v0.0.20
 	github.com/mattn/go-runewidth v0.0.15
 	github.com/nsf/termbox-go v1.1.1
@@ -34,7 +34,7 @@ require (
 	github.com/spf13/pflag v1.0.5
 	go.opencensus.io v0.24.0
 	golang.org/x/net v0.20.0
-	golang.org/x/tools v0.16.1
+	golang.org/x/tools v0.17.0
 	google.golang.org/grpc v1.60.1
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.3.0
 	google.golang.org/protobuf v1.32.0
@@ -142,7 +142,7 @@ require (
 	golang.org/x/crypto v0.18.0 // indirect
 	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/oauth2 v0.16.0 // indirect
-	golang.org/x/sync v0.5.0 // indirect
+	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect
 	golang.org/x/term v0.16.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
diff --git a/go.sum b/go.sum
index 7a115c486..470140acb 100644
--- a/go.sum
+++ b/go.sum
@@ -123,8 +123,8 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
-github.com/evanphx/json-patch v5.7.0+incompatible h1:vgGkfT/9f8zE6tvSCe74nfpAVDQ2tG6yudJd8LBksgI=
-github.com/evanphx/json-patch v5.7.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
+github.com/evanphx/json-patch v5.8.0+incompatible h1:1Av9pn2FyxPdvrWNQszj1g6D6YthSmvCfcN6SYclTJg=
+github.com/evanphx/json-patch v5.8.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM=
 github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE=
 github.com/felixge/httpsnoop v1.0.3 h1:s/nj+GCswXYzN5v2DpNMuMQYe+0DDwt5WVCU6CWBdXk=
@@ -296,8 +296,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de h1:9TO3cAIGXtEhnIaL+V+BEER86oLrvS+kWobKpbJuye0=
 github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de/go.mod h1:zAbeS9B/r2mtpb6U+EI2rYA5OAXxsYw6wTamcNW+zcE=
-github.com/linkerd/linkerd2-proxy-api v0.12.0 h1:ZB9a2OF7JAVk3tjtoukPz7zfBRZ3U+jhkdFCpBzJAR4=
-github.com/linkerd/linkerd2-proxy-api v0.12.0/go.mod h1:n9g59OaIKK7y0VF1lIwRkIhuvFtE1KHRtA8iz6Wenh8=
+github.com/linkerd/linkerd2-proxy-api v0.13.0 h1:CYHnqeH7LZ8rB7kKSjK7KwHy6mSPG2LS2Ctn4ihFSsc=
+github.com/linkerd/linkerd2-proxy-api v0.13.0/go.mod h1:CVUFCu0OXK11ecBtCU+FiGqUX0KyOR3zvxPF0vwSbPs=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
@@ -553,8 +553,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
-golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
+golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -645,8 +645,8 @@ golang.org/x/tools v0.0.0-20200505023115-26f46d2f7ef8/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
-golang.org/x/tools v0.16.1 h1:TLyB3WofjdOEepBHAU20JdNC1Zbg87elYofWYAY5oZA=
-golang.org/x/tools v0.16.1/go.mod h1:kYVVN6I1mBNoB1OX+noeBjbRk4IUEPa7JJ+TJMEooJ0=
+golang.org/x/tools v0.17.0 h1:FvmRgNOcs3kOa+T20R1uhfP9F6HgG2mfxDv1vrx1Htc=
+golang.org/x/tools v0.17.0/go.mod h1:xsh6VxdV005rRVaS6SSAf9oiAqljS7UZUacMZ8Bnsps=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/jaeger/charts/linkerd-jaeger/Chart.yaml b/jaeger/charts/linkerd-jaeger/Chart.yaml
index f72b117b2..80f5a0439 100644
--- a/jaeger/charts/linkerd-jaeger/Chart.yaml
+++ b/jaeger/charts/linkerd-jaeger/Chart.yaml
@@ -11,7 +11,7 @@ kubeVersion: ">=1.22.0-0"
 name: linkerd-jaeger
 sources:
   - https://github.com/linkerd/linkerd2/
-version: 30.14.4-edge
+version: 30.14.5-edge
 icon: https://linkerd.io/images/logo-only-200h.png
 maintainers:
   - name: Linkerd authors
diff --git a/jaeger/charts/linkerd-jaeger/README.md b/jaeger/charts/linkerd-jaeger/README.md
index a23fedf1a..ef2e0c418 100644
--- a/jaeger/charts/linkerd-jaeger/README.md
+++ b/jaeger/charts/linkerd-jaeger/README.md
@@ -3,7 +3,7 @@
 The Linkerd-Jaeger extension adds distributed tracing to Linkerd using
 OpenCensus and Jaeger.
 
-![Version: 30.14.4-edge](https://img.shields.io/badge/Version-30.14.4--edge-informational?style=flat-square)
+![Version: 30.14.5-edge](https://img.shields.io/badge/Version-30.14.5--edge-informational?style=flat-square)
 
 ![AppVersion: edge-XX.X.X](https://img.shields.io/badge/AppVersion-edge--XX.X.X-informational?style=flat-square)
 
diff --git a/multicluster/charts/linkerd-multicluster/Chart.yaml b/multicluster/charts/linkerd-multicluster/Chart.yaml
index b00d5c2cb..0dab84db2 100644
--- a/multicluster/charts/linkerd-multicluster/Chart.yaml
+++ b/multicluster/charts/linkerd-multicluster/Chart.yaml
@@ -11,7 +11,7 @@ kubeVersion: ">=1.22.0-0"
 name: "linkerd-multicluster"
 sources:
   - https://github.com/linkerd/linkerd2/
-version: 30.12.13-edge
+version: 30.12.14-edge
 icon: https://linkerd.io/images/logo-only-200h.png
 maintainers:
   - name: Linkerd authors
diff --git a/multicluster/charts/linkerd-multicluster/README.md b/multicluster/charts/linkerd-multicluster/README.md
index 2fad13df6..a8a4b78da 100644
--- a/multicluster/charts/linkerd-multicluster/README.md
+++ b/multicluster/charts/linkerd-multicluster/README.md
@@ -3,7 +3,7 @@
 The Linkerd-Multicluster extension contains resources to support multicluster
 linking to remote clusters
 
-![Version: 30.12.13-edge](https://img.shields.io/badge/Version-30.12.13--edge-informational?style=flat-square)
+![Version: 30.12.14-edge](https://img.shields.io/badge/Version-30.12.14--edge-informational?style=flat-square)
 
 ![AppVersion: edge-XX.X.X](https://img.shields.io/badge/AppVersion-edge--XX.X.X-informational?style=flat-square)
 
diff --git a/pkg/k8s/fake.go b/pkg/k8s/fake.go
index 6ca91fcec..b05040849 100644
--- a/pkg/k8s/fake.go
+++ b/pkg/k8s/fake.go
@@ -99,6 +99,8 @@ func NewFakeClientSets(configs ...string) (
 			spObjs = append(spObjs, obj)
 		case Server:
 			spObjs = append(spObjs, obj)
+		case ExtWorkload:
+			spObjs = append(spObjs, obj)
 		default:
 			objs = append(objs, obj)
 		}
diff --git a/pkg/k8s/k8s.go b/pkg/k8s/k8s.go
index 0d94a1bcc..bbe482986 100644
--- a/pkg/k8s/k8s.go
+++ b/pkg/k8s/k8s.go
@@ -55,6 +55,7 @@ const (
 	ServerKind      = "Server"
 	HTTPRouteKind   = "HTTPRoute"
 	ExtWorkloadKind = "ExternalWorkload"
+	PodKind         = "Pod"
 
 	WorkloadAPIGroup   = "workload.linkerd.io"
 	WorkloadAPIVersion = "v1alpha1"
diff --git a/pkg/k8s/labels.go b/pkg/k8s/labels.go
index d255ef058..861b727f3 100644
--- a/pkg/k8s/labels.go
+++ b/pkg/k8s/labels.go
@@ -8,6 +8,7 @@ package k8s
 import (
 	"fmt"
 
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
 	"github.com/linkerd/linkerd2/pkg/version"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -513,6 +514,16 @@ func GetPodLabels(ownerKind, ownerName string, pod *corev1.Pod) map[string]strin
 	return labels
 }
 
+// GetExternalWorkloadLabels returns the set of prometheus owner labels for a given ExternalWorkload
+func GetExternalWorkloadLabels(ownerKind, ownerName string, ew *ewv1alpha1.ExternalWorkload) map[string]string {
+	labels := map[string]string{"external_workload": ew.Name}
+
+	if ownerKind != "" && ownerName != "" {
+		labels[ownerKind] = ownerName
+	}
+	return labels
+}
+
 // IsMeshed returns whether a given Pod is in a given controller's service mesh.
 func IsMeshed(pod *corev1.Pod, controllerNS string) bool {
 	return pod.Labels[ControllerNSLabel] == controllerNS
diff --git a/policy-controller/grpc/src/inbound.rs b/policy-controller/grpc/src/inbound.rs
index 2a7bcd7e8..966237ab0 100644
--- a/policy-controller/grpc/src/inbound.rs
+++ b/policy-controller/grpc/src/inbound.rs
@@ -1,4 +1,4 @@
-use crate::{http_route, workload::Kind, workload::Workload};
+use crate::{http_route, workload::Workload};
 use futures::prelude::*;
 use linkerd2_proxy_api::{
     self as api,
@@ -31,7 +31,7 @@ pub struct InboundPolicyServer<T> {
 
 impl<T> InboundPolicyServer<T>
 where
-    T: DiscoverInboundServer<(String, String, NonZeroU16)> + Send + Sync + 'static,
+    T: DiscoverInboundServer<(Workload, NonZeroU16)> + Send + Sync + 'static,
 {
     pub fn new(discover: T, cluster_networks: Vec<IpNet>, drain: drain::Watch) -> Self {
         Self {
@@ -48,33 +48,21 @@ where
     fn check_target(
         &self,
         proto::PortSpec { workload, port }: proto::PortSpec,
-    ) -> Result<(String, String, NonZeroU16), tonic::Status> {
-        let (ns, name) = match Workload::from_str(&workload)? {
-            Workload {
-                namespace,
-                kind: Kind::Pod(pod),
-            } => (namespace, pod),
-            _ => {
-                // TODO: handle external workloads
-                return Err(tonic::Status::invalid_argument(
-                    "only pod workload supported at the moment",
-                ));
-            }
-        };
-
+    ) -> Result<(Workload, NonZeroU16), tonic::Status> {
+        let workload = Workload::from_str(&workload)?;
         // Ensure that the port is in the valid range.
         let port = u16::try_from(port)
             .and_then(NonZeroU16::try_from)
             .map_err(|_| tonic::Status::invalid_argument(format!("Invalid port: {port}")))?;
 
-        Ok((ns.to_string(), name.to_string(), port))
+        Ok((workload, port))
     }
 }
 
 #[async_trait::async_trait]
 impl<T> InboundServerPolicies for InboundPolicyServer<T>
 where
-    T: DiscoverInboundServer<(String, String, NonZeroU16)> + Send + Sync + 'static,
+    T: DiscoverInboundServer<(Workload, NonZeroU16)> + Send + Sync + 'static,
 {
     async fn get_port(
         &self,
diff --git a/policy-controller/grpc/src/lib.rs b/policy-controller/grpc/src/lib.rs
index 404d2c9f2..834a81472 100644
--- a/policy-controller/grpc/src/lib.rs
+++ b/policy-controller/grpc/src/lib.rs
@@ -2,7 +2,7 @@
 #![forbid(unsafe_code)]
 
 mod http_route;
-mod workload;
 
 pub mod inbound;
 pub mod outbound;
+pub mod workload;
diff --git a/policy-controller/k8s/index/src/inbound.rs b/policy-controller/k8s/index/src/inbound.rs
index c4d5901f4..d2e2c33f6 100644
--- a/policy-controller/k8s/index/src/inbound.rs
+++ b/policy-controller/k8s/index/src/inbound.rs
@@ -3,9 +3,9 @@ mod http_route;
 pub mod index;
 mod meshtls_authentication;
 mod network_authentication;
-mod pod;
 mod server;
 pub mod server_authorization;
+mod workload;
 
 pub use index::{Index, SharedIndex};
 
diff --git a/policy-controller/k8s/index/src/inbound/index.rs b/policy-controller/k8s/index/src/inbound/index.rs
index 8b94c1c49..d2861d0d2 100644
--- a/policy-controller/k8s/index/src/inbound/index.rs
+++ b/policy-controller/k8s/index/src/inbound/index.rs
@@ -8,7 +8,7 @@
 
 use super::{
     authorization_policy, http_route::RouteBinding, meshtls_authentication, network_authentication,
-    pod, server, server_authorization,
+    server, server_authorization, workload,
 };
 use crate::{
     http_route::{gkn_for_gateway_http_route, gkn_for_linkerd_http_route, gkn_for_resource},
@@ -65,10 +65,11 @@ struct AuthenticationNsIndex {
     by_ns: HashMap<String, AuthenticationIndex>,
 }
 
-/// Holds `Pod`, `Server`, and `ServerAuthorization` indices for a single namespace.
+/// Holds `Pod`, `ExternalWorkload`, `Server`, and `ServerAuthorization` indices for a single namespace.
 #[derive(Debug)]
 struct Namespace {
     pods: PodIndex,
+    external_workloads: ExternalWorkloadIndex,
     policy: PolicyIndex,
 }
 
@@ -85,7 +86,7 @@ struct PodIndex {
 /// or as `Server` resources select a port.
 #[derive(Debug)]
 struct Pod {
-    meta: pod::Meta,
+    meta: workload::Meta,
 
     /// The pod's named container ports. Used by `Server` port selectors.
     ///
@@ -97,7 +98,7 @@ struct Pod {
     /// `Namespace::reindex`--when a port is selected by a `Server`--or by
     /// `Namespace::get_pod_server` when a client discovers a port that has no
     /// configured server (and i.e. uses the default policy).
-    port_servers: PortMap<PodPortServer>,
+    port_servers: PortMap<WorkloadPortServer>,
 
     /// The pod's probe ports and their respective paths.
     ///
@@ -107,18 +108,45 @@ struct Pod {
     probes: PortMap<BTreeSet<String>>,
 }
 
-/// Holds the state of a single port on a pod.
+/// Holds the state of a single port on a workload (e.g. a pod or an external
+/// workload).
 #[derive(Debug)]
-struct PodPortServer {
+struct WorkloadPortServer {
     /// The name of the server resource that matches this port. Unset when no
     /// server resources match this pod/port (and, i.e., the default policy is
     /// used).
     name: Option<String>,
 
-    /// A sender used to broadcast pod port server updates.
+    /// A sender used to broadcast workload port server updates.
     watch: watch::Sender<InboundServer>,
 }
 
+/// Holds all external workload data for a single namespace
+#[derive(Debug)]
+struct ExternalWorkloadIndex {
+    namespace: String,
+    by_name: HashMap<String, ExternalWorkload>,
+}
+
+/// Holds data for a single external workload, with server watches for all known
+/// ports.
+///
+/// The set of ports / servers is updated as clients discover server
+/// configuration or as `Server` resources select a port.
+#[derive(Debug)]
+struct ExternalWorkload {
+    meta: workload::Meta,
+
+    // The workload's named container ports. Used by `Server` port selectors.
+    //
+    // A workload will not have multiple ports with the same name, e.g. two
+    // `admin-http` ports pointing to different numerical values.
+    port_names: HashMap<String, NonZeroU16>,
+
+    /// All known TCP server ports.
+    port_servers: PortMap<WorkloadPortServer>,
+}
+
 /// Holds the state of policy resources for a single namespace.
 #[derive(Debug)]
 struct PolicyIndex {
@@ -184,6 +212,34 @@ impl Index {
             .subscribe())
     }
 
+    /// Obtains an external_workload:port's server receiver.
+    ///
+    /// An error is returned if the external workload is not found. If the port
+    /// is not found, a default server is created.
+    pub fn external_workload_server_rx(
+        &mut self,
+        namespace: &str,
+        workload: &str,
+        port: NonZeroU16,
+    ) -> Result<watch::Receiver<InboundServer>> {
+        let ns = self
+            .namespaces
+            .by_ns
+            .get_mut(namespace)
+            .ok_or_else(|| anyhow::anyhow!("namespace not found: {}", namespace))?;
+        let external_workload =
+            ns.external_workloads
+                .by_name
+                .get_mut(workload)
+                .ok_or_else(|| {
+                    anyhow::anyhow!("external workload {}.{} not found", workload, namespace)
+                })?;
+        Ok(external_workload
+            .port_server_or_default(port, &self.cluster_info)
+            .watch
+            .subscribe())
+    }
+
     fn ns_with_reindex(&mut self, namespace: String, f: impl FnOnce(&mut Namespace) -> bool) {
         self.namespaces
             .get_with_reindex(namespace, &self.authentications, f)
@@ -311,15 +367,15 @@ impl kubert::index::IndexNamespacedResource<k8s::Pod> for Index {
         let port_names = pod
             .spec
             .as_ref()
-            .map(pod::tcp_ports_by_name)
+            .map(workload::pod_tcp_ports_by_name)
             .unwrap_or_default();
         let probes = pod
             .spec
             .as_ref()
-            .map(pod::pod_http_probes)
+            .map(workload::pod_http_probes)
             .unwrap_or_default();
 
-        let meta = pod::Meta::from_metadata(pod.metadata);
+        let meta = workload::Meta::from_metadata(pod.metadata);
 
         // Add or update the pod. If the pod was not already present in the
         // index with the same metadata, index it against the policy resources,
@@ -350,6 +406,57 @@ impl kubert::index::IndexNamespacedResource<k8s::Pod> for Index {
     // handle resets specially.
 }
 
+impl kubert::index::IndexNamespacedResource<k8s::external_workload::ExternalWorkload> for Index {
+    fn apply(&mut self, ext_workload: k8s::external_workload::ExternalWorkload) {
+        let ns = ext_workload.namespace().unwrap();
+        let name = ext_workload.name_unchecked();
+        let _span = info_span!("apply", %ns, %name).entered();
+
+        // Extract ports and settings.
+        // Note: external workloads do not have any probe paths to synthesise
+        // default policies for.
+        let port_names = workload::external_tcp_ports_by_name(&ext_workload.spec);
+        let meta = workload::Meta::from_metadata(ext_workload.metadata);
+
+        // Add or update the workload.
+        //
+        // If the resource is present in the index, but its metadata has
+        // changed, then it means the watches need to get an update.
+        let ns = self.namespaces.get_or_default(ns);
+        match ns.external_workloads.update(name, meta, port_names) {
+            // No update
+            Ok(None) => {}
+            // Update, so re-index
+            Ok(Some(workload)) => workload.reindex_servers(&ns.policy, &self.authentications),
+            Err(error) => {
+                tracing::error!(%error, "Illegal external workload update");
+            }
+        }
+    }
+
+    fn delete(&mut self, ns: String, name: String) {
+        tracing::debug!(%ns, %name, "delete");
+        if let Entry::Occupied(mut ns) = self.namespaces.by_ns.entry(ns) {
+            // Once the external workload is removed, there's nothing else to
+            // update. Any open watches will complete. No other parts of the
+            // index need to be updated.
+            if ns
+                .get_mut()
+                .external_workloads
+                .by_name
+                .remove(&name)
+                .is_some()
+                && ns.get().is_empty()
+            {
+                ns.remove();
+            }
+        }
+    }
+
+    // Since apply only reindexes a single external workload at a time, there's no need to
+    // handle resets specially.
+}
+
 impl kubert::index::IndexNamespacedResource<k8s::policy::Server> for Index {
     fn apply(&mut self, srv: k8s::policy::Server) {
         let ns = srv.namespace().expect("server must be namespaced");
@@ -828,6 +935,10 @@ impl Namespace {
                 namespace: namespace.clone(),
                 by_name: HashMap::default(),
             },
+            external_workloads: ExternalWorkloadIndex {
+                namespace: namespace.clone(),
+                by_name: HashMap::default(),
+            },
             policy: PolicyIndex {
                 namespace,
                 cluster_info,
@@ -842,12 +953,13 @@ impl Namespace {
     /// Returns true if the index does not include any resources.
     #[inline]
     fn is_empty(&self) -> bool {
-        self.pods.is_empty() && self.policy.is_empty()
+        self.pods.is_empty() && self.policy.is_empty() && self.external_workloads.is_empty()
     }
 
     #[inline]
     fn reindex(&mut self, authns: &AuthenticationNsIndex) {
         self.pods.reindex(&self.policy, authns);
+        self.external_workloads.reindex(&self.policy, authns);
     }
 }
 
@@ -862,7 +974,7 @@ impl PodIndex {
     fn update(
         &mut self,
         name: String,
-        meta: pod::Meta,
+        meta: workload::Meta,
         port_names: HashMap<String, PortSet>,
         probes: PortMap<BTreeSet<String>>,
     ) -> Result<Option<&mut Pod>> {
@@ -975,7 +1087,7 @@ impl Pod {
             Entry::Vacant(entry) => {
                 tracing::trace!(port = %port, server = %name, "Creating server");
                 let (watch, _) = watch::channel(server);
-                entry.insert(PodPortServer {
+                entry.insert(WorkloadPortServer {
                     name: Some(name.to_string()),
                     watch,
                 });
@@ -1013,7 +1125,7 @@ impl Pod {
 
     /// Updates a pod-port to use the given named server.
     fn set_default_server(&mut self, port: NonZeroU16, config: &ClusterInfo) {
-        let server = Self::default_inbound_server(
+        let server = PolicyIndex::default_inbound_server(
             port,
             &self.meta.settings,
             self.probes
@@ -1027,7 +1139,7 @@ impl Pod {
             Entry::Vacant(entry) => {
                 tracing::debug!(%port, server = %config.default_policy, "Creating default server");
                 let (watch, _) = watch::channel(server);
-                entry.insert(PodPortServer { name: None, watch });
+                entry.insert(WorkloadPortServer { name: None, watch });
             }
 
             Entry::Occupied(mut entry) => {
@@ -1068,11 +1180,11 @@ impl Pod {
         &mut self,
         port: NonZeroU16,
         config: &ClusterInfo,
-    ) -> &mut PodPortServer {
+    ) -> &mut WorkloadPortServer {
         match self.port_servers.entry(port) {
             Entry::Occupied(entry) => entry.into_mut(),
             Entry::Vacant(entry) => {
-                let (watch, _) = watch::channel(Self::default_inbound_server(
+                let (watch, _) = watch::channel(PolicyIndex::default_inbound_server(
                     port,
                     &self.meta.settings,
                     self.probes
@@ -1082,45 +1194,257 @@ impl Pod {
                         .map(|p| p.as_str()),
                     config,
                 ));
-                entry.insert(PodPortServer { name: None, watch })
+                entry.insert(WorkloadPortServer { name: None, watch })
             }
         }
     }
+}
 
-    fn default_inbound_server<'p>(
-        port: NonZeroU16,
-        settings: &pod::Settings,
-        probe_paths: impl Iterator<Item = &'p str>,
-        config: &ClusterInfo,
-    ) -> InboundServer {
-        let protocol = if settings.opaque_ports.contains(&port) {
-            ProxyProtocol::Opaque
-        } else {
-            ProxyProtocol::Detect {
-                timeout: config.default_detect_timeout,
+// === impl ExternalWorkloadIndex ===
+
+impl ExternalWorkloadIndex {
+    #[inline]
+    fn is_empty(&self) -> bool {
+        self.by_name.is_empty()
+    }
+
+    /// Indexes an external workload resource and computes any changes in the
+    /// workload's state.
+    ///
+    /// If the workload is indexed for the first time, or some of its settings
+    /// have changed (e.g. metadata) then indexing will trigger a namespace-wide
+    /// server re-index to push new state to the clients.
+    ///
+    /// Otherwise, if nothing has changed, do not trigger re-indexing.
+    fn update(
+        &mut self,
+        name: String,
+        meta: workload::Meta,
+        port_names: HashMap<String, NonZeroU16>,
+    ) -> Result<Option<&mut ExternalWorkload>> {
+        let workload = match self.by_name.entry(name.clone()) {
+            Entry::Vacant(entry) => entry.insert(ExternalWorkload {
+                meta,
+                port_names,
+                port_servers: PortMap::default(),
+            }),
+            Entry::Occupied(entry) => {
+                let workload = entry.into_mut();
+
+                if workload.meta == meta && workload.port_names == port_names {
+                    tracing::debug!(external_workload = %name, "No changes");
+                    return Ok(None);
+                }
+
+                if workload.meta != meta {
+                    tracing::trace!(external_workload = %name, "Updating workload's metadata");
+                    workload.meta = meta;
+                }
+
+                if workload.port_names != port_names {
+                    tracing::trace!(external_workload = %name, "Updating workload's ports");
+                    workload.port_names = port_names;
+                }
+
+                tracing::debug!(external_workload = %name, "Updating");
+                workload
             }
         };
+        Ok(Some(workload))
+    }
 
-        let mut policy = settings.default_policy.unwrap_or(config.default_policy);
-        if settings.require_id_ports.contains(&port) {
-            if let DefaultPolicy::Allow {
-                ref mut authenticated_only,
-                ..
-            } = policy
-            {
-                *authenticated_only = true;
+    /// For each external workload in a namespace, re-compute the server and
+    /// authorization policy states to determine if new changes need to be
+    /// pushed to clients.
+    fn reindex(&mut self, policy: &PolicyIndex, authns: &AuthenticationNsIndex) {
+        let _span = info_span!("reindex", ns = %self.namespace).entered();
+        for (name, ext_workload) in self.by_name.iter_mut() {
+            let _span = info_span!("external_workload", external_workload = %name).entered();
+            ext_workload.reindex_servers(policy, authns);
+        }
+    }
+}
+
+//
+impl ExternalWorkload {
+    /// Determines the policies for ports on this workload
+    fn reindex_servers(&mut self, policy: &PolicyIndex, authentications: &AuthenticationNsIndex) {
+        // Keep track of ports that are already known so that they may receive
+        // default policies if they are still not selected by a server.
+        //
+        // TODO (matei): we could eagerly fill this out if we require users to
+        // always document a workload's ports. i.e. right now a non-named port
+        // will be lazily discovered only when traffic is sent, we could eagerly
+        // set policies for it (but I feel like for now, this should be similar in
+        // behaviour to a pod)
+        //
+        let mut unmatched_ports = self.port_servers.keys().copied().collect::<PortSet>();
+
+        // Keep track of which ports have been matched with servers so that we
+        // can detect when more than one server matches a single port.
+        let mut matched_ports = PortMap::with_capacity_and_hasher(
+            unmatched_ports.len(),
+            std::hash::BuildHasherDefault::<PortHasher>::default(),
+        );
+
+        for (srvname, server) in policy.servers.iter() {
+            if let Selector::ExternalWorkload(selector) = &server.selector {
+                if selector.matches(&self.meta.labels) {
+                    // Each server selects exactly one port on an
+                    // external workload
+                    //
+                    // Note: an external workload has only one set of ports. A
+                    // pod has a union, each container declares its own set.
+                    let port = if let Some(srv_port) = self.selects_port(&server.port_ref) {
+                        srv_port
+                    } else {
+                        // If server references a named port, and our workload
+                        // contains no such port, then skip this server.
+                        continue;
+                    };
+
+                    if let Some(prior) = matched_ports.get(&port) {
+                        // If a different server has already matched this
+                        tracing::warn!(
+                        %port,
+                        server = %prior,
+                        conflict = %srvname,
+                        "Port already matched by another server; skipping"
+                        );
+                        continue;
+                    }
+
+                    let s = policy.inbound_server(
+                        srvname.clone(),
+                        server,
+                        authentications,
+                        Vec::new().into_iter(),
+                    );
+
+                    self.update_server(port, srvname, s);
+                    matched_ports.insert(port, srvname.clone());
+                    unmatched_ports.remove(&port);
+                }
             }
         }
 
-        let authorizations = policy.default_authzs(config);
+        // Reset all other ports that were previously selected to defaults
+        for port in unmatched_ports.into_iter() {
+            self.set_default_server(port, &policy.cluster_info);
+        }
+    }
 
-        let http_routes = config.default_inbound_http_routes(probe_paths);
+    /// Updates an external workload-port to use a given named server.
+    ///
+    /// We use name explicitly (and not derived from the 'server') to ensure we
+    /// are not handling a default server.
+    fn update_server(&mut self, port: NonZeroU16, name: &str, server: InboundServer) {
+        match self.port_servers.entry(port) {
+            Entry::Vacant(entry) => {
+                tracing::trace!(port = %port, server = %name, "Creating server");
+                let (watch, _) = watch::channel(server);
+                entry.insert(WorkloadPortServer {
+                    name: Some(name.to_string()),
+                    watch,
+                });
+            }
 
-        InboundServer {
-            reference: ServerRef::Default(policy.as_str()),
-            protocol,
-            authorizations,
-            http_routes,
+            Entry::Occupied(mut entry) => {
+                let ps = entry.get_mut();
+
+                ps.watch.send_if_modified(|current| {
+                    if ps.name.as_deref() == Some(name) && *current == server {
+                        tracing::trace!(port = %port, server = %name, "Skipped redundant server update");
+                        tracing::trace!(?server);
+                        return false;
+                    }
+
+                    // If the port's server previously matched a different
+                    // server, this can either mean that multiple servers
+                    // currently match the external_workload:port, or that we're
+                    // in the middle of an update. We make the opportunistic
+                    // choice to assume the cluster is configured coherently so
+                    // we take the update. The admission controller should
+                    // prevent conflicts.
+                    tracing::trace!(port = %port, server = %name, "Updating server");
+                    if ps.name.as_deref() != Some(name) {
+                        ps.name = Some(name.to_string());
+                    }
+
+                    *current = server;
+                    true
+                });
+            }
+        }
+
+        tracing::debug!(port = %port, server = %name, "Updated server");
+    }
+
+    /// Updates a workload-port to use a given named server.
+    fn set_default_server(&mut self, port: NonZeroU16, config: &ClusterInfo) {
+        // Create a default server policy, without authorising any probe paths
+        let server = PolicyIndex::default_inbound_server(
+            port,
+            &self.meta.settings,
+            Vec::new().into_iter(),
+            config,
+        );
+        match self.port_servers.entry(port) {
+            Entry::Vacant(entry) => {
+                tracing::debug!(%port, server = %config.default_policy, "Creating default server");
+                let (watch, _) = watch::channel(server);
+                entry.insert(WorkloadPortServer { name: None, watch });
+            }
+
+            Entry::Occupied(mut entry) => {
+                // A server must have selected this before; override with
+                // default policy
+                let ps = entry.get_mut();
+                ps.watch.send_if_modified(|current| {
+                    // Avoid sending redundant updates
+                    if *current == server {
+                        tracing::trace!(%port, server = %config.default_policy, "Default server already set");
+                        return false;
+                    }
+
+                    tracing::debug!(%port, server = %config.default_policy, "Setting default server");
+                    ps.name = None;
+                    *current = server;
+                    true
+                });
+            }
+        }
+    }
+
+    /// Returns an optional port when a server references a known external
+    /// workload port.
+    ///
+    /// Unlike a pod, an external workload has only one set of ports. Names
+    /// within the set are unique, and as a result, only one port will ever
+    /// match a given name.
+    fn selects_port(&mut self, port_ref: &Port) -> Option<NonZeroU16> {
+        match port_ref {
+            Port::Number(p) => Some(*p),
+            Port::Name(name) => self.port_names.get(name).cloned(),
+        }
+    }
+
+    fn port_server_or_default(
+        &mut self,
+        port: NonZeroU16,
+        config: &ClusterInfo,
+    ) -> &mut WorkloadPortServer {
+        match self.port_servers.entry(port) {
+            Entry::Occupied(entry) => entry.into_mut(),
+            Entry::Vacant(entry) => {
+                let (watch, _) = watch::channel(PolicyIndex::default_inbound_server(
+                    port,
+                    &self.meta.settings,
+                    Vec::new().into_iter(),
+                    config,
+                ));
+                entry.insert(WorkloadPortServer { name: None, watch })
+            }
         }
     }
 }
@@ -1187,6 +1511,43 @@ impl PolicyIndex {
         true
     }
 
+    fn default_inbound_server<'p>(
+        port: NonZeroU16,
+        settings: &workload::Settings,
+        probe_paths: impl Iterator<Item = &'p str>,
+        config: &ClusterInfo,
+    ) -> InboundServer {
+        let protocol = if settings.opaque_ports.contains(&port) {
+            ProxyProtocol::Opaque
+        } else {
+            ProxyProtocol::Detect {
+                timeout: config.default_detect_timeout,
+            }
+        };
+
+        let mut policy = settings.default_policy.unwrap_or(config.default_policy);
+        if settings.require_id_ports.contains(&port) {
+            if let DefaultPolicy::Allow {
+                ref mut authenticated_only,
+                ..
+            } = policy
+            {
+                *authenticated_only = true;
+            }
+        }
+
+        let authorizations = policy.default_authzs(config);
+
+        let http_routes = config.default_inbound_http_routes(probe_paths);
+
+        InboundServer {
+            reference: ServerRef::Default(policy.as_str()),
+            protocol,
+            authorizations,
+            http_routes,
+        }
+    }
+
     fn inbound_server<'p>(
         &self,
         name: String,
diff --git a/policy-controller/k8s/index/src/inbound/pod.rs b/policy-controller/k8s/index/src/inbound/workload.rs
similarity index 90%
rename from policy-controller/k8s/index/src/inbound/pod.rs
rename to policy-controller/k8s/index/src/inbound/workload.rs
index 0aa5f37ee..233ec0a3f 100644
--- a/policy-controller/k8s/index/src/inbound/pod.rs
+++ b/policy-controller/k8s/index/src/inbound/workload.rs
@@ -5,17 +5,17 @@ use anyhow::Result;
 use linkerd_policy_controller_k8s_api as k8s;
 use std::{collections::BTreeSet, num::NonZeroU16};
 
-/// Holds pod metadata/config that can change.
+/// Holds workload metadata/config that can change.
 #[derive(Debug, PartialEq)]
 pub(crate) struct Meta {
-    /// The pod's labels. Used by `Server` pod selectors.
+    /// The workload's labels. Used by `Server` selectors.
     pub labels: k8s::Labels,
 
-    // Pod-specific settings (i.e., derived from annotations).
+    // Workload-specific settings (i.e., derived from annotations).
     pub settings: Settings,
 }
 
-/// Per-pod settings, as configured by the pod's annotations.
+/// Per-workload settings, as configured by the workload's annotations.
 #[derive(Debug, Default, PartialEq)]
 pub(crate) struct Settings {
     pub require_id_ports: PortSet,
@@ -24,7 +24,7 @@ pub(crate) struct Settings {
 }
 
 /// Gets the set of named ports with `protocol: TCP` from a pod spec.
-pub(crate) fn tcp_ports_by_name(spec: &k8s::PodSpec) -> HashMap<String, PortSet> {
+pub(crate) fn pod_tcp_ports_by_name(spec: &k8s::PodSpec) -> HashMap<String, PortSet> {
     let mut ports = HashMap::<String, PortSet>::default();
     for (port, name) in spec
         .containers
@@ -37,6 +37,26 @@ pub(crate) fn tcp_ports_by_name(spec: &k8s::PodSpec) -> HashMap<String, PortSet>
     ports
 }
 
+/// Gets the set of named ports withn `protocol: TCP` from an external workload
+/// spec.
+///
+/// Since an external workload has only one set of ports, each name is
+/// guaranteed to be unique.
+pub(crate) fn external_tcp_ports_by_name(
+    spec: &k8s::external_workload::ExternalWorkloadSpec,
+) -> HashMap<String, NonZeroU16> {
+    let mut ports = HashMap::<String, NonZeroU16>::default();
+    for (port, name) in spec
+        .ports
+        .iter()
+        .flatten()
+        .filter_map(named_external_tcp_port)
+    {
+        ports.insert(name.into(), port);
+    }
+    ports
+}
+
 /// Gets the container probe ports for a Pod.
 ///
 /// The result is a mapping for each probe port exposed by a container in the
@@ -103,6 +123,16 @@ fn named_tcp_port(port: &k8s::ContainerPort) -> Option<(NonZeroU16, &str)> {
     Some((p, n))
 }
 
+fn named_external_tcp_port(spec: &k8s::external_workload::PortSpec) -> Option<(NonZeroU16, &str)> {
+    if let Some(ref proto) = spec.protocol {
+        if !proto.eq_ignore_ascii_case("TCP") {
+            return None;
+        }
+    }
+    let n = spec.name.as_deref()?;
+    Some((spec.port, n))
+}
+
 // === impl Meta ===
 
 impl Meta {
diff --git a/policy-controller/src/lib.rs b/policy-controller/src/lib.rs
index 8efbb1cb9..d524a6578 100644
--- a/policy-controller/src/lib.rs
+++ b/policy-controller/src/lib.rs
@@ -36,26 +36,49 @@ impl OutboundDiscover {
 }
 
 #[async_trait::async_trait]
-impl DiscoverInboundServer<(String, String, NonZeroU16)> for InboundDiscover {
+impl DiscoverInboundServer<(grpc::workload::Workload, NonZeroU16)> for InboundDiscover {
     async fn get_inbound_server(
         &self,
-        (namespace, pod, port): (String, String, NonZeroU16),
+        (workload, port): (grpc::workload::Workload, NonZeroU16),
     ) -> Result<Option<InboundServer>> {
-        let rx = match self.0.write().pod_server_rx(&namespace, &pod, port) {
-            Ok(rx) => rx,
-            Err(_) => return Ok(None),
+        let grpc::workload::Workload { namespace, kind } = workload;
+        let rx = match kind {
+            grpc::workload::Kind::External(name) => self
+                .0
+                .write()
+                .external_workload_server_rx(&namespace, &name, port),
+            grpc::workload::Kind::Pod(name) => {
+                self.0.write().pod_server_rx(&namespace, &name, port)
+            }
         };
-        let server = (*rx.borrow()).clone();
-        Ok(Some(server))
+
+        if let Ok(rx) = rx {
+            let server = (*rx.borrow()).clone();
+            Ok(Some(server))
+        } else {
+            Ok(None)
+        }
     }
 
     async fn watch_inbound_server(
         &self,
-        (namespace, pod, port): (String, String, NonZeroU16),
+        (workload, port): (grpc::workload::Workload, NonZeroU16),
     ) -> Result<Option<InboundServerStream>> {
-        match self.0.write().pod_server_rx(&namespace, &pod, port) {
-            Ok(rx) => Ok(Some(Box::pin(tokio_stream::wrappers::WatchStream::new(rx)))),
-            Err(_) => Ok(None),
+        let grpc::workload::Workload { namespace, kind } = workload;
+        let rx = match kind {
+            grpc::workload::Kind::External(name) => self
+                .0
+                .write()
+                .external_workload_server_rx(&namespace, &name, port),
+            grpc::workload::Kind::Pod(name) => {
+                self.0.write().pod_server_rx(&namespace, &name, port)
+            }
+        };
+
+        if let Ok(rx) = rx {
+            Ok(Some(Box::pin(tokio_stream::wrappers::WatchStream::new(rx))))
+        } else {
+            Ok(None)
         }
     }
 }
diff --git a/policy-controller/src/main.rs b/policy-controller/src/main.rs
index 9f40c6f0d..2e923bbfa 100644
--- a/policy-controller/src/main.rs
+++ b/policy-controller/src/main.rs
@@ -170,6 +170,13 @@ async fn main() -> Result<()> {
         kubert::index::namespaced(inbound_index.clone(), pods).instrument(info_span!("pods")),
     );
 
+    let external_workloads =
+        runtime.watch_all::<k8s::external_workload::ExternalWorkload>(watcher::Config::default());
+    tokio::spawn(
+        kubert::index::namespaced(inbound_index.clone(), external_workloads)
+            .instrument(info_span!("external_workloads")),
+    );
+
     let servers = runtime.watch_all::<k8s::policy::Server>(watcher::Config::default());
     let servers_indexes = IndexList::new(inbound_index.clone())
         .push(status_index.clone())
diff --git a/viz/charts/linkerd-viz/Chart.yaml b/viz/charts/linkerd-viz/Chart.yaml
index 944e41212..149f59e15 100644
--- a/viz/charts/linkerd-viz/Chart.yaml
+++ b/viz/charts/linkerd-viz/Chart.yaml
@@ -11,7 +11,7 @@ kubeVersion: ">=1.22.0-0"
 name: "linkerd-viz"
 sources:
   - https://github.com/linkerd/linkerd2/
-version: 30.14.4-edge
+version: 30.14.5-edge
 icon: https://linkerd.io/images/logo-only-200h.png
 maintainers:
   - name: Linkerd authors
diff --git a/viz/charts/linkerd-viz/README.md b/viz/charts/linkerd-viz/README.md
index eceeb2ef1..8f89c70df 100644
--- a/viz/charts/linkerd-viz/README.md
+++ b/viz/charts/linkerd-viz/README.md
@@ -3,7 +3,7 @@
 The Linkerd-Viz extension contains observability and visualization
 components for Linkerd.
 
-![Version: 30.14.4-edge](https://img.shields.io/badge/Version-30.14.4--edge-informational?style=flat-square)
+![Version: 30.14.5-edge](https://img.shields.io/badge/Version-30.14.5--edge-informational?style=flat-square)
 
 ![AppVersion: edge-XX.X.X](https://img.shields.io/badge/AppVersion-edge--XX.X.X-informational?style=flat-square)
 
diff --git a/web/app/package.json b/web/app/package.json
index 1297bf38b..384910f21 100644
--- a/web/app/package.json
+++ b/web/app/package.json
@@ -16,13 +16,13 @@
     "@material-ui/icons": "4.11.3",
     "@material-ui/lab": "^4.0.0-alpha.61",
     "classnames": "2.5.1",
-    "core-js": "^3.35.0",
+    "core-js": "^3.35.1",
     "css-mediaquery": "^0.1.2",
     "d3-drag": "3.0.0",
     "d3-force": "3.0.0",
     "d3-format": "3.1.0",
     "d3-selection": "3.0.0",
-    "date-fns": "3.1.0",
+    "date-fns": "3.3.0",
     "jest-progress-bar-reporter": "^1.0.25",
     "locales-detector": "3.0.2",
     "lodash": "4.17.21",
@@ -44,7 +44,7 @@
     "@babel/core": "^7.23.7",
     "@babel/eslint-parser": "^7.23.3",
     "@babel/plugin-proposal-class-properties": "^7.17.12",
-    "@babel/preset-env": "^7.23.7",
+    "@babel/preset-env": "^7.23.8",
     "@babel/preset-react": "^7.23.3",
     "@babel/runtime": "^7.23.8",
     "@lingui/cli": "3.17.2",
@@ -56,7 +56,7 @@
     "babel-plugin-transform-react-remove-prop-types": "0.4.24",
     "chai": "4.3.10",
     "clean-webpack-plugin": "4.0.0",
-    "css-loader": "^6.9.0",
+    "css-loader": "^6.9.1",
     "enzyme": "3.11.0",
     "enzyme-adapter-react-16": "^1.15.7",
     "eslint": "^8.56.0",
diff --git a/web/app/yarn.lock b/web/app/yarn.lock
index 603f44fb5..aac8d76af 100644
--- a/web/app/yarn.lock
+++ b/web/app/yarn.lock
@@ -629,16 +629,15 @@
     "@babel/helper-plugin-utils" "^7.22.5"
     "@babel/plugin-syntax-class-static-block" "^7.14.5"
 
-"@babel/plugin-transform-classes@^7.23.5":
-  version "7.23.5"
-  resolved "https://registry.yarnpkg.com/@babel/plugin-transform-classes/-/plugin-transform-classes-7.23.5.tgz#e7a75f815e0c534cc4c9a39c56636c84fc0d64f2"
-  integrity sha512-jvOTR4nicqYC9yzOHIhXG5emiFEOpappSJAl73SDSEDcybD+Puuze8Tnpb9p9qEyYup24tq891gkaygIFvWDqg==
+"@babel/plugin-transform-classes@^7.23.8":
+  version "7.23.8"
+  resolved "https://registry.yarnpkg.com/@babel/plugin-transform-classes/-/plugin-transform-classes-7.23.8.tgz#d08ae096c240347badd68cdf1b6d1624a6435d92"
+  integrity sha512-yAYslGsY1bX6Knmg46RjiCiNSwJKv2IUC8qOdYKqMMr0491SXFhcHqOdRDeCRohOOIzwN/90C6mQ9qAKgrP7dg==
   dependencies:
     "@babel/helper-annotate-as-pure" "^7.22.5"
-    "@babel/helper-compilation-targets" "^7.22.15"
+    "@babel/helper-compilation-targets" "^7.23.6"
     "@babel/helper-environment-visitor" "^7.22.20"
     "@babel/helper-function-name" "^7.23.0"
-    "@babel/helper-optimise-call-expression" "^7.22.5"
     "@babel/helper-plugin-utils" "^7.22.5"
     "@babel/helper-replace-supers" "^7.22.20"
     "@babel/helper-split-export-declaration" "^7.22.6"
@@ -994,10 +993,10 @@
     "@babel/helper-create-regexp-features-plugin" "^7.22.15"
     "@babel/helper-plugin-utils" "^7.22.5"
 
-"@babel/preset-env@^7.23.7":
-  version "7.23.7"
-  resolved "https://registry.yarnpkg.com/@babel/preset-env/-/preset-env-7.23.7.tgz#e5d69b9f14db8a13bae4d8e5ce7f360973626241"
-  integrity sha512-SY27X/GtTz/L4UryMNJ6p4fH4nsgWbz84y9FE0bQeWJP6O5BhgVCt53CotQKHCOeXJel8VyhlhujhlltKms/CA==
+"@babel/preset-env@^7.23.8":
+  version "7.23.8"
+  resolved "https://registry.yarnpkg.com/@babel/preset-env/-/preset-env-7.23.8.tgz#7d6f8171ea7c221ecd28059e65ad37c20e441e3e"
+  integrity sha512-lFlpmkApLkEP6woIKprO6DO60RImpatTQKtz4sUcDjVcK8M8mQ4sZsuxaTMNOZf0sqAq/ReYW1ZBHnOQwKpLWA==
   dependencies:
     "@babel/compat-data" "^7.23.5"
     "@babel/helper-compilation-targets" "^7.23.6"
@@ -1032,7 +1031,7 @@
     "@babel/plugin-transform-block-scoping" "^7.23.4"
     "@babel/plugin-transform-class-properties" "^7.23.3"
     "@babel/plugin-transform-class-static-block" "^7.23.4"
-    "@babel/plugin-transform-classes" "^7.23.5"
+    "@babel/plugin-transform-classes" "^7.23.8"
     "@babel/plugin-transform-computed-properties" "^7.23.3"
     "@babel/plugin-transform-destructuring" "^7.23.3"
     "@babel/plugin-transform-dotall-regex" "^7.23.3"
@@ -3251,10 +3250,10 @@ core-js-compat@^3.31.0, core-js-compat@^3.33.1:
   dependencies:
     browserslist "^4.22.2"
 
-core-js@^3.35.0:
-  version "3.35.0"
-  resolved "https://registry.yarnpkg.com/core-js/-/core-js-3.35.0.tgz#58e651688484f83c34196ca13f099574ee53d6b4"
-  integrity sha512-ntakECeqg81KqMueeGJ79Q5ZgQNR+6eaE8sxGCx62zMbAIj65q+uYvatToew3m6eAGdU4gNZwpZ34NMe4GYswg==
+core-js@^3.35.1:
+  version "3.35.1"
+  resolved "https://registry.yarnpkg.com/core-js/-/core-js-3.35.1.tgz#9c28f8b7ccee482796f8590cc8d15739eaaf980c"
+  integrity sha512-IgdsbxNyMskrTFxa9lWHyMwAJU5gXOPP+1yO+K59d50VLVAIDAbs7gIv705KzALModfK3ZrSZTPNpC0PQgIZuw==
 
 core-util-is@~1.0.0:
   version "1.0.3"
@@ -3309,16 +3308,16 @@ cross-spawn@^7.0.2, cross-spawn@^7.0.3:
     shebang-command "^2.0.0"
     which "^2.0.1"
 
-css-loader@^6.9.0:
-  version "6.9.0"
-  resolved "https://registry.yarnpkg.com/css-loader/-/css-loader-6.9.0.tgz#0cc2f14df94ed97c526c5ae42b6b13916d1d8d0e"
-  integrity sha512-3I5Nu4ytWlHvOP6zItjiHlefBNtrH+oehq8tnQa2kO305qpVyx9XNIT1CXIj5bgCJs7qICBCkgCYxQLKPANoLA==
+css-loader@^6.9.1:
+  version "6.9.1"
+  resolved "https://registry.yarnpkg.com/css-loader/-/css-loader-6.9.1.tgz#9ec9a434368f2bdfeffbf8f6901a1ce773586c6b"
+  integrity sha512-OzABOh0+26JKFdMzlK6PY1u5Zx8+Ck7CVRlcGNZoY9qwJjdfu2VWFuprTIpPW+Av5TZTVViYWcFQaEEQURLknQ==
   dependencies:
     icss-utils "^5.1.0"
-    postcss "^8.4.31"
+    postcss "^8.4.33"
     postcss-modules-extract-imports "^3.0.0"
-    postcss-modules-local-by-default "^4.0.3"
-    postcss-modules-scope "^3.1.0"
+    postcss-modules-local-by-default "^4.0.4"
+    postcss-modules-scope "^3.1.1"
     postcss-modules-values "^4.0.0"
     postcss-value-parser "^4.2.0"
     semver "^7.5.4"
@@ -3440,10 +3439,10 @@ data-urls@^3.0.2:
     whatwg-mimetype "^3.0.0"
     whatwg-url "^11.0.0"
 
-date-fns@3.1.0:
-  version "3.1.0"
-  resolved "https://registry.yarnpkg.com/date-fns/-/date-fns-3.1.0.tgz#576df8df93202942aeebd49ecc042218feb40426"
-  integrity sha512-ZO7yefXV/wCWzd3I9haCHmfzlfA3i1a2HHO7ZXjtJrRjXt8FULKJ2Vl8wji3XYF4dQ0ZJ/tokXDZeYlFvgms9Q==
+date-fns@3.3.0:
+  version "3.3.0"
+  resolved "https://registry.yarnpkg.com/date-fns/-/date-fns-3.3.0.tgz#c1681691cf751a1d371279099a45e71409c7c761"
+  integrity sha512-xuouT0GuI2W8yXhCMn/AXbSl1Av3wu2hJXxMnnILTY3bYY0UgNK0qOwVXqdFBrobW5qbX1TuOTgMw7c2H2OuhA==
 
 date-fns@^2.16.1:
   version "2.30.0"
@@ -7259,19 +7258,19 @@ postcss-modules-extract-imports@^3.0.0:
   resolved "https://registry.yarnpkg.com/postcss-modules-extract-imports/-/postcss-modules-extract-imports-3.0.0.tgz#cda1f047c0ae80c97dbe28c3e76a43b88025741d"
   integrity sha512-bdHleFnP3kZ4NYDhuGlVK+CMrQ/pqUm8bx/oGL93K6gVwiclvX5x0n76fYMKuIGKzlABOy13zsvqjb0f92TEXw==
 
-postcss-modules-local-by-default@^4.0.3:
-  version "4.0.3"
-  resolved "https://registry.yarnpkg.com/postcss-modules-local-by-default/-/postcss-modules-local-by-default-4.0.3.tgz#b08eb4f083050708998ba2c6061b50c2870ca524"
-  integrity sha512-2/u2zraspoACtrbFRnTijMiQtb4GW4BvatjaG/bCjYQo8kLTdevCUlwuBHx2sCnSyrI3x3qj4ZK1j5LQBgzmwA==
+postcss-modules-local-by-default@^4.0.4:
+  version "4.0.4"
+  resolved "https://registry.yarnpkg.com/postcss-modules-local-by-default/-/postcss-modules-local-by-default-4.0.4.tgz#7cbed92abd312b94aaea85b68226d3dec39a14e6"
+  integrity sha512-L4QzMnOdVwRm1Qb8m4x8jsZzKAaPAgrUF1r/hjDR2Xj7R+8Zsf97jAlSQzWtKx5YNiNGN8QxmPFIc/sh+RQl+Q==
   dependencies:
     icss-utils "^5.0.0"
     postcss-selector-parser "^6.0.2"
     postcss-value-parser "^4.1.0"
 
-postcss-modules-scope@^3.1.0:
-  version "3.1.0"
-  resolved "https://registry.yarnpkg.com/postcss-modules-scope/-/postcss-modules-scope-3.1.0.tgz#fbfddfda93a31f310f1d152c2bb4d3f3c5592ee0"
-  integrity sha512-SaIbK8XW+MZbd0xHPf7kdfA/3eOt7vxJ72IRecn3EzuZVLr1r0orzf0MX/pN8m+NMDoo6X/SQd8oeKqGZd8PXg==
+postcss-modules-scope@^3.1.1:
+  version "3.1.1"
+  resolved "https://registry.yarnpkg.com/postcss-modules-scope/-/postcss-modules-scope-3.1.1.tgz#32cfab55e84887c079a19bbb215e721d683ef134"
+  integrity sha512-uZgqzdTleelWjzJY+Fhti6F3C9iF1JR/dODLs/JDefozYcKTBCdD8BIl6nNPbTbcLnGrk56hzwZC2DaGNvYjzA==
   dependencies:
     postcss-selector-parser "^6.0.4"
 
@@ -7295,7 +7294,7 @@ postcss-value-parser@^4.1.0, postcss-value-parser@^4.2.0:
   resolved "https://registry.yarnpkg.com/postcss-value-parser/-/postcss-value-parser-4.2.0.tgz#723c09920836ba6d3e5af019f92bc0971c02e514"
   integrity sha512-1NNCs6uurfkVbeXG4S8JFT9t19m45ICnif8zWLd5oPSZ50QnwMfK+H3jv408d4jw/7Bttv5axS5IiHoLaVNHeQ==
 
-postcss@^8.4.31:
+postcss@^8.4.33:
   version "8.4.33"
   resolved "https://registry.yarnpkg.com/postcss/-/postcss-8.4.33.tgz#1378e859c9f69bf6f638b990a0212f43e2aaa742"
   integrity sha512-Kkpbhhdjw2qQs2O2DGX+8m5OVqEcbB9HRBvuYM9pgrjEFUg30A9LmXNlTAUj4S9kgtGyrMbTzVjH7E+s5Re2yg==


Test patch:
diff --git a/cli/cmd/testdata/install_controlplane_tracing_output.golden b/cli/cmd/testdata/install_controlplane_tracing_output.golden
index 5c58c6a22..75f20491f 100644
--- a/cli/cmd/testdata/install_controlplane_tracing_output.golden
+++ b/cli/cmd/testdata/install_controlplane_tracing_output.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1232,7 +1246,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_custom_domain.golden b/cli/cmd/testdata/install_custom_domain.golden
index 01096f0c3..8324a574b 100644
--- a/cli/cmd/testdata/install_custom_domain.golden
+++ b/cli/cmd/testdata/install_custom_domain.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1231,7 +1245,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_custom_registry.golden b/cli/cmd/testdata/install_custom_registry.golden
index be2a34800..48f10e6bf 100644
--- a/cli/cmd/testdata/install_custom_registry.golden
+++ b/cli/cmd/testdata/install_custom_registry.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1231,7 +1245,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_default.golden b/cli/cmd/testdata/install_default.golden
index 01096f0c3..8324a574b 100644
--- a/cli/cmd/testdata/install_default.golden
+++ b/cli/cmd/testdata/install_default.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1231,7 +1245,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_default_override_dst_get_nets.golden b/cli/cmd/testdata/install_default_override_dst_get_nets.golden
index 64a99407b..37952cbe8 100644
--- a/cli/cmd/testdata/install_default_override_dst_get_nets.golden
+++ b/cli/cmd/testdata/install_default_override_dst_get_nets.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1231,7 +1245,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_default_token.golden b/cli/cmd/testdata/install_default_token.golden
index dd73e0c20..ec0d8aa40 100644
--- a/cli/cmd/testdata/install_default_token.golden
+++ b/cli/cmd/testdata/install_default_token.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1222,7 +1236,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_ha_output.golden b/cli/cmd/testdata/install_ha_output.golden
index 6226c4124..49303684a 100644
--- a/cli/cmd/testdata/install_ha_output.golden
+++ b/cli/cmd/testdata/install_ha_output.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1329,7 +1343,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: d678d93a3ae6be52fc98976b988d287fb3f8bc41d1eb9cc5f06d63fc3a3b9ee7
+        checksum/config: 8f967c9ecfe4fb09646dfa503d3831a391288c65fb8b77200d2b3c9390620c1e
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_ha_with_overrides_output.golden b/cli/cmd/testdata/install_ha_with_overrides_output.golden
index a7c633867..684aa2e1e 100644
--- a/cli/cmd/testdata/install_ha_with_overrides_output.golden
+++ b/cli/cmd/testdata/install_ha_with_overrides_output.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1329,7 +1343,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: d678d93a3ae6be52fc98976b988d287fb3f8bc41d1eb9cc5f06d63fc3a3b9ee7
+        checksum/config: 8f967c9ecfe4fb09646dfa503d3831a391288c65fb8b77200d2b3c9390620c1e
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_heartbeat_disabled_output.golden b/cli/cmd/testdata/install_heartbeat_disabled_output.golden
index e17f5634f..91bab1d21 100644
--- a/cli/cmd/testdata/install_heartbeat_disabled_output.golden
+++ b/cli/cmd/testdata/install_heartbeat_disabled_output.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1162,7 +1176,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_helm_control_plane_output.golden b/cli/cmd/testdata/install_helm_control_plane_output.golden
index fb2546893..7da3a4b50 100644
--- a/cli/cmd/testdata/install_helm_control_plane_output.golden
+++ b/cli/cmd/testdata/install_helm_control_plane_output.golden
@@ -73,9 +73,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -229,6 +235,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1206,7 +1220,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 387ff1ed65e66e400e556c47490d20187adcb885526d44d9f0aa75dc97f23223
+        checksum/config: bdf627c4f8fc3bc85e69064de22d6032ab30887fa55d18238a677e9b9b594dd8
         linkerd.io/created-by: linkerd/helm linkerd-version
         linkerd.io/proxy-version: test-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_helm_control_plane_output_ha.golden b/cli/cmd/testdata/install_helm_control_plane_output_ha.golden
index ab4f21c77..ee0986c39 100644
--- a/cli/cmd/testdata/install_helm_control_plane_output_ha.golden
+++ b/cli/cmd/testdata/install_helm_control_plane_output_ha.golden
@@ -73,9 +73,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -229,6 +235,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1304,7 +1318,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: ff01f9fed5a80929d1215ebec514a25888ac7fed762a5a753bd896928e26b951
+        checksum/config: b0c26a237398c80aaed48f6954a403fc169549f3b7e927bdd86c71f8d13c8762
         linkerd.io/created-by: linkerd/helm linkerd-version
         linkerd.io/proxy-version: test-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_helm_output_ha_labels.golden b/cli/cmd/testdata/install_helm_output_ha_labels.golden
index 7f3885fe9..a189adbae 100644
--- a/cli/cmd/testdata/install_helm_output_ha_labels.golden
+++ b/cli/cmd/testdata/install_helm_output_ha_labels.golden
@@ -73,9 +73,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -229,6 +235,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1312,7 +1326,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: ff01f9fed5a80929d1215ebec514a25888ac7fed762a5a753bd896928e26b951
+        checksum/config: b0c26a237398c80aaed48f6954a403fc169549f3b7e927bdd86c71f8d13c8762
         linkerd.io/created-by: linkerd/helm linkerd-version
         linkerd.io/proxy-version: test-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden b/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden
index 8ac004e00..97031c08f 100644
--- a/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden
+++ b/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden
@@ -73,9 +73,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -229,6 +235,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1294,7 +1308,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: cf05a64f720a18752cb8e0e759cb4cd6e22602eb0129a2144a2093ac6ea87659
+        checksum/config: 467bcfd10f346970575701d5199fde803e9f61e9039e05d4c765d3c5524804b4
         linkerd.io/created-by: linkerd/helm linkerd-version
         linkerd.io/proxy-version: test-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_no_init_container.golden b/cli/cmd/testdata/install_no_init_container.golden
index f4d992fb5..251150fa7 100644
--- a/cli/cmd/testdata/install_no_init_container.golden
+++ b/cli/cmd/testdata/install_no_init_container.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1225,7 +1239,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_output.golden b/cli/cmd/testdata/install_output.golden
index 2ae231777..82c343d94 100644
--- a/cli/cmd/testdata/install_output.golden
+++ b/cli/cmd/testdata/install_output.golden
@@ -82,6 +82,12 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -235,6 +241,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1197,7 +1211,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: d3eac3abc6cf5f1943e0f005fd15ad0107229283fa6f8db72f9ccb4766bb2228
+        checksum/config: 91d0273ac7d213bf95872f62c460ba146a459106e21d12f75e2ebe6ad7562b7f
         linkerd.io/created-by: CliVersion
         linkerd.io/proxy-version: ProxyVersion
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_proxy_ignores.golden b/cli/cmd/testdata/install_proxy_ignores.golden
index fdb1a225c..c6ec681a4 100644
--- a/cli/cmd/testdata/install_proxy_ignores.golden
+++ b/cli/cmd/testdata/install_proxy_ignores.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1231,7 +1245,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/cli/cmd/testdata/install_values_file.golden b/cli/cmd/testdata/install_values_file.golden
index 5859872de..bab001a87 100644
--- a/cli/cmd/testdata/install_values_file.golden
+++ b/cli/cmd/testdata/install_values_file.golden
@@ -82,9 +82,15 @@ rules:
 - apiGroups: ["linkerd.io"]
   resources: ["serviceprofiles"]
   verbs: ["list", "get", "watch"]
+- apiGroups: ["workload.linkerd.io"]
+  resources: ["externalworkloads"]
+  verbs: ["list", "get", "watch"]
+- apiGroups: ["coordination.k8s.io"]
+  resources: ["leases"]
+  verbs: ["create", "get", "update", "patch"]
 - apiGroups: ["discovery.k8s.io"]
   resources: ["endpointslices"]
-  verbs: ["list", "get", "watch"]
+  verbs: ["list", "get", "watch", "create", "update", "patch", "delete"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -238,6 +244,14 @@ rules:
       - httproutes/status
     verbs:
       - patch
+  - apiGroups:
+      - workload.linkerd.io
+    resources:
+      - externalworkloads
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - coordination.k8s.io
     resources:
@@ -1231,7 +1245,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 6c6be36634a1edea1ebcad1c3981959c48eb6b78684e7093058a50eca6ba2f97
+        checksum/config: 579a2f931900ccee9dc86afedf01af0e4ca273ef6ec649b2630c358daa1e067f
         linkerd.io/created-by: linkerd/cli dev-undefined
         linkerd.io/proxy-version: install-proxy-version
         cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
diff --git a/controller/api/destination/endpoint_translator_test.go b/controller/api/destination/endpoint_translator_test.go
index 6f3bf80b5..9e7ca43f8 100644
--- a/controller/api/destination/endpoint_translator_test.go
+++ b/controller/api/destination/endpoint_translator_test.go
@@ -11,6 +11,7 @@ import (
 	pb "github.com/linkerd/linkerd2-proxy-api/go/destination"
 	"github.com/linkerd/linkerd2-proxy-api/go/net"
 	"github.com/linkerd/linkerd2/controller/api/destination/watcher"
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
 	"github.com/linkerd/linkerd2/pkg/addr"
 	"github.com/linkerd/linkerd2/pkg/k8s"
 	corev1 "k8s.io/api/core/v1"
@@ -101,6 +102,95 @@ var (
 		OpaqueProtocol: true,
 	}
 
+	ew1 = watcher.Address{
+		IP:   "1.1.1.1",
+		Port: 1,
+		ExternalWorkload: &ewv1alpha1.ExternalWorkload{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "ew-1",
+				Namespace: "ns",
+			},
+			Spec: ewv1alpha1.ExternalWorkloadSpec{
+				MeshTls: ewv1alpha1.MeshTls{
+					Identity:   "spiffe://some-domain/ew-1",
+					ServerName: "server.local",
+				},
+			},
+		},
+		OwnerKind: "workloadgroup",
+		OwnerName: "wg-name",
+	}
+
+	ew2 = watcher.Address{
+		IP:   "1.1.1.2",
+		Port: 2,
+		ExternalWorkload: &ewv1alpha1.ExternalWorkload{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "ew-2",
+				Namespace: "ns",
+				Labels: map[string]string{
+					k8s.ControllerNSLabel:    "linkerd",
+					k8s.ProxyDeploymentLabel: "deployment-name",
+				},
+			},
+			Spec: ewv1alpha1.ExternalWorkloadSpec{
+				MeshTls: ewv1alpha1.MeshTls{
+					Identity:   "spiffe://some-domain/ew-2",
+					ServerName: "server.local",
+				},
+			},
+		},
+	}
+
+	ew3 = watcher.Address{
+		IP:   "1.1.1.3",
+		Port: 3,
+		ExternalWorkload: &ewv1alpha1.ExternalWorkload{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "ew-3",
+				Namespace: "ns",
+				Labels: map[string]string{
+					k8s.ControllerNSLabel:    "linkerd",
+					k8s.ProxyDeploymentLabel: "deployment-name",
+				},
+			},
+			Spec: ewv1alpha1.ExternalWorkloadSpec{
+				MeshTls: ewv1alpha1.MeshTls{
+					Identity:   "spiffe://some-domain/ew-3",
+					ServerName: "server.local",
+				},
+			},
+		},
+	}
+
+	ewOpaque = watcher.Address{
+		IP:   "1.1.1.4",
+		Port: 4,
+		ExternalWorkload: &ewv1alpha1.ExternalWorkload{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "pod4",
+				Namespace: "ns",
+				Annotations: map[string]string{
+					k8s.ProxyOpaquePortsAnnotation: "4",
+				},
+			},
+			Spec: ewv1alpha1.ExternalWorkloadSpec{
+				MeshTls: ewv1alpha1.MeshTls{
+					Identity:   "spiffe://some-domain/ew-opaque",
+					ServerName: "server.local",
+				},
+
+				Ports: []ewv1alpha1.PortSpec{
+					{
+						Port: 4143,
+						Name: "linkerd-proxy",
+					},
+				},
+			},
+		},
+		OpaqueProtocol: true,
+	}
+
 	remoteGateway1 = watcher.Address{
 		IP:   "1.1.1.1",
 		Port: 1,
@@ -429,6 +519,136 @@ func TestEndpointTranslatorForPods(t *testing.T) {
 	})
 }
 
+func TestEndpointTranslatorExternalWorkloads(t *testing.T) {
+	t.Run("Sends one update for add and another for remove", func(t *testing.T) {
+		mockGetServer, translator := makeEndpointTranslator(t)
+		translator.Start()
+		defer translator.Stop()
+
+		translator.Add(mkAddressSetForExternalWorkloads(ew1, ew2))
+		translator.Remove(mkAddressSetForExternalWorkloads(ew2))
+
+		expectedNumUpdates := 2
+		<-mockGetServer.updatesReceived // Add
+		<-mockGetServer.updatesReceived // Remove
+
+		if len(mockGetServer.updatesReceived) != 0 {
+			t.Fatalf("Expecting [%d] updates, got [%d].", expectedNumUpdates, expectedNumUpdates+len(mockGetServer.updatesReceived))
+		}
+	})
+
+	t.Run("Sends addresses as removed or added", func(t *testing.T) {
+		mockGetServer, translator := makeEndpointTranslator(t)
+		translator.Start()
+		defer translator.Stop()
+
+		translator.Add(mkAddressSetForExternalWorkloads(ew1, ew2, ew3))
+		translator.Remove(mkAddressSetForExternalWorkloads(ew3))
+
+		addressesAdded := (<-mockGetServer.updatesReceived).GetAdd().Addrs
+		actualNumberOfAdded := len(addressesAdded)
+		expectedNumberOfAdded := 3
+		if actualNumberOfAdded != expectedNumberOfAdded {
+			t.Fatalf("Expecting [%d] addresses to be added, got [%d]: %v", expectedNumberOfAdded, actualNumberOfAdded, addressesAdded)
+		}
+
+		addressesRemoved := (<-mockGetServer.updatesReceived).GetRemove().Addrs
+		actualNumberOfRemoved := len(addressesRemoved)
+		expectedNumberOfRemoved := 1
+		if actualNumberOfRemoved != expectedNumberOfRemoved {
+			t.Fatalf("Expecting [%d] addresses to be removed, got [%d]: %v", expectedNumberOfRemoved, actualNumberOfRemoved, addressesRemoved)
+		}
+
+		sort.Slice(addressesAdded, func(i, j int) bool {
+			return addressesAdded[i].GetAddr().Port < addressesAdded[j].GetAddr().Port
+		})
+		checkAddressAndWeight(t, addressesAdded[0], ew1, defaultWeight)
+		checkAddressAndWeight(t, addressesAdded[1], ew2, defaultWeight)
+		checkAddress(t, addressesRemoved[0], ew3)
+	})
+
+	t.Run("Sends metric labels with added addresses", func(t *testing.T) {
+		mockGetServer, translator := makeEndpointTranslator(t)
+		translator.Start()
+		defer translator.Stop()
+
+		translator.Add(mkAddressSetForExternalWorkloads(ew1))
+
+		update := <-mockGetServer.updatesReceived
+
+		actualGlobalMetricLabels := update.GetAdd().MetricLabels
+		expectedGlobalMetricLabels := map[string]string{"namespace": "service-ns", "service": "service-name"}
+		if diff := deep.Equal(actualGlobalMetricLabels, expectedGlobalMetricLabels); diff != nil {
+			t.Fatalf("Expected global metric labels sent to be [%v] but was [%v]", expectedGlobalMetricLabels, actualGlobalMetricLabels)
+		}
+
+		actualAddedAddress1MetricLabels := update.GetAdd().Addrs[0].MetricLabels
+		expectedAddedAddress1MetricLabels := map[string]string{
+			"external_workload": "ew-1",
+			"zone":              "",
+			"workloadgroup":     "wg-name",
+		}
+		if diff := deep.Equal(actualAddedAddress1MetricLabels, expectedAddedAddress1MetricLabels); diff != nil {
+			t.Fatalf("Expected global metric labels sent to be [%v] but was [%v]", expectedAddedAddress1MetricLabels, actualAddedAddress1MetricLabels)
+		}
+	})
+
+	t.Run("Sends TlsIdentity and Server Name when enabled", func(t *testing.T) {
+		expectedTLSIdentity := &pb.TlsIdentity{
+			Strategy: &pb.TlsIdentity_UriLikeIdentity_{
+				UriLikeIdentity: &pb.TlsIdentity_UriLikeIdentity{
+					Uri: "spiffe://some-domain/ew-1",
+				},
+			},
+			ServerName: &pb.TlsIdentity_DnsLikeIdentity{
+				Name: "server.local",
+			},
+		}
+
+		mockGetServer, translator := makeEndpointTranslator(t)
+		translator.Start()
+		defer translator.Stop()
+
+		translator.Add(mkAddressSetForExternalWorkloads(ew1))
+		addrs := (<-mockGetServer.updatesReceived).GetAdd().GetAddrs()
+		if len(addrs) != 1 {
+			t.Fatalf("Expected [1] address returned, got %v", addrs)
+		}
+
+		actualTLSIdentity := addrs[0].GetTlsIdentity()
+		if diff := deep.Equal(actualTLSIdentity, expectedTLSIdentity); diff != nil {
+			t.Fatalf("Expected TlsIdentity to be [%v] but was [%v]", expectedTLSIdentity, actualTLSIdentity)
+		}
+	})
+
+	t.Run("Sends Opaque ProtocolHint for opaque ports", func(t *testing.T) {
+		expectedProtocolHint := &pb.ProtocolHint{
+			Protocol: &pb.ProtocolHint_Opaque_{
+				Opaque: &pb.ProtocolHint_Opaque{},
+			},
+			OpaqueTransport: &pb.ProtocolHint_OpaqueTransport{
+				InboundPort: 4143,
+			},
+		}
+
+		mockGetServer, translator := makeEndpointTranslator(t)
+		translator.Start()
+		defer translator.Stop()
+
+		translator.Add(mkAddressSetForExternalWorkloads(ewOpaque))
+
+		addrs := (<-mockGetServer.updatesReceived).GetAdd().GetAddrs()
+		if len(addrs) != 1 {
+			t.Fatalf("Expected [1] address returned, got %v", addrs)
+		}
+
+		actualProtocolHint := addrs[0].GetProtocolHint()
+		if diff := deep.Equal(actualProtocolHint, expectedProtocolHint); diff != nil {
+			t.Fatalf("ProtocolHint: %v", diff)
+		}
+	})
+}
+
 func TestEndpointTranslatorTopologyAwareFilter(t *testing.T) {
 	t.Run("Sends one update for add and none for remove", func(t *testing.T) {
 		mockGetServer, translator := makeEndpointTranslator(t)
@@ -575,6 +795,18 @@ func mkAddressSetForPods(podAddresses ...watcher.Address) watcher.AddressSet {
 	return set
 }
 
+func mkAddressSetForExternalWorkloads(ewAddresses ...watcher.Address) watcher.AddressSet {
+	set := watcher.AddressSet{
+		Addresses: make(map[watcher.PodID]watcher.Address),
+		Labels:    map[string]string{"service": "service-name", "namespace": "service-ns"},
+	}
+	for _, ew := range ewAddresses {
+		id := watcher.ExternalWorkloadID{Name: ew.ExternalWorkload.Name, Namespace: ew.ExternalWorkload.Namespace}
+		set.Addresses[id] = ew
+	}
+	return set
+}
+
 func checkAddressAndWeight(t *testing.T, actual *pb.WeightedAddr, expected watcher.Address, weight uint32) {
 	t.Helper()
 
diff --git a/controller/api/destination/external-workload/endpoints_controller_test.go b/controller/api/destination/external-workload/endpoints_controller_test.go
new file mode 100644
index 000000000..25eaec309
--- /dev/null
+++ b/controller/api/destination/external-workload/endpoints_controller_test.go
@@ -0,0 +1,354 @@
+package externalworkload
+
+import (
+	"testing"
+
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
+	"github.com/linkerd/linkerd2/controller/k8s"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func makeExternalWorkload(labels map[string]string, ports map[int32]string, ips []string) *ewv1alpha1.ExternalWorkload {
+	portSpecs := []ewv1alpha1.PortSpec{}
+	for port, name := range ports {
+		spec := ewv1alpha1.PortSpec{
+			Port: port,
+		}
+		if name != "" {
+			spec.Name = name
+		}
+		portSpecs = append(portSpecs, spec)
+	}
+
+	wIps := []ewv1alpha1.WorkloadIP{}
+	for _, ip := range ips {
+		wIps = append(wIps, ewv1alpha1.WorkloadIP{Ip: ip})
+	}
+
+	return &ewv1alpha1.ExternalWorkload{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "wkld1",
+			Namespace: "ns",
+			Labels:    labels,
+		},
+		Spec: ewv1alpha1.ExternalWorkloadSpec{
+			MeshTls: ewv1alpha1.MeshTls{
+				Identity:   "some-identity",
+				ServerName: "some-sni",
+			},
+			Ports:       portSpecs,
+			WorkloadIPs: wIps,
+		},
+	}
+}
+
+// Test diffing logic that determines if two workloads with the same name and
+// namespace have changed enough to warrant reconciliation
+func TestWorkloadSpecChanged(t *testing.T) {
+	for _, tt := range []struct {
+		name          string
+		old           *ewv1alpha1.ExternalWorkload
+		updated       *ewv1alpha1.ExternalWorkload
+		expectChanged bool
+	}{
+		{
+			name: "no change",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			expectChanged: false,
+		},
+		{
+			name: "updated workload adds an IP address",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0", "192.0.3.0"},
+			),
+			expectChanged: true,
+		},
+		{
+			name: "updated workload removes an IP address",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0", "192.0.3.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			expectChanged: true,
+		},
+		{
+			name: "updated workload changes an IP address",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.3.0"},
+			),
+			expectChanged: true,
+		},
+		{
+			name: "updated workload adds new port",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1", 2: "port-2"},
+				[]string{"192.0.2.0"},
+			),
+			expectChanged: true,
+		},
+		{
+			name: "updated workload removes port",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1", 2: "port-2"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			expectChanged: true,
+		},
+		{
+			name: "updated workload changes port number",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{2: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			expectChanged: true,
+		},
+		{
+			name: "updated workload changes port name",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-foo"},
+				[]string{"192.0.2.0"},
+			),
+			expectChanged: true,
+		},
+		{
+			name: "updated workload removes port name",
+			old: makeExternalWorkload(
+				nil,
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				nil,
+				map[int32]string{1: ""},
+				[]string{"192.0.2.0"},
+			),
+			expectChanged: true,
+		},
+	} {
+		tt := tt // Pin
+		t.Run(tt.name, func(t *testing.T) {
+			changed := specChanged(tt.old, tt.updated)
+			if tt.expectChanged != changed {
+				t.Errorf("expected changed '%v', got '%v'", tt.expectChanged, changed)
+			}
+		})
+
+	}
+}
+
+// Test diffing logic that determines if two workloads with the same name and
+// namespace have changed enough to warrant reconciliation
+func TestWorkloadServicesToUpdate(t *testing.T) {
+	for _, tt := range []struct {
+		name           string
+		old            *ewv1alpha1.ExternalWorkload
+		updated        *ewv1alpha1.ExternalWorkload
+		k8sConfigs     []string
+		expectServices map[string]struct{}
+	}{
+		{
+			name: "no change",
+			old: makeExternalWorkload(
+				map[string]string{"app": "test"},
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				map[string]string{"app": "test"},
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			k8sConfigs: []string{`
+            apiVersion: v1
+            kind: Service
+            metadata:
+              name: svc-1
+              namespace: ns
+            spec:
+              selector:
+                app: test`,
+			},
+			expectServices: map[string]struct{}{},
+		},
+		{
+			name: "labels and spec have changed",
+			old: makeExternalWorkload(
+				map[string]string{"app": "test-1"},
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				map[string]string{"app": "test-2"},
+				map[int32]string{2: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			k8sConfigs: []string{`
+            apiVersion: v1
+            kind: Service
+            metadata:
+              name: svc-1
+              namespace: ns
+            spec:
+              selector:
+                app: test-1`, `
+            apiVersion: v1
+            kind: Service
+            metadata:
+              name: svc-2
+              namespace: ns
+            spec:
+              selector:
+                app: test-2`,
+			},
+			expectServices: map[string]struct{}{"ns/svc-1": {}, "ns/svc-2": {}},
+		},
+		{
+			name: "spec has changed",
+			old: makeExternalWorkload(
+				map[string]string{"app": "test-1"},
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				map[string]string{"app": "test-1"},
+				map[int32]string{2: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			k8sConfigs: []string{`
+            apiVersion: v1
+            kind: Service
+            metadata:
+              name: svc-1
+              namespace: ns
+            spec:
+              selector:
+                app: test-1`,
+			},
+			expectServices: map[string]struct{}{"ns/svc-1": {}},
+		},
+		{
+			name: "labels have changed",
+			old: makeExternalWorkload(
+				map[string]string{"app": "test-1", "env": "staging"},
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			updated: makeExternalWorkload(
+				map[string]string{"app": "test-1", "env": "prod"},
+				map[int32]string{1: "port-1"},
+				[]string{"192.0.2.0"},
+			),
+			k8sConfigs: []string{`
+            apiVersion: v1
+            kind: Service
+            metadata:
+              name: internal
+              namespace: ns
+            spec:
+              selector:
+                app: test-1`, `
+            apiVersion: v1
+            kind: Service
+            metadata:
+              name: staging
+              namespace: ns
+            spec:
+              selector:
+                env: staging`, `
+            apiVersion: v1
+            kind: Service
+            metadata:
+              name: prod
+              namespace: ns
+            spec:
+              selector:
+                env: prod`,
+			},
+			expectServices: map[string]struct{}{"ns/staging": {}, "ns/prod": {}},
+		}} {
+		tt := tt // Pin
+		t.Run(tt.name, func(t *testing.T) {
+			k8sAPI, err := k8s.NewFakeAPI(tt.k8sConfigs...)
+			if err != nil {
+				t.Fatalf("unexpected error %v", err)
+			}
+
+			ec, err := NewEndpointsController(k8sAPI, "my-hostname", "controlplane-ns", make(chan struct{}), false)
+			if err != nil {
+				t.Fatalf("unexpected error %v", err)
+			}
+
+			ec.Start()
+			k8sAPI.Sync(nil)
+
+			services, err := ec.servicesToUpdate(tt.old, tt.updated)
+			if err != nil {
+				t.Fatalf("unexpected error %v", err)
+			}
+
+			if len(services) != len(tt.expectServices) {
+				t.Fatalf("expected %d services to update, got %d services instead", len(tt.expectServices), len(services))
+			}
+
+			for _, svc := range services {
+				if _, ok := tt.expectServices[svc]; !ok {
+					t.Errorf("unexpected service key %s found in list of results", svc)
+				}
+			}
+		})
+
+	}
+}
diff --git a/controller/api/destination/watcher/endpoints_watcher_test.go b/controller/api/destination/watcher/endpoints_watcher_test.go
index afc3af6fb..9a60e7a06 100644
--- a/controller/api/destination/watcher/endpoints_watcher_test.go
+++ b/controller/api/destination/watcher/endpoints_watcher_test.go
@@ -1340,6 +1340,418 @@ status:
 	}
 }
 
+func TestEndpointsWatcherWithEndpointSlicesExternalWorkload(t *testing.T) {
+	for _, tt := range []struct {
+		serviceType                      string
+		k8sConfigs                       []string
+		id                               ServiceID
+		hostname                         string
+		port                             Port
+		expectedAddresses                []string
+		expectedNoEndpoints              bool
+		expectedNoEndpointsServiceExists bool
+		expectedError                    bool
+		expectedLocalTrafficPolicy       bool
+	}{
+		{
+			serviceType: "local services with EndpointSlice",
+			k8sConfigs: []string{`
+kind: APIResourceList
+apiVersion: v1
+groupVersion: discovery.k8s.io/v1
+resources:
+  - name: endpointslices
+    singularName: endpointslice
+    namespaced: true
+    kind: EndpointSlice
+    verbs:
+      - delete
+      - deletecollection
+      - get
+      - list
+      - patch
+      - create
+      - update
+      - watch
+`, `
+apiVersion: v1
+kind: Service
+metadata:
+  name: name-1
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8989
+  internalTrafficPolicy: Local`,
+				`
+addressType: IPv4
+apiVersion: discovery.k8s.io/v1
+endpoints:
+- addresses:
+  - 172.17.0.12
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name-1-1
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+- addresses:
+  - 172.17.0.19
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name-1-2
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+- addresses:
+  - 172.17.0.20
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name-1-3
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-2
+- addresses:
+  - 172.17.0.21
+  conditions:
+    ready: true
+  topology:
+    kubernetes.io/hostname: node-2
+kind: EndpointSlice
+metadata:
+  labels:
+    kubernetes.io/service-name: name-1
+  name: name-1-bhnqh
+  namespace: ns
+ports:
+- name: ""
+  port: 8989`,
+				`
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name-1-1
+  namespace: ns
+status:
+  conditions:
+  ready: true`,
+				`
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name-1-2
+  namespace: ns
+status:
+  conditions:
+  ready: true`,
+				`
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name-1-3
+  namespace: ns
+status:
+  conditions:
+  ready: true`,
+			},
+			id:   ExternalWorkloadID{Name: "name-1", Namespace: "ns"},
+			port: 8989,
+			expectedAddresses: []string{
+				"172.17.0.12:8989",
+				"172.17.0.19:8989",
+				"172.17.0.20:8989",
+				"172.17.0.21:8989",
+			},
+			expectedNoEndpoints:              false,
+			expectedNoEndpointsServiceExists: false,
+			expectedError:                    false,
+			expectedLocalTrafficPolicy:       true,
+		},
+		{
+			serviceType: "local services with missing addresses and EndpointSlice",
+			k8sConfigs: []string{`
+kind: APIResourceList
+apiVersion: v1
+groupVersion: discovery.k8s.io/v1
+resources:
+  - name: endpointslices
+    singularName: endpointslice
+    namespaced: true
+    kind: EndpointSlice
+    verbs:
+      - delete
+      - deletecollection
+      - get
+      - list
+      - patch
+      - create
+      - update
+      - watch
+`, `
+apiVersion: v1
+kind: Service
+metadata:
+  name: name-1
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8989`, `
+addressType: IPv4
+apiVersion: discovery.k8s.io/v1
+endpoints:
+- addresses:
+  - 172.17.0.23
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name-1-1
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+- addresses:
+  - 172.17.0.24
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name-1-2
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+- addresses:
+  - 172.17.0.25
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name-1-3
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-2
+kind: EndpointSlice
+metadata:
+  labels:
+    kubernetes.io/service-name: name-1
+  name: name1-f5fad
+  namespace: ns
+ports:
+- name: ""
+  port: 8989`, `
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name-1-3
+  namespace: ns
+status:
+  conditions:
+  ready: true`,
+			},
+			id:                               ServiceID{Name: "name-1", Namespace: "ns"},
+			port:                             8989,
+			expectedAddresses:                []string{"172.17.0.25:8989"},
+			expectedNoEndpoints:              false,
+			expectedNoEndpointsServiceExists: false,
+			expectedError:                    false,
+		},
+		{
+			serviceType: "service with EndpointSlice without labels",
+			k8sConfigs: []string{`
+kind: APIResourceList
+apiVersion: v1
+groupVersion: discovery.k8s.io/v1
+resources:
+  - name: endpointslices
+    singularName: endpointslice
+    namespaced: true
+    kind: EndpointSlice
+    verbs:
+      - delete
+      - deletecollection
+      - get
+      - list
+      - patch
+      - create
+      - update
+      - watch
+`, `
+apiVersion: v1
+kind: Service
+metadata:
+  name: name-5
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8989`, `
+addressType: IPv4
+apiVersion: discovery.k8s.io/v1
+endpoints:
+- addresses:
+  - 172.17.0.12
+  conditions:
+    ready: true
+  hostname: name-1-1
+  targetRef:
+    kind: ExternalWorkload
+    name: name-1-1
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+kind: EndpointSlice
+metadata:
+  labels:
+  name: name-1-f5fad
+  namespace: ns
+ports:
+- name: ""
+  port: 8989`, `
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name-1-1
+  namespace: ns
+status:
+  conditions:
+  ready: true`,
+			},
+			id:                               ServiceID{Name: "name-5", Namespace: "ns"},
+			port:                             8989,
+			expectedAddresses:                []string{},
+			expectedNoEndpoints:              true,
+			expectedNoEndpointsServiceExists: true,
+			expectedError:                    false,
+		},
+
+		{
+			serviceType: "service with IPv6 address type EndpointSlice",
+			k8sConfigs: []string{`
+kind: APIResourceList
+apiVersion: v1
+groupVersion: discovery.k8s.io/v1
+resources:
+  - name: endpointslices
+    singularName: endpointslice
+    namespaced: true
+    kind: EndpointSlice
+    verbs:
+      - delete
+      - deletecollection
+      - get
+      - list
+      - patch
+      - create
+      - update
+      - watch
+`, `
+apiVersion: v1
+kind: Service
+metadata:
+  name: name-5
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 9000`, `
+addressType: IPv6
+apiVersion: discovery.k8s.io/v1
+endpoints:
+- addresses:
+  - 0:0:0:0:0:0:0:1
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name-5-1
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+kind: EndpointSlice
+metadata:
+  labels:
+  name: name-5-f65dv
+  namespace: ns
+  ownerReferences:
+  - apiVersion: v1
+    kind: Service
+    name: name-5
+ports:
+- name: ""
+  port: 9000`, `
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name-5-1
+  namespace: ns
+status:
+  conditions:
+  ready: true`,
+			},
+			id:                               ServiceID{Name: "name-5", Namespace: "ns"},
+			port:                             9000,
+			expectedAddresses:                []string{},
+			expectedNoEndpoints:              true,
+			expectedNoEndpointsServiceExists: true,
+			expectedError:                    false,
+		},
+	} {
+		tt := tt // pin
+		t.Run("subscribes listener to "+tt.serviceType, func(t *testing.T) {
+			k8sAPI, err := k8s.NewFakeAPI(tt.k8sConfigs...)
+			if err != nil {
+				t.Fatalf("NewFakeAPI returned an error: %s", err)
+			}
+
+			metadataAPI, err := k8s.NewFakeMetadataAPI(nil)
+			if err != nil {
+				t.Fatalf("NewFakeMetadataAPI returned an error: %s", err)
+			}
+
+			watcher, err := NewEndpointsWatcher(k8sAPI, metadataAPI, logging.WithField("test", t.Name()), true, "local")
+			if err != nil {
+				t.Fatalf("can't create Endpoints watcher: %s", err)
+			}
+
+			k8sAPI.Sync(nil)
+			metadataAPI.Sync(nil)
+
+			listener := newBufferingEndpointListener()
+
+			err = watcher.Subscribe(tt.id, tt.port, tt.hostname, listener)
+			if tt.expectedError && err == nil {
+				t.Fatal("Expected error but was ok")
+			}
+			if !tt.expectedError && err != nil {
+				t.Fatalf("Expected no error, got [%s]", err)
+			}
+
+			if listener.localTrafficPolicy != tt.expectedLocalTrafficPolicy {
+				t.Fatalf("Expected localTrafficPolicy [%v], got [%v]", tt.expectedLocalTrafficPolicy, listener.localTrafficPolicy)
+			}
+
+			listener.ExpectAdded(tt.expectedAddresses, t)
+
+			if listener.endpointsAreNotCalled() != tt.expectedNoEndpoints {
+				t.Fatalf("Expected noEndpointsCalled to be [%t], got [%t]",
+					tt.expectedNoEndpoints, listener.endpointsAreNotCalled())
+			}
+
+			if listener.endpointsDoNotExist() != tt.expectedNoEndpointsServiceExists {
+				t.Fatalf("Expected noEndpointsExist to be [%t], got [%t]",
+					tt.expectedNoEndpointsServiceExists, listener.endpointsDoNotExist())
+			}
+		})
+	}
+}
+
 func TestEndpointsWatcherDeletion(t *testing.T) {
 	k8sConfigs := []string{`
 apiVersion: v1
@@ -1570,7 +1982,177 @@ status:
 			id:                ServiceID{Name: "name1", Namespace: "ns"},
 			port:              8989,
 			hostname:          "name1-1",
-			objectToDelete:    createTestEndpointSlice(),
+			objectToDelete:    createTestEndpointSlice(consts.PodKind),
+			hasSliceAccess:    true,
+			noEndpointsCalled: true,
+		},
+		{
+			serviceType:       "can delete an EndpointSlice when wrapped in a DeletedFinalStateUnknown",
+			k8sConfigs:        k8sConfigsWithES,
+			id:                ServiceID{Name: "name1", Namespace: "ns"},
+			port:              8989,
+			hostname:          "name1-1",
+			objectToDelete:    createTestEndpointSlice(consts.PodKind),
+			hasSliceAccess:    true,
+			noEndpointsCalled: true,
+		},
+		{
+			serviceType:       "can delete an EndpointSlice when there are multiple ones",
+			k8sConfigs:        k8sConfigWithMultipleES,
+			id:                ServiceID{Name: "name1", Namespace: "ns"},
+			port:              8989,
+			hostname:          "name1-1",
+			objectToDelete:    createTestEndpointSlice(consts.PodKind),
+			hasSliceAccess:    true,
+			noEndpointsCalled: false,
+		},
+	} {
+		tt := tt // pin
+		t.Run("subscribes listener to "+tt.serviceType, func(t *testing.T) {
+			k8sAPI, err := k8s.NewFakeAPI(tt.k8sConfigs...)
+			if err != nil {
+				t.Fatalf("NewFakeAPI returned an error: %s", err)
+			}
+
+			metadataAPI, err := k8s.NewFakeMetadataAPI(nil)
+			if err != nil {
+				t.Fatalf("NewFakeMetadataAPI returned an error: %s", err)
+			}
+
+			watcher, err := NewEndpointsWatcher(k8sAPI, metadataAPI, logging.WithField("test", t.Name()), true, "local")
+			if err != nil {
+				t.Fatalf("can't create Endpoints watcher: %s", err)
+			}
+
+			k8sAPI.Sync(nil)
+			metadataAPI.Sync(nil)
+
+			listener := newBufferingEndpointListener()
+
+			err = watcher.Subscribe(tt.id, tt.port, tt.hostname, listener)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			watcher.deleteEndpointSlice(tt.objectToDelete)
+
+			if listener.endpointsAreNotCalled() != tt.noEndpointsCalled {
+				t.Fatalf("Expected noEndpointsCalled to be [%t], got [%t]",
+					tt.noEndpointsCalled, listener.endpointsAreNotCalled())
+			}
+		})
+	}
+}
+
+func TestEndpointsWatcherDeletionWithEndpointSlicesExternalWorkload(t *testing.T) {
+	k8sConfigsWithES := []string{`
+kind: APIResourceList
+apiVersion: v1
+groupVersion: discovery.k8s.io/v1
+resources:
+  - name: endpointslices
+    singularName: endpointslice
+    namespaced: true
+    kind: EndpointSlice
+    verbs:
+      - delete
+      - deletecollection
+      - get
+      - list
+      - patch
+      - create
+      - update
+      - watch
+`, `
+apiVersion: v1
+kind: Service
+metadata:
+  name: name1
+  namespace: ns
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8989`, `
+addressType: IPv4
+apiVersion: discovery.k8s.io/v1
+endpoints:
+- addresses:
+  - 172.17.0.12
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name1-1
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+kind: EndpointSlice
+metadata:
+  labels:
+    kubernetes.io/service-name: name1
+  name: name1-del
+  namespace: ns
+ports:
+- name: ""
+  port: 8989`, `
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name1-1
+  namespace: ns
+status:
+  conditions:
+  ready: true`}
+
+	k8sConfigWithMultipleES := append(k8sConfigsWithES, []string{`
+addressType: IPv4
+apiVersion: discovery.k8s.io/v1
+endpoints:
+- addresses:
+  - 172.17.0.13
+  conditions:
+    ready: true
+  targetRef:
+    kind: ExternalWorkload
+    name: name1-2
+    namespace: ns
+  topology:
+    kubernetes.io/hostname: node-1
+kind: EndpointSlice
+metadata:
+  labels:
+    kubernetes.io/service-name: name1
+  name: name1-live
+  namespace: ns
+ports:
+- name: ""
+  port: 8989`, `apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: name1-2
+  namespace: ns
+status:
+  conditions:
+  ready: true`}...)
+
+	for _, tt := range []struct {
+		serviceType       string
+		k8sConfigs        []string
+		id                ServiceID
+		hostname          string
+		port              Port
+		objectToDelete    interface{}
+		deletingServices  bool
+		hasSliceAccess    bool
+		noEndpointsCalled bool
+	}{
+		{
+			serviceType:       "can delete an EndpointSlice",
+			k8sConfigs:        k8sConfigsWithES,
+			id:                ServiceID{Name: "name1", Namespace: "ns"},
+			port:              8989,
+			hostname:          "name1-1",
+			objectToDelete:    createTestEndpointSlice(consts.ExtWorkloadKind),
 			hasSliceAccess:    true,
 			noEndpointsCalled: true,
 		},
@@ -1580,7 +2162,7 @@ status:
 			id:                ServiceID{Name: "name1", Namespace: "ns"},
 			port:              8989,
 			hostname:          "name1-1",
-			objectToDelete:    createTestEndpointSlice(),
+			objectToDelete:    createTestEndpointSlice(consts.ExtWorkloadKind),
 			hasSliceAccess:    true,
 			noEndpointsCalled: true,
 		},
@@ -1590,7 +2172,7 @@ status:
 			id:                ServiceID{Name: "name1", Namespace: "ns"},
 			port:              8989,
 			hostname:          "name1-1",
-			objectToDelete:    createTestEndpointSlice(),
+			objectToDelete:    createTestEndpointSlice(consts.ExtWorkloadKind),
 			hasSliceAccess:    true,
 			noEndpointsCalled: false,
 		},
@@ -1927,7 +2509,7 @@ func endpoints(identity string) *corev1.Endpoints {
 	}
 }
 
-func createTestEndpointSlice() *dv1.EndpointSlice {
+func createTestEndpointSlice(targetRefKind string) *dv1.EndpointSlice {
 	return &dv1.EndpointSlice{
 		AddressType: "IPv4",
 		ObjectMeta:  metav1.ObjectMeta{Name: "name1-del", Namespace: "ns", Labels: map[string]string{dv1.LabelServiceName: "name1"}},
@@ -1935,7 +2517,7 @@ func createTestEndpointSlice() *dv1.EndpointSlice {
 			{
 				Addresses:  []string{"172.17.0.12"},
 				Conditions: dv1.EndpointConditions{Ready: func(b bool) *bool { return &b }(true)},
-				TargetRef:  &corev1.ObjectReference{Name: "name1-1", Namespace: "ns", Kind: "Pod"},
+				TargetRef:  &corev1.ObjectReference{Name: "name1-1", Namespace: "ns", Kind: targetRefKind},
 			},
 		},
 		Ports: []dv1.EndpointPort{
diff --git a/controller/k8s/test_helper.go b/controller/k8s/test_helper.go
index 725cb65da..4145ccaaf 100644
--- a/controller/k8s/test_helper.go
+++ b/controller/k8s/test_helper.go
@@ -48,6 +48,7 @@ func NewFakeClusterScopedAPI(clientSet kubernetes.Interface, l5dClientSet l5dcrd
 		MWC,
 		NS,
 		Pod,
+		ExtWorkload,
 		RC,
 		RS,
 		SP,
@@ -57,6 +58,7 @@ func NewFakeClusterScopedAPI(clientSet kubernetes.Interface, l5dClientSet l5dcrd
 		ES,
 		Srv,
 		Secret,
+		ExtWorkload,
 	)
 }
 
diff --git a/policy-test/Cargo.toml b/policy-test/Cargo.toml
index b8505a5a3..9ddc8f144 100644
--- a/policy-test/Cargo.toml
+++ b/policy-test/Cargo.toml
@@ -14,6 +14,7 @@ k8s-gateway-api = "0.15"
 k8s-openapi = { version = "0.20", features = ["v1_22"] }
 linkerd-policy-controller-core = { path = "../policy-controller/core" }
 linkerd-policy-controller-k8s-api = { path = "../policy-controller/k8s/api" }
+linkerd-policy-controller-grpc = { path = "../policy-controller/grpc"}
 maplit = "1"
 rand = "0.8"
 serde = "1"
diff --git a/policy-test/src/grpc.rs b/policy-test/src/grpc.rs
index 58d7903ec..348c3aaa6 100644
--- a/policy-test/src/grpc.rs
+++ b/policy-test/src/grpc.rs
@@ -9,6 +9,7 @@ use linkerd2_proxy_api::{
     inbound::inbound_server_policies_client::InboundServerPoliciesClient,
     outbound::outbound_policies_client::OutboundPoliciesClient,
 };
+use linkerd_policy_controller_grpc::workload;
 use linkerd_policy_controller_k8s_api::{self as k8s, ResourceExt};
 use tokio::io;
 
@@ -58,6 +59,25 @@ macro_rules! assert_protocol_detect {
     }};
 }
 
+#[macro_export]
+macro_rules! assert_protocol_detect_external {
+    ($config:expr) => {{
+        use linkerd2_proxy_api::inbound;
+
+        assert_eq!(
+            $config.protocol,
+            Some(inbound::ProxyProtocol {
+                kind: Some(inbound::proxy_protocol::Kind::Detect(
+                    inbound::proxy_protocol::Detect {
+                        timeout: Some(std::time::Duration::from_secs(10).try_into().unwrap()),
+                        http_routes: vec![$crate::grpc::defaults::http_route()],
+                    }
+                ))
+            })
+        )
+    }};
+}
+
 #[macro_export]
 macro_rules! assert_default_accrual_backoff {
     ($backoff:expr) => {{
@@ -157,6 +177,7 @@ impl InboundPolicyClient {
         Ok(rsp.into_inner())
     }
 
+    //TODO (matei): we should move our tests over to the new token format
     pub async fn watch_port(
         &mut self,
         ns: &str,
@@ -172,6 +193,52 @@ impl InboundPolicyClient {
             .await?;
         Ok(rsp.into_inner())
     }
+
+    //TODO (matei): see if we can collapse this into `get_port` once it supports
+    //new token format
+    pub async fn get_port_for_external_workload(
+        &mut self,
+        ns: &str,
+        name: &str,
+        port: u16,
+    ) -> Result<inbound::Server, tonic::Status> {
+        let token = serde_json::to_string(&workload::Workload {
+            kind: workload::Kind::External(name.into()),
+            namespace: ns.into(),
+        })
+        .unwrap();
+        let rsp = self
+            .client
+            .get_port(tonic::Request::new(inbound::PortSpec {
+                workload: token,
+                port: port as u32,
+            }))
+            .await?;
+
+        Ok(rsp.into_inner())
+    }
+
+    pub async fn watch_port_for_external_workload(
+        &mut self,
+        ns: &str,
+        name: &str,
+        port: u16,
+    ) -> Result<tonic::Streaming<inbound::Server>, tonic::Status> {
+        let token = serde_json::to_string(&workload::Workload {
+            kind: workload::Kind::External(name.into()),
+            namespace: ns.into(),
+        })
+        .unwrap();
+        let rsp = self
+            .client
+            .watch_port(tonic::Request::new(inbound::PortSpec {
+                workload: token,
+                port: port as u32,
+            }))
+            .await?;
+
+        Ok(rsp.into_inner())
+    }
 }
 
 // === impl OutboundPolicyClient ===
@@ -309,6 +376,15 @@ pub mod defaults {
         }
     }
 
+    pub fn proxy_protocol_external() -> inbound::ProxyProtocol {
+        use inbound::proxy_protocol::{Http1, Kind};
+        inbound::ProxyProtocol {
+            kind: Some(Kind::Http1(Http1 {
+                routes: vec![http_route()],
+            })),
+        }
+    }
+
     pub fn http_route() -> inbound::HttpRoute {
         use http_route::{path_match, HttpRouteMatch, PathMatch};
         use inbound::{http_route::Rule, HttpRoute};
diff --git a/policy-test/tests/inbound_api_external_workload.rs b/policy-test/tests/inbound_api_external_workload.rs
new file mode 100644
index 000000000..d4bb9edaf
--- /dev/null
+++ b/policy-test/tests/inbound_api_external_workload.rs
@@ -0,0 +1,464 @@
+use std::{num::NonZeroU16, time::Duration};
+
+use futures::prelude::*;
+use k8s::policy::LocalTargetRef;
+use kube::ResourceExt;
+use linkerd_policy_controller_core::{Ipv4Net, Ipv6Net};
+use linkerd_policy_controller_k8s_api as k8s;
+use linkerd_policy_test::{
+    assert_default_all_unauthenticated_labels, assert_is_default_all_unauthenticated,
+    assert_protocol_detect_external, create, grpc, with_temp_ns,
+};
+use maplit::{btreemap, convert_args, hashmap};
+use tokio::time;
+
+#[tokio::test(flavor = "current_thread")]
+async fn external_workload_srv_with_authorization_policy() {
+    with_temp_ns(|client, ns| async move {
+        // Create an external workload object.
+        let ext_workload = create(&client, mk_external_workload(&ns, "wkld-1")).await;
+
+        tracing::trace!(
+            external_workload = %ext_workload.name_any(),
+            ip = ?ext_workload.spec.workload_ips.as_ref().unwrap()[0]
+        );
+
+        let mut rx = retry_watch_server(&client, &ns, &ext_workload.name_any()).await;
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        tracing::trace!(?config);
+        assert_is_default_all_unauthenticated!(config);
+        assert_protocol_detect_external!(config);
+
+        // Create a server that selects the http port on the workload and ensure
+        // the update now uses this server (which has no authorizations)
+        let server = create(&client, mk_http_server(&ns, "external-http")).await;
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        assert_eq!(
+            config.protocol,
+            Some(grpc::defaults::proxy_protocol_external())
+        );
+        assert_eq!(config.authorizations, vec![]);
+        assert_eq!(
+            config.labels,
+            convert_args!(hashmap!(
+                "group" => "policy.linkerd.io",
+                "kind" => "server",
+                "name" => "external-http"
+            )),
+        );
+
+        // Create a server authorization that refers to the `linkerd-admin`
+        // server (by name) and ensure that the update now reflects this
+        // authorization.
+        create(
+            &client,
+            k8s::policy::AuthorizationPolicy {
+                metadata: kube::api::ObjectMeta {
+                    namespace: Some(ns.clone()),
+                    name: Some("all-http".to_string()),
+                    ..Default::default()
+                },
+                spec: k8s::policy::AuthorizationPolicySpec {
+                    target_ref: LocalTargetRef {
+                        group: Some("policy.linkerd.io".to_string()),
+                        kind: "server".to_string(),
+                        name: server.name_any(),
+                    },
+                    required_authentication_refs: vec![],
+                },
+            },
+        )
+        .await;
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an updated config");
+        tracing::trace!(?config);
+        assert_eq!(
+            config.protocol,
+            Some(grpc::defaults::proxy_protocol_external())
+        );
+        assert_eq!(
+            config.authorizations.first().unwrap().labels,
+            convert_args!(hashmap!(
+                "group" => "policy.linkerd.io",
+                "kind" => "authorizationpolicy",
+                "name" => "all-http",
+            )),
+        );
+        assert_eq!(
+            *config
+                .authorizations
+                .first()
+                .unwrap()
+                .authentication
+                .as_ref()
+                .unwrap(),
+            grpc::inbound::Authn {
+                permit: Some(grpc::inbound::authn::Permit::Unauthenticated(
+                    grpc::inbound::authn::PermitUnauthenticated {}
+                )),
+            }
+        );
+        assert_eq!(
+            config.labels,
+            convert_args!(hashmap!(
+                "group" => "policy.linkerd.io",
+                "kind" => "server",
+                "name" => server.name_unchecked()
+            ))
+        );
+
+        // Delete the `Server` and ensure that the update reverts to the
+        // default.
+        kube::Api::<k8s::policy::Server>::namespaced(client.clone(), &ns)
+            .delete(
+                &server.name_unchecked(),
+                &kube::api::DeleteParams::default(),
+            )
+            .await
+            .expect("Server must be deleted");
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an updated config");
+
+        assert_is_default_all_unauthenticated!(config);
+        assert_protocol_detect_external!(config);
+    })
+    .await
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn external_workload_srv_with_http_route() {
+    with_temp_ns(|client, ns| async move {
+        // Create an external workload object.
+        let ext_workload = create(&client, mk_external_workload(&ns, "wkld-1")).await;
+
+        tracing::trace!(
+            external_workload = %ext_workload.name_any(),
+            ip = ?ext_workload.spec.workload_ips.as_ref().unwrap()[0]
+        );
+
+        let mut rx = retry_watch_server(&client, &ns, &ext_workload.name_any()).await;
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        tracing::trace!(?config);
+        assert_is_default_all_unauthenticated!(config);
+        assert_protocol_detect_external!(config);
+
+        // Create a server that selects the http port on the workload and ensure
+        // the update now uses this server (which has no authorizations)
+        let server = create(&client, mk_http_server(&ns, "external-http")).await;
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        assert_eq!(
+            config.protocol,
+            Some(grpc::defaults::proxy_protocol_external())
+        );
+        assert_eq!(config.authorizations, vec![]);
+        assert_eq!(
+            config.labels,
+            convert_args!(hashmap!(
+                "group" => "policy.linkerd.io",
+                "kind" => "server",
+                "name" => "external-http"
+            )),
+        );
+
+        let created_route = {
+            use k8s::policy::httproute as api;
+            let http_route = api::HttpRoute {
+                metadata: kube::api::ObjectMeta {
+                    namespace: Some(ns.to_string()),
+                    name: Some("http-route".to_string()),
+                    ..Default::default()
+                },
+                spec: api::HttpRouteSpec {
+                    inner: api::CommonRouteSpec {
+                        parent_refs: Some(vec![api::ParentReference {
+                            group: Some("policy.linkerd.io".to_string()),
+                            kind: Some("Server".to_string()),
+                            name: server.name_any(),
+                            namespace: None,
+                            section_name: None,
+                            port: None,
+                        }]),
+                    },
+                    hostnames: None,
+                    rules: Some(vec![api::HttpRouteRule {
+                        matches: Some(vec![api::HttpRouteMatch {
+                            path: Some(api::HttpPathMatch::Exact {
+                                value: "/endpoint".to_string(),
+                            }),
+                            headers: None,
+                            query_params: None,
+                            method: Some("GET".to_string()),
+                        }]),
+                        filters: None,
+                        backend_refs: None,
+                        timeouts: None,
+                    }]),
+                },
+                status: None,
+            };
+
+            create(&client, http_route).await
+        };
+
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        let kind = config
+            .protocol
+            .as_ref()
+            .expect("must have proxy protocol")
+            .kind
+            .as_ref()
+            .expect("must have kind");
+        let routes = if let grpc::inbound::proxy_protocol::Kind::Http1(ref http1) = kind {
+            &http1.routes[..]
+        } else {
+            panic!("proxy protocol must be 'Http1'; actually got:\n{kind:#?}");
+        };
+
+        assert_eq!(routes.len(), 1);
+        let route = routes.first().expect("must have route");
+        // Route should have no authz policy by default
+        assert_eq!(route.authorizations, vec![]);
+        assert_eq!(
+            route.metadata.to_owned().expect("route must have metadata"),
+            grpc::meta::Metadata {
+                kind: Some(grpc::meta::metadata::Kind::Resource(grpc::meta::Resource {
+                    group: "policy.linkerd.io".to_string(),
+                    kind: "HTTPRoute".to_string(),
+                    name: "http-route".to_string(),
+                    ..Default::default()
+                }))
+            }
+        );
+
+        // Route has path match
+        let rule_match = route
+            .rules
+            .first()
+            .expect("must have rule")
+            .matches
+            .first()
+            .expect("must have match");
+        assert_eq!(
+            rule_match
+                .path
+                .to_owned()
+                .expect("must have path match")
+                .kind
+                .expect("must have kind"),
+            grpc::http_route::path_match::Kind::Exact("/endpoint".to_string())
+        );
+
+        // Create a network authn and a policy that refers to the route
+        let all_networks = create(
+            &client,
+            k8s::policy::NetworkAuthentication {
+                metadata: kube::api::ObjectMeta {
+                    namespace: Some(ns.clone()),
+                    name: Some("all-net".to_string()),
+                    ..Default::default()
+                },
+                spec: k8s::policy::NetworkAuthenticationSpec {
+                    networks: vec![
+                        k8s::policy::network_authentication::Network {
+                            cidr: Ipv4Net::default().into(),
+                            except: None,
+                        },
+                        k8s::policy::network_authentication::Network {
+                            cidr: Ipv6Net::default().into(),
+                            except: None,
+                        },
+                    ],
+                },
+            },
+        )
+        .await;
+        create(
+            &client,
+            k8s::policy::AuthorizationPolicy {
+                metadata: kube::api::ObjectMeta {
+                    namespace: Some(ns.clone()),
+                    name: Some("all-net".to_string()),
+                    ..Default::default()
+                },
+                spec: k8s::policy::AuthorizationPolicySpec {
+                    target_ref: k8s::policy::LocalTargetRef::from_resource(&created_route),
+                    required_authentication_refs: vec![
+                        k8s::policy::NamespacedTargetRef::from_resource(&all_networks),
+                    ],
+                },
+            },
+        )
+        .await;
+
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        let http1 = if let grpc::inbound::proxy_protocol::Kind::Http1(http1) = config
+            .protocol
+            .expect("must have proxy protocol")
+            .kind
+            .expect("must have kind")
+        {
+            http1
+        } else {
+            panic!("proxy protocol must be HTTP1");
+        };
+        let h1_route = http1.routes.first().expect("must have route");
+        assert_eq!(h1_route.authorizations.len(), 1, "must have authorizations");
+
+        // Delete the `HttpRoute` and ensure that the update reverts to the
+        // default.
+        kube::Api::<k8s::policy::HttpRoute>::namespaced(client.clone(), &ns)
+            .delete("http-route", &kube::api::DeleteParams::default())
+            .await
+            .expect("HttpRoute must be deleted");
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        assert_eq!(
+            config.protocol,
+            Some(grpc::defaults::proxy_protocol_external())
+        );
+    })
+    .await;
+}
+#[tokio::test(flavor = "current_thread")]
+async fn external_workload_default_http_route() {
+    with_temp_ns(|client, ns| async move {
+        // Create an external workload object.
+        let ext_workload = create(&client, mk_external_workload(&ns, "wkld-1")).await;
+
+        tracing::trace!(
+            external_workload = %ext_workload.name_any(),
+            ip = ?ext_workload.spec.workload_ips.as_ref().unwrap()[0]
+        );
+
+        let mut rx = retry_watch_server(&client, &ns, &ext_workload.name_any()).await;
+        let config = rx
+            .next()
+            .await
+            .expect("watch must not fail")
+            .expect("watch must return an initial config");
+        tracing::trace!(?config);
+        assert_is_default_all_unauthenticated!(config);
+        assert_protocol_detect_external!(config);
+
+        let kind = config
+            .protocol
+            .as_ref()
+            .expect("must have proxy protocol")
+            .kind
+            .as_ref()
+            .expect("must have kind");
+        let routes = if let grpc::inbound::proxy_protocol::Kind::Detect(ref detect) = kind {
+            &detect.http_routes[..]
+        } else {
+            panic!("proxy protocol must be 'Detect'; actually got:\n{kind:#?}");
+        };
+
+        assert_eq!(routes.len(), 1);
+        let route_authzs = &routes[0].authorizations;
+        assert_eq!(route_authzs.len(), 0);
+    })
+    .await
+}
+
+fn mk_external_workload(ns: &str, name: &str) -> k8s::external_workload::ExternalWorkload {
+    k8s::external_workload::ExternalWorkload {
+        metadata: k8s::ObjectMeta {
+            namespace: Some(ns.into()),
+            name: Some(name.into()),
+            labels: Some(convert_args!(btreemap!(
+                        "app" => "ext",
+            ))),
+            ..Default::default()
+        },
+        spec: k8s::external_workload::ExternalWorkloadSpec {
+            mesh_tls: k8s::external_workload::MeshTls {
+                identity: "some-identity".to_string(),
+                server_name: "some-sni".to_string(),
+            },
+            ports: Some(vec![k8s::external_workload::PortSpec {
+                name: Some("http".into()),
+                port: NonZeroU16::new(80).unwrap(),
+                protocol: Default::default(),
+            }]),
+            workload_ips: Some(vec![k8s::external_workload::WorkloadIP {
+                ip: "192.0.2.0".to_string(),
+            }]),
+        },
+        status: None,
+    }
+}
+
+async fn retry_watch_server(
+    client: &kube::Client,
+    ns: &str,
+    workload_name: &str,
+) -> tonic::Streaming<grpc::inbound::Server> {
+    let mut policy_api = grpc::InboundPolicyClient::port_forwarded(client).await;
+    loop {
+        match policy_api
+            .watch_port_for_external_workload(ns, workload_name, 80)
+            .await
+        {
+            Ok(rx) => return rx,
+            Err(error) => {
+                tracing::error!(
+                    ?error,
+                    ns,
+                    workload_name,
+                    "failed to watch policy for port 80"
+                );
+                time::sleep(Duration::from_secs(1)).await;
+            }
+        }
+    }
+}
+
+fn mk_http_server(ns: &str, name: &str) -> k8s::policy::Server {
+    k8s::policy::Server {
+        metadata: k8s::ObjectMeta {
+            namespace: Some(ns.to_string()),
+            name: Some(name.to_string()),
+            ..Default::default()
+        },
+        spec: k8s::policy::ServerSpec {
+            selector: k8s::policy::server::Selector::ExternalWorkload(
+                k8s::labels::Selector::default(),
+            ),
+            port: k8s::policy::server::Port::Name("http".to_string()),
+            proxy_protocol: Some(k8s::policy::server::ProxyProtocol::Http1),
+        },
+    }
+}