Repository: linkerd/linkerd2
Pull request #11952 (2024-01-19 01:31:42)

Add support for ExternalWorkloads in endpoint profiles
When a meshed client attempts to establish a connection directly to the workload IP of an ExternalWorkload, the destination controller should return an endpoint profile for that ExternalWorkload with a single endpoint and the metadata associated with that ExternalWorkload including:
* mesh TLS identity
* workload metric labels
* opaque / protocol hints

<!--  Thanks for sending a pull request!

If you already have a well-structured git commit message, chances are GitHub
set the title and description of this PR to the git commit message subject and
body, respectively. If so, you may delete these instructions and submit your PR.

If this is your first time, please read our contributor guide:
https://github.com/linkerd/linkerd2/blob/main/CONTRIBUTING.md

The title and description of your Pull Request should match the git commit
subject and body, respectively. Git commit messages are structured as follows:

```
Subject

Problem

Solution

Validation

Fixes #[GitHub issue ID]

DCO Sign off
```

Example git commit message:

```
Introduce Pull Request Template

GitHub's community guidelines recommend a pull request template, the repo was
lacking one.

Introduce a `PULL_REQUEST_TEMPLATE.md` file.

Once merged, the
[Community profile checklist](https://github.com/linkerd/linkerd2/community)
should indicate the repo now provides a pull request template.

Fixes #3321

Signed-off-by: Jane Smith <jane.smith@example.com>
```

Note the git commit message subject becomes the pull request title.

For more details around git commits, see the section on Committing in our
contributor guide:
https://github.com/linkerd/linkerd2/blob/main/CONTRIBUTING.md#committing
-->



Patch:
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 2b4bc30c4..c4a3811d7 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@62f4729b5df35e6e0e01265fa70a82ccaf196b4b
+      - uses: tj-actions/changed-files@ae82ed4ae04587b665efad2f206578aa6f0e8539
         id: changed
         with:
           files: |
@@ -287,7 +287,7 @@ jobs:
     timeout-minutes: 5
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
-      - uses: tj-actions/changed-files@62f4729b5df35e6e0e01265fa70a82ccaf196b4b
+      - uses: tj-actions/changed-files@ae82ed4ae04587b665efad2f206578aa6f0e8539
         id: changed
         with:
           files: |
@@ -330,7 +330,7 @@ jobs:
     timeout-minutes: 5
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
-      - uses: tj-actions/changed-files@62f4729b5df35e6e0e01265fa70a82ccaf196b4b
+      - uses: tj-actions/changed-files@ae82ed4ae04587b665efad2f206578aa6f0e8539
         id: changed
         with:
           files: |
diff --git a/.github/workflows/markdown.yml b/.github/workflows/markdown.yml
index 2b2721dd4..112112b8a 100644
--- a/.github/workflows/markdown.yml
+++ b/.github/workflows/markdown.yml
@@ -15,7 +15,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
-      - uses: DavidAnson/markdownlint-cli2-action@455b6612a7b7a80f28be9e019b70abdd11696e4e
+      - uses: DavidAnson/markdownlint-cli2-action@510b996878fc0d1a46c8a04ec86b06dbfba09de7
         with:
           globs: |
             **/*.md
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 9293796e9..095a10c9b 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -36,7 +36,7 @@ jobs:
     continue-on-error: ${{ matrix.checks == 'advisories' }}
     steps:
     - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11
-    - uses: EmbarkStudios/cargo-deny-action@1e59595bed8fc55c969333d08d7817b36888f0c5
+    - uses: EmbarkStudios/cargo-deny-action@1350841d0991444487fd89747f1e39de5349455f
       with:
         command: check ${{ matrix.checks }}
 
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 bc4f23a43..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",
@@ -1589,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",
@@ -1602,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",
@@ -2562,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/RELEASE.md b/RELEASE.md
index f612415de..d37a42934 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -24,27 +24,26 @@ bin/git-commit-proxy-version <linkerd2-proxy-sha>
 The script will update the `.proxy-version` file. Submit a PR to obtain reviews
 and approval.
 
-## 2. Bump the proxy-init version
+## 2. Bump the proxy-init or CNI plugin version
 
-If the `linkerd2/proxy-init` project has a new release (which is rare), the
-following updates are needed:
-
-- `go.mod`
-
-   ```mod
-   github.com/linkerd/linkerd2-proxy-init v1.2.0
-   ```
+If the `linkerd2/proxy-init` or `linkerd2/cni-plugin` projects have a new
+release (which is rare), the following updates are needed:
 
 - `pkg/version/version.go` (this also implies changes in unit test fixtures)
 
    ```go
-   var ProxyInitVersion = "v1.2.0"
+   var ProxyInitVersion = "v2.2.3"
+   var LinkerdCNIVersion = "v1.3.0"
    ```
 
 - `charts/linkerd-control-plane/values.yaml`
 
    Upgrade the version in `global.proxyInit.image.version`
 
+- `charts/linkerd2-cni/values.yaml`
+
+   Upgrade the version in `image.version`
+
 Create a new branch in the `linkerd2` repo,
 `username/proxy-init-version-bump`.
 
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/controller/api/destination/endpoint_profile_translator.go b/controller/api/destination/endpoint_profile_translator.go
index fea388e6f..7352a4025 100644
--- a/controller/api/destination/endpoint_profile_translator.go
+++ b/controller/api/destination/endpoint_profile_translator.go
@@ -111,7 +111,12 @@ func (ept *endpointProfileTranslator) Update(address *watcher.Address) error {
 }
 
 func (ept *endpointProfileTranslator) update(address *watcher.Address) {
-	opaquePorts := watcher.GetAnnotatedOpaquePorts(address.Pod, ept.defaultOpaquePorts)
+	var opaquePorts map[uint32]struct{}
+	if address.Pod != nil {
+		opaquePorts = watcher.GetAnnotatedOpaquePorts(address.Pod, ept.defaultOpaquePorts)
+	} else {
+		opaquePorts = watcher.GetAnnotatedOpaquePortsForExternalWorkload(address.ExternalWorkload, ept.defaultOpaquePorts)
+	}
 	endpoint, err := ept.createEndpoint(*address, opaquePorts)
 	if err != nil {
 		ept.log.Errorf("Failed to create endpoint for %s:%d: %s",
@@ -139,7 +144,13 @@ func (ept *endpointProfileTranslator) update(address *watcher.Address) {
 }
 
 func (ept *endpointProfileTranslator) createEndpoint(address watcher.Address, opaquePorts map[uint32]struct{}) (*pb.WeightedAddr, error) {
-	weightedAddr, err := createWeightedAddr(address, opaquePorts, ept.enableH2Upgrade, ept.identityTrustDomain, ept.controllerNS)
+	var weightedAddr *pb.WeightedAddr
+	var err error
+	if address.ExternalWorkload != nil {
+		weightedAddr, err = createWeightedAddrForExternalWorkload(address, opaquePorts)
+	} else {
+		weightedAddr, err = createWeightedAddr(address, opaquePorts, ept.enableH2Upgrade, ept.identityTrustDomain, ept.controllerNS)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -148,6 +159,8 @@ func (ept *endpointProfileTranslator) createEndpoint(address watcher.Address, op
 	// metadata, so it needs to be special-cased.
 	if address.Pod != nil {
 		weightedAddr.MetricLabels["namespace"] = address.Pod.Namespace
+	} else if address.ExternalWorkload != nil {
+		weightedAddr.MetricLabels["namespace"] = address.ExternalWorkload.Namespace
 	}
 
 	return weightedAddr, err
diff --git a/controller/api/destination/endpoint_translator.go b/controller/api/destination/endpoint_translator.go
index 88acf8078..d1018d5f9 100644
--- a/controller/api/destination/endpoint_translator.go
+++ b/controller/api/destination/endpoint_translator.go
@@ -13,7 +13,6 @@ import (
 	"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"
@@ -385,7 +384,7 @@ func (et *endpointTranslator) sendClientAdd(set watcher.AddressSet) {
 				continue
 			}
 		} else if address.ExternalWorkload != nil {
-			opaquePorts = getAnnotatedOpaquePortsForExternalWorkload(address.ExternalWorkload, et.defaultOpaquePorts)
+			opaquePorts = watcher.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)
@@ -699,53 +698,3 @@ func getInboundPortFromExternalWorkload(ewSpec *ewv1alpha1.ExternalWorkloadSpec)
 
 	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
index 9c2a7b23c..eafbd7ee1 100644
--- a/controller/api/destination/external-workload/endpoints_controller.go
+++ b/controller/api/destination/external-workload/endpoints_controller.go
@@ -63,14 +63,18 @@ type EndpointsController struct {
 //
 // 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{}) (*EndpointsController, error) {
-	// TODO: pass in a metrics provider to the queue config
+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(), workqueue.RateLimitingQueueConfig{
-			Name: "endpoints_controller_workqueue",
-		}),
-		stop: stopCh,
+		queue:  workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workQueueConfig),
+		stop:   stopCh,
 		log: logging.WithFields(logging.Fields{
 			"component": "external-endpoints-controller",
 		}),
@@ -358,7 +362,7 @@ func (ec *EndpointsController) updateService(obj interface{}) {
 	ec.queue.Add(key)
 }
 
-func isReady(ew *ewv1alpha1.ExternalWorkload) bool {
+func IsReady(ew *ewv1alpha1.ExternalWorkload) bool {
 	if len(ew.Status.Conditions) == 0 {
 		return false
 	}
@@ -401,7 +405,7 @@ func labelsChanged(old, updated *ewv1alpha1.ExternalWorkload) bool {
 // 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) {
+	if IsReady(old) != IsReady(updated) {
 		return true
 	}
 
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/server.go b/controller/api/destination/server.go
index 84f94d9d0..d62fb5fe9 100644
--- a/controller/api/destination/server.go
+++ b/controller/api/destination/server.go
@@ -42,7 +42,7 @@ type (
 
 		config Config
 
-		pods         *watcher.PodWatcher
+		workloads    *watcher.WorkloadWatcher
 		endpoints    *watcher.EndpointsWatcher
 		opaquePorts  *watcher.OpaquePortsWatcher
 		profiles     *watcher.ProfileWatcher
@@ -86,7 +86,7 @@ func NewServer(
 		return nil, err
 	}
 
-	pods, err := watcher.NewPodWatcher(k8sAPI, metadataAPI, log, config.DefaultOpaquePorts)
+	workloads, err := watcher.NewWorkloadWatcher(k8sAPI, metadataAPI, log, config.DefaultOpaquePorts)
 	if err != nil {
 		return nil, err
 	}
@@ -106,7 +106,7 @@ func NewServer(
 	srv := server{
 		pb.UnimplementedDestinationServer{},
 		config,
-		pods,
+		workloads,
 		endpoints,
 		opaquePorts,
 		profiles,
@@ -507,11 +507,11 @@ func (s *server) subscribeToEndpointProfile(
 	defer translator.Stop()
 
 	var err error
-	ip, err = s.pods.Subscribe(service, hostname, ip, port, translator)
+	ip, err = s.workloads.Subscribe(service, hostname, ip, port, translator)
 	if err != nil {
 		return err
 	}
-	defer s.pods.Unsubscribe(ip, port, translator)
+	defer s.workloads.Unsubscribe(ip, port, translator)
 
 	select {
 	case <-s.shutdown:
diff --git a/controller/api/destination/watcher/endpoints_watcher.go b/controller/api/destination/watcher/endpoints_watcher.go
index e127d59a5..acea2c6f4 100644
--- a/controller/api/destination/watcher/endpoints_watcher.go
+++ b/controller/api/destination/watcher/endpoints_watcher.go
@@ -916,7 +916,7 @@ func (pp *portPublisher) endpointSliceToAddresses(es *discovery.EndpointSlice) A
 					continue
 				}
 
-				err = setToServerProtocolExternalWorkload(pp.k8sAPI, &address, resolvedPort)
+				err = SetToServerProtocolExternalWorkload(pp.k8sAPI, &address, resolvedPort)
 				if err != nil {
 					pp.log.Errorf("failed to set address OpaqueProtocol: %s", err)
 					continue
@@ -1499,7 +1499,7 @@ func SetToServerProtocol(k8sAPI *k8s.API, address *Address, port Port) error {
 
 // 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 {
+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)
 	}
diff --git a/controller/api/destination/watcher/k8s.go b/controller/api/destination/watcher/k8s.go
index ef698921b..5c4f03800 100644
--- a/controller/api/destination/watcher/k8s.go
+++ b/controller/api/destination/watcher/k8s.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net"
 
+	ext "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
 	"github.com/linkerd/linkerd2/controller/k8s"
 	"github.com/prometheus/client_golang/prometheus"
 	corev1 "k8s.io/api/core/v1"
@@ -17,6 +18,8 @@ const (
 	PodIPIndex = "ip"
 	// HostIPIndex is the key for the index based on Host IP of pods with host network enabled
 	HostIPIndex = "hostIP"
+	// ExternalWorkloadIPIndex is the key for the index based on IP of externalworkloads
+	ExternalWorkloadIPIndex = "externalWorkloadIP"
 )
 
 type (
@@ -156,6 +159,25 @@ func InitializeIndexers(k8sAPI *k8s.API) error {
 		return fmt.Errorf("could not create an indexer for pods: %w", err)
 	}
 
+	err = k8sAPI.ExtWorkload().Informer().AddIndexers(cache.Indexers{ExternalWorkloadIPIndex: func(obj interface{}) ([]string, error) {
+		ew, ok := obj.(*ext.ExternalWorkload)
+		if !ok {
+			return nil, errors.New("object is not an externalworkload")
+		}
+
+		addrs := []string{}
+		for _, ip := range ew.Spec.WorkloadIPs {
+			for _, port := range ew.Spec.Ports {
+				addrs = append(addrs, net.JoinHostPort(ip.Ip, fmt.Sprintf("%d", port.Port)))
+			}
+		}
+		return addrs, nil
+	}})
+
+	if err != nil {
+		return fmt.Errorf("could not create an indexer for externalworkloads: %w", err)
+	}
+
 	return nil
 }
 
@@ -175,6 +197,19 @@ func getIndexedPods(k8sAPI *k8s.API, indexName string, key string) ([]*corev1.Po
 	return pods, nil
 }
 
+func getIndexedExternalWorkloads(k8sAPI *k8s.API, indexName string, key string) ([]*ext.ExternalWorkload, error) {
+	objs, err := k8sAPI.ExtWorkload().Informer().GetIndexer().ByIndex(indexName, key)
+	if err != nil {
+		return nil, fmt.Errorf("failed getting %s indexed externalworkloads: %w", indexName, err)
+	}
+	workloads := make([]*ext.ExternalWorkload, 0)
+	for _, obj := range objs {
+		workload := obj.(*ext.ExternalWorkload)
+		workloads = append(workloads, workload)
+	}
+	return workloads, nil
+}
+
 func podNotTerminating(pod *corev1.Pod) bool {
 	phase := pod.Status.Phase
 	podTerminated := phase == corev1.PodSucceeded || phase == corev1.PodFailed
diff --git a/controller/api/destination/watcher/pod_watcher.go b/controller/api/destination/watcher/pod_watcher.go
deleted file mode 100644
index 382f0c25d..000000000
--- a/controller/api/destination/watcher/pod_watcher.go
+++ /dev/null
@@ -1,583 +0,0 @@
-package watcher
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/linkerd/linkerd2/controller/gen/apis/server/v1beta2"
-	"github.com/linkerd/linkerd2/controller/k8s"
-	consts "github.com/linkerd/linkerd2/pkg/k8s"
-	"github.com/linkerd/linkerd2/pkg/util"
-	"github.com/prometheus/client_golang/prometheus"
-	logging "github.com/sirupsen/logrus"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/client-go/tools/cache"
-)
-
-type (
-	// PodWatcher watches all pods in the cluster. It keeps a map of publishers
-	// keyed by IP and port.
-	PodWatcher struct {
-		defaultOpaquePorts map[uint32]struct{}
-		k8sAPI             *k8s.API
-		metadataAPI        *k8s.MetadataAPI
-		publishers         map[IPPort]*podPublisher
-		log                *logging.Entry
-
-		mu sync.RWMutex
-	}
-
-	// podPublisher represents an ip:port along with the backing pod (if any).
-	// It keeps a list of listeners to be notified whenever the pod or the
-	// associated opaque protocol config changes.
-	podPublisher struct {
-		defaultOpaquePorts map[uint32]struct{}
-		k8sAPI             *k8s.API
-		metadataAPI        *k8s.MetadataAPI
-		ip                 string
-		port               Port
-		pod                *corev1.Pod
-		listeners          []PodUpdateListener
-		metrics            metrics
-		log                *logging.Entry
-
-		mu sync.RWMutex
-	}
-
-	// PodUpdateListener is the interface subscribers must implement.
-	PodUpdateListener interface {
-		Update(*Address) error
-	}
-)
-
-var ipPortVecs = newMetricsVecs("ip_port", []string{"ip", "port"})
-
-func NewPodWatcher(k8sAPI *k8s.API, metadataAPI *k8s.MetadataAPI, log *logging.Entry, defaultOpaquePorts map[uint32]struct{}) (*PodWatcher, error) {
-	pw := &PodWatcher{
-		defaultOpaquePorts: defaultOpaquePorts,
-		k8sAPI:             k8sAPI,
-		metadataAPI:        metadataAPI,
-		publishers:         make(map[IPPort]*podPublisher),
-		log: log.WithFields(logging.Fields{
-			"component": "pod-watcher",
-		}),
-	}
-
-	_, err := k8sAPI.Pod().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    pw.addPod,
-		DeleteFunc: pw.deletePod,
-		UpdateFunc: pw.updatePod,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	_, err = k8sAPI.Srv().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    pw.updateServers,
-		DeleteFunc: pw.updateServers,
-		UpdateFunc: pw.updateServer,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return pw, nil
-}
-
-// Subscribe notifies the listener on changes on any pod backing the passed
-// host/ip:port or changes to its associated opaque protocol config. If service
-// and hostname are empty then ip should be set and vice-versa. If ip is empty,
-// the corresponding ip is found for the given service/hostname, and returned.
-func (pw *PodWatcher) Subscribe(service *ServiceID, hostname, ip string, port Port, listener PodUpdateListener) (string, error) {
-	if hostname != "" {
-		pw.log.Debugf("Establishing watch on pod %s.%s.%s:%d", hostname, service.Name, service.Namespace, port)
-	} else if service != nil {
-		pw.log.Debugf("Establishing watch on pod %s.%s:%d", service.Name, service.Namespace, port)
-	} else {
-		pw.log.Debugf("Establishing watch on pod %s:%d", ip, port)
-	}
-	pp, err := pw.getOrNewPodPublisher(service, hostname, ip, port)
-	if err != nil {
-		return "", err
-	}
-
-	pp.subscribe(listener)
-
-	address, err := pp.createAddress()
-	if err != nil {
-		return "", err
-	}
-
-	if err = listener.Update(&address); err != nil {
-		return "", fmt.Errorf("failed to send initial update: %w", err)
-	}
-	pp.metrics.incUpdates()
-
-	return pp.ip, nil
-}
-
-// Subscribe stops notifying the listener on chages on any pod backing the
-// passed ip:port or its associated protocol config
-func (pw *PodWatcher) Unsubscribe(ip string, port Port, listener PodUpdateListener) {
-	pw.mu.Lock()
-	defer pw.mu.Unlock()
-
-	pw.log.Debugf("Stopping watch on %s:%d", ip, port)
-	pp, ok := pw.getPodPublisher(ip, port)
-	if !ok {
-		pw.log.Errorf("Cannot unsubscribe from unknown ip:port [%s:%d]", ip, port)
-		return
-	}
-	pp.unsubscribe(listener)
-
-	if len(pp.listeners) == 0 {
-		delete(pw.publishers, IPPort{pp.ip, pp.port})
-	}
-}
-
-// addPod is an event handler so it cannot block
-func (pw *PodWatcher) addPod(obj any) {
-	pod := obj.(*corev1.Pod)
-	pw.log.Tracef("Added pod %s.%s", pod.Name, pod.Namespace)
-	go pw.submitPodUpdate(pod, false)
-}
-
-// deletePod is an event handler so it cannot block
-func (pw *PodWatcher) deletePod(obj any) {
-	pod, ok := obj.(*corev1.Pod)
-	if !ok {
-		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-		if !ok {
-			pw.log.Errorf("Couldn't get object from DeletedFinalStateUnknown %#v", obj)
-			return
-		}
-		pod, ok = tombstone.Obj.(*corev1.Pod)
-		if !ok {
-			pw.log.Errorf("DeletedFinalStateUnknown contained object that is not a Pod %#v", obj)
-			return
-		}
-	}
-	pw.log.Tracef("Deleted pod %s.%s", pod.Name, pod.Namespace)
-	go pw.submitPodUpdate(pod, true)
-}
-
-// updatePod is an event handler so it cannot block
-func (pw *PodWatcher) updatePod(oldObj any, newObj any) {
-	oldPod := oldObj.(*corev1.Pod)
-	newPod := newObj.(*corev1.Pod)
-	if oldPod.DeletionTimestamp == nil && newPod.DeletionTimestamp != nil {
-		// this is just a mark, wait for actual deletion event
-		return
-	}
-
-	oldUpdated := latestUpdated(oldPod.ManagedFields)
-	updated := latestUpdated(newPod.ManagedFields)
-	if !updated.IsZero() && updated != oldUpdated {
-		delta := time.Since(updated)
-		podInformerLag.Observe(delta.Seconds())
-	}
-
-	pw.log.Tracef("Updated pod %s.%s", newPod.Name, newPod.Namespace)
-	go pw.submitPodUpdate(newPod, false)
-}
-
-func (pw *PodWatcher) submitPodUpdate(pod *corev1.Pod, remove bool) {
-	pw.mu.RLock()
-	defer pw.mu.RUnlock()
-
-	submitPod := pod
-	if remove {
-		submitPod = nil
-	}
-
-	for _, container := range pod.Spec.Containers {
-		for _, containerPort := range container.Ports {
-			if containerPort.ContainerPort != 0 {
-				for _, pip := range pod.Status.PodIPs {
-					if pp, ok := pw.getPodPublisher(pip.IP, Port(containerPort.ContainerPort)); ok {
-						pp.updatePod(submitPod)
-					}
-				}
-				if len(pod.Status.PodIPs) == 0 && pod.Status.PodIP != "" {
-					if pp, ok := pw.getPodPublisher(pod.Status.PodIP, Port(containerPort.ContainerPort)); ok {
-						pp.updatePod(submitPod)
-					}
-				}
-			}
-
-			if containerPort.HostPort != 0 {
-				for _, hip := range pod.Status.HostIPs {
-					if pp, ok := pw.getPodPublisher(hip.IP, Port(containerPort.HostPort)); ok {
-						pp.updatePod(submitPod)
-					}
-				}
-				if len(pod.Status.HostIPs) == 0 && pod.Status.HostIP != "" {
-					if pp, ok := pw.getPodPublisher(pod.Status.HostIP, Port(containerPort.HostPort)); ok {
-						pp.updatePod(submitPod)
-					}
-				}
-			}
-		}
-	}
-}
-
-func (pw *PodWatcher) updateServer(oldObj interface{}, newObj interface{}) {
-	oldServer := oldObj.(*v1beta2.Server)
-	newServer := newObj.(*v1beta2.Server)
-
-	oldUpdated := latestUpdated(oldServer.ManagedFields)
-	updated := latestUpdated(newServer.ManagedFields)
-	if !updated.IsZero() && updated != oldUpdated {
-		delta := time.Since(updated)
-		serverInformerLag.Observe(delta.Seconds())
-	}
-
-	pw.updateServers(newObj)
-}
-
-// updateServer triggers an Update() call to the listeners of the podPublishers
-// whose pod matches the Server's selector. This function is an event handler
-// so it cannot block.
-func (pw *PodWatcher) updateServers(_ any) {
-	pw.mu.RLock()
-	defer pw.mu.RUnlock()
-
-	for _, pp := range pw.publishers {
-		if pp.pod == nil {
-			continue
-		}
-		opaquePorts := GetAnnotatedOpaquePorts(pp.pod, pw.defaultOpaquePorts)
-		_, isOpaque := opaquePorts[pp.port]
-		// if port is annotated to be always opaque we can disregard Server updates
-		if isOpaque {
-			continue
-		}
-
-		go func(pp *podPublisher) {
-			pp.mu.RLock()
-			defer pp.mu.RUnlock()
-
-			updated := false
-			for _, listener := range pp.listeners {
-				// the Server in question doesn't carry information about other
-				// Servers that might target this podPublisher; createAddress()
-				// queries all the relevant Servers to determine the full state
-				addr, err := pp.createAddress()
-				if err != nil {
-					pw.log.Errorf("Error creating address for pod: %s", err)
-					continue
-				}
-				if err = listener.Update(&addr); err != nil {
-					pw.log.Warnf("Error sending update to listener: %s", err)
-					continue
-				}
-				updated = true
-			}
-			if updated {
-				pp.metrics.incUpdates()
-			}
-		}(pp)
-	}
-}
-
-// getOrNewPodPublisher returns the podPublisher for the given target if it
-// exists. Otherwise, it creates a new one and returns it.
-func (pw *PodWatcher) getOrNewPodPublisher(service *ServiceID, hostname, ip string, port Port) (*podPublisher, error) {
-	pw.mu.Lock()
-	defer pw.mu.Unlock()
-
-	var pod *corev1.Pod
-	var err error
-	if hostname != "" {
-		pod, err = pw.getEndpointByHostname(hostname, service)
-		if err != nil {
-			return nil, err
-		}
-		ip = pod.Status.PodIP
-	} else {
-		pod, err = pw.getPodByPodIP(ip, port)
-		if err != nil {
-			return nil, err
-		}
-		if pod == nil {
-			pod, err = pw.getPodByHostIP(ip, port)
-			if err != nil {
-				return nil, err
-			}
-		}
-	}
-
-	ipPort := IPPort{ip, port}
-	pp, ok := pw.publishers[ipPort]
-	if !ok {
-		pp = &podPublisher{
-			defaultOpaquePorts: pw.defaultOpaquePorts,
-			k8sAPI:             pw.k8sAPI,
-			metadataAPI:        pw.metadataAPI,
-			ip:                 ip,
-			port:               port,
-			pod:                pod,
-			metrics: ipPortVecs.newMetrics(prometheus.Labels{
-				"ip":   ip,
-				"port": strconv.FormatUint(uint64(port), 10),
-			}),
-			log: pw.log.WithFields(logging.Fields{
-				"component": "pod-publisher",
-				"ip":        ip,
-				"port":      port,
-			}),
-		}
-		pw.publishers[ipPort] = pp
-	}
-	return pp, nil
-}
-
-func (pw *PodWatcher) getPodPublisher(ip string, port Port) (pp *podPublisher, ok bool) {
-	ipPort := IPPort{ip, port}
-	pp, ok = pw.publishers[ipPort]
-	return
-}
-
-// getPodByPodIP returns a pod that maps to the given IP address in the pod network
-func (pw *PodWatcher) getPodByPodIP(podIP string, port uint32) (*corev1.Pod, error) {
-	podIPPods, err := getIndexedPods(pw.k8sAPI, PodIPIndex, podIP)
-	if err != nil {
-		return nil, status.Error(codes.Unknown, err.Error())
-	}
-	if len(podIPPods) == 1 {
-		pw.log.Debugf("found %s on the pod network", podIP)
-		return podIPPods[0], nil
-	}
-	if len(podIPPods) > 1 {
-		conflictingPods := []string{}
-		for _, pod := range podIPPods {
-			conflictingPods = append(conflictingPods, fmt.Sprintf("%s:%s", pod.Namespace, pod.Name))
-		}
-		pw.log.Warnf("found conflicting %s IP on the pod network: %s", podIP, strings.Join(conflictingPods, ","))
-		return nil, status.Errorf(codes.FailedPrecondition, "found %d pods with a conflicting pod network IP %s", len(podIPPods), podIP)
-	}
-
-	pw.log.Debugf("no pod found for %s:%d", podIP, port)
-	return nil, nil
-}
-
-// getPodByHostIP returns a pod that maps to the given IP address in the host
-// network. It must have a container port that exposes `port` as a host port.
-func (pw *PodWatcher) getPodByHostIP(hostIP string, port uint32) (*corev1.Pod, error) {
-	addr := net.JoinHostPort(hostIP, fmt.Sprintf("%d", port))
-	hostIPPods, err := getIndexedPods(pw.k8sAPI, HostIPIndex, addr)
-	if err != nil {
-		return nil, status.Error(codes.Unknown, err.Error())
-	}
-	if len(hostIPPods) == 1 {
-		pw.log.Debugf("found %s:%d on the host network", hostIP, port)
-		return hostIPPods[0], nil
-	}
-	if len(hostIPPods) > 1 {
-		conflictingPods := []string{}
-		for _, pod := range hostIPPods {
-			conflictingPods = append(conflictingPods, fmt.Sprintf("%s:%s", pod.Namespace, pod.Name))
-		}
-		pw.log.Warnf("found conflicting %s:%d endpoint on the host network: %s", hostIP, port, strings.Join(conflictingPods, ","))
-		return nil, status.Errorf(codes.FailedPrecondition, "found %d pods with a conflicting host network endpoint %s:%d", len(hostIPPods), hostIP, port)
-	}
-
-	return nil, nil
-}
-
-// getEndpointByHostname returns a pod that maps to the given hostname (or an
-// instanceID). The hostname is generally the prefix of the pod's DNS name;
-// since it may be arbitrary we need to look at the corresponding service's
-// Endpoints object to see whether the hostname matches a pod.
-func (pw *PodWatcher) getEndpointByHostname(hostname string, svcID *ServiceID) (*corev1.Pod, error) {
-	ep, err := pw.k8sAPI.Endpoint().Lister().Endpoints(svcID.Namespace).Get(svcID.Name)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, subset := range ep.Subsets {
-		for _, addr := range subset.Addresses {
-
-			if hostname == addr.Hostname {
-				if addr.TargetRef != nil && addr.TargetRef.Kind == "Pod" {
-					podName := addr.TargetRef.Name
-					podNamespace := addr.TargetRef.Namespace
-					pod, err := pw.k8sAPI.Pod().Lister().Pods(podNamespace).Get(podName)
-					if err != nil {
-						return nil, err
-					}
-					return pod, nil
-				}
-				return nil, nil
-			}
-		}
-	}
-
-	return nil, status.Errorf(codes.NotFound, "no pod found in Endpoints %s/%s for hostname %s", svcID.Namespace, svcID.Name, hostname)
-}
-
-func (pp *podPublisher) subscribe(listener PodUpdateListener) {
-	pp.mu.Lock()
-	defer pp.mu.Unlock()
-
-	pp.listeners = append(pp.listeners, listener)
-	pp.metrics.setSubscribers(len(pp.listeners))
-}
-
-func (pp *podPublisher) unsubscribe(listener PodUpdateListener) {
-	pp.mu.Lock()
-	defer pp.mu.Unlock()
-
-	for i, e := range pp.listeners {
-		if e == listener {
-			n := len(pp.listeners)
-			pp.listeners[i] = pp.listeners[n-1]
-			pp.listeners[n-1] = nil
-			pp.listeners = pp.listeners[:n-1]
-			break
-		}
-	}
-
-	pp.metrics.setSubscribers(len(pp.listeners))
-}
-
-// updatePod creates an Address instance for the given pod, that is passed to
-// the listener's Update() method, only if the pod's readiness state has
-// changed. If the passed pod is nil, it means the pod (still referred to in
-// pp.pod) has been deleted.
-func (pp *podPublisher) updatePod(pod *corev1.Pod) {
-	pp.mu.Lock()
-	defer pp.mu.Unlock()
-
-	// pod wasn't ready or there was no backing pod - check if passed pod is ready
-	if pp.pod == nil {
-		if pod == nil {
-			pp.log.Trace("Pod deletion event already consumed - ignore")
-			return
-		}
-
-		if !isRunningAndReady(pod) {
-			pp.log.Tracef("Pod %s.%s not ready - ignore", pod.Name, pod.Namespace)
-			return
-		}
-
-		pp.log.Debugf("Pod %s.%s became ready", pod.Name, pod.Namespace)
-		pp.pod = pod
-		updated := false
-		for _, l := range pp.listeners {
-			addr, err := pp.createAddress()
-			if err != nil {
-				pp.log.Errorf("Error creating address for pod: %s", err)
-				continue
-			}
-			if err = l.Update(&addr); err != nil {
-				pp.log.Warnf("Error sending update to listener: %s", err)
-				continue
-			}
-			updated = true
-		}
-		if updated {
-			pp.metrics.incUpdates()
-		}
-		return
-	}
-
-	// backing pod becoming unready or getting deleted
-	if pod == nil || !isRunningAndReady(pod) {
-		pp.log.Debugf("Pod %s.%s deleted or it became unready - remove", pp.pod.Name, pp.pod.Namespace)
-		pp.pod = nil
-		updated := false
-		for _, l := range pp.listeners {
-			addr, err := pp.createAddress()
-			if err != nil {
-				pp.log.Errorf("Error creating address for pod: %s", err)
-				continue
-			}
-			if err = l.Update(&addr); err != nil {
-				pp.log.Warnf("Error sending update to listener: %s", err)
-				continue
-			}
-			updated = true
-		}
-		if updated {
-			pp.metrics.incUpdates()
-		}
-		return
-	}
-
-	pp.log.Tracef("Ignored event on pod %s.%s", pod.Name, pod.Namespace)
-}
-
-// createAddress returns an Address instance for the given ip, port and pod. It
-// completes the ownership and opaque protocol information
-func (pp *podPublisher) createAddress() (Address, error) {
-	var ownerKind, ownerName string
-	var err error
-	if pp.pod != nil {
-		ownerKind, ownerName, err = pp.metadataAPI.GetOwnerKindAndName(context.Background(), pp.pod, true)
-		if err != nil {
-			return Address{}, err
-		}
-	}
-
-	address := Address{
-		IP:        pp.ip,
-		Port:      pp.port,
-		Pod:       pp.pod,
-		OwnerName: ownerName,
-		OwnerKind: ownerKind,
-	}
-
-	// Override opaqueProtocol if the endpoint's port is annotated as opaque
-	opaquePorts := GetAnnotatedOpaquePorts(pp.pod, pp.defaultOpaquePorts)
-	if _, ok := opaquePorts[pp.port]; ok {
-		address.OpaqueProtocol = true
-	} else if pp.pod != nil {
-		if err := SetToServerProtocol(pp.k8sAPI, &address, pp.port); err != nil {
-			return Address{}, fmt.Errorf("failed to set address OpaqueProtocol: %w", err)
-		}
-	}
-
-	return address, nil
-}
-
-// GetAnnotatedOpaquePorts returns the opaque ports for the pod given its
-// annotations, or the default opaque ports if it's not annotated
-func GetAnnotatedOpaquePorts(pod *corev1.Pod, defaultPorts map[uint32]struct{}) map[uint32]struct{} {
-	if pod == nil {
-		return defaultPorts
-	}
-	annotation, ok := pod.Annotations[consts.ProxyOpaquePortsAnnotation]
-	if !ok {
-		return defaultPorts
-	}
-	opaquePorts := make(map[uint32]struct{})
-	if annotation != "" {
-		for _, pr := range util.ParseContainerOpaquePorts(annotation, pod.Spec.Containers) {
-			for _, port := range pr.Ports() {
-				opaquePorts[uint32(port)] = struct{}{}
-			}
-		}
-	}
-	return opaquePorts
-}
-
-func isRunningAndReady(pod *corev1.Pod) bool {
-	if pod == nil || pod.Status.Phase != corev1.PodRunning {
-		return false
-	}
-	for _, condition := range pod.Status.Conditions {
-		if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
-			return true
-		}
-	}
-
-	return false
-}
diff --git a/controller/api/destination/watcher/prometheus.go b/controller/api/destination/watcher/prometheus.go
index 014a01dc5..a62d5a5bd 100644
--- a/controller/api/destination/watcher/prometheus.go
+++ b/controller/api/destination/watcher/prometheus.go
@@ -87,6 +87,14 @@ var (
 		},
 	)
 
+	externalWorkloadInformerLag = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "externalworkload_informer_lag_seconds",
+			Help:    "The amount of time between when an ExternalWorkload resource is updated and when an informer observes it",
+			Buckets: informer_lag_seconds_buckets,
+		},
+	)
+
 	serviceProfileInformerLag = promauto.NewHistogram(
 		prometheus.HistogramOpts{
 			Name:    "serviceprofiles_informer_lag_seconds",
diff --git a/controller/api/destination/watcher/workload_watcher.go b/controller/api/destination/watcher/workload_watcher.go
new file mode 100644
index 000000000..4fce08cdf
--- /dev/null
+++ b/controller/api/destination/watcher/workload_watcher.go
@@ -0,0 +1,840 @@
+package watcher
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	externalworkload "github.com/linkerd/linkerd2/controller/api/destination/external-workload"
+	ext "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"
+	"github.com/linkerd/linkerd2/pkg/util"
+	"github.com/prometheus/client_golang/prometheus"
+	logging "github.com/sirupsen/logrus"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/tools/cache"
+)
+
+type (
+	// WorkloadWatcher watches all pods and externalworkloads in the cluster.
+	// It keeps a map of publishers keyed by IP and port.
+	WorkloadWatcher struct {
+		defaultOpaquePorts map[uint32]struct{}
+		k8sAPI             *k8s.API
+		metadataAPI        *k8s.MetadataAPI
+		publishers         map[IPPort]*workloadPublisher
+		log                *logging.Entry
+
+		mu sync.RWMutex
+	}
+
+	// workloadPublisher represents an ip:port along with the backing pod
+	// or externalworkload (if any). It keeps a list of listeners to be notified
+	// whenever the workload or the associated opaque protocol config changes.
+	workloadPublisher struct {
+		defaultOpaquePorts map[uint32]struct{}
+		k8sAPI             *k8s.API
+		metadataAPI        *k8s.MetadataAPI
+		ip                 string
+		port               Port
+		pod                *corev1.Pod
+		externalWorkload   *ext.ExternalWorkload
+		listeners          []WorkloadUpdateListener
+		metrics            metrics
+		log                *logging.Entry
+
+		mu sync.RWMutex
+	}
+
+	// PodUpdateListener is the interface subscribers must implement.
+	WorkloadUpdateListener interface {
+		Update(*Address) error
+	}
+)
+
+var ipPortVecs = newMetricsVecs("ip_port", []string{"ip", "port"})
+
+func NewWorkloadWatcher(k8sAPI *k8s.API, metadataAPI *k8s.MetadataAPI, log *logging.Entry, defaultOpaquePorts map[uint32]struct{}) (*WorkloadWatcher, error) {
+	ww := &WorkloadWatcher{
+		defaultOpaquePorts: defaultOpaquePorts,
+		k8sAPI:             k8sAPI,
+		metadataAPI:        metadataAPI,
+		publishers:         make(map[IPPort]*workloadPublisher),
+		log: log.WithFields(logging.Fields{
+			"component": "workload-watcher",
+		}),
+	}
+
+	_, err := k8sAPI.Pod().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    ww.addPod,
+		DeleteFunc: ww.deletePod,
+		UpdateFunc: ww.updatePod,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = k8sAPI.ExtWorkload().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    ww.addExternalWorkload,
+		DeleteFunc: ww.deleteExternalWorkload,
+		UpdateFunc: ww.updateExternalWorkload,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = k8sAPI.Srv().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    ww.updateServers,
+		DeleteFunc: ww.updateServers,
+		UpdateFunc: ww.updateServer,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return ww, nil
+}
+
+// Subscribe notifies the listener on changes on any workload backing the passed
+// host/ip:port or changes to its associated opaque protocol config. If service
+// and hostname are empty then ip should be set and vice-versa. If ip is empty,
+// the corresponding ip is found for the given service/hostname, and returned.
+func (ww *WorkloadWatcher) Subscribe(service *ServiceID, hostname, ip string, port Port, listener WorkloadUpdateListener) (string, error) {
+	if hostname != "" {
+		ww.log.Debugf("Establishing watch on workload %s.%s.%s:%d", hostname, service.Name, service.Namespace, port)
+	} else if service != nil {
+		ww.log.Debugf("Establishing watch on workload %s.%s:%d", service.Name, service.Namespace, port)
+	} else {
+		ww.log.Debugf("Establishing watch on workload %s:%d", ip, port)
+	}
+	pp, err := ww.getOrNewWorkloadPublisher(service, hostname, ip, port)
+	if err != nil {
+		return "", err
+	}
+
+	pp.subscribe(listener)
+
+	address, err := pp.createAddress()
+	if err != nil {
+		return "", err
+	}
+
+	if err = listener.Update(&address); err != nil {
+		return "", fmt.Errorf("failed to send initial update: %w", err)
+	}
+	pp.metrics.incUpdates()
+
+	return pp.ip, nil
+}
+
+// Subscribe stops notifying the listener on chages on any pod backing the
+// passed ip:port or its associated protocol config
+func (ww *WorkloadWatcher) Unsubscribe(ip string, port Port, listener WorkloadUpdateListener) {
+	ww.mu.Lock()
+	defer ww.mu.Unlock()
+
+	ww.log.Debugf("Stopping watch on %s:%d", ip, port)
+	wp, ok := ww.getWorkloadPublisher(ip, port)
+	if !ok {
+		ww.log.Errorf("Cannot unsubscribe from unknown ip:port [%s:%d]", ip, port)
+		return
+	}
+	wp.unsubscribe(listener)
+
+	if len(wp.listeners) == 0 {
+		delete(ww.publishers, IPPort{wp.ip, wp.port})
+	}
+}
+
+// addPod is an event handler so it cannot block
+func (ww *WorkloadWatcher) addPod(obj any) {
+	pod := obj.(*corev1.Pod)
+	ww.log.Tracef("Added pod %s.%s", pod.Name, pod.Namespace)
+	go ww.submitPodUpdate(pod, false)
+}
+
+// deletePod is an event handler so it cannot block
+func (ww *WorkloadWatcher) deletePod(obj any) {
+	pod, ok := obj.(*corev1.Pod)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			ww.log.Errorf("Couldn't get object from DeletedFinalStateUnknown %#v", obj)
+			return
+		}
+		pod, ok = tombstone.Obj.(*corev1.Pod)
+		if !ok {
+			ww.log.Errorf("DeletedFinalStateUnknown contained object that is not a Pod %#v", obj)
+			return
+		}
+	}
+	ww.log.Tracef("Deleted pod %s.%s", pod.Name, pod.Namespace)
+	go ww.submitPodUpdate(pod, true)
+}
+
+// updatePod is an event handler so it cannot block
+func (ww *WorkloadWatcher) updatePod(oldObj any, newObj any) {
+	oldPod := oldObj.(*corev1.Pod)
+	newPod := newObj.(*corev1.Pod)
+	if oldPod.DeletionTimestamp == nil && newPod.DeletionTimestamp != nil {
+		// this is just a mark, wait for actual deletion event
+		return
+	}
+
+	oldUpdated := latestUpdated(oldPod.ManagedFields)
+	updated := latestUpdated(newPod.ManagedFields)
+	if !updated.IsZero() && updated != oldUpdated {
+		delta := time.Since(updated)
+		podInformerLag.Observe(delta.Seconds())
+	}
+
+	ww.log.Tracef("Updated pod %s.%s", newPod.Name, newPod.Namespace)
+	go ww.submitPodUpdate(newPod, false)
+}
+
+// addExternalWorkload is an event handler so it cannot block
+func (ww *WorkloadWatcher) addExternalWorkload(obj any) {
+	externalWorkload := obj.(*ext.ExternalWorkload)
+	ww.log.Tracef("Added externalworkload %s.%s", externalWorkload.Name, externalWorkload.Namespace)
+	go ww.submitExternalWorkloadUpdate(externalWorkload, false)
+}
+
+// deleteExternalWorkload is an event handler so it cannot block
+func (ww *WorkloadWatcher) deleteExternalWorkload(obj any) {
+	externalWorkload, ok := obj.(*ext.ExternalWorkload)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			ww.log.Errorf("Couldn't get object from DeletedFinalStateUnknown %#v", obj)
+			return
+		}
+		externalWorkload, ok = tombstone.Obj.(*ext.ExternalWorkload)
+		if !ok {
+			ww.log.Errorf("DeletedFinalStateUnknown contained object that is not an ExternalWorkload %#v", obj)
+			return
+		}
+	}
+	ww.log.Tracef("Deleted externalworklod %s.%s", externalWorkload.Name, externalWorkload.Namespace)
+	go ww.submitExternalWorkloadUpdate(externalWorkload, true)
+}
+
+// updateExternalWorkload is an event handler so it cannot block
+func (ww *WorkloadWatcher) updateExternalWorkload(oldObj any, newObj any) {
+	oldExternalWorkload := oldObj.(*ext.ExternalWorkload)
+	newExternalWorkload := newObj.(*ext.ExternalWorkload)
+	if oldExternalWorkload.DeletionTimestamp == nil && newExternalWorkload.DeletionTimestamp != nil {
+		// this is just a mark, wait for actual deletion event
+		return
+	}
+
+	oldUpdated := latestUpdated(oldExternalWorkload.ManagedFields)
+	updated := latestUpdated(newExternalWorkload.ManagedFields)
+	if !updated.IsZero() && updated != oldUpdated {
+		delta := time.Since(updated)
+		externalWorkloadInformerLag.Observe(delta.Seconds())
+	}
+
+	ww.log.Tracef("Updated pod %s.%s", newExternalWorkload.Name, newExternalWorkload.Namespace)
+	go ww.submitExternalWorkloadUpdate(newExternalWorkload, false)
+}
+
+func (ww *WorkloadWatcher) submitPodUpdate(pod *corev1.Pod, remove bool) {
+	ww.mu.RLock()
+	defer ww.mu.RUnlock()
+
+	submitPod := pod
+	if remove {
+		submitPod = nil
+	}
+
+	for _, container := range pod.Spec.Containers {
+		for _, containerPort := range container.Ports {
+			if containerPort.ContainerPort != 0 {
+				for _, pip := range pod.Status.PodIPs {
+					if wp, ok := ww.getWorkloadPublisher(pip.IP, Port(containerPort.ContainerPort)); ok {
+						wp.updatePod(submitPod)
+					}
+				}
+				if len(pod.Status.PodIPs) == 0 && pod.Status.PodIP != "" {
+					if wp, ok := ww.getWorkloadPublisher(pod.Status.PodIP, Port(containerPort.ContainerPort)); ok {
+						wp.updatePod(submitPod)
+					}
+				}
+			}
+
+			if containerPort.HostPort != 0 {
+				for _, hip := range pod.Status.HostIPs {
+					if pp, ok := ww.getWorkloadPublisher(hip.IP, Port(containerPort.HostPort)); ok {
+						pp.updatePod(submitPod)
+					}
+				}
+				if len(pod.Status.HostIPs) == 0 && pod.Status.HostIP != "" {
+					if pp, ok := ww.getWorkloadPublisher(pod.Status.HostIP, Port(containerPort.HostPort)); ok {
+						pp.updatePod(submitPod)
+					}
+				}
+			}
+		}
+	}
+}
+
+func (ww *WorkloadWatcher) submitExternalWorkloadUpdate(externalWorkload *ext.ExternalWorkload, remove bool) {
+	ww.mu.RLock()
+	defer ww.mu.RUnlock()
+
+	submitWorkload := externalWorkload
+	if remove {
+		submitWorkload = nil
+	}
+
+	for _, port := range externalWorkload.Spec.Ports {
+		for _, ip := range externalWorkload.Spec.WorkloadIPs {
+			if wp, ok := ww.getWorkloadPublisher(ip.Ip, Port(port.Port)); ok {
+				wp.updateExternalWorkload(submitWorkload)
+			}
+		}
+	}
+}
+
+func (ww *WorkloadWatcher) updateServer(oldObj interface{}, newObj interface{}) {
+	oldServer := oldObj.(*v1beta2.Server)
+	newServer := newObj.(*v1beta2.Server)
+
+	oldUpdated := latestUpdated(oldServer.ManagedFields)
+	updated := latestUpdated(newServer.ManagedFields)
+	if !updated.IsZero() && updated != oldUpdated {
+		delta := time.Since(updated)
+		serverInformerLag.Observe(delta.Seconds())
+	}
+
+	ww.updateServers(newObj)
+}
+
+// updateServer triggers an Update() call to the listeners of the workloadPublishers
+// whose pod matches the Server's podSelector or whose externalworkload matches
+// the Server's externalworkload selection. This function is an event handler
+// so it cannot block.
+func (ww *WorkloadWatcher) updateServers(_ any) {
+	ww.mu.RLock()
+	defer ww.mu.RUnlock()
+
+	for _, wp := range ww.publishers {
+		var opaquePorts map[uint32]struct{}
+		if wp.pod != nil {
+			opaquePorts = GetAnnotatedOpaquePorts(wp.pod, ww.defaultOpaquePorts)
+		} else if wp.externalWorkload != nil {
+			opaquePorts = GetAnnotatedOpaquePortsForExternalWorkload(wp.externalWorkload, ww.defaultOpaquePorts)
+		} else {
+			continue
+		}
+
+		_, isOpaque := opaquePorts[wp.port]
+		// if port is annotated to be always opaque we can disregard Server updates
+		if isOpaque {
+			continue
+		}
+
+		go func(wp *workloadPublisher) {
+			wp.mu.RLock()
+			defer wp.mu.RUnlock()
+
+			updated := false
+			for _, listener := range wp.listeners {
+				// the Server in question doesn't carry information about other
+				// Servers that might target this workloadPublisher; createAddress()
+				// queries all the relevant Servers to determine the full state
+				addr, err := wp.createAddress()
+				if err != nil {
+					ww.log.Errorf("Error creating address for workload: %s", err)
+					continue
+				}
+				if err = listener.Update(&addr); err != nil {
+					ww.log.Warnf("Error sending update to listener: %s", err)
+					continue
+				}
+				updated = true
+			}
+			if updated {
+				wp.metrics.incUpdates()
+			}
+		}(wp)
+	}
+}
+
+// getOrNewWorkloadPublisher returns the workloadPublisher for the given target if it
+// exists. Otherwise, it creates a new one and returns it.
+func (ww *WorkloadWatcher) getOrNewWorkloadPublisher(service *ServiceID, hostname, ip string, port Port) (*workloadPublisher, error) {
+	ww.mu.Lock()
+	defer ww.mu.Unlock()
+
+	var pod *corev1.Pod
+	var externalWorkload *ext.ExternalWorkload
+	var err error
+	if hostname != "" {
+		pod, err = ww.getEndpointByHostname(hostname, service)
+		if err != nil {
+			return nil, err
+		}
+		ip = pod.Status.PodIP
+	} else {
+		pod, err = ww.getPodByPodIP(ip, port)
+		if err != nil {
+			return nil, err
+		}
+		if pod == nil {
+			pod, err = ww.getPodByHostIP(ip, port)
+			if err != nil {
+				return nil, err
+			}
+		}
+		if pod == nil {
+			externalWorkload, err = ww.getExternalWorkloadByIP(ip, port)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	ipPort := IPPort{ip, port}
+	wp, ok := ww.publishers[ipPort]
+	if !ok {
+		wp = &workloadPublisher{
+			defaultOpaquePorts: ww.defaultOpaquePorts,
+			k8sAPI:             ww.k8sAPI,
+			metadataAPI:        ww.metadataAPI,
+			ip:                 ip,
+			port:               port,
+			pod:                pod,
+			externalWorkload:   externalWorkload,
+			metrics: ipPortVecs.newMetrics(prometheus.Labels{
+				"ip":   ip,
+				"port": strconv.FormatUint(uint64(port), 10),
+			}),
+			log: ww.log.WithFields(logging.Fields{
+				"component": "workload-publisher",
+				"ip":        ip,
+				"port":      port,
+			}),
+		}
+		ww.publishers[ipPort] = wp
+	}
+	return wp, nil
+}
+
+func (ww *WorkloadWatcher) getWorkloadPublisher(ip string, port Port) (wp *workloadPublisher, ok bool) {
+	ipPort := IPPort{ip, port}
+	wp, ok = ww.publishers[ipPort]
+	return
+}
+
+// getPodByPodIP returns a pod that maps to the given IP address in the pod network
+func (ww *WorkloadWatcher) getPodByPodIP(podIP string, port uint32) (*corev1.Pod, error) {
+	podIPPods, err := getIndexedPods(ww.k8sAPI, PodIPIndex, podIP)
+	if err != nil {
+		return nil, status.Error(codes.Unknown, err.Error())
+	}
+	if len(podIPPods) == 1 {
+		ww.log.Debugf("found %s on the pod network", podIP)
+		return podIPPods[0], nil
+	}
+	if len(podIPPods) > 1 {
+		conflictingPods := []string{}
+		for _, pod := range podIPPods {
+			conflictingPods = append(conflictingPods, fmt.Sprintf("%s:%s", pod.Namespace, pod.Name))
+		}
+		ww.log.Warnf("found conflicting %s IP on the pod network: %s", podIP, strings.Join(conflictingPods, ","))
+		return nil, status.Errorf(codes.FailedPrecondition, "found %d pods with a conflicting pod network IP %s", len(podIPPods), podIP)
+	}
+
+	ww.log.Debugf("no pod found for %s:%d", podIP, port)
+	return nil, nil
+}
+
+// getPodByHostIP returns a pod that maps to the given IP address in the host
+// network. It must have a container port that exposes `port` as a host port.
+func (ww *WorkloadWatcher) getPodByHostIP(hostIP string, port uint32) (*corev1.Pod, error) {
+	addr := net.JoinHostPort(hostIP, fmt.Sprintf("%d", port))
+	hostIPPods, err := getIndexedPods(ww.k8sAPI, HostIPIndex, addr)
+	if err != nil {
+		return nil, status.Error(codes.Unknown, err.Error())
+	}
+	if len(hostIPPods) == 1 {
+		ww.log.Debugf("found %s:%d on the host network", hostIP, port)
+		return hostIPPods[0], nil
+	}
+	if len(hostIPPods) > 1 {
+		conflictingPods := []string{}
+		for _, pod := range hostIPPods {
+			conflictingPods = append(conflictingPods, fmt.Sprintf("%s:%s", pod.Namespace, pod.Name))
+		}
+		ww.log.Warnf("found conflicting %s:%d endpoint on the host network: %s", hostIP, port, strings.Join(conflictingPods, ","))
+		return nil, status.Errorf(codes.FailedPrecondition, "found %d pods with a conflicting host network endpoint %s:%d", len(hostIPPods), hostIP, port)
+	}
+
+	return nil, nil
+}
+
+// getExternalWorkloadByIP returns an externalworkload with the given IP
+// address.
+func (ww *WorkloadWatcher) getExternalWorkloadByIP(ip string, port uint32) (*ext.ExternalWorkload, error) {
+	addr := net.JoinHostPort(ip, fmt.Sprintf("%d", port))
+	workloads, err := getIndexedExternalWorkloads(ww.k8sAPI, ExternalWorkloadIPIndex, addr)
+	if err != nil {
+		return nil, status.Error(codes.Unknown, err.Error())
+	}
+	if len(workloads) == 0 {
+		ww.log.Debugf("no externalworkload found for %s:%d", ip, port)
+		return nil, nil
+	}
+	if len(workloads) == 1 {
+		ww.log.Debugf("found externalworkload %s:%d", ip, port)
+		return workloads[0], nil
+	}
+	if len(workloads) > 1 {
+		conflictingWorkloads := []string{}
+		for _, ew := range workloads {
+			conflictingWorkloads = append(conflictingWorkloads, fmt.Sprintf("%s:%s", ew.Namespace, ew.Name))
+		}
+		ww.log.Warnf("found conflicting %s:%d externalworkload: %s", ip, port, strings.Join(conflictingWorkloads, ","))
+		return nil, status.Errorf(codes.FailedPrecondition, "found %d externalworkloads with a conflicting ip %s:%d", len(workloads), ip, port)
+	}
+
+	return nil, nil
+}
+
+// getEndpointByHostname returns a pod that maps to the given hostname (or an
+// instanceID). The hostname is generally the prefix of the pod's DNS name;
+// since it may be arbitrary we need to look at the corresponding service's
+// Endpoints object to see whether the hostname matches a pod.
+func (ww *WorkloadWatcher) getEndpointByHostname(hostname string, svcID *ServiceID) (*corev1.Pod, error) {
+	ep, err := ww.k8sAPI.Endpoint().Lister().Endpoints(svcID.Namespace).Get(svcID.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, subset := range ep.Subsets {
+		for _, addr := range subset.Addresses {
+
+			if hostname == addr.Hostname {
+				if addr.TargetRef != nil && addr.TargetRef.Kind == "Pod" {
+					podName := addr.TargetRef.Name
+					podNamespace := addr.TargetRef.Namespace
+					pod, err := ww.k8sAPI.Pod().Lister().Pods(podNamespace).Get(podName)
+					if err != nil {
+						return nil, err
+					}
+					return pod, nil
+				}
+				return nil, nil
+			}
+		}
+	}
+
+	return nil, status.Errorf(codes.NotFound, "no pod found in Endpoints %s/%s for hostname %s", svcID.Namespace, svcID.Name, hostname)
+}
+
+func (wp *workloadPublisher) subscribe(listener WorkloadUpdateListener) {
+	wp.mu.Lock()
+	defer wp.mu.Unlock()
+
+	wp.listeners = append(wp.listeners, listener)
+	wp.metrics.setSubscribers(len(wp.listeners))
+}
+
+func (wp *workloadPublisher) unsubscribe(listener WorkloadUpdateListener) {
+	wp.mu.Lock()
+	defer wp.mu.Unlock()
+
+	for i, e := range wp.listeners {
+		if e == listener {
+			n := len(wp.listeners)
+			wp.listeners[i] = wp.listeners[n-1]
+			wp.listeners[n-1] = nil
+			wp.listeners = wp.listeners[:n-1]
+			break
+		}
+	}
+
+	wp.metrics.setSubscribers(len(wp.listeners))
+}
+
+// updatePod creates an Address instance for the given pod, that is passed to
+// the listener's Update() method, only if the pod's readiness state has
+// changed. If the passed pod is nil, it means the pod (still referred to in
+// wp.pod) has been deleted.
+func (wp *workloadPublisher) updatePod(pod *corev1.Pod) {
+	wp.mu.Lock()
+	defer wp.mu.Unlock()
+
+	// pod wasn't ready or there was no backing pod - check if passed pod is ready
+	if wp.pod == nil {
+		if pod == nil {
+			wp.log.Trace("Pod deletion event already consumed - ignore")
+			return
+		}
+
+		if !isRunningAndReady(pod) {
+			wp.log.Tracef("Pod %s.%s not ready - ignore", pod.Name, pod.Namespace)
+			return
+		}
+
+		wp.log.Debugf("Pod %s.%s became ready", pod.Name, pod.Namespace)
+		wp.pod = pod
+		updated := false
+		for _, l := range wp.listeners {
+			addr, err := wp.createAddress()
+			if err != nil {
+				wp.log.Errorf("Error creating address for pod: %s", err)
+				continue
+			}
+			if err = l.Update(&addr); err != nil {
+				wp.log.Warnf("Error sending update to listener: %s", err)
+				continue
+			}
+			updated = true
+		}
+		if updated {
+			wp.metrics.incUpdates()
+		}
+		return
+	}
+
+	// backing pod becoming unready or getting deleted
+	if pod == nil || !isRunningAndReady(pod) {
+		wp.log.Debugf("Pod %s.%s deleted or it became unready - remove", wp.pod.Name, wp.pod.Namespace)
+		wp.pod = nil
+		updated := false
+		for _, l := range wp.listeners {
+			addr, err := wp.createAddress()
+			if err != nil {
+				wp.log.Errorf("Error creating address for pod: %s", err)
+				continue
+			}
+			if err = l.Update(&addr); err != nil {
+				wp.log.Warnf("Error sending update to listener: %s", err)
+				continue
+			}
+			updated = true
+		}
+		if updated {
+			wp.metrics.incUpdates()
+		}
+		return
+	}
+
+	wp.log.Tracef("Ignored event on pod %s.%s", pod.Name, pod.Namespace)
+}
+
+// updateExternalWorkload creates an Address instance for the given externalworkload,
+// that is passed to the listener's Update() method, only if the workload's
+// readiness state has changed. If the passed workload is nil, it means the
+// workload (still referred to in wp.externalWorkload) has been deleted.
+func (wp *workloadPublisher) updateExternalWorkload(externalWorkload *ext.ExternalWorkload) {
+	wp.mu.Lock()
+	defer wp.mu.Unlock()
+
+	// externalWorkload wasn't ready or there was no backing externalWorkload.
+	// check if passed externalWorkload is ready
+	if wp.externalWorkload == nil {
+		if externalWorkload == nil {
+			wp.log.Trace("ExternalWorkload deletion event already consumed - ignore")
+			return
+		}
+
+		if !externalworkload.IsReady(externalWorkload) {
+			wp.log.Tracef("ExternalWorkload %s.%s not ready - ignore", externalWorkload.Name, externalWorkload.Namespace)
+			return
+		}
+
+		wp.log.Debugf("ExternalWorkload %s.%s became ready", externalWorkload.Name, externalWorkload.Namespace)
+		wp.externalWorkload = externalWorkload
+		updated := false
+		for _, l := range wp.listeners {
+			addr, err := wp.createAddress()
+			if err != nil {
+				wp.log.Errorf("Error creating address for externalWorkload: %s", err)
+				continue
+			}
+			if err = l.Update(&addr); err != nil {
+				wp.log.Warnf("Error sending update to listener: %s", err)
+				continue
+			}
+			updated = true
+		}
+		if updated {
+			wp.metrics.incUpdates()
+		}
+		return
+	}
+
+	// backing pod becoming unready or getting deleted
+	if externalWorkload == nil || !externalworkload.IsReady(externalWorkload) {
+		wp.log.Debugf("ExternalWorkload %s.%s deleted or it became unready - remove", wp.externalWorkload.Name, wp.externalWorkload.Namespace)
+		wp.externalWorkload = nil
+		updated := false
+		for _, l := range wp.listeners {
+			addr, err := wp.createAddress()
+			if err != nil {
+				wp.log.Errorf("Error creating address for pod: %s", err)
+				continue
+			}
+			if err = l.Update(&addr); err != nil {
+				wp.log.Warnf("Error sending update to listener: %s", err)
+				continue
+			}
+			updated = true
+		}
+		if updated {
+			wp.metrics.incUpdates()
+		}
+		return
+	}
+
+	wp.log.Tracef("Ignored event on externalWorkload %s.%s", externalWorkload.Name, externalWorkload.Namespace)
+}
+
+// createAddress returns an Address instance for the given ip, port and workload. It
+// completes the ownership and opaque protocol information
+func (wp *workloadPublisher) createAddress() (Address, error) {
+	var ownerKind, ownerName string
+	var err error
+	if wp.pod != nil {
+		ownerKind, ownerName, err = wp.metadataAPI.GetOwnerKindAndName(context.Background(), wp.pod, true)
+		if err != nil {
+			return Address{}, err
+		}
+	} else if wp.externalWorkload != nil {
+		if len(wp.externalWorkload.GetOwnerReferences()) == 1 {
+			ownerKind = wp.externalWorkload.GetOwnerReferences()[0].Kind
+			ownerName = wp.externalWorkload.GetOwnerReferences()[0].Name
+		}
+	}
+
+	address := Address{
+		IP:               wp.ip,
+		Port:             wp.port,
+		Pod:              wp.pod,
+		ExternalWorkload: wp.externalWorkload,
+		OwnerName:        ownerName,
+		OwnerKind:        ownerKind,
+	}
+
+	// Override opaqueProtocol if the endpoint's port is annotated as opaque
+	if wp.pod != nil {
+		opaquePorts := GetAnnotatedOpaquePorts(wp.pod, wp.defaultOpaquePorts)
+		if _, ok := opaquePorts[wp.port]; ok {
+			address.OpaqueProtocol = true
+		} else {
+			if err := SetToServerProtocol(wp.k8sAPI, &address, wp.port); err != nil {
+				return Address{}, fmt.Errorf("failed to set address OpaqueProtocol: %w", err)
+			}
+		}
+	} else if wp.externalWorkload != nil {
+		opaquePorts := GetAnnotatedOpaquePortsForExternalWorkload(wp.externalWorkload, wp.defaultOpaquePorts)
+		if _, ok := opaquePorts[wp.port]; ok {
+			address.OpaqueProtocol = true
+		} else {
+			if err := SetToServerProtocolExternalWorkload(wp.k8sAPI, &address, wp.port); err != nil {
+				return Address{}, fmt.Errorf("failed to set address OpaqueProtocol: %w", err)
+			}
+		}
+	} else {
+		if _, ok := wp.defaultOpaquePorts[wp.port]; ok {
+			address.OpaqueProtocol = true
+		}
+	}
+
+	return address, nil
+}
+
+// GetAnnotatedOpaquePorts returns the opaque ports for the pod given its
+// annotations, or the default opaque ports if it's not annotated
+func GetAnnotatedOpaquePorts(pod *corev1.Pod, defaultPorts map[uint32]struct{}) map[uint32]struct{} {
+	if pod == nil {
+		return defaultPorts
+	}
+	annotation, ok := pod.Annotations[consts.ProxyOpaquePortsAnnotation]
+	if !ok {
+		return defaultPorts
+	}
+	opaquePorts := make(map[uint32]struct{})
+	if annotation != "" {
+		for _, pr := range util.ParseContainerOpaquePorts(annotation, pod.Spec.Containers) {
+			for _, port := range pr.Ports() {
+				opaquePorts[uint32(port)] = struct{}{}
+			}
+		}
+	}
+	return opaquePorts
+}
+
+// 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 *ext.ExternalWorkload, defaultPorts map[uint32]struct{}) map[uint32]struct{} {
+	if ew == nil {
+		return defaultPorts
+	}
+	annotation, ok := ew.Annotations[consts.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 *ext.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 *ext.ExternalWorkload) (int32, bool) {
+	for _, p := range ew.Spec.Ports {
+		if p.Name == pr {
+			return p.Port, true
+		}
+	}
+
+	return 0, false
+}
+
+func isRunningAndReady(pod *corev1.Pod) bool {
+	if pod == nil || pod.Status.Phase != corev1.PodRunning {
+		return false
+	}
+	for _, condition := range pod.Status.Conditions {
+		if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/go.mod b/go.mod
index 323ced862..029cfb321 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.8.0+incompatible
+	github.com/evanphx/json-patch v5.8.1+incompatible
 	github.com/fatih/color v1.16.0
 	github.com/fsnotify/fsnotify v1.7.0
 	github.com/go-openapi/spec v0.20.14
@@ -34,17 +34,17 @@ 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
-	helm.sh/helm/v3 v3.13.3
-	k8s.io/api v0.29.0
+	helm.sh/helm/v3 v3.14.0
+	k8s.io/api v0.29.1
 	k8s.io/apiextensions-apiserver v0.29.0
-	k8s.io/apimachinery v0.29.0
+	k8s.io/apimachinery v0.29.1
 	k8s.io/client-go v0.29.0
 	k8s.io/code-generator v0.29.0
-	k8s.io/klog/v2 v2.110.1
+	k8s.io/klog/v2 v2.120.1
 	k8s.io/kube-aggregator v0.29.0
 	sigs.k8s.io/gateway-api v0.8.1
 	sigs.k8s.io/yaml v1.4.0
@@ -83,7 +83,7 @@ require (
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/go-errors/errors v1.4.2 // indirect
-	github.com/go-logr/logr v1.3.0 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.20.2 // indirect
 	github.com/go-openapi/jsonreference v0.20.4 // indirect
@@ -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
@@ -153,7 +153,7 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/cli-runtime v0.28.4 // indirect
+	k8s.io/cli-runtime v0.29.0 // indirect
 	k8s.io/gengo v0.0.0-20230829151522-9cce18d56c01 // indirect
 	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
 	oras.land/oras-go v1.2.4 // indirect
@@ -162,11 +162,3 @@ require (
 	sigs.k8s.io/kustomize/kyaml v0.14.3 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 )
-
-// Required to unblock these bumps:
-// k8s.io/apiextensions-apiserver@v0.29.0
-// k8s.io/apimachinery@v0.29.0
-// k8s.io/kube-aggregator@v0.29.0
-// k8s.io/api@0.29.0
-// Remove when helm v3.13.4 is out
-replace k8s.io/cli-runtime => k8s.io/cli-runtime v0.29.0
diff --git a/go.sum b/go.sum
index 93fed8627..e3dfaabeb 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.8.0+incompatible h1:1Av9pn2FyxPdvrWNQszj1g6D6YthSmvCfcN6SYclTJg=
-github.com/evanphx/json-patch v5.8.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
+github.com/evanphx/json-patch v5.8.1+incompatible h1:2toJaoe7/rNa1zpeQx0UnVEjqk6z2ecyA20V/zg8vTU=
+github.com/evanphx/json-patch v5.8.1+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=
@@ -146,8 +146,8 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
-github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY=
-github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
+github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-openapi/jsonpointer v0.20.2 h1:mQc3nmndL8ZBzStEo3JYF8wzmeWffDH4VbXz58sAx6Q=
@@ -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=
@@ -748,20 +748,20 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools/v3 v3.4.0 h1:ZazjZUfuVeZGLAmlKKuyv3IKP5orXcwtOwDQH6YVr6o=
 gotest.tools/v3 v3.4.0/go.mod h1:CtbdzLSsqVhDgMtKsx03ird5YTGB3ar27v0u/yKBW5g=
-helm.sh/helm/v3 v3.13.3 h1:0zPEdGqHcubehJHP9emCtzRmu8oYsJFRrlVF3TFj8xY=
-helm.sh/helm/v3 v3.13.3/go.mod h1:3OKO33yI3p4YEXtTITN2+4oScsHeQe71KuzhlZ+aPfg=
+helm.sh/helm/v3 v3.14.0 h1:TaZIH6uOchn7L27ptwnnuHJiFrT/BsD4dFdp/HLT2nM=
+helm.sh/helm/v3 v3.14.0/go.mod h1:2itvvDv2WSZXTllknfQo6j7u3VVgMAvm8POCDgYH424=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-k8s.io/api v0.29.0 h1:NiCdQMY1QOp1H8lfRyeEf8eOwV6+0xA6XEE44ohDX2A=
-k8s.io/api v0.29.0/go.mod h1:sdVmXoz2Bo/cb77Pxi71IPTSErEW32xa4aXwKH7gfBA=
+k8s.io/api v0.29.1 h1:DAjwWX/9YT7NQD4INu49ROJuZAAAP/Ijki48GUPzxqw=
+k8s.io/api v0.29.1/go.mod h1:7Kl10vBRUXhnQQI8YR/R327zXC8eJ7887/+Ybta+RoQ=
 k8s.io/apiextensions-apiserver v0.29.0 h1:0VuspFG7Hj+SxyF/Z/2T0uFbI5gb5LRgEyUVE3Q4lV0=
 k8s.io/apiextensions-apiserver v0.29.0/go.mod h1:TKmpy3bTS0mr9pylH0nOt/QzQRrW7/h7yLdRForMZwc=
-k8s.io/apimachinery v0.29.0 h1:+ACVktwyicPz0oc6MTMLwa2Pw3ouLAfAon1wPLtG48o=
-k8s.io/apimachinery v0.29.0/go.mod h1:eVBxQ/cwiJxH58eK/jd/vAk4mrxmVlnpBH5J2GbMeis=
+k8s.io/apimachinery v0.29.1 h1:KY4/E6km/wLBguvCZv8cKTeOwwOBqFNjwJIdMkMbbRc=
+k8s.io/apimachinery v0.29.1/go.mod h1:6HVkd1FwxIagpYrHSwJlQqZI3G9LfYWRPAkUvLnXTKU=
 k8s.io/cli-runtime v0.29.0 h1:q2kC3cex4rOBLfPOnMSzV2BIrrQlx97gxHJs21KxKS4=
 k8s.io/cli-runtime v0.29.0/go.mod h1:VKudXp3X7wR45L+nER85YUzOQIru28HQpXr0mTdeCrk=
 k8s.io/client-go v0.29.0 h1:KmlDtFcrdUzOYrBhXHgKw5ycWzc3ryPX5mQe0SkG3y8=
@@ -771,8 +771,8 @@ k8s.io/code-generator v0.29.0/go.mod h1:5bqIZoCxs2zTRKMWNYqyQWW/bajc+ah4rh0tMY8z
 k8s.io/gengo v0.0.0-20230829151522-9cce18d56c01 h1:pWEwq4Asjm4vjW7vcsmijwBhOr1/shsbSYiWXmNGlks=
 k8s.io/gengo v0.0.0-20230829151522-9cce18d56c01/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
-k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0=
-k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo=
+k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw=
+k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-aggregator v0.29.0 h1:N4fmtePxOZ+bwiK1RhVEztOU+gkoVkvterHgpwAuiTw=
 k8s.io/kube-aggregator v0.29.0/go.mod h1:bjatII63ORkFg5yUFP2qm2OC49R0wwxZhRVIyJ4Z4X0=
 k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 h1:aVUu9fTY98ivBPKR9Y5w/AuzbMm96cd3YHRTU83I780=
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..3c125ab9d 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)
 
@@ -91,6 +91,7 @@ Kubernetes: `>=1.22.0-0`
 | collector.resources.memory.request | string | `nil` | Amount of memory that the collector container requests |
 | collector.tolerations | string | `nil` | Tolerations section, See the [K8S documentation](https://kubernetes.io/docs/concepts/scheduling-eviction/taint-and-toleration/) for more information |
 | commonLabels | object | `{}` | Labels to apply to all resources |
+| createNamespaceMetadataJob | bool | `true` | Creates a Job that adds necessary metadata to the extension's namespace during install; disable if lack of privileges require doing this manually |
 | defaultUID | int | `2103` | Default UID for all the jaeger components |
 | enablePSP | bool | `false` | Create Roles and RoleBindings to associate this extension's ServiceAccounts to the control plane PSP resource. This requires that `enabledPSP` is set to true on the control plane install. Note PSP has been deprecated since k8s v1.21 |
 | enablePodAntiAffinity | bool | `false` | Enables Pod Anti Affinity logic to balance the placement of replicas across hosts and zones for High Availability. Enable this only when you have multiple replicas of components. |
diff --git a/jaeger/charts/linkerd-jaeger/templates/namespace-metadata-rbac.yaml b/jaeger/charts/linkerd-jaeger/templates/namespace-metadata-rbac.yaml
index 6c1413f31..60f25c471 100644
--- a/jaeger/charts/linkerd-jaeger/templates/namespace-metadata-rbac.yaml
+++ b/jaeger/charts/linkerd-jaeger/templates/namespace-metadata-rbac.yaml
@@ -1,4 +1,5 @@
 ---
+{{- if .Values.createNamespaceMetadataJob}}
 kind: ServiceAccount
 apiVersion: v1
 metadata:
@@ -76,3 +77,4 @@ subjects:
 - kind: ServiceAccount
   name: namespace-metadata
   namespace: {{.Release.Namespace}}
+{{- end }}
diff --git a/jaeger/charts/linkerd-jaeger/templates/namespace-metadata.yaml b/jaeger/charts/linkerd-jaeger/templates/namespace-metadata.yaml
index e280c872b..878d621ed 100644
--- a/jaeger/charts/linkerd-jaeger/templates/namespace-metadata.yaml
+++ b/jaeger/charts/linkerd-jaeger/templates/namespace-metadata.yaml
@@ -1,4 +1,5 @@
 ---
+{{- if .Values.createNamespaceMetadataJob}}
 apiVersion: batch/v1
 kind: Job
 metadata:
@@ -60,3 +61,4 @@ spec:
         - {{.Release.Namespace}}
         - --linkerd-namespace
         - {{.Values.linkerdNamespace}}
+{{- end }}
diff --git a/jaeger/charts/linkerd-jaeger/values.yaml b/jaeger/charts/linkerd-jaeger/values.yaml
index ca9efa0cd..2cbebc27a 100644
--- a/jaeger/charts/linkerd-jaeger/values.yaml
+++ b/jaeger/charts/linkerd-jaeger/values.yaml
@@ -32,6 +32,10 @@ enablePodAntiAffinity: false
 # for more information
 #nodeAffinity:
 
+# -- Creates a Job that adds necessary metadata to the extension's namespace
+# during install; disable if lack of privileges require doing this manually
+createNamespaceMetadataJob: true
+
 # -- Create Roles and RoleBindings to associate this extension's
 # ServiceAccounts to the control plane PSP resource. This requires that
 # `enabledPSP` is set to true on the control plane install. Note PSP has been
diff --git a/jaeger/injector/mutator/patch.go b/jaeger/injector/mutator/patch.go
index 7af89466b..01f2b8835 100644
--- a/jaeger/injector/mutator/patch.go
+++ b/jaeger/injector/mutator/patch.go
@@ -35,7 +35,7 @@ const tpl = `[
     "path": "/spec/{{.ProxyPath}}/volumeMounts/-",
     "value": {
       "mountPath": "var/run/linkerd/podinfo",
-      "name": "podinfo"
+      "name": "linkerd-podinfo"
     }
   },
   {
@@ -52,7 +52,7 @@ const tpl = `[
      }
    ]
        },
-       "name": "podinfo"
+       "name": "linkerd-podinfo"
      }
   }
 ]`
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..6916e9319 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)
 
@@ -71,6 +71,7 @@ Kubernetes: `>=1.22.0-0`
 | Key | Type | Default | Description |
 |-----|------|---------|-------------|
 | commonLabels | object | `{}` | Labels to apply to all resources |
+| createNamespaceMetadataJob | bool | `true` | Creates a Job that adds necessary metadata to the extension's namespace during install; disable if lack of privileges require doing this manually |
 | enablePSP | bool | `false` | Create Roles and RoleBindings to associate this extension's ServiceAccounts to the control plane PSP resource. This requires that `enabledPSP` is set to true on the control plane install. Note PSP has been deprecated since k8s v1.21 |
 | enablePodAntiAffinity | bool | `false` | Enables Pod Anti Affinity logic to balance the placement of replicas across hosts and zones for High Availability. Enable this only when you have multiple replicas of components. |
 | gateway.UID | int | `2103` | User id under which the gateway shall be ran |
diff --git a/multicluster/charts/linkerd-multicluster/templates/namespace-metadata-rbac.yaml b/multicluster/charts/linkerd-multicluster/templates/namespace-metadata-rbac.yaml
index ff2a9718d..31f3fbc8b 100644
--- a/multicluster/charts/linkerd-multicluster/templates/namespace-metadata-rbac.yaml
+++ b/multicluster/charts/linkerd-multicluster/templates/namespace-metadata-rbac.yaml
@@ -1,3 +1,4 @@
+{{- if .Values.createNamespaceMetadataJob}}
 kind: ServiceAccount
 apiVersion: v1
 metadata:
@@ -75,3 +76,4 @@ subjects:
 - kind: ServiceAccount
   name: namespace-metadata
   namespace: {{.Release.Namespace}}
+{{- end }}
diff --git a/multicluster/charts/linkerd-multicluster/templates/namespace-metadata.yaml b/multicluster/charts/linkerd-multicluster/templates/namespace-metadata.yaml
index e61a61f1c..d7a0a8cef 100644
--- a/multicluster/charts/linkerd-multicluster/templates/namespace-metadata.yaml
+++ b/multicluster/charts/linkerd-multicluster/templates/namespace-metadata.yaml
@@ -1,3 +1,4 @@
+{{- if .Values.createNamespaceMetadataJob}}
 apiVersion: batch/v1
 kind: Job
 metadata:
@@ -59,3 +60,4 @@ spec:
         - {{.Release.Namespace}}
         - --linkerd-namespace
         - {{.Values.linkerdNamespace}}
+{{- end }}
diff --git a/multicluster/charts/linkerd-multicluster/values.yaml b/multicluster/charts/linkerd-multicluster/values.yaml
index 3de8edc4b..c4a9f8f08 100644
--- a/multicluster/charts/linkerd-multicluster/values.yaml
+++ b/multicluster/charts/linkerd-multicluster/values.yaml
@@ -98,3 +98,7 @@ enablePodAntiAffinity: false
 # [K8S documentation](https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/#node-affinity)
 # for more information
 # nodeAffinity:
+
+# -- Creates a Job that adds necessary metadata to the extension's namespace
+# during install; disable if lack of privileges require doing this manually
+createNamespaceMetadataJob: true
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..0e8852af8 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)
 
@@ -72,6 +72,7 @@ Kubernetes: `>=1.22.0-0`
 |-----|------|---------|-------------|
 | clusterDomain | string | `"cluster.local"` | Kubernetes DNS Domain name to use |
 | commonLabels | object | `{}` | Labels to apply to all resources |
+| createNamespaceMetadataJob | bool | `true` | Creates a Job that adds necessary metadata to the extension's namespace during install; disable if lack of privileges require doing this manually |
 | dashboard.UID | string | `nil` | UID for the dashboard resource |
 | dashboard.enforcedHostRegexp | string | `""` | Host header validation regex for the dashboard. See the [Linkerd documentation](https://linkerd.io/2/tasks/exposing-dashboard) for more information |
 | dashboard.image.name | string | `"web"` | Docker image name for the web instance |
diff --git a/viz/charts/linkerd-viz/templates/namespace-metadata-rbac.yaml b/viz/charts/linkerd-viz/templates/namespace-metadata-rbac.yaml
index 433f350ca..18f2cc117 100644
--- a/viz/charts/linkerd-viz/templates/namespace-metadata-rbac.yaml
+++ b/viz/charts/linkerd-viz/templates/namespace-metadata-rbac.yaml
@@ -1,3 +1,4 @@
+{{- if .Values.createNamespaceMetadataJob}}
 kind: ServiceAccount
 apiVersion: v1
 metadata:
@@ -75,3 +76,4 @@ subjects:
 - kind: ServiceAccount
   name: namespace-metadata
   namespace: {{.Release.Namespace}}
+{{- end }}
diff --git a/viz/charts/linkerd-viz/templates/namespace-metadata.yaml b/viz/charts/linkerd-viz/templates/namespace-metadata.yaml
index b6b577a2d..a9f165718 100644
--- a/viz/charts/linkerd-viz/templates/namespace-metadata.yaml
+++ b/viz/charts/linkerd-viz/templates/namespace-metadata.yaml
@@ -1,3 +1,4 @@
+{{- if .Values.createNamespaceMetadataJob}}
 apiVersion: batch/v1
 kind: Job
 metadata:
@@ -65,3 +66,4 @@ spec:
         - --prometheus-url
         - {{.}}
         {{- end }}
+{{- end }}
diff --git a/viz/charts/linkerd-viz/values.yaml b/viz/charts/linkerd-viz/values.yaml
index 2aa82a673..4bac52526 100644
--- a/viz/charts/linkerd-viz/values.yaml
+++ b/viz/charts/linkerd-viz/values.yaml
@@ -58,6 +58,10 @@ enablePodDisruptionBudget: false
 # for more information
 # nodeAffinity:
 
+# -- Creates a Job that adds necessary metadata to the extension's namespace
+# during install; disable if lack of privileges require doing this manually
+createNamespaceMetadataJob: true
+
 # -- Create Roles and RoleBindings to associate this extension's
 # ServiceAccounts to the control plane PSP resource. This requires that
 # `enabledPSP` is set to true on the control plane install. Note PSP has been
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/controller/api/destination/external-workload/endpoints_controller_test.go b/controller/api/destination/external-workload/endpoints_controller_test.go
index 720212cb5..25eaec309 100644
--- a/controller/api/destination/external-workload/endpoints_controller_test.go
+++ b/controller/api/destination/external-workload/endpoints_controller_test.go
@@ -326,7 +326,7 @@ func TestWorkloadServicesToUpdate(t *testing.T) {
 				t.Fatalf("unexpected error %v", err)
 			}
 
-			ec, err := NewEndpointsController(k8sAPI, "my-hostname", "controlplane-ns", make(chan struct{}))
+			ec, err := NewEndpointsController(k8sAPI, "my-hostname", "controlplane-ns", make(chan struct{}), false)
 			if err != nil {
 				t.Fatalf("unexpected error %v", err)
 			}
diff --git a/controller/api/destination/server_test.go b/controller/api/destination/server_test.go
index 5265804d9..0cb2dfcb5 100644
--- a/controller/api/destination/server_test.go
+++ b/controller/api/destination/server_test.go
@@ -39,6 +39,8 @@ const podIPPolicy = "172.17.0.16"
 const podIPStatefulSet = "172.17.13.15"
 const externalIP = "192.168.1.20"
 const externalIPv6 = "2001:db8::78"
+const externalWorkloadIP = "200.1.1.1"
+const externalWorkloadIPPolicy = "200.1.1.2"
 const port uint32 = 8989
 const opaquePort uint32 = 4242
 const skippedPort uint32 = 24224
@@ -405,6 +407,47 @@ func TestGetProfiles(t *testing.T) {
 		}
 	})
 
+	t.Run("Return profile with endpoint when using externalworkload IP", func(t *testing.T) {
+		server := makeServer(t)
+		defer server.clusterStore.UnregisterGauges()
+
+		stream := profileStream(t, server, externalWorkloadIP, port, "ns:ns")
+		defer stream.Cancel()
+
+		epAddr, err := toAddress(externalWorkloadIP, port)
+		if err != nil {
+			t.Fatalf("Got error: %s", err)
+		}
+
+		// An explanation for why we expect 1 to 3 updates is in test cases
+		// above
+		updates := stream.Updates()
+		if len(updates) == 0 || len(updates) > 3 {
+			t.Fatalf("Expected 1 to 3 updates but got %d: %v", len(updates), updates)
+		}
+
+		first := updates[0]
+		if first.Endpoint == nil {
+			t.Fatalf("Expected response to have endpoint field")
+		}
+		if first.OpaqueProtocol {
+			t.Fatalf("Expected port %d to not be an opaque protocol, but it was", port)
+		}
+		_, exists := first.Endpoint.MetricLabels["namespace"]
+		if !exists {
+			t.Fatalf("Expected 'namespace' metric label to exist but it did not %v", first.Endpoint)
+		}
+		if first.GetEndpoint().GetProtocolHint() == nil {
+			t.Fatalf("Expected protocol hint but found none")
+		}
+		if first.GetEndpoint().GetProtocolHint().GetOpaqueTransport() != nil {
+			t.Fatalf("Expected externalworkload to not support opaque traffic on port %d", port)
+		}
+		if first.Endpoint.Addr.String() != epAddr.String() {
+			t.Fatalf("Expected endpoint IP to be %s, but it was %s", epAddr.Ip, first.Endpoint.Addr.Ip)
+		}
+	})
+
 	t.Run("Return default profile when IP does not map to service or pod", func(t *testing.T) {
 		server := makeServer(t)
 		defer server.clusterStore.UnregisterGauges()
@@ -558,6 +601,47 @@ func TestGetProfiles(t *testing.T) {
 		}
 	})
 
+	t.Run("Return opaque protocol profile with endpoint when using externalworkload IP and opaque protocol port", func(t *testing.T) {
+		server := makeServer(t)
+		defer server.clusterStore.UnregisterGauges()
+
+		stream := profileStream(t, server, externalWorkloadIP, opaquePort, "")
+		defer stream.Cancel()
+
+		epAddr, err := toAddress(externalWorkloadIP, opaquePort)
+		if err != nil {
+			t.Fatalf("Got error: %s", err)
+		}
+
+		// An explanation for why we expect 1 to 3 updates is in test cases
+		// above
+		updates := stream.Updates()
+		if len(updates) == 0 || len(updates) > 3 {
+			t.Fatalf("Expected 1 to 3 updates but got %d: %v", len(updates), updates)
+		}
+
+		profile := assertSingleProfile(t, updates)
+		if profile.Endpoint == nil {
+			t.Fatalf("Expected response to have endpoint field")
+		}
+		if !profile.OpaqueProtocol {
+			t.Fatalf("Expected port %d to be an opaque protocol, but it was not", opaquePort)
+		}
+		_, exists := profile.Endpoint.MetricLabels["namespace"]
+		if !exists {
+			t.Fatalf("Expected 'namespace' metric label to exist but it did not")
+		}
+		if profile.Endpoint.ProtocolHint == nil {
+			t.Fatalf("Expected protocol hint but found none")
+		}
+		if profile.Endpoint.GetProtocolHint().GetOpaqueTransport().GetInboundPort() != 4143 {
+			t.Fatalf("Expected pod to support opaque traffic on port 4143")
+		}
+		if profile.Endpoint.Addr.String() != epAddr.String() {
+			t.Fatalf("Expected endpoint IP port to be %d, but it was %d", epAddr.Port, profile.Endpoint.Addr.Port)
+		}
+	})
+
 	t.Run("Return profile with opaque protocol when using Pod IP selected by a Server", func(t *testing.T) {
 		server := makeServer(t)
 		defer server.clusterStore.UnregisterGauges()
@@ -579,6 +663,27 @@ func TestGetProfiles(t *testing.T) {
 		}
 	})
 
+	t.Run("Return profile with opaque protocol when using externalworkload IP selected by a Server", func(t *testing.T) {
+		server := makeServer(t)
+		defer server.clusterStore.UnregisterGauges()
+
+		stream := profileStream(t, server, externalWorkloadIPPolicy, 80, "")
+		defer stream.Cancel()
+		profile := assertSingleProfile(t, stream.Updates())
+		if profile.Endpoint == nil {
+			t.Fatalf("Expected response to have endpoint field")
+		}
+		if !profile.OpaqueProtocol {
+			t.Fatalf("Expected port %d to be an opaque protocol, but it was not", 80)
+		}
+		if profile.Endpoint.GetProtocolHint() == nil {
+			t.Fatalf("Expected protocol hint but found none")
+		}
+		if profile.Endpoint.GetProtocolHint().GetOpaqueTransport().GetInboundPort() != 4143 {
+			t.Fatalf("Expected pod to support opaque traffic on port 4143")
+		}
+	})
+
 	t.Run("Return profile with opaque protocol when using an opaque port with an external IP", func(t *testing.T) {
 		server := makeServer(t)
 		defer server.clusterStore.UnregisterGauges()
diff --git a/controller/api/destination/test_util.go b/controller/api/destination/test_util.go
index d0e796128..f27e8e34a 100644
--- a/controller/api/destination/test_util.go
+++ b/controller/api/destination/test_util.go
@@ -337,6 +337,9 @@ spec:
   podSelector:
     matchLabels:
       app: policy-test
+  externalWorkloadSelector:
+    matchLabels:
+      app: external-workload-policy-test
   port: 80
   proxyProtocol: opaque`,
 	}
@@ -445,6 +448,51 @@ spec:
   - port: 80`,
 	}
 
+	externalWorkloads := []string{`
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: my-cool-workload
+  namespace: ns
+  annotations:
+    config.linkerd.io/opaque-ports: "4242"
+spec:
+  meshTls:
+    identity: spiffe://some-domain/cool
+    serverName: server.local
+  workloadIPs:
+  - ip: 200.1.1.1
+  ports:
+  - port: 8989
+  - port: 4242
+  - name: linkerd-proxy
+    port: 4143
+status:
+  conditions:
+  ready: true`,
+		`
+apiVersion: workload.linkerd.io/v1alpha1
+kind: ExternalWorkload
+metadata:
+  name: policy-test-workload
+  namespace: ns
+  labels:
+    app: external-workload-policy-test
+spec:
+  meshTls:
+    identity: spiffe://some-domain/cool
+    serverName: server.local
+  workloadIPs:
+  - ip: 200.1.1.2
+  ports:
+  - port: 80
+  - name: linkerd-proxy
+    port: 4143
+status:
+  conditions:
+  ready: true`,
+	}
+
 	res := append(meshedPodResources, clientSP...)
 	res = append(res, unmeshedPod)
 	res = append(res, meshedOpaquePodResources...)
@@ -455,6 +503,7 @@ spec:
 	res = append(res, hostPortMapping...)
 	res = append(res, mirrorServiceResources...)
 	res = append(res, destinationCredentialsResources...)
+	res = append(res, externalWorkloads...)
 	k8sAPI, l5dClient, err := k8s.NewFakeAPIWithL5dClient(res...)
 	if err != nil {
 		t.Fatalf("NewFakeAPIWithL5dClient returned an error: %s", err)
@@ -479,9 +528,9 @@ spec:
 		t.Fatalf("initializeIndexers returned an error: %s", err)
 	}
 
-	pods, err := watcher.NewPodWatcher(k8sAPI, metadataAPI, log, defaultOpaquePorts)
+	workloads, err := watcher.NewWorkloadWatcher(k8sAPI, metadataAPI, log, defaultOpaquePorts)
 	if err != nil {
-		t.Fatalf("can't create Pods watcher: %s", err)
+		t.Fatalf("can't create Workloads watcher: %s", err)
 	}
 	endpoints, err := watcher.NewEndpointsWatcher(k8sAPI, metadataAPI, log, false, "local")
 	if err != nil {
@@ -516,7 +565,7 @@ spec:
 			IdentityTrustDomain: "trust.domain",
 			DefaultOpaquePorts:  defaultOpaquePorts,
 		},
-		pods,
+		workloads,
 		endpoints,
 		opaquePorts,
 		profiles,
diff --git a/controller/api/destination/watcher/pod_watcher_test.go b/controller/api/destination/watcher/workload_watcher_test.go
similarity index 92%
rename from controller/api/destination/watcher/pod_watcher_test.go
rename to controller/api/destination/watcher/workload_watcher_test.go
index 3e74811ee..ed4011044 100644
--- a/controller/api/destination/watcher/pod_watcher_test.go
+++ b/controller/api/destination/watcher/workload_watcher_test.go
@@ -60,7 +60,7 @@ status:
 		}
 
 		k8sAPI.Sync(nil)
-		pw := PodWatcher{
+		ww := WorkloadWatcher{
 			k8sAPI: k8sAPI,
 			log: log.WithFields(log.Fields{
 				"component": "pod-watcher",
@@ -68,7 +68,7 @@ status:
 		}
 
 		// Get host IP pod that is mapped to the port `hostPort1`
-		pod, err := pw.getPodByHostIP(hostIP, hostPort1)
+		pod, err := ww.getPodByHostIP(hostIP, hostPort1)
 		if err != nil {
 			t.Fatalf("failed to get pod: %s", err)
 		}
@@ -81,7 +81,7 @@ status:
 		// Get host IP pod that is mapped to the port `hostPort2`; this tests
 		// that the indexer properly adds multiple containers from a single
 		// pod.
-		pod, err = pw.getPodByHostIP(hostIP, hostPort2)
+		pod, err = ww.getPodByHostIP(hostIP, hostPort2)
 		if err != nil {
 			t.Fatalf("failed to get pod: %s", err)
 		}
@@ -92,7 +92,7 @@ status:
 			t.Fatalf("expected pod name to be %s, but got %s", expectedPodName, pod.Name)
 		}
 		// Get host IP pod with unmapped host port
-		pod, err = pw.getPodByHostIP(hostIP, 12347)
+		pod, err = ww.getPodByHostIP(hostIP, 12347)
 		if err != nil {
 			t.Fatalf("expected no error when getting host IP pod with unmapped host port, but got: %s", err)
 		}
@@ -100,7 +100,7 @@ status:
 			t.Fatal("expected no pod to be found with unmapped host port")
 		}
 		// Get pod IP pod and expect an error
-		_, err = pw.getPodByPodIP(podIP, 12346)
+		_, err = ww.getPodByPodIP(podIP, 12346)
 		if err == nil {
 			t.Fatal("expected error when getting by pod IP and unmapped host port, but got none")
 		}
diff --git a/test/integration/multicluster/install_test.go b/test/integration/multicluster/install_test.go
index dcff44023..a19b26587 100644
--- a/test/integration/multicluster/install_test.go
+++ b/test/integration/multicluster/install_test.go
@@ -16,6 +16,7 @@ import (
 
 	mcHealthcheck "github.com/linkerd/linkerd2/multicluster/cmd"
 	"github.com/linkerd/linkerd2/pkg/healthcheck"
+	"github.com/linkerd/linkerd2/pkg/version"
 	"github.com/linkerd/linkerd2/testutil"
 )
 
@@ -113,7 +114,7 @@ func TestInstall(t *testing.T) {
 		"install",
 		"--controller-log-level", "debug",
 		"--set", "proxyInit.image.name=ghcr.io/linkerd/proxy-init",
-		"--set", "proxyInit.image.version=v2.2.3",
+		"--set", fmt.Sprintf("proxyInit.image.version=%s", version.ProxyInitVersion),
 		"--set", fmt.Sprintf("proxy.image.version=%s", TestHelper.GetVersion()),
 		"--set", "heartbeatSchedule=1 2 3 4 5",
 		"--identity-trust-anchors-file", rootPath,