Repository: linkerd/linkerd2
Pull request #11975 (2024-01-24 03:17:28)

build(deps): bump windows_x86_64_gnullvm from 0.48.2 to 0.48.5
Bumps [windows_x86_64_gnullvm](https://github.com/microsoft/windows-rs) from 0.48.2 to 0.48.5.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/microsoft/windows-rs/releases">windows_x86_64_gnullvm's releases</a>.</em></p>
<blockquote>
<h2>0.48.5</h2>
<p>As discussed in <a href="https://redirect.github.com/microsoft/windows-rs/pull/2613">microsoft/windows-rs#2613</a>, this update bumps <code>windows-targets</code> to 0.48.5 and will be the final release to support 1.48.</p>
<p>I will then switch everything in <code>windows-rs</code> to use edition 2021 and MSRV 1.56.</p>
<h2>What's Changed</h2>
<ul>
<li>0.48.5 by <a href="https://github.com/kennykerr"><code>@​kennykerr</code></a> in <a href="https://redirect.github.com/microsoft/windows-rs/pull/2617">microsoft/windows-rs#2617</a></li>
</ul>
<p><strong>Full Changelog</strong>: <a href="https://github.com/microsoft/windows-rs/compare/0.48.4...0.48.5">https://github.com/microsoft/windows-rs/compare/0.48.4...0.48.5</a></p>
<h2>0.48.4</h2>
<p>A user reported issues with using Rust 1.48 with <code>windows-targets</code>. It is quite difficult to validate as that is such an old version of Rust, but this tweak should address the compatibility concern.</p>
<p>This just flips <code>windows-targets</code> and the arch-specific crates to use Rust edition 2018.</p>
<h2>What's Changed</h2>
<ul>
<li>0.48.4 by <a href="https://github.com/kennykerr"><code>@​kennykerr</code></a> in <a href="https://redirect.github.com/microsoft/windows-rs/pull/2613">microsoft/windows-rs#2613</a></li>
</ul>
<p><strong>Full Changelog</strong>: <a href="https://github.com/microsoft/windows-rs/compare/0.51.1...0.48.4">https://github.com/microsoft/windows-rs/compare/0.51.1...0.48.4</a></p>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li>See full diff in <a href="https://github.com/microsoft/windows-rs/commits/0.48.5">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=windows_x86_64_gnullvm&package-manager=cargo&previous-version=0.48.2&new-version=0.48.5)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

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

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

---

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

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


</details>


Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 3d522a522..ae5cd5612 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1792,13 +1792,13 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.10.2"
+version = "1.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "380b951a9c5e80ddfd6136919eef32310721aa4aacd4889a8d39124b026ab343"
+checksum = "b62dbe01f0b06f9d8dc7d49e05a0785f153b00b2c227856282f671e0318c9b15"
 dependencies = [
  "aho-corasick",
  "memchr",
- "regex-automata 0.4.3",
+ "regex-automata 0.4.4",
  "regex-syntax 0.8.2",
 ]
 
@@ -1813,9 +1813,9 @@ dependencies = [
 
 [[package]]
 name = "regex-automata"
-version = "0.4.3"
+version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5f804c7828047e88b2d32e2d7fe5a105da8ee3264f01902f796c8e067dc2483f"
+checksum = "3b7fa1134405e2ec9353fd416b17f8dacd46c473d7d3fd1cf202706a14eb792a"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -2768,7 +2768,7 @@ dependencies = [
  "windows_i686_gnu 0.48.2",
  "windows_i686_msvc 0.48.5",
  "windows_x86_64_gnu 0.48.2",
- "windows_x86_64_gnullvm 0.48.2",
+ "windows_x86_64_gnullvm 0.48.5",
  "windows_x86_64_msvc 0.48.2",
 ]
 
@@ -2849,9 +2849,9 @@ checksum = "3d6fa32db2bc4a2f5abeacf2b69f7992cd09dca97498da74a151a3132c26befd"
 
 [[package]]
 name = "windows_x86_64_gnullvm"
-version = "0.48.2"
+version = "0.48.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f1a05a1ece9a7a0d5a7ccf30ba2c33e3a61a30e042ffd247567d1de1d94120d"
+checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
 
 [[package]]
 name = "windows_x86_64_gnullvm"
diff --git a/charts/linkerd-crds/templates/workload/external-workload.yaml b/charts/linkerd-crds/templates/workload/external-workload.yaml
index d55447260..273898ac4 100644
--- a/charts/linkerd-crds/templates/workload/external-workload.yaml
+++ b/charts/linkerd-crds/templates/workload/external-workload.yaml
@@ -154,8 +154,5 @@ spec:
                       maxLength: 32768
                       type: string
                 required:
-                - lastTransitionTime
                 - status
                 - type
-                - reason
-                - message
diff --git a/controller/api/destination/external-workload/controller_util.go b/controller/api/destination/external-workload/controller_util.go
new file mode 100644
index 000000000..98937174d
--- /dev/null
+++ b/controller/api/destination/external-workload/controller_util.go
@@ -0,0 +1,231 @@
+package externalworkload
+
+import (
+	"reflect"
+
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/cache"
+)
+
+func (ec *EndpointsController) getServicesToUpdateOnExternalWorkloadChange(old, cur interface{}) sets.Set[string] {
+	newEw, newEwOk := cur.(*ewv1alpha1.ExternalWorkload)
+	oldEw, oldEwOk := old.(*ewv1alpha1.ExternalWorkload)
+
+	if !oldEwOk {
+		ec.log.Errorf("Expected (cur) to be an EndpointSlice in getServicesToUpdateOnExternalWorkloadChange(), got type: %T", cur)
+		return sets.Set[string]{}
+	}
+
+	if !newEwOk {
+		ec.log.Errorf("Expected (old) to be an EndpointSlice in getServicesToUpdateOnExternalWorkloadChange(), got type: %T", old)
+		return sets.Set[string]{}
+	}
+
+	if newEw.ResourceVersion == oldEw.ResourceVersion {
+		// Periodic resync will send update events for all known ExternalWorkloads.
+		// Two different versions of the same pod will always have different RVs
+		return sets.Set[string]{}
+	}
+
+	ewChanged, labelsChanged := ewEndpointsChanged(oldEw, newEw)
+	if !ewChanged && !labelsChanged {
+		ec.log.Errorf("skipping update; nothing has changed between old rv %s and new rv %s", oldEw.ResourceVersion, newEw.ResourceVersion)
+		return sets.Set[string]{}
+	}
+
+	services, err := ec.getExternalWorkloadSvcMembership(newEw)
+	if err != nil {
+		ec.log.Errorf("unable to get pod %s/%s's service memberships: %v", newEw.Namespace, newEw.Name, err)
+		return sets.Set[string]{}
+	}
+
+	if labelsChanged {
+		oldServices, err := ec.getExternalWorkloadSvcMembership(oldEw)
+		if err != nil {
+			ec.log.Errorf("unable to get pod %s/%s's service memberships: %v", oldEw.Namespace, oldEw.Name, err)
+		}
+		services = determineNeededServiceUpdates(oldServices, services, ewChanged)
+	}
+
+	return services
+}
+
+func determineNeededServiceUpdates(oldServices, services sets.Set[string], specChanged bool) sets.Set[string] {
+	if specChanged {
+		// if the labels and spec changed, all services need to be updated
+		services = services.Union(oldServices)
+	} else {
+		// if only the labels changed, services not common to both the new
+		// and old service set (the disjuntive union) need to be updated
+		services = services.Difference(oldServices).Union(oldServices.Difference(services))
+	}
+	return services
+}
+
+// getExternalWorkloadSvcMembership accepts a pointer to an external workload
+// resource and returns a set of service keys (<namespace>/<name>). The set
+// includes all services local to the workload's namespace that match the workload.
+func (ec *EndpointsController) getExternalWorkloadSvcMembership(workload *ewv1alpha1.ExternalWorkload) (sets.Set[string], error) {
+	keys := sets.Set[string]{}
+	services, err := ec.k8sAPI.Svc().Lister().Services(workload.Namespace).List(labels.Everything())
+	if err != nil {
+		return keys, err
+	}
+
+	for _, svc := range services {
+		if svc.Spec.Selector == nil {
+			continue
+		}
+
+		// Taken from upstream k8s code, this checks whether a given object has
+		// a deleted state before returning a `namespace/name` key. This is
+		// important since we do not want to consider a service that has been
+		// deleted and is waiting for cache eviction
+		key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(svc)
+		if err != nil {
+			return sets.Set[string]{}, err
+		}
+
+		// Check if service selects our ExternalWorkload.
+		if labels.ValidatedSetSelector(svc.Spec.Selector).Matches(labels.Set(workload.Labels)) {
+			keys.Insert(key)
+		}
+	}
+
+	return keys, nil
+}
+
+// getEndpointSliceFromDeleteAction parses an EndpointSlice from a delete action.
+func (ec *EndpointsController) getEndpointSliceFromDeleteAction(obj interface{}) *discoveryv1.EndpointSlice {
+	if endpointSlice, ok := obj.(*discoveryv1.EndpointSlice); ok {
+		// Enqueue all the services that the pod used to be a member of.
+		// This is the same thing we do when we add a pod.
+		return endpointSlice
+	}
+	// If we reached here it means the pod was deleted but its final state is unrecorded.
+	tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+	if !ok {
+		ec.log.Errorf("Couldn't get object from tombstone")
+		return nil
+	}
+	endpointSlice, ok := tombstone.Obj.(*discoveryv1.EndpointSlice)
+	if !ok {
+		ec.log.Errorf("Tombstone contained object that is not a EndpointSlice")
+		return nil
+	}
+	return endpointSlice
+}
+
+// getExternalWorkloadFromDeleteAction parses an ExternalWorkload from a delete action.
+func (ec *EndpointsController) getExternalWorkloadFromDeleteAction(obj interface{}) *ewv1alpha1.ExternalWorkload {
+	if ew, ok := obj.(*ewv1alpha1.ExternalWorkload); ok {
+		return ew
+	}
+
+	// If we reached here it means the pod was deleted but its final state is unrecorded.
+	tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+	if !ok {
+		ec.log.Errorf("couldn't get object from tombstone %#v", obj)
+		return nil
+	}
+
+	ew, ok := tombstone.Obj.(*ewv1alpha1.ExternalWorkload)
+	if !ok {
+		ec.log.Errorf("tombstone contained object that is not a ExternalWorkload: %#v", obj)
+		return nil
+	}
+	return ew
+}
+
+// ewEndpointsChanged returns two boolean values. The first is true if the ExternalWorkload has
+// changed in a way that may change existing endpoints. The second value is true if the
+// ExternalWorkload has changed in a way that may affect which Services it matches.
+func ewEndpointsChanged(oldEw, newEw *ewv1alpha1.ExternalWorkload) (bool, bool) {
+	// Check if the ExternalWorkload labels have changed, indicating a possible
+	// change in the service membership
+	labelsChanged := false
+	if !reflect.DeepEqual(newEw.Labels, oldEw.Labels) {
+		labelsChanged = true
+	}
+
+	// If the ExternalWorkload's deletion timestamp is set, remove endpoint from ready address.
+	if newEw.DeletionTimestamp != oldEw.DeletionTimestamp {
+		return true, labelsChanged
+	}
+	// If the ExternalWorkload's readiness has changed, the associated endpoint address
+	// will move from the unready endpoints set to the ready endpoints.
+	// So for the purposes of an endpoint, a readiness change on an ExternalWorkload
+	// means we have a changed ExternalWorkload.
+	if IsEwReady(oldEw) != IsEwReady(newEw) {
+		return true, labelsChanged
+	}
+
+	// Check if the ExternalWorkload IPs have changed
+	if len(oldEw.Spec.WorkloadIPs) != len(newEw.Spec.WorkloadIPs) {
+		return true, labelsChanged
+	}
+	for i := range oldEw.Spec.WorkloadIPs {
+		if oldEw.Spec.WorkloadIPs[i].Ip != newEw.Spec.WorkloadIPs[i].Ip {
+			return true, labelsChanged
+		}
+	}
+
+	// Check if the Ports  have changed
+	if len(oldEw.Spec.Ports) != len(newEw.Spec.Ports) {
+		return true, labelsChanged
+	}
+
+	// Determine if the ports have changed between workload resources
+	portSet := make(map[int32]ewv1alpha1.PortSpec)
+	for _, ps := range newEw.Spec.Ports {
+		portSet[ps.Port] = ps
+	}
+
+	for _, oldPs := range oldEw.Spec.Ports {
+		// If the port number is present in the new workload but not the old
+		// one, then we have a diff and we return early
+		newPs, ok := portSet[oldPs.Port]
+		if !ok {
+			return true, labelsChanged
+		}
+
+		// If the port is present in both workloads, we check to see if any of
+		// the port spec's values have changed, e.g. name or protocol
+		if newPs.Name != oldPs.Name || newPs.Protocol != oldPs.Protocol {
+			return true, labelsChanged
+		}
+	}
+
+	return false, labelsChanged
+}
+
+func managedByController(es *discoveryv1.EndpointSlice) bool {
+	esManagedBy := es.Labels[discoveryv1.LabelManagedBy]
+	return managedBy == esManagedBy
+}
+
+func managedByChanged(endpointSlice1, endpointSlice2 *discoveryv1.EndpointSlice) bool {
+	return managedByController(endpointSlice1) != managedByController(endpointSlice2)
+}
+
+func IsEwReady(ew *ewv1alpha1.ExternalWorkload) bool {
+	if len(ew.Status.Conditions) == 0 {
+		return false
+	}
+
+	// Loop through the conditions and look at each condition in turn starting
+	// from the top.
+	for i := range ew.Status.Conditions {
+		cond := ew.Status.Conditions[i]
+		// Stop once we find a 'Ready' condition. We expect a resource to only
+		// have one 'Ready' type condition.
+		if cond.Type == ewv1alpha1.WorkloadReady && cond.Status == ewv1alpha1.ConditionTrue {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/controller/api/destination/external-workload/endpoints_controller.go b/controller/api/destination/external-workload/endpoints_controller.go
index eafbd7ee1..a8a3331c3 100644
--- a/controller/api/destination/external-workload/endpoints_controller.go
+++ b/controller/api/destination/external-workload/endpoints_controller.go
@@ -2,6 +2,7 @@ package externalworkload
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -9,12 +10,16 @@ import (
 	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
 	"github.com/linkerd/linkerd2/controller/k8s"
 	logging "github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/leaderelection"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/util/workqueue"
+	endpointslicerec "k8s.io/endpointslice"
 )
 
 const (
@@ -32,6 +37,13 @@ const (
 	// Core controllers have a value of 2 seconds.
 	leaseRetryPeriod = 2 * time.Second
 
+	// Name of the controller. Used as an annotation value for all created
+	// EndpointSlice objects
+	managedBy = "linkerd-external-workloads-controller"
+
+	// Max number of endpoints per EndpointSlice
+	maxEndpointsQuota = 100
+
 	// Max retries for a service to be reconciled
 	maxRetryBudget = 15
 )
@@ -40,11 +52,12 @@ const (
 // by writing EndpointSlice objects for Services that select one or more
 // external endpoints.
 type EndpointsController struct {
-	k8sAPI   *k8s.API
-	log      *logging.Entry
-	queue    workqueue.RateLimitingInterface
-	stop     chan struct{}
-	isLeader atomic.Bool
+	k8sAPI     *k8s.API
+	log        *logging.Entry
+	queue      workqueue.RateLimitingInterface
+	reconciler *endpointsReconciler
+	stop       chan struct{}
+	isLeader   atomic.Bool
 
 	lec leaderelection.LeaderElectionConfig
 	sync.RWMutex
@@ -72,19 +85,20 @@ func NewEndpointsController(k8sAPI *k8s.API, hostname, controllerNs string, stop
 	}
 
 	ec := &EndpointsController{
-		k8sAPI: k8sAPI,
-		queue:  workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workQueueConfig),
-		stop:   stopCh,
+		k8sAPI:     k8sAPI,
+		reconciler: newEndpointsReconciler(k8sAPI, managedBy, maxEndpointsQuota),
+		queue:      workqueue.NewRateLimitingQueueWithConfig(workqueue.DefaultControllerRateLimiter(), workQueueConfig),
+		stop:       stopCh,
 		log: logging.WithFields(logging.Fields{
 			"component": "external-endpoints-controller",
 		}),
 	}
 
 	_, err := k8sAPI.Svc().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    ec.updateService,
-		DeleteFunc: ec.updateService,
+		AddFunc:    ec.onServiceUpdate,
+		DeleteFunc: ec.onServiceUpdate,
 		UpdateFunc: func(_, newObj interface{}) {
-			ec.updateService(newObj)
+			ec.onServiceUpdate(newObj)
 		},
 	})
 
@@ -92,89 +106,20 @@ func NewEndpointsController(k8sAPI *k8s.API, hostname, controllerNs string, stop
 		return nil, err
 	}
 
-	_, err = k8sAPI.ExtWorkload().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			ew, ok := obj.(*ewv1alpha1.ExternalWorkload)
-			if !ok {
-				ec.log.Errorf("couldn't get ExternalWorkload from object %#v", obj)
-				return
-			}
-
-			if len(ew.Spec.WorkloadIPs) == 0 {
-				ec.log.Debugf("ExternalWorkload %s/%s has no IP addresses", ew.Namespace, ew.Name)
-				return
-			}
-
-			if len(ew.Spec.Ports) == 0 {
-				ec.log.Debugf("ExternalWorkload %s/%s has no ports", ew.Namespace, ew.Name)
-				return
-			}
-
-			services, err := ec.getSvcMembership(ew)
-			if err != nil {
-				ec.log.Errorf("failed to get service membership for %s/%s: %v", ew.Namespace, ew.Name, err)
-				return
-			}
-
-			for _, svc := range services {
-				ec.queue.Add(svc)
-			}
-		},
-		DeleteFunc: func(obj interface{}) {
-			ew, ok := obj.(*ewv1alpha1.ExternalWorkload)
-			if !ok {
-				tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-				if !ok {
-					ec.log.Errorf("couldn't get object from tombstone %+v", obj)
-					return
-				}
-				ew, ok = tombstone.Obj.(*ewv1alpha1.ExternalWorkload)
-				if !ok {
-					ec.log.Errorf("tombstone contained object that is not an ExternalWorkload %+v", obj)
-					return
-				}
-			}
-
-			services, err := ec.getSvcMembership(ew)
-			if err != nil {
-				ec.log.Errorf("failed to get service membership for %s/%s: %v", ew.Namespace, ew.Name, err)
-				return
-			}
-
-			for _, svc := range services {
-				ec.queue.Add(svc)
-			}
-		},
-		UpdateFunc: func(oldObj, newObj interface{}) {
-			old, ok := oldObj.(*ewv1alpha1.ExternalWorkload)
-			if !ok {
-				ec.log.Errorf("couldn't get ExternalWorkload from object %#v", oldObj)
-				return
-			}
-
-			updated, ok := newObj.(*ewv1alpha1.ExternalWorkload)
-			if !ok {
-				ec.log.Errorf("couldn't get ExternalWorkload from object %#v", newObj)
-				return
-			}
-
-			// Ignore resync updates. If nothing has changed in the object, then
-			// the update processing is redudant.
-			if old.ResourceVersion == updated.ResourceVersion {
-				ec.log.Tracef("skipping ExternalWorkload resync update event")
-				return
-			}
+	_, err = k8sAPI.ES().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    ec.onEndpointSliceAdd,
+		UpdateFunc: ec.onEndpointSliceUpdate,
+		DeleteFunc: ec.onEndpointSliceDelete,
+	})
 
-			services, err := ec.servicesToUpdate(old, updated)
-			if err != nil {
-				ec.log.Errorf("failed to get list of services to update: %v", err)
-				return
-			}
+	if err != nil {
+		return nil, err
+	}
 
-			for _, svc := range services {
-				ec.queue.Add(svc)
-			}
-		},
+	_, err = k8sAPI.ExtWorkload().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    ec.onAddExternalWorkload,
+		DeleteFunc: ec.onDeleteExternalWorkload,
+		UpdateFunc: ec.onUpdateExternalWorkload,
 	})
 
 	if err != nil {
@@ -287,8 +232,12 @@ func (ec *EndpointsController) processQueue() {
 			return
 		}
 
-		key := item.(string)
-		err := ec.processUpdate(key)
+		key, ok := item.(string)
+		if !ok {
+			ec.log.Errorf("Found queue element of type %T, was expecting a string", item)
+			continue
+		}
+		err := ec.syncService(key)
 		ec.handleError(err, key)
 
 		// Tell the queue that we are done processing this key. This will
@@ -299,19 +248,6 @@ func (ec *EndpointsController) processQueue() {
 	}
 }
 
-// processUpdate will run a reconciliation function for a single Service object
-// that needs to have its EndpointSlice objects reconciled.
-//
-// TODO (matei): remove lint during impl of processUpdate. CI complains error is
-// always nil
-//
-//nolint:all
-func (ec *EndpointsController) processUpdate(update string) error {
-	// TODO (matei): reconciliation logic
-	ec.log.Infof("received %s", update)
-	return nil
-}
-
 // handleError will look at the result of the queue update processing step and
 // decide whether an update should be re-tried or marked as done.
 //
@@ -336,17 +272,77 @@ func (ec *EndpointsController) handleError(err error, key string) {
 	ec.log.Errorf("dropped Service %s out of update queue: %v", key, err)
 }
 
-// === Callbacks ===
+// syncService will run a reconciliation function for a single Service object
+// that needs to have its EndpointSlice objects reconciled.
+func (ec *EndpointsController) syncService(update string) error {
+	namespace, name, err := cache.SplitMetaNamespaceKey(update)
+	if err != nil {
+		return err
+	}
+
+	svc, err := ec.k8sAPI.Svc().Lister().Services(namespace).Get(name)
+	if err != nil {
+		// If the error is anything except a 'NotFound' then bubble up the error
+		// and re-queue the entry; the service will be re-processed at some
+		// point in the future.
+		if !kerrors.IsNotFound(err) {
+			return err
+		}
+
+		ec.reconciler.endpointTracker.DeleteService(namespace, name)
+		// The service has been deleted, return nil so that it won't be retried.
+		return nil
+	}
+
+	if svc.Spec.Type == corev1.ServiceTypeExternalName {
+		// services with Type ExternalName do not receive any endpoints
+		return nil
+	}
+
+	if svc.Spec.Selector == nil {
+		// services without a selector will not get any endpoints automatically
+		// created; this is done out-of-band by the service operator
+		return nil
+	}
+
+	ewSelector := labels.Set(svc.Spec.Selector).AsSelectorPreValidated()
+	ews, err := ec.k8sAPI.ExtWorkload().Lister().List(ewSelector)
+	if err != nil {
+		// This operation should be infallible since we retrieve from the cache
+		// (we can guarantee we will receive at least an empty list), for good
+		// measure, bubble up the error if one will be returned by the informer.
+		return err
+	}
+
+	esSelector := labels.Set(map[string]string{
+		discoveryv1.LabelServiceName: svc.Name,
+		discoveryv1.LabelManagedBy:   managedBy,
+	}).AsSelectorPreValidated()
+	epSlices, err := ec.k8sAPI.ES().Lister().List(esSelector)
+	if err != nil {
+		return err
+	}
+
+	epSlices = dropEndpointSlicesPendingDeletion(epSlices)
+	if ec.reconciler.endpointTracker.StaleSlices(svc, epSlices) {
+		return errors.New("EndpointSlice informer cache is out of date")
+	}
+	err = ec.reconciler.reconcile(svc, ews, epSlices)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
 
 // When a service update has been received (regardless of the event type, i.e.
 // can be Added, Modified, Deleted) send it to the endpoint controller for
 // processing.
-func (ec *EndpointsController) updateService(obj interface{}) {
-	// Use client-go's generic key function to make a key of the format
-	// <namespace>/<name>
+func (ec *EndpointsController) onServiceUpdate(obj interface{}) {
 	key, err := cache.MetaNamespaceKeyFunc(obj)
 	if err != nil {
 		ec.log.Infof("failed to get key for object %+v: %v", obj, err)
+		return
 	}
 
 	namespace, _, err := cache.SplitMetaNamespaceKey(key)
@@ -362,203 +358,116 @@ func (ec *EndpointsController) updateService(obj interface{}) {
 	ec.queue.Add(key)
 }
 
-func IsReady(ew *ewv1alpha1.ExternalWorkload) bool {
-	if len(ew.Status.Conditions) == 0 {
-		return false
-	}
-
-	// Loop through the conditions and look at each condition in turn starting
-	// from the top.
-	for i := range ew.Status.Conditions {
-		cond := ew.Status.Conditions[i]
-		// Stop once we find a 'Ready' condition. We expect a resource to only
-		// have one 'Ready' type condition.
-		if cond.Type == ewv1alpha1.WorkloadReady && cond.Status == ewv1alpha1.ConditionTrue {
-			return true
-		}
-	}
-
-	return false
-}
-
-// === Util ===
-
-// Check whether two label sets are matching
-func labelsChanged(old, updated *ewv1alpha1.ExternalWorkload) bool {
-	if len(old.Labels) != len(updated.Labels) {
-		return true
+// onEndpointSliceAdd queues a sync for the relevant Service for a sync if the
+// EndpointSlice resource version does not match the expected version in the
+// endpointSliceTracker.
+func (ec *EndpointsController) onEndpointSliceAdd(obj interface{}) {
+	es := obj.(*discoveryv1.EndpointSlice)
+	if es == nil {
+		ec.log.Info("Invalid EndpointSlice provided to onEndpointSliceAdd()")
+		return
 	}
 
-	for upK, upV := range updated.Labels {
-		oldV, ok := old.Labels[upK]
-		if !ok || oldV != upV {
-			return true
-		}
+	if managedByController(es) && ec.reconciler.endpointTracker.ShouldSync(es) {
+		ec.queueServiceForEndpointSlice(es)
 	}
-
-	return false
 }
 
-// specChanged will check whether two workload resource specs have changed
-//
-// Note: we are interested in changes to the ports, ips and readiness fields
-// since these are going to influence a change in a service's underlying
-// endpoint slice
-func specChanged(old, updated *ewv1alpha1.ExternalWorkload) bool {
-	if IsReady(old) != IsReady(updated) {
-		return true
-	}
-
-	if len(old.Spec.Ports) != len(updated.Spec.Ports) ||
-		len(old.Spec.WorkloadIPs) != len(updated.Spec.WorkloadIPs) {
-		return true
-	}
-
-	// Determine if the ports have changed between workload resources
-	portSet := make(map[int32]ewv1alpha1.PortSpec)
-	for _, ps := range updated.Spec.Ports {
-		portSet[ps.Port] = ps
+// onEndpointSliceUpdate queues a sync for the relevant Service for a sync if
+// the EndpointSlice resource version does not match the expected version in the
+// endpointSliceTracker or the managed-by value of the EndpointSlice has changed
+// from or to this controller.
+func (ec *EndpointsController) onEndpointSliceUpdate(prevObj, obj interface{}) {
+	prevEndpointSlice := prevObj.(*discoveryv1.EndpointSlice)
+	endpointSlice := obj.(*discoveryv1.EndpointSlice)
+	if endpointSlice == nil || prevEndpointSlice == nil {
+		ec.log.Info("Invalid EndpointSlice provided to onEndpointSliceUpdate()")
+		return
 	}
 
-	for _, oldPs := range old.Spec.Ports {
-		// If the port number is present in the new workload but not the old
-		// one, then we have a diff and we return early
-		newPs, ok := portSet[oldPs.Port]
-		if !ok {
-			return true
-		}
-
-		// If the port is present in both workloads, we check to see if any of
-		// the port spec's values have changed, e.g. name or protocol
-		if newPs.Name != oldPs.Name || newPs.Protocol != oldPs.Protocol {
-			return true
-		}
+	// EndpointSlice generation does not change when labels change. Although the
+	// controller will never change LabelServiceName, users might. This check
+	// ensures that we handle changes to this label.
+	svcName := endpointSlice.Labels[discoveryv1.LabelServiceName]
+	prevSvcName := prevEndpointSlice.Labels[discoveryv1.LabelServiceName]
+	if svcName != prevSvcName {
+		ec.log.Infof("label changed label: %s, oldService: %s, newService: %s, endpointsliece: %s", discoveryv1.LabelServiceName, prevSvcName, svcName, endpointSlice.Name)
+		ec.queueServiceForEndpointSlice(endpointSlice)
+		ec.queueServiceForEndpointSlice(prevEndpointSlice)
+		return
 	}
-
-	// Determine if the ips have changed between workload resources. If an IP
-	// is documented for one workload but not the other, then we have a diff.
-	ipSet := make(map[string]struct{})
-	for _, addr := range updated.Spec.WorkloadIPs {
-		ipSet[addr.Ip] = struct{}{}
+	if managedByChanged(prevEndpointSlice, endpointSlice) ||
+		(managedByController(endpointSlice) && ec.reconciler.endpointTracker.ShouldSync(endpointSlice)) {
+		ec.queueServiceForEndpointSlice(endpointSlice)
 	}
+}
 
-	for _, addr := range old.Spec.WorkloadIPs {
-		if _, ok := ipSet[addr.Ip]; !ok {
-			return true
+// onEndpointSliceDelete queues a sync for the relevant Service for a sync if the
+// EndpointSlice resource version does not match the expected version in the
+// endpointSliceTracker.
+func (ec *EndpointsController) onEndpointSliceDelete(obj interface{}) {
+	endpointSlice := ec.getEndpointSliceFromDeleteAction(obj)
+	if endpointSlice != nil && managedByController(endpointSlice) && ec.reconciler.endpointTracker.Has(endpointSlice) {
+		// This returns false if we didn't expect the EndpointSlice to be
+		// deleted. If that is the case, we queue the Service for another sync.
+		if !ec.reconciler.endpointTracker.HandleDeletion(endpointSlice) {
+			ec.queueServiceForEndpointSlice(endpointSlice)
 		}
 	}
-
-	return false
 }
 
-func toSet(s []string) map[string]struct{} {
-	set := map[string]struct{}{}
-	for _, k := range s {
-		set[k] = struct{}{}
+// queueServiceForEndpointSlice attempts to queue the corresponding Service for
+// the provided EndpointSlice.
+func (ec *EndpointsController) queueServiceForEndpointSlice(endpointSlice *discoveryv1.EndpointSlice) {
+	key, err := endpointslicerec.ServiceControllerKey(endpointSlice)
+	if err != nil {
+		ec.log.Errorf("Couldn't get key for EndpointSlice %+v: %v", endpointSlice, err)
+		return
 	}
-	return set
+
+	ec.queue.Add(key)
 }
 
-// servicesToUpdate will look at an old and an updated external workload
-// resource and determine which services need to be reconciled. The outcome is
-// determined by what has changed in-between resources (selections, spec, or
-// both).
-func (ec *EndpointsController) servicesToUpdate(old, updated *ewv1alpha1.ExternalWorkload) ([]string, error) {
-	labelsChanged := labelsChanged(old, updated)
-	specChanged := specChanged(old, updated)
-	if !labelsChanged && !specChanged {
-		ec.log.Debugf("skipping update; nothing has changed between old rv %s and new rv %s", old.ResourceVersion, updated.ResourceVersion)
-		return nil, nil
+func (ec *EndpointsController) onAddExternalWorkload(obj interface{}) {
+	ew, ok := obj.(*ewv1alpha1.ExternalWorkload)
+	if !ok {
+		ec.log.Errorf("couldn't get ExternalWorkload from object %#v", obj)
+		return
 	}
 
-	newSelection, err := ec.getSvcMembership(updated)
+	services, err := ec.getExternalWorkloadSvcMembership(ew)
 	if err != nil {
-		return nil, err
-
+		ec.log.Errorf("failed to get service membership for %s/%s: %v", ew.Namespace, ew.Name, err)
+		return
 	}
 
-	oldSelection, err := ec.getSvcMembership(old)
-	if err != nil {
-		return nil, err
+	for svc := range services {
+		ec.queue.Add(svc)
 	}
+}
 
-	result := map[string]struct{}{}
-	// Determine the list of services we need to update based on the difference
-	// between our old and updated workload.
-	//
-	// Service selections (i.e. services that select a workload through a label
-	// selector) may reference an old workload, a new workload, or both,
-	// depending on the workload's labels.
-	if labelsChanged && specChanged {
-		// When the selection has changed, and the workload has changed, all
-		// services need to be updated so we consider the union of selections.
-		result = toSet(append(newSelection, oldSelection...))
-	} else if specChanged {
-		// When the workload resource has changed, it is enough to consider
-		// either the oldSelection slice or the newSelection slice, since they
-		// are equal. We have the same set of services to update since no
-		// selection has been changed by the update.
-		return newSelection, nil
-	} else {
-		// When the selection has changed, then we need to consider only
-		// services that reference the old workload's labels, or the new
-		// workload's labels, but not both. Services that select both are
-		// unchanged since the workload has not changed.
-		newSelectionSet := toSet(newSelection)
-		oldSelectionSet := toSet(oldSelection)
-
-		// Determine selections that reference only the old workload resource
-		for _, oldSvc := range oldSelection {
-			if _, ok := newSelectionSet[oldSvc]; !ok {
-				result[oldSvc] = struct{}{}
-			}
-		}
-
-		// Determine selections that reference only the new workload resource
-		for _, newSvc := range newSelection {
-			if _, ok := oldSelectionSet[newSvc]; !ok {
-				result[newSvc] = struct{}{}
-			}
-		}
-	}
+func (ec *EndpointsController) onUpdateExternalWorkload(old, cur interface{}) {
+	services := ec.getServicesToUpdateOnExternalWorkloadChange(old, cur)
 
-	var resultSlice []string
-	for svc := range result {
-		resultSlice = append(resultSlice, svc)
+	for svc := range services {
+		ec.queue.Add(svc)
 	}
-
-	return resultSlice, nil
 }
 
-// getSvcMembership accepts a pointer to an external workload resource and
-// returns a set of service keys (<namespace>/<name>). The set includes all
-// services local to the workload's namespace that match the workload.
-func (ec *EndpointsController) getSvcMembership(workload *ewv1alpha1.ExternalWorkload) ([]string, error) {
-	keys := []string{}
-	services, err := ec.k8sAPI.Svc().Lister().Services(workload.Namespace).List(labels.Everything())
-	if err != nil {
-		return keys, err
+func (ec *EndpointsController) onDeleteExternalWorkload(obj interface{}) {
+	ew := ec.getExternalWorkloadFromDeleteAction(obj)
+	if ew != nil {
+		ec.onAddExternalWorkload(ew)
 	}
+}
 
-	for _, svc := range services {
-		if svc.Spec.Selector == nil {
-			continue
-		}
-
-		// Taken from upstream k8s code, this checks whether a given object has
-		// a deleted state before returning a `namespace/name` key. This is
-		// important since we do not want to consider a service that has been
-		// deleted and is waiting for cache eviction
-		key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(svc)
-		if err != nil {
-			return []string{}, err
-		}
-
-		// Check if service selects our ExternalWorkload.
-		if labels.ValidatedSetSelector(svc.Spec.Selector).Matches(labels.Set(workload.Labels)) {
-			keys = append(keys, key)
+func dropEndpointSlicesPendingDeletion(endpointSlices []*discoveryv1.EndpointSlice) []*discoveryv1.EndpointSlice {
+	n := 0
+	for _, endpointSlice := range endpointSlices {
+		if endpointSlice.DeletionTimestamp == nil {
+			endpointSlices[n] = endpointSlice
+			n++
 		}
 	}
-
-	return keys, nil
+	return endpointSlices[:n]
 }
diff --git a/controller/api/destination/external-workload/endpoints_reconciler.go b/controller/api/destination/external-workload/endpoints_reconciler.go
new file mode 100644
index 000000000..8b3d1a5ab
--- /dev/null
+++ b/controller/api/destination/external-workload/endpoints_reconciler.go
@@ -0,0 +1,703 @@
+package externalworkload
+
+import (
+	"context"
+	"fmt"
+	"sort"
+
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
+	"github.com/linkerd/linkerd2/controller/k8s"
+	logging "github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	epsliceutil "k8s.io/endpointslice/util"
+	utilnet "k8s.io/utils/net"
+)
+
+// endpointsReconciler is a subcomponent of the EndpointsController.
+//
+// Its main responsibility is to reconcile a service's endpoints (by diffing
+// states) and keeping track of any drifts between the informer cache and what
+// has been written to the API Server
+type endpointsReconciler struct {
+	k8sAPI         *k8s.API
+	log            *logging.Entry
+	controllerName string
+	// Upstream utility component that will internally track the most recent
+	// resourceVersion observed for an EndpointSlice
+	endpointTracker *epsliceutil.EndpointSliceTracker
+	maxEndpoints    int
+	// TODO (matei): add metrics around events
+}
+
+// endpointMeta is a helper struct that incldues attributes slices will be
+// grouped on (i.e. ports and the address family supported).
+//
+// Note: this is inspired from the upstream EndpointSlice controller impl.
+type endpointMeta struct {
+	ports       []discoveryv1.EndpointPort
+	addressType discoveryv1.AddressType
+}
+
+// newEndpointsReconciler takes an API client and returns a reconciler with
+// logging and a tracker set-up
+func newEndpointsReconciler(k8sAPI *k8s.API, controllerName string, maxEndpoints int) *endpointsReconciler {
+	return &endpointsReconciler{
+		k8sAPI,
+		logging.WithFields(logging.Fields{
+			"component": "external-endpoints-reconciler",
+		}),
+		controllerName,
+		epsliceutil.NewEndpointSliceTracker(),
+		maxEndpoints,
+	}
+
+}
+
+// === Reconciler ===
+
+// reconcile is the main entry-point for the reconciler's work.
+//
+// It accepts a slice of external workloads and their corresponding service.
+// Optionally, if the controller has previously created any slices for this
+// service, these will also be passed in. The reconciler will:
+//
+// * Determine what address types the service supports
+// * For each address type, it will determine which slices to process (an
+// EndpointSlice is specialised and supports only one type)
+func (r *endpointsReconciler) reconcile(svc *corev1.Service, ews []*ewv1alpha1.ExternalWorkload, existingSlices []*discoveryv1.EndpointSlice) error {
+	toDelete := []*discoveryv1.EndpointSlice{}
+	slicesByAddrType := make(map[discoveryv1.AddressType][]*discoveryv1.EndpointSlice)
+	errs := []error{}
+
+	// Get the list of supported address types for the service
+	supportedAddrTypes := getSupportedAddressTypes(svc)
+	for _, slice := range existingSlices {
+		// If a slice has an address type that the slice does not support, then
+		// it should be deleted
+		_, supported := supportedAddrTypes[slice.AddressType]
+		if !supported {
+			toDelete = append(toDelete, slice)
+			continue
+		}
+
+		// If this is the first time we see this address type, create the list
+		// in the set.
+		if _, ok := slicesByAddrType[slice.AddressType]; !ok {
+			slicesByAddrType[slice.AddressType] = []*discoveryv1.EndpointSlice{}
+		}
+
+		slicesByAddrType[slice.AddressType] = append(slicesByAddrType[slice.AddressType], slice)
+	}
+
+	// For each supported address type, reconcile endpoint slices that match the
+	// given type
+	for addrType := range supportedAddrTypes {
+		existingSlices := slicesByAddrType[addrType]
+		err := r.reconcileByAddressType(svc, ews, existingSlices, addrType)
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	// delete services whose address type is no longer supported by the service
+	for _, slice := range toDelete {
+		err := r.k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Delete(context.TODO(), slice.Name, metav1.DeleteOptions{})
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	return utilerrors.NewAggregate(errs)
+}
+
+// reconcileIPv4Endpoints operates on a set of external workloads, their
+// service, and any endpointslices that have been created by the controller. It
+// will compute the diff that needs to be written to the API Server.
+func (r *endpointsReconciler) reconcileByAddressType(svc *corev1.Service, extWorkloads []*ewv1alpha1.ExternalWorkload, existingSlices []*discoveryv1.EndpointSlice, addrType discoveryv1.AddressType) error {
+	slicesToCreate := []*discoveryv1.EndpointSlice{}
+	slicesToUpdate := []*discoveryv1.EndpointSlice{}
+	slicesToDelete := []*discoveryv1.EndpointSlice{}
+
+	// We start the reconciliation by checking ownerRefs
+	//
+	// We follow the upstream here and look at our existing slices and segment
+	// by ports.
+	existingSlicesByPorts := map[epsliceutil.PortMapKey][]*discoveryv1.EndpointSlice{}
+	for _, slice := range existingSlices {
+		// Loop through the endpointslices and figure out which endpointslice
+		// does not have an ownerRef set to the service. If a slice has been
+		// selected but does not point to the service, we delete it.
+		if ownedBy(slice, svc) {
+			hash := epsliceutil.NewPortMapKey(slice.Ports)
+			existingSlicesByPorts[hash] = append(existingSlicesByPorts[hash], slice)
+		} else {
+			slicesToDelete = append(slicesToDelete, slice)
+		}
+	}
+
+	// desiredEndpointsByPortMap represents a set of endpoints grouped together
+	// by the list of ports they use. These are the endpoints that we will keep
+	// and write to the API server.
+	desiredEndpointsByPortMap := map[epsliceutil.PortMapKey]epsliceutil.EndpointSet{}
+	// desiredMetaByPortMap represents grouping metadata keyed off by the same
+	// hashed port list as the endpoints.
+	desiredMetaByPortMap := map[epsliceutil.PortMapKey]*endpointMeta{}
+
+	for _, extWorkload := range extWorkloads {
+		// We skip workloads with no IPs.
+		//
+		// Note: workloads only have a 'Ready' status so we do not care about
+		// other status conditions.
+		if len(extWorkload.Spec.WorkloadIPs) == 0 {
+			continue
+		}
+
+		// Find which ports a service selects (or maps to) on an external workload
+		// Note: we require all workload ports are documented. Pods do not have
+		// to document all of their container ports.
+		ports := r.findEndpointPorts(svc, extWorkload)
+		portHash := epsliceutil.NewPortMapKey(ports)
+		if _, ok := desiredMetaByPortMap[portHash]; !ok {
+			desiredMetaByPortMap[portHash] = &endpointMeta{ports, addrType}
+		}
+
+		if _, ok := desiredEndpointsByPortMap[portHash]; !ok {
+			desiredEndpointsByPortMap[portHash] = epsliceutil.EndpointSet{}
+		}
+
+		ep := externalWorkloadToEndpoint(addrType, extWorkload, svc)
+		if len(ep.Addresses) > 0 {
+			desiredEndpointsByPortMap[portHash].Insert(&ep)
+		}
+	}
+
+	for portKey, desiredEndpoints := range desiredEndpointsByPortMap {
+		create, update, del := r.reconcileEndpointsByPortMap(svc, existingSlicesByPorts[portKey], desiredEndpoints, desiredMetaByPortMap[portKey])
+		slicesToCreate = append(slicesToCreate, create...)
+		slicesToUpdate = append(slicesToUpdate, update...)
+		slicesToDelete = append(slicesToDelete, del...)
+	}
+
+	// If there are any slices whose ports no longer match what we want in our
+	// current reconciliation, delete them
+	for portHash, existingSlices := range existingSlicesByPorts {
+		if _, ok := desiredEndpointsByPortMap[portHash]; !ok {
+			slicesToDelete = append(slicesToDelete, existingSlices...)
+		}
+	}
+
+	return r.finalize(svc, slicesToCreate, slicesToUpdate, slicesToDelete)
+}
+
+// reconcileEndpointsByPortMap will compute the state diff to be written to the
+// API Server for a service. The function takes into account any existing
+// endpoint slices and any external workloads matched by the service.
+// The function works on slices and workloads that have been already grouped by
+// a common set of ports.
+func (r *endpointsReconciler) reconcileEndpointsByPortMap(svc *corev1.Service, existingSlices []*discoveryv1.EndpointSlice, desiredEps epsliceutil.EndpointSet, desiredMeta *endpointMeta) ([]*discoveryv1.EndpointSlice, []*discoveryv1.EndpointSlice, []*discoveryv1.EndpointSlice) {
+	slicesByName := map[string]*discoveryv1.EndpointSlice{}
+	sliceNamesUnchanged := map[string]struct{}{}
+	sliceNamesToUpdate := map[string]struct{}{}
+	sliceNamesToDelete := map[string]struct{}{}
+
+	// 1. Figure out which endpoints are no longer required in the existing
+	// slices, and update endpoints that have changed
+	for _, existingSlice := range existingSlices {
+		slicesByName[existingSlice.Name] = existingSlice
+		keepEndpoints := []discoveryv1.Endpoint{}
+		epUpdated := false
+		for _, endpoint := range existingSlice.Endpoints {
+			endpoint := endpoint // pin
+			found := desiredEps.Get(&endpoint)
+			// If the endpoint is desired (i.e. a workload exists with an IP and
+			// we want to add it to the service's endpoints), then we should
+			// keep it.
+			if found != nil {
+				keepEndpoints = append(keepEndpoints, *found)
+				// We know the slice already contains an endpoint we want, but
+				// has the endpoint changed? If yes, we need to persist it
+				if !epsliceutil.EndpointsEqualBeyondHash(found, &endpoint) {
+					epUpdated = true
+				}
+
+				// Once an endpoint has been found in a slice, we can delete it
+				desiredEps.Delete(&endpoint)
+			}
+		}
+
+		// Re-generate labels and see whether service's labels have changed
+		labels, labelsChanged := setEndpointSliceLabels(existingSlice, svc, r.controllerName)
+
+		// Consider what kind of reconciliation we should proceed with:
+		//
+		// 1. We can have a set of endpoints that have changed; this can either
+		// mean we need to update the endpoints, or it can also mean we have no
+		// endpoints to keep.
+		// 2. We need to update the slice's metadata because labels have
+		// changed.
+		// 3. Slice remains unchanged so we have a noop on our hands
+		if epUpdated || len(existingSlice.Endpoints) != len(keepEndpoints) {
+			if len(keepEndpoints) == 0 {
+				// When there are no endpoints to keep, then the slice should be
+				// deleted
+				sliceNamesToDelete[existingSlice.Name] = struct{}{}
+			} else {
+				// There is at least one endpoint to keep / update
+				slice := existingSlice.DeepCopy()
+				slice.Labels = labels
+				slice.Endpoints = keepEndpoints
+				sliceNamesToUpdate[slice.Name] = struct{}{}
+				slicesByName[slice.Name] = slice
+			}
+		} else if labelsChanged {
+			slice := existingSlice.DeepCopy()
+			slice.Labels = labels
+			sliceNamesToUpdate[slice.Name] = struct{}{}
+			slicesByName[slice.Name] = slice
+		} else {
+			// Unchanged, we save it for later.
+			// unchanged slices may receive new endpoints that are leftover if
+			// they're not past their quotaca
+			sliceNamesUnchanged[existingSlice.Name] = struct{}{}
+		}
+	}
+
+	// 2. If we still have desired endpoints left, but they haven't matched any
+	// endpoint that already exists in a slice, we need to add it somewhere.
+	//
+	// We start by adding our leftover endpoints to the list of endpoints we
+	// will update anyway (to save a write).
+	if desiredEps.Len() > 0 && len(sliceNamesToUpdate) > 0 {
+		slices := []*discoveryv1.EndpointSlice{}
+		for sliceName := range sliceNamesToUpdate {
+			slices = append(slices, slicesByName[sliceName])
+		}
+
+		// Sort in descending order of capacity; fullest first.
+		sort.Slice(slices, func(i, j int) bool {
+			return len(slices[i].Endpoints) > len(slices[j].Endpoints)
+		})
+
+		// Iterate and fill up the slices
+		for _, slice := range slices {
+			for desiredEps.Len() > 0 && len(slice.Endpoints) < r.maxEndpoints {
+				ep, _ := desiredEps.PopAny()
+				slice.Endpoints = append(slice.Endpoints, *ep)
+			}
+		}
+	}
+
+	// If we have remaining endpoints, we need to deal with them
+	// by using unchanged slices or creating new ones
+	slicesToCreate := []*discoveryv1.EndpointSlice{}
+	for desiredEps.Len() > 0 {
+		var sliceToFill *discoveryv1.EndpointSlice
+
+		// Deal with any remaining endpoints by:
+		// (a) adding to unchanged slices first
+		if desiredEps.Len() < r.maxEndpoints && len(sliceNamesUnchanged) > 0 {
+			unchangedSlices := []*discoveryv1.EndpointSlice{}
+			for unchangedSlice := range sliceNamesUnchanged {
+				unchangedSlices = append(unchangedSlices, slicesByName[unchangedSlice])
+			}
+
+			sliceToFill = getSliceToFill(unchangedSlices, desiredEps.Len(), r.maxEndpoints)
+		}
+
+		// If we have no unchanged slice to fill, then
+		// (b) create a new slice
+		if sliceToFill == nil {
+			sliceToFill = newEndpointSlice(svc, desiredMeta, r.controllerName)
+		} else {
+			// deep copy required to mutate slice
+			sliceToFill = sliceToFill.DeepCopy()
+			slicesByName[sliceToFill.Name] = sliceToFill
+		}
+
+		// Fill out the slice
+		for desiredEps.Len() > 0 && len(sliceToFill.Endpoints) < r.maxEndpoints {
+			ep, _ := desiredEps.PopAny()
+			sliceToFill.Endpoints = append(sliceToFill.Endpoints, *ep)
+		}
+
+		// Figure out what kind of slice we just filled and update the diffed
+		// state
+		if sliceToFill.Name != "" {
+			sliceNamesToUpdate[sliceToFill.Name] = struct{}{}
+			delete(sliceNamesUnchanged, sliceToFill.Name)
+		} else {
+			slicesToCreate = append(slicesToCreate, sliceToFill)
+		}
+	}
+
+	slicesToUpdate := []*discoveryv1.EndpointSlice{}
+	for name := range sliceNamesToUpdate {
+		slicesToUpdate = append(slicesToUpdate, slicesByName[name])
+	}
+
+	slicesToDelete := []*discoveryv1.EndpointSlice{}
+	for name := range sliceNamesToDelete {
+		slicesToDelete = append(slicesToDelete, slicesByName[name])
+	}
+
+	return slicesToCreate, slicesToUpdate, slicesToDelete
+}
+
+// finalize performs writes to the API Server to update the state after it's
+// been diffed.
+func (r *endpointsReconciler) finalize(svc *corev1.Service, slicesToCreate, slicesToUpdate, slicesToDelete []*discoveryv1.EndpointSlice) error {
+	// If there are slices to create and delete, change the creates to updates
+	// of the slices that would otherwise be deleted.
+	for i := 0; i < len(slicesToDelete); {
+		if len(slicesToCreate) == 0 {
+			break
+		}
+		sliceToDelete := slicesToDelete[i]
+		slice := slicesToCreate[len(slicesToCreate)-1]
+		// Only update EndpointSlices that are owned by this Service and have
+		// the same AddressType. We need to avoid updating EndpointSlices that
+		// are being garbage collected for an old Service with the same name.
+		// The AddressType field is immutable. Since Services also consider
+		// IPFamily immutable, the only case where this should matter will be
+		// the migration from IP to IPv4 and IPv6 AddressTypes, where there's a
+		// chance EndpointSlices with an IP AddressType would otherwise be
+		// updated to IPv4 or IPv6 without this check.
+		if sliceToDelete.AddressType == slice.AddressType && ownedBy(sliceToDelete, svc) {
+			slice.Name = sliceToDelete.Name
+			slicesToCreate = slicesToCreate[:len(slicesToCreate)-1]
+			slicesToUpdate = append(slicesToUpdate, slice)
+			slicesToDelete = append(slicesToDelete[:i], slicesToDelete[i+1:]...)
+		} else {
+			i++
+		}
+	}
+
+	r.log.Debugf("reconciliation result for %s/%s: %d to add, %d to update, %d to remove", svc.Namespace, svc.Name, len(slicesToCreate), len(slicesToUpdate), len(slicesToDelete))
+
+	// Create EndpointSlices only if the service has not been marked for
+	// deletion; according to the upstream implementation not doing so has the
+	// potential to cause race conditions
+	if svc.DeletionTimestamp == nil {
+		// TODO: context with timeout
+		for _, slice := range slicesToCreate {
+			r.log.Tracef("starting create: %s/%s", slice.Namespace, slice.Name)
+			createdSlice, err := r.k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Create(context.TODO(), slice, metav1.CreateOptions{})
+			if err != nil {
+				// If the namespace  is terminating, operations will not
+				// succeed. Drop the entire reconiliation effort
+				if errors.HasStatusCause(err, corev1.NamespaceTerminatingCause) {
+					return nil
+				}
+
+				return err
+			}
+			r.endpointTracker.Update(createdSlice)
+			r.log.Tracef("finished creating: %s/%s", createdSlice.Namespace, createdSlice.Name)
+		}
+	}
+
+	for _, slice := range slicesToUpdate {
+		r.log.Tracef("starting update: %s/%s", slice.Namespace, slice.Name)
+		updatedSlice, err := r.k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Update(context.TODO(), slice, metav1.UpdateOptions{})
+		if err != nil {
+			return err
+		}
+		r.endpointTracker.Update(updatedSlice)
+		r.log.Tracef("finished updating: %s/%s", updatedSlice.Namespace, updatedSlice.Name)
+	}
+
+	for _, slice := range slicesToDelete {
+		r.log.Tracef("starting delete: %s/%s", slice.Namespace, slice.Name)
+		err := r.k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Delete(context.TODO(), slice.Name, metav1.DeleteOptions{})
+		if err != nil {
+			return err
+		}
+		r.endpointTracker.ExpectDeletion(slice)
+		r.log.Tracef("finished deleting: %s/%s", slice.Namespace, slice.Name)
+	}
+
+	return nil
+}
+
+// === Utility ===
+
+// Creates a new endpointslice object
+func newEndpointSlice(svc *corev1.Service, meta *endpointMeta, controllerName string) *discoveryv1.EndpointSlice {
+	// We need an ownerRef to point to our service
+	ownerRef := metav1.NewControllerRef(svc, schema.GroupVersionKind{Version: "v1", Kind: "Service"})
+	slice := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName:    fmt.Sprintf("linkerd-external-%s", svc.Name),
+			Namespace:       svc.Namespace,
+			Labels:          map[string]string{},
+			OwnerReferences: []metav1.OwnerReference{*ownerRef},
+		},
+		AddressType: meta.addressType,
+		Endpoints:   []discoveryv1.Endpoint{},
+		Ports:       meta.ports,
+	}
+	labels, _ := setEndpointSliceLabels(slice, svc, controllerName)
+	slice.Labels = labels
+	return slice
+}
+
+// getSliceToFill will return an endpoint slice from a list of endpoint slices
+// whose capacity is closest to being full when numEndpoints are added. If no
+// slice fits the criteria a nil pointer is returned
+func getSliceToFill(slices []*discoveryv1.EndpointSlice, numEndpoints, maxEndpoints int) *discoveryv1.EndpointSlice {
+	closestDiff := maxEndpoints
+	var closestSlice *discoveryv1.EndpointSlice
+	for _, slice := range slices {
+		diff := maxEndpoints - (numEndpoints + len(slice.Endpoints))
+		if diff >= 0 && diff < closestDiff {
+			closestDiff = diff
+			closestSlice = slice
+			if closestDiff == 0 {
+				return closestSlice
+			}
+		}
+	}
+	return closestSlice
+}
+
+// setEndpointSliceLabels returns a new map with the new endpoint slice labels,
+// and returns true if there was an update.
+//
+// Slice labels should always be equivalent to Service labels, except for a
+// reserved IsHeadlessService, LabelServiceName, and LabelManagedBy. If any
+// reserved labels have changed on the service, they are not copied over.
+//
+// copied from https://github.com/kubernetes/endpointslice/commit/a09c1c9580d13f5020248d25c7fd11f5dde6dd9b
+// copyright 2019 The Kubernetes Authors
+func setEndpointSliceLabels(es *discoveryv1.EndpointSlice, service *corev1.Service, controllerName string) (map[string]string, bool) {
+	isReserved := func(label string) bool {
+		if label == discoveryv1.LabelServiceName ||
+			label == discoveryv1.LabelManagedBy ||
+			label == corev1.IsHeadlessService {
+			return true
+		}
+		return false
+	}
+
+	updated := false
+	epLabels := make(map[string]string)
+	svcLabels := make(map[string]string)
+
+	// check if the endpoint slice and the service have the same labels
+	// clone current slice labels except the reserved labels
+	for key, value := range es.Labels {
+		if isReserved(key) {
+			continue
+		}
+		// copy endpoint slice labels
+		epLabels[key] = value
+	}
+
+	for key, value := range service.Labels {
+		if isReserved(key) {
+			continue
+		}
+		// copy service labels
+		svcLabels[key] = value
+	}
+
+	// if the labels are not identical update the slice with the corresponding service labels
+	for svcLabelKey, svcLabelVal := range svcLabels {
+		epLabelVal, found := epLabels[svcLabelKey]
+		if !found {
+			updated = true
+			break
+		}
+
+		if svcLabelVal != epLabelVal {
+			updated = true
+			break
+		}
+	}
+
+	// add or remove headless label depending on the service Type
+	if service.Spec.ClusterIP == corev1.ClusterIPNone {
+		svcLabels[corev1.IsHeadlessService] = ""
+	} else {
+		delete(svcLabels, corev1.IsHeadlessService)
+	}
+
+	// override endpoint slices reserved labels
+	svcLabels[discoveryv1.LabelServiceName] = service.Name
+	svcLabels[discoveryv1.LabelManagedBy] = controllerName
+
+	return svcLabels, updated
+}
+
+func externalWorkloadToEndpoint(addrType discoveryv1.AddressType, ew *ewv1alpha1.ExternalWorkload, svc *corev1.Service) discoveryv1.Endpoint {
+	// Note: an ExternalWorkload does not have the same lifecycle as a pod; we
+	// do not mark a workload as "Terminating". Because of that, our code is
+	// simpler than the upstream and we never have to consider:
+	// * publishNotReadyAddresses (found on a service)
+	// * deletionTimestamps (found normally on a pod)
+	// * or a terminating flag on the endpoint
+	serving := IsEwReady(ew)
+
+	addresses := []string{}
+	// We assume the workload has been validated beforehand and contains a valid
+	// IP address regardless of its address family.
+	for _, addr := range ew.Spec.WorkloadIPs {
+		ip := addr.Ip
+		isIPv6 := utilnet.IsIPv6String(ip)
+		if isIPv6 && addrType == discoveryv1.AddressTypeIPv6 {
+			addresses = append(addresses, ip)
+		} else if !isIPv6 && addrType == discoveryv1.AddressTypeIPv4 {
+			addresses = append(addresses, ip)
+		}
+	}
+
+	terminating := false
+	ep := discoveryv1.Endpoint{
+		Addresses: addresses,
+		Conditions: discoveryv1.EndpointConditions{
+			Ready:       &serving,
+			Serving:     &serving,
+			Terminating: &terminating,
+		},
+		TargetRef: &corev1.ObjectReference{
+			Kind:      "ExternalWorkload",
+			Namespace: ew.Namespace,
+			Name:      ew.Name,
+			UID:       ew.UID,
+		},
+	}
+
+	zone, ok := ew.Labels[corev1.LabelTopologyZone]
+	if ok {
+		ep.Zone = &zone
+	}
+
+	// Add a hostname conditionally
+	// Note: upstream does this a bit differently; pods may include a hostname
+	// as part of their spec. We consider a hostname as long as the service is
+	// headless since that's what we would use a hostname for when routing in
+	// linkerd (we care about DNS record creation)
+	if svc.Spec.ClusterIP == corev1.ClusterIPNone && ew.Namespace == svc.Namespace {
+		ep.Hostname = &ew.Name
+	}
+
+	return ep
+}
+
+func ownedBy(slice *discoveryv1.EndpointSlice, svc *corev1.Service) bool {
+	for _, o := range slice.OwnerReferences {
+		if o.UID == svc.UID && o.Kind == "Service" && o.APIVersion == "v1" {
+			return true
+		}
+	}
+	return false
+}
+
+// findEndpointPorts is a utility function that will return a list of ports
+// that are documented on an external workload and selected by a service
+func (r *endpointsReconciler) findEndpointPorts(svc *corev1.Service, ew *ewv1alpha1.ExternalWorkload) []discoveryv1.EndpointPort {
+	epPorts := []discoveryv1.EndpointPort{}
+	// If we are dealing with a headless service, upstream implementation allows
+	// the service not to have any ports
+	if len(svc.Spec.Ports) == 0 && svc.Spec.ClusterIP == corev1.ClusterIPNone {
+		return epPorts
+	}
+
+	for _, svcPort := range svc.Spec.Ports {
+		svcPort := svcPort // pin
+		portNum, err := findWorkloadPort(ew, &svcPort)
+		if err != nil {
+			r.log.Errorf("failed to find port for service %s/%s: %v", svc.Namespace, svc.Name, err)
+			continue
+		}
+
+		portName := &svcPort.Name
+		if *portName == "" {
+			portName = nil
+		}
+		portProto := &svcPort.Protocol
+		if *portProto == "" {
+			portProto = nil
+		}
+		epPorts = append(epPorts, discoveryv1.EndpointPort{
+			Name:     portName,
+			Port:     &portNum,
+			Protocol: portProto,
+		})
+	}
+
+	return epPorts
+}
+
+// findWorkloadPort is provided a service port and an external workload and
+// checks whether the workload documents in its spec the target port referenced
+// by the service.
+//
+// adapted from copied from k8s.io/kubernetes/pkg/api/v1/pod
+func findWorkloadPort(ew *ewv1alpha1.ExternalWorkload, svcPort *corev1.ServicePort) (int32, error) {
+	targetPort := svcPort.TargetPort
+	switch targetPort.Type {
+	case intstr.String:
+		name := targetPort.StrVal
+		for _, wPort := range ew.Spec.Ports {
+			if wPort.Name == name && wPort.Protocol == svcPort.Protocol {
+				return wPort.Port, nil
+			}
+		}
+	case intstr.Int:
+		// Ensure the port is documented in the workload spec, since we
+		// require it.
+		// Upstream version allows for undocumented container ports here (i.e.
+		// it returns the int value).
+		for _, wPort := range ew.Spec.Ports {
+			port := int32(targetPort.IntValue())
+			if wPort.Port == port && wPort.Protocol == svcPort.Protocol {
+				return port, nil
+			}
+		}
+	}
+	return 0, fmt.Errorf("no suitable port for targetPort %s on workload %s/%s", targetPort.String(), ew.Namespace, ew.Name)
+}
+
+// getSupportedAddressTypes will return a set of address families (AF) supported
+// by this service. A service may be IPv4 or IPv6 only, or it may be dual-stack.
+func getSupportedAddressTypes(svc *corev1.Service) map[discoveryv1.AddressType]struct{} {
+	afs := map[discoveryv1.AddressType]struct{}{}
+	// Field only applies to LoadBalancer, ClusterIP and NodePort services. A
+	// headless service will not receive any IP families; it may hold max 2
+	// entries and can be mutated (although the 'primary' choice is never
+	// removed).
+	// See client-go type documentation for more info.
+	for _, af := range svc.Spec.IPFamilies {
+		if af == corev1.IPv4Protocol {
+			afs[discoveryv1.AddressTypeIPv4] = struct{}{}
+		} else if af == corev1.IPv6Protocol {
+			afs[discoveryv1.AddressTypeIPv6] = struct{}{}
+		}
+	}
+
+	if len(afs) > 0 {
+		// If we appended at least one address family, it means we didn't have
+		// to deal with a headless service.
+		return afs
+	}
+
+	// Note: our logic will differ from the upstream Kubernetes controller.
+	// Specifically, our minimum k8s version is greater than v1.20. Upstream
+	// controller needs to handle an upgrade path from v1.19 to newer APIs,
+	// which we disregard since we can assume all services will see contain the
+	// `IPFamilies` field
+	//
+	// Our only other option is to have a headless service. Our ExternalWorkload
+	// CRD is generic over the AF used so we may create slices for both AF_INET
+	// and AF_INET6
+	afs[discoveryv1.AddressTypeIPv4] = struct{}{}
+	afs[discoveryv1.AddressTypeIPv6] = struct{}{}
+	return afs
+}
diff --git a/controller/api/destination/watcher/workload_watcher.go b/controller/api/destination/watcher/workload_watcher.go
index 4fce08cdf..d4c13da2e 100644
--- a/controller/api/destination/watcher/workload_watcher.go
+++ b/controller/api/destination/watcher/workload_watcher.go
@@ -649,7 +649,7 @@ func (wp *workloadPublisher) updateExternalWorkload(externalWorkload *ext.Extern
 			return
 		}
 
-		if !externalworkload.IsReady(externalWorkload) {
+		if !externalworkload.IsEwReady(externalWorkload) {
 			wp.log.Tracef("ExternalWorkload %s.%s not ready - ignore", externalWorkload.Name, externalWorkload.Namespace)
 			return
 		}
@@ -676,7 +676,7 @@ func (wp *workloadPublisher) updateExternalWorkload(externalWorkload *ext.Extern
 	}
 
 	// backing pod becoming unready or getting deleted
-	if externalWorkload == nil || !externalworkload.IsReady(externalWorkload) {
+	if externalWorkload == nil || !externalworkload.IsEwReady(externalWorkload) {
 		wp.log.Debugf("ExternalWorkload %s.%s deleted or it became unready - remove", wp.externalWorkload.Name, wp.externalWorkload.Namespace)
 		wp.externalWorkload = nil
 		updated := false
diff --git a/controller/cmd/destination/main.go b/controller/cmd/destination/main.go
index 4edddbde5..b3c86ebe2 100644
--- a/controller/cmd/destination/main.go
+++ b/controller/cmd/destination/main.go
@@ -9,6 +9,7 @@ import (
 	"syscall"
 
 	"github.com/linkerd/linkerd2/controller/api/destination"
+	externalworkload "github.com/linkerd/linkerd2/controller/api/destination/external-workload"
 	"github.com/linkerd/linkerd2/controller/api/destination/watcher"
 	"github.com/linkerd/linkerd2/controller/k8s"
 	"github.com/linkerd/linkerd2/pkg/admin"
@@ -35,6 +36,9 @@ func Main(args []string) {
 	clusterDomain := cmd.String("cluster-domain", "", "kubernetes cluster domain")
 	defaultOpaquePorts := cmd.String("default-opaque-ports", "", "configures the default opaque ports")
 	enablePprof := cmd.Bool("enable-pprof", false, "Enable pprof endpoints on the admin server")
+	// This will default to true. It can be overridden with experimental CLI
+	// flags. Currently not exposed as a configuration value through Helm.
+	exportControllerQueueMetrics := cmd.Bool("export-queue-metrics", true, "Exports queue metrics for the external workload controller")
 
 	traceCollector := flags.AddTraceFlags(cmd)
 
@@ -159,6 +163,21 @@ func Main(args []string) {
 	metadataAPI.Sync(nil)
 	clusterStore.Sync(nil)
 
+	// Start mesh expansion external workload controller to write endpointslices
+	// to API Server.
+	if *enableEndpointSlices {
+		hostname, ok := os.LookupEnv("HOSTNAME")
+		if !ok {
+			log.Fatal("Failed to initialize External Workload Endpoints Controller, \"HOSTNAME\" value not found")
+		}
+		externalWorkloadController, err := externalworkload.NewEndpointsController(k8sAPI, hostname, *controllerNamespace, done, *exportControllerQueueMetrics)
+		if err != nil {
+			log.Fatalf("Failed to initialize External Workload Endpoints Controller: %v", err)
+		}
+
+		externalWorkloadController.Start()
+	}
+
 	go func() {
 		log.Infof("starting gRPC server on %s", *addr)
 		if err := server.Serve(lis); err != nil {
diff --git a/controller/gen/apis/externalworkload/v1alpha1/types.go b/controller/gen/apis/externalworkload/v1alpha1/types.go
index 719deee26..09ad65da2 100644
--- a/controller/gen/apis/externalworkload/v1alpha1/types.go
+++ b/controller/gen/apis/externalworkload/v1alpha1/types.go
@@ -111,12 +111,15 @@ type WorkloadCondition struct {
 	// +optional
 	LastProbeTime metav1.Time `json:"lastProbeTime,omitempty"`
 	// Last time a condition transitioned from one status to another.
-	LastTransitionTime metav1.Time `json:"lastTransitionTime"`
+	// +optional
+	LastTransitionTime metav1.Time `json:"lastTransitionTime,omitempty"`
 	// Unique one word reason in CamelCase that describes the reason for a
 	// transition.
-	Reason string `json:"reason"`
+	// +optional
+	Reason string `json:"reason,omitempty"`
 	// Human readable message that describes details about last transition.
-	Message string `json:"message"`
+	// +optional
+	Message string `json:"message,omitempty"`
 }
 
 // WorkloadConditionType is a value for the type of a condition in an
diff --git a/go.mod b/go.mod
index 1874fca80..4c016e614 100644
--- a/go.mod
+++ b/go.mod
@@ -27,15 +27,15 @@ require (
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/common v0.46.0
 	github.com/sergi/go-diff v1.3.1
-	github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749 // indirect
-	github.com/shurcooL/vfsgen v0.0.0-20181202132449-6a9ea43bcacd
+	github.com/shurcooL/httpfs v0.0.0-20230704072500-f1e31cf0ba5c // indirect
+	github.com/shurcooL/vfsgen v0.0.0-20230704071429-0000e147ea92
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.8.0
 	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.17.0
-	google.golang.org/grpc v1.60.1
+	google.golang.org/grpc v1.61.0
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.3.0
 	google.golang.org/protobuf v1.32.0
 	helm.sh/helm/v3 v3.14.0
@@ -66,6 +66,7 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.3 // indirect
 	github.com/Microsoft/hcsshim v0.11.4 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/containerd/containerd v1.7.11 // indirect
@@ -80,14 +81,14 @@ require (
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/docker/go-metrics v0.0.1 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
-	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
+	github.com/emicklei/go-restful/v3 v3.11.2 // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/go-errors/errors v1.4.2 // 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
-	github.com/go-openapi/swag v0.22.6 // indirect
+	github.com/go-openapi/swag v0.22.8 // indirect
 	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
@@ -96,7 +97,7 @@ require (
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
-	github.com/google/uuid v1.3.1 // indirect
+	github.com/google/uuid v1.4.0 // indirect
 	github.com/gorilla/mux v1.8.0 // indirect
 	github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
@@ -149,13 +150,15 @@ require (
 	golang.org/x/time v0.3.0 // indirect
 	google.golang.org/api v0.143.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20231002182017-d307bd883b97 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // 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
+	k8s.io/component-base v0.29.1 // indirect
+	k8s.io/endpointslice v0.29.1 // indirect
+	k8s.io/gengo v0.0.0-20240110203215-22eea95d1e7a // indirect
+	k8s.io/kube-openapi v0.0.0-20240117194847-208609032b15 // indirect
 	oras.land/oras-go v1.2.4 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/kustomize/api v0.13.5-0.20230601165947-6ce0bf390ce3 // indirect
diff --git a/go.sum b/go.sum
index 83a6e912a..164d93ea5 100644
--- a/go.sum
+++ b/go.sum
@@ -53,6 +53,8 @@ github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
+github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/bombsimon/logrusr/v4 v4.1.0 h1:uZNPbwusB0eUXlO8hIUwStE6Lr5bLN6IgYgG+75kuh4=
 github.com/bombsimon/logrusr/v4 v4.1.0/go.mod h1:pjfHC5e59CvjTBIU3V3sGhFWFAnsnhOR03TRc6im0l8=
 github.com/briandowns/spinner v0.0.0-20190212173954-5cf08d0ac778 h1:Dmz6bJXocvwkw7BOz4jpyVZReGrkjs+fBDWKn5tBES4=
@@ -117,6 +119,8 @@ github.com/docker/libtrust v0.0.0-20150114040149-fa567046d9b1 h1:ZClxb8laGDf5arX
 github.com/docker/libtrust v0.0.0-20150114040149-fa567046d9b1/go.mod h1:cyGadeNEkKy96OOhEzfZl+yxihPEzKnqJwvfuSUqbZE=
 github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
 github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
+github.com/emicklei/go-restful/v3 v3.11.2 h1:1onLa9DcsMYO9P+CXaL0dStDqQ2EHHXLiz+BtnqkLAU=
+github.com/emicklei/go-restful/v3 v3.11.2/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/emicklei/proto v1.13.0 h1:YtC/om6EdkJ0me1JPw4h2g10k+ELITjYFb7tpzm8i8k=
 github.com/emicklei/proto v1.13.0/go.mod h1:rn1FgRS/FANiZdD2djyH7TMA9jdRDcYQ9IEN9yvjX0A=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -158,6 +162,8 @@ github.com/go-openapi/spec v0.20.14 h1:7CBlRnw+mtjFGlPDRZmAMnq35cRzI91xj03HVyUi/
 github.com/go-openapi/spec v0.20.14/go.mod h1:8EOhTpBoFiask8rrgwbLC3zmJfz4zsCUueRuPM6GNkw=
 github.com/go-openapi/swag v0.22.6 h1:dnqg1XfHXL9aBxSbktBqFR5CxVyVI+7fYWhAf1JOeTw=
 github.com/go-openapi/swag v0.22.6/go.mod h1:Gl91UqO+btAM0plGGxHqJcQZ1ZTy6jbmridBTsDy8A0=
+github.com/go-openapi/swag v0.22.8 h1:/9RjDSQ0vbFR+NyjGMkFTsA1IA0fmhKSThmfGZjicbw=
+github.com/go-openapi/swag v0.22.8/go.mod h1:6QT22icPLEqAM/z/TChgb4WAveCHF92+2gF0CNjHpPI=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls=
@@ -234,8 +240,8 @@ github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaU
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
-github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
+github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH4=
@@ -402,8 +408,12 @@ github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5g
 github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749 h1:bUGsEnyNbVPw06Bs80sCeARAlK8lhwqGyi6UT8ymuGk=
 github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749/go.mod h1:ZY1cvUeJuFPAdZ/B6v7RHavJWZn2YPVFQ1OSXhCGOkg=
+github.com/shurcooL/httpfs v0.0.0-20230704072500-f1e31cf0ba5c h1:aqg5Vm5dwtvL+YgDpBcK1ITf3o96N/K7/wsRXQnUTEs=
+github.com/shurcooL/httpfs v0.0.0-20230704072500-f1e31cf0ba5c/go.mod h1:owqhoLW1qZoYLZzLnBw+QkPP9WZnjlSWihhxAJC1+/M=
 github.com/shurcooL/vfsgen v0.0.0-20181202132449-6a9ea43bcacd h1:ug7PpSOB5RBPK1Kg6qskGBoP3Vnj/aNYFTznWvlkGo0=
 github.com/shurcooL/vfsgen v0.0.0-20181202132449-6a9ea43bcacd/go.mod h1:TrYk7fJVaAttu97ZZKrO9UbRa8izdowaMIZcxYMbVaw=
+github.com/shurcooL/vfsgen v0.0.0-20230704071429-0000e147ea92 h1:OfRzdxCzDhp+rsKWXuOO2I/quKMJ/+TQwVbIP/gltZg=
+github.com/shurcooL/vfsgen v0.0.0-20230704071429-0000e147ea92/go.mod h1:7/OT02F6S6I7v6WXb+IjhMuZEYfH/RJ5RwEWnEo5BMg=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
@@ -692,12 +702,12 @@ google.golang.org/genproto v0.0.0-20200331122359-1ee6d9798940/go.mod h1:55QSHmfG
 google.golang.org/genproto v0.0.0-20200513103714-09dca8ec2884/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
 google.golang.org/genproto v0.0.0-20200527145253-8367513e4ece/go.mod h1:jDfRM7FcilCzHH/e9qn6dsT145K34l5v+OpcnNgKAAA=
-google.golang.org/genproto v0.0.0-20231002182017-d307bd883b97 h1:SeZZZx0cP0fqUyA+oRzP9k7cSwJlvDFiROO72uwD6i0=
-google.golang.org/genproto v0.0.0-20231002182017-d307bd883b97/go.mod h1:t1VqOqqvce95G3hIDCT5FeO3YUc6Q4Oe24L/+rNMxRk=
-google.golang.org/genproto/googleapis/api v0.0.0-20231002182017-d307bd883b97 h1:W18sezcAYs+3tDZX4F80yctqa12jcP1PUS2gQu1zTPU=
-google.golang.org/genproto/googleapis/api v0.0.0-20231002182017-d307bd883b97/go.mod h1:iargEX0SFPm3xcfMI0d1domjg0ZF4Aa0p2awqyxhvF0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97 h1:6GQBEOdGkX6MMTLT9V+TjtIRZCw9VPD5Z+yHY9wMgS0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97/go.mod h1:v7nGkzlmW8P3n/bKmWBn2WpBjpOEx8Q6gMueudAmKfY=
+google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17 h1:wpZ8pe2x1Q3f2KyT5f8oP/fa9rHAKgFPr/HZdNuS+PQ=
+google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:J7XzRzVy1+IPwWHZUzoD0IccYZIrXILAQpc+Qy9CMhY=
+google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 h1:JpwMPBpFN3uKhdaekDpiNlImDdkUAyiJ6ez/uxGaUSo=
+google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:0xJLfVdJqpAPl8tDg1ujOCGzx6LFLttXT5NhllGOXY4=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 h1:Jyp0Hsi0bmHXG6k9eATXoYtjd6e2UzZ1SCn/wIupY14=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:oQ5rr10WTTMvP4A36n8JpR1OrO1BEiV4f78CneXZxkA=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
@@ -709,8 +719,8 @@ google.golang.org/grpc v1.27.1/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8
 google.golang.org/grpc v1.28.0/go.mod h1:rpkK4SK4GF4Ach/+MFLZUBavHOvF2JJB5uozKKal+60=
 google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3IjizoKk=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.60.1 h1:26+wFr+cNqSGFcOXcabYC0lUVJVRa2Sb2ortSK7VrEU=
-google.golang.org/grpc v1.60.1/go.mod h1:OlCHIeLYqSSsLi6i49B5QGdzaMZK9+M7LXN2FKz4eGM=
+google.golang.org/grpc v1.61.0 h1:TOvOcuXn30kRao+gfcvsebNEa5iZIiLkisYEkf7R7o0=
+google.golang.org/grpc v1.61.0/go.mod h1:VUbo7IFqmF1QtCAstipjG0GIoq49KvMe9+h1jFLBNJs=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.3.0 h1:rNBFJjBCOgVr9pWD7rs/knKL4FRTKgpZmsRfV214zcA=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.3.0/go.mod h1:Dk1tviKTvMCz5tvh7t+fh94dhmQVHuCt2OzJB3CTW9Y=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
@@ -768,8 +778,14 @@ k8s.io/client-go v0.29.1 h1:19B/+2NGEwnFLzt0uB5kNJnfTsbV8w6TgQRz9l7ti7A=
 k8s.io/client-go v0.29.1/go.mod h1:TDG/psL9hdet0TI9mGyHJSgRkW3H9JZk2dNEUS7bRks=
 k8s.io/code-generator v0.29.1 h1:8ba8BdtSmAVHgAMpzThb/fuyQeTRtN7NtN7VjMcDLew=
 k8s.io/code-generator v0.29.1/go.mod h1:FwFi3C9jCrmbPjekhaCYcYG1n07CYiW1+PAPCockaos=
+k8s.io/component-base v0.29.1 h1:MUimqJPCRnnHsskTTjKD+IC1EHBbRCVyi37IoFBrkYw=
+k8s.io/component-base v0.29.1/go.mod h1:fP9GFjxYrLERq1GcWWZAE3bqbNcDKDytn2srWuHTtKc=
+k8s.io/endpointslice v0.29.1 h1:RPwp1lyFeRptuoFaouDLn4CksQX5wDqrMJC3hnSzd0c=
+k8s.io/endpointslice v0.29.1/go.mod h1:Ckn6puMEjlogNzcEjjo/ArHvGJFIl1kCbOAhjU+rATM=
 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/gengo v0.0.0-20240110203215-22eea95d1e7a h1:zCwpCC6Ghs+RstFfEJZ3arc7dLZ9z0tlmLHDXGCkINY=
+k8s.io/gengo v0.0.0-20240110203215-22eea95d1e7a/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.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw=
 k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
@@ -777,6 +793,8 @@ k8s.io/kube-aggregator v0.29.1 h1:ArCHuHNT2vNOQbrFBjt23nUs+08w1KcLABuWUinOD4U=
 k8s.io/kube-aggregator v0.29.1/go.mod h1:Wdf0L0CWYwhUKs+KaYiM+NwqkZTp0Erd/wgefvyZBwQ=
 k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 h1:aVUu9fTY98ivBPKR9Y5w/AuzbMm96cd3YHRTU83I780=
 k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00/go.mod h1:AsvuZPBlUDVuCdzJ87iajxtXuR9oktsTctW/R9wwouA=
+k8s.io/kube-openapi v0.0.0-20240117194847-208609032b15 h1:m6dl1pkxz3HuE2mP9MUYPCCGyy6IIFlv/vTlLBDxIwA=
+k8s.io/kube-openapi v0.0.0-20240117194847-208609032b15/go.mod h1:Pa1PvrP7ACSkuX6I7KYomY6cmMA0Tx86waBhDUgoKPw=
 k8s.io/utils v0.0.0-20230726121419-3b25d923346b h1:sgn3ZU783SCgtaSJjpcVVlRqd6GSnlTLKgpAAttJvpI=
 k8s.io/utils v0.0.0-20230726121419-3b25d923346b/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 oras.land/oras-go v1.2.4 h1:djpBY2/2Cs1PV87GSJlxv4voajVOMZxqqtq9AB8YNvY=
diff --git a/pkg/k8s/fake.go b/pkg/k8s/fake.go
index b05040849..d5d800d8d 100644
--- a/pkg/k8s/fake.go
+++ b/pkg/k8s/fake.go
@@ -3,6 +3,7 @@ package k8s
 import (
 	"bufio"
 	"errors"
+	"fmt"
 	"io"
 	"strings"
 
@@ -19,12 +20,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/rand"
 	yamlDecoder "k8s.io/apimachinery/pkg/util/yaml"
 	discoveryfake "k8s.io/client-go/discovery/fake"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/testing"
 	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
 	apiregistrationclient "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset"
 	apiregistrationfake "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/fake"
@@ -72,7 +75,7 @@ func NewFakeAPIFromManifests(readers []io.Reader) (*KubernetesAPI, error) {
 // NewFakeClientSets provides mock Kubernetes ClientSets.
 // TODO: make this private once KubernetesAPI (and NewFakeAPI) supports spClient
 func NewFakeClientSets(configs ...string) (
-	kubernetes.Interface,
+	*fake.Clientset,
 	apiextensionsclient.Interface,
 	apiregistrationclient.Interface,
 	spclient.Interface,
@@ -137,6 +140,29 @@ metadata:
 		},
 	})
 
+	// Add helpers to work with endpoint slice objects.
+	cs.PrependReactor("create", "endpointslices", testing.ReactionFunc(func(action testing.Action) (bool, runtime.Object, error) {
+		es := action.(testing.CreateAction).GetObject().(*discovery.EndpointSlice)
+
+		// The API Server cannot generate a name when we use mocks, intercept
+		// the object and change its name
+		if es.GenerateName != "" {
+			es.Name = fmt.Sprintf("%s-%s", es.GenerateName, rand.String(8))
+			es.GenerateName = ""
+		}
+		es.Generation = 1
+
+		return false, es, nil
+	}))
+
+	cs.PrependReactor("update", "endpointslices", testing.ReactionFunc(func(action testing.Action) (bool, runtime.Object, error) {
+		// An update won't increase the generation since the API Server is
+		// mocked, so do a typecast and increment it here.
+		es := action.(testing.CreateAction).GetObject().(*discovery.EndpointSlice)
+		es.Generation++
+		return false, es, nil
+	}))
+
 	return cs,
 		apiextensionsfake.NewSimpleClientset(apiextObjs...),
 		apiregistrationfake.NewSimpleClientset(apiRegObjs...),
diff --git a/policy-controller/k8s/api/src/external_workload.rs b/policy-controller/k8s/api/src/external_workload.rs
index 65ba1b17a..76f7fd0ed 100644
--- a/policy-controller/k8s/api/src/external_workload.rs
+++ b/policy-controller/k8s/api/src/external_workload.rs
@@ -77,14 +77,14 @@ pub struct Condition {
     /// Can be True, False, Unknown
     status: ConditionStatus,
     /// Last time a condition transitioned from one status to another.
-    last_transition_time: crate::apimachinery::pkg::apis::meta::v1::Time,
+    last_transition_time: Option<crate::apimachinery::pkg::apis::meta::v1::Time>,
     /// Last time an ExternalWorkload was probed for a condition.
     last_probe_time: Option<crate::apimachinery::pkg::apis::meta::v1::Time>,
     /// Unique one word reason in CamelCase that describes the reason for a
     /// transition.
-    reason: String,
+    reason: Option<String>,
     /// Human readable message that describes details about last transition.
-    message: String,
+    message: Option<String>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]


Test patch:
diff --git a/cli/cmd/testdata/install_crds.golden b/cli/cmd/testdata/install_crds.golden
index 9896e11e4..053fb024d 100644
--- a/cli/cmd/testdata/install_crds.golden
+++ b/cli/cmd/testdata/install_crds.golden
@@ -10382,8 +10382,5 @@ spec:
                       maxLength: 32768
                       type: string
                 required:
-                - lastTransitionTime
                 - status
                 - type
-                - reason
-                - message
diff --git a/cli/cmd/testdata/install_helm_crds_output.golden b/cli/cmd/testdata/install_helm_crds_output.golden
index 050b51875..b11bf1728 100644
--- a/cli/cmd/testdata/install_helm_crds_output.golden
+++ b/cli/cmd/testdata/install_helm_crds_output.golden
@@ -10400,8 +10400,5 @@ spec:
                       maxLength: 32768
                       type: string
                 required:
-                - lastTransitionTime
                 - status
                 - type
-                - reason
-                - message
diff --git a/cli/cmd/testdata/install_helm_crds_output_ha.golden b/cli/cmd/testdata/install_helm_crds_output_ha.golden
index 050b51875..b11bf1728 100644
--- a/cli/cmd/testdata/install_helm_crds_output_ha.golden
+++ b/cli/cmd/testdata/install_helm_crds_output_ha.golden
@@ -10400,8 +10400,5 @@ spec:
                       maxLength: 32768
                       type: string
                 required:
-                - lastTransitionTime
                 - status
                 - type
-                - reason
-                - message
diff --git a/controller/api/destination/external-workload/endpoints_controller_test.go b/controller/api/destination/external-workload/endpoints_controller_test.go
index 25eaec309..7f388a5aa 100644
--- a/controller/api/destination/external-workload/endpoints_controller_test.go
+++ b/controller/api/destination/external-workload/endpoints_controller_test.go
@@ -1,14 +1,315 @@
 package externalworkload
 
 import (
+	"context"
+	"fmt"
+	"reflect"
+	"sort"
 	"testing"
+	"time"
 
 	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
 	"github.com/linkerd/linkerd2/controller/k8s"
+	v1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/utils/ptr"
 )
 
-func makeExternalWorkload(labels map[string]string, ports map[int32]string, ips []string) *ewv1alpha1.ExternalWorkload {
+type endpointSliceController struct {
+	*EndpointsController
+	endpointSliceStore     cache.Store
+	externalWorkloadsStore cache.Store
+	serviceStore           cache.Store
+}
+
+func newController(t *testing.T) (*k8s.API, func() []k8stesting.Action, *endpointSliceController) {
+	t.Helper()
+
+	k8sAPI, actions, err := k8s.NewFakeAPIWithActions()
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	esController, err := NewEndpointsController(k8sAPI, "hostname", "linkerd", make(chan struct{}), false)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	return k8sAPI, actions, &endpointSliceController{
+		esController,
+		k8sAPI.ES().Informer().GetStore(),
+		k8sAPI.ExtWorkload().Informer().GetStore(),
+		k8sAPI.Svc().Informer().GetStore(),
+	}
+
+}
+
+func newExternalWorkload(n int, namespace string, ready bool, terminating bool) *ewv1alpha1.ExternalWorkload {
+	status := ewv1alpha1.ConditionTrue
+	if !ready {
+		status = ewv1alpha1.ConditionFalse
+	}
+
+	var deletionTimestamp *metav1.Time
+	if terminating {
+		deletionTimestamp = &metav1.Time{
+			Time: time.Now(),
+		}
+	}
+
+	ew := &ewv1alpha1.ExternalWorkload{
+		TypeMeta: metav1.TypeMeta{APIVersion: "v1"},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:         namespace,
+			Name:              fmt.Sprintf("ew-%d", n),
+			Labels:            map[string]string{"foo": "bar"},
+			DeletionTimestamp: deletionTimestamp,
+			ResourceVersion:   fmt.Sprint(n),
+		},
+		Spec: ewv1alpha1.ExternalWorkloadSpec{
+			Ports: []ewv1alpha1.PortSpec{
+				{
+					Name: "name",
+					Port: 444,
+				},
+			},
+			WorkloadIPs: []ewv1alpha1.WorkloadIP{
+				{Ip: "1.2.3.4"},
+			},
+		},
+		Status: ewv1alpha1.ExternalWorkloadStatus{
+			Conditions: []ewv1alpha1.WorkloadCondition{
+				{
+					Type:   ewv1alpha1.WorkloadReady,
+					Status: status,
+				},
+			},
+		},
+	}
+
+	return ew
+}
+
+// Ensure SyncService for service with no selector results in no action
+func TestSyncServiceNoSelector(t *testing.T) {
+	ns := metav1.NamespaceDefault
+	serviceName := "testing-1"
+	_, actions, esController := newController(t)
+	esController.serviceStore.Add(&v1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: ns},
+		Spec: v1.ServiceSpec{
+			Ports: []v1.ServicePort{{TargetPort: intstr.FromInt32(80)}},
+		},
+	})
+
+	err := esController.syncService(fmt.Sprintf("%s/%s", ns, serviceName))
+	if err != nil {
+		t.Errorf("unexpected error: %s", err)
+	}
+
+	if len(actions()) != 0 {
+		t.Errorf("expected 0 actions, got: %d", len(actions()))
+	}
+}
+
+func TestServiceExternalNameTypeSync(t *testing.T) {
+	serviceName := "testing-1"
+	namespace := "zahari"
+
+	testCases := []struct {
+		desc    string
+		service *v1.Service
+	}{
+		{
+			desc: "External name with selector and ports should not receive endpoint slices",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: namespace},
+				Spec: v1.ServiceSpec{
+					Selector: map[string]string{"foo": "bar"},
+					Ports:    []v1.ServicePort{{Port: 80}},
+					Type:     v1.ServiceTypeExternalName,
+				},
+			},
+		},
+		{
+			desc: "External name with ports should not receive endpoint slices",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: namespace},
+				Spec: v1.ServiceSpec{
+					Ports: []v1.ServicePort{{Port: 80}},
+					Type:  v1.ServiceTypeExternalName,
+				},
+			},
+		},
+		{
+			desc: "External name with selector should not receive endpoint slices",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: namespace},
+				Spec: v1.ServiceSpec{
+					Selector: map[string]string{"foo": "bar"},
+					Type:     v1.ServiceTypeExternalName,
+				},
+			},
+		},
+		{
+			desc: "External name without selector and ports should not receive endpoint slices",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: namespace},
+				Spec: v1.ServiceSpec{
+					Type: v1.ServiceTypeExternalName,
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			client, actions, esController := newController(t)
+			ew := newExternalWorkload(1, namespace, true, false)
+			err := esController.externalWorkloadsStore.Add(ew)
+			if err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			err = esController.serviceStore.Add(tc.service)
+			if err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			err = esController.syncService(fmt.Sprintf("%s/%s", namespace, serviceName))
+			if err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			if len(actions()) != 0 {
+				t.Errorf("expected 0 actions, got: %d", len(actions()))
+			}
+
+			sliceList, err := client.Client.DiscoveryV1().EndpointSlices(namespace).List(context.TODO(), metav1.ListOptions{})
+			if err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+			if len(sliceList.Items) != 0 {
+				t.Errorf("Expected 0 endpoint slices, got: %d", len(sliceList.Items))
+			}
+		})
+	}
+}
+
+// Ensure SyncService for service with pending deletion results in no action
+func TestSyncServicePendingDeletion(t *testing.T) {
+	ns := metav1.NamespaceDefault
+	serviceName := "testing-1"
+	deletionTimestamp := metav1.Now()
+	_, actions, esController := newController(t)
+	esController.serviceStore.Add(&v1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: serviceName, Namespace: ns, DeletionTimestamp: &deletionTimestamp},
+		Spec: v1.ServiceSpec{
+			Selector: map[string]string{"foo": "bar"},
+			Ports:    []v1.ServicePort{{TargetPort: intstr.FromInt32(80)}},
+		},
+	})
+
+	err := esController.syncService(fmt.Sprintf("%s/%s", ns, serviceName))
+	if err != nil {
+		t.Errorf("unexpected error: %s", err)
+	}
+	if len(actions()) != 0 {
+		t.Errorf("Expected 0 actions, got: %d", len(actions()))
+	}
+}
+
+// Ensure SyncService correctly selects ExternalWorkload.
+func TestSyncServiceExternalWorkloadSelection(t *testing.T) {
+	client, actions, esController := newController(t)
+	ns := "test-ns"
+
+	ew1 := newExternalWorkload(1, ns, true, false)
+	esController.externalWorkloadsStore.Add(ew1)
+
+	// ensure this ew will not match the selector
+	ew2 := newExternalWorkload(2, ns, true, false)
+	ew2.Labels["foo"] = "boo"
+	esController.externalWorkloadsStore.Add(ew2)
+
+	standardSyncService(t, esController, ns, "testing-1")
+	expectActions(t, actions(), 1, "create", "endpointslices")
+
+	// an endpoint slice should be created, it should only reference ew1 (not ew2)
+	slices, err := client.Client.DiscoveryV1().EndpointSlices(ns).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		t.Errorf("Expected no error fetching endpoint slices, got: %s", err)
+	}
+	if len(slices.Items) != 1 {
+		t.Errorf("Expected 1 endpoint slices, got: %d", len(slices.Items))
+	}
+
+	slice := slices.Items[0]
+	if len(slice.Endpoints) != 1 {
+		t.Errorf("Expected 1 endpoint in first slice, got: %d", len(slice.Endpoints))
+	}
+	endpoint := slice.Endpoints[0]
+	if endpoint.TargetRef.Kind != "ExternalWorkload" || endpoint.TargetRef.Namespace != ns || endpoint.TargetRef.Name != ew1.Name {
+		t.Errorf("Expected endpoint to target ExternalWorkload")
+	}
+}
+
+func TestSyncServiceEndpointSlicePendingDeletion(t *testing.T) {
+	client, actions, esController := newController(t)
+	ns := "test-ns"
+	serviceName := "testing-1"
+	service := createService(t, esController, ns, serviceName)
+	err := esController.syncService(fmt.Sprintf("%s/%s", ns, serviceName))
+	if err != nil {
+		t.Fatalf("Expected no error creating EndpointSlice: %v", err)
+	}
+
+	gvk := schema.GroupVersionKind{Version: "v1", Kind: "Service"}
+	ownerRef := metav1.NewControllerRef(service, gvk)
+
+	deletedTs := metav1.Now()
+	endpointSlice := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "epSlice-1",
+			Namespace:       ns,
+			OwnerReferences: []metav1.OwnerReference{*ownerRef},
+			Labels: map[string]string{
+				discoveryv1.LabelServiceName: serviceName,
+				discoveryv1.LabelManagedBy:   managedBy,
+			},
+			DeletionTimestamp: &deletedTs,
+		},
+		AddressType: discoveryv1.AddressTypeIPv4,
+	}
+	err = esController.endpointSliceStore.Add(endpointSlice)
+	if err != nil {
+		t.Fatalf("Expected no error adding EndpointSlice: %v", err)
+	}
+	_, err = client.Client.DiscoveryV1().EndpointSlices(ns).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("Expected no error creating EndpointSlice: %v", err)
+	}
+
+	numActionsBefore := len(actions())
+	err = esController.syncService(fmt.Sprintf("%s/%s", ns, serviceName))
+	if err != nil {
+		t.Errorf("Expected no error syncing service, got: %s", err)
+	}
+
+	// The EndpointSlice marked for deletion should be ignored by the controller, and thus
+	// should not result in any action.
+	if len(actions()) != numActionsBefore {
+		t.Errorf("Expected 0 more actions, got %d", len(actions())-numActionsBefore)
+	}
+}
+
+func makeExternalWorkload(resVersion, name string, labels map[string]string, ports map[int32]string, ips []string) *ewv1alpha1.ExternalWorkload {
 	portSpecs := []ewv1alpha1.PortSpec{}
 	for port, name := range ports {
 		spec := ewv1alpha1.PortSpec{
@@ -25,11 +326,12 @@ func makeExternalWorkload(labels map[string]string, ports map[int32]string, ips
 		wIps = append(wIps, ewv1alpha1.WorkloadIP{Ip: ip})
 	}
 
-	return &ewv1alpha1.ExternalWorkload{
+	ew := &ewv1alpha1.ExternalWorkload{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      "wkld1",
-			Namespace: "ns",
-			Labels:    labels,
+			Name:            name,
+			Namespace:       "ns",
+			Labels:          labels,
+			ResourceVersion: resVersion,
 		},
 		Spec: ewv1alpha1.ExternalWorkloadSpec{
 			MeshTls: ewv1alpha1.MeshTls{
@@ -39,153 +341,724 @@ func makeExternalWorkload(labels map[string]string, ports map[int32]string, ips
 			Ports:       portSpecs,
 			WorkloadIPs: wIps,
 		},
+		Status: ewv1alpha1.ExternalWorkloadStatus{
+			Conditions: []ewv1alpha1.WorkloadCondition{
+				{
+					Type:   ewv1alpha1.WorkloadReady,
+					Status: ewv1alpha1.ConditionTrue,
+				},
+			},
+		},
+	}
+
+	ew.ObjectMeta.UID = types.UID(fmt.Sprintf("%s-%s", ew.Namespace, ew.Name))
+	return ew
+}
+
+func TestSyncService(t *testing.T) {
+	creationTimestamp := metav1.Now()
+	namespace := "test-ns"
+	testcases := []struct {
+		name                  string
+		service               *v1.Service
+		externalWorkloads     []*ewv1alpha1.ExternalWorkload
+		expectedEndpointPorts []discoveryv1.EndpointPort
+		expectedEndpoints     []discoveryv1.Endpoint
+	}{
+		{
+			name: "EW with multiple IPs and Service with ipFamilies=ipv4",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "foobar",
+					Namespace:         namespace,
+					CreationTimestamp: creationTimestamp,
+				},
+				Spec: v1.ServiceSpec{
+					Ports: []v1.ServicePort{
+						{Name: "tcp-example", TargetPort: intstr.FromInt32(80), Protocol: v1.ProtocolTCP},
+						{Name: "udp-example", TargetPort: intstr.FromInt32(161), Protocol: v1.ProtocolUDP},
+						{Name: "sctp-example", TargetPort: intstr.FromInt32(3456), Protocol: v1.ProtocolSCTP},
+					},
+					Selector:   map[string]string{"foo": "bar"},
+					IPFamilies: []v1.IPFamily{v1.IPv4Protocol},
+				},
+			},
+			externalWorkloads: []*ewv1alpha1.ExternalWorkload{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew0",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.1",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionTrue,
+							},
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew1",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.2",
+							},
+							{
+								Ip: "fd08::5678:0000:0000:9abc:def0",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionTrue,
+							},
+						},
+					},
+				},
+			},
+			expectedEndpointPorts: []discoveryv1.EndpointPort{
+				{
+					Name:     ptr.To("udp-example"),
+					Protocol: protoPtr(v1.ProtocolUDP),
+					Port:     ptr.To(int32(161)),
+				},
+				{
+					Name:     ptr.To("sctp-example"),
+					Protocol: protoPtr(v1.ProtocolSCTP),
+					Port:     ptr.To(int32(3456)),
+				},
+				{
+					Name:     ptr.To("tcp-example"),
+					Protocol: protoPtr(v1.ProtocolTCP),
+					Port:     ptr.To(int32(80)),
+				},
+			},
+			expectedEndpoints: []discoveryv1.Endpoint{
+				{
+					Conditions: discoveryv1.EndpointConditions{
+						Ready:       ptr.To(true),
+						Serving:     ptr.To(true),
+						Terminating: ptr.To(false),
+					},
+					Addresses: []string{"10.0.0.1"},
+					TargetRef: &v1.ObjectReference{Kind: "ExternalWorkload", Namespace: namespace, Name: "ew0"},
+				},
+				{
+					Conditions: discoveryv1.EndpointConditions{
+						Ready:       ptr.To(true),
+						Serving:     ptr.To(true),
+						Terminating: ptr.To(false),
+					},
+					Addresses: []string{"10.0.0.2"},
+					TargetRef: &v1.ObjectReference{Kind: "ExternalWorkload", Namespace: namespace, Name: "ew1"},
+				},
+			},
+		},
+		{
+			name: "EWs with multiple IPs and Service with ipFamilies=ipv6",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "foobar",
+					Namespace:         namespace,
+					CreationTimestamp: creationTimestamp,
+				},
+				Spec: v1.ServiceSpec{
+					Ports: []v1.ServicePort{
+						{Name: "tcp-example", TargetPort: intstr.FromInt32(80), Protocol: v1.ProtocolTCP},
+						{Name: "udp-example", TargetPort: intstr.FromInt32(161), Protocol: v1.ProtocolUDP},
+						{Name: "sctp-example", TargetPort: intstr.FromInt32(3456), Protocol: v1.ProtocolSCTP},
+					},
+					Selector:   map[string]string{"foo": "bar"},
+					IPFamilies: []v1.IPFamily{v1.IPv6Protocol},
+				},
+			},
+			externalWorkloads: []*ewv1alpha1.ExternalWorkload{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew0",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.1",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionTrue,
+							},
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew1",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.2",
+							},
+							{
+
+								Ip: "fd08::5678:0000:0000:9abc:def0",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionTrue,
+							},
+						},
+					},
+				},
+			},
+			expectedEndpointPorts: []discoveryv1.EndpointPort{
+				{
+					Name:     ptr.To("udp-example"),
+					Protocol: protoPtr(v1.ProtocolUDP),
+					Port:     ptr.To(int32(161)),
+				},
+				{
+					Name:     ptr.To("sctp-example"),
+					Protocol: protoPtr(v1.ProtocolSCTP),
+					Port:     ptr.To(int32(3456)),
+				},
+				{
+					Name:     ptr.To("tcp-example"),
+					Protocol: protoPtr(v1.ProtocolTCP),
+					Port:     ptr.To(int32(80)),
+				},
+			},
+			expectedEndpoints: []discoveryv1.Endpoint{
+				{
+					Conditions: discoveryv1.EndpointConditions{
+						Ready:       ptr.To(true),
+						Serving:     ptr.To(true),
+						Terminating: ptr.To(false),
+					},
+					Addresses: []string{"fd08::5678:0000:0000:9abc:def0"},
+					TargetRef: &v1.ObjectReference{Kind: "ExternalWorkload", Namespace: namespace, Name: "ew1"},
+				},
+			},
+		},
+		{
+			name: "Not ready workloads",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "foobar",
+					Namespace:         namespace,
+					CreationTimestamp: creationTimestamp,
+				},
+				Spec: v1.ServiceSpec{
+					Ports: []v1.ServicePort{
+						{Name: "tcp-example", TargetPort: intstr.FromInt32(80), Protocol: v1.ProtocolTCP},
+						{Name: "udp-example", TargetPort: intstr.FromInt32(161), Protocol: v1.ProtocolUDP},
+						{Name: "sctp-example", TargetPort: intstr.FromInt32(3456), Protocol: v1.ProtocolSCTP},
+					},
+					Selector:   map[string]string{"foo": "bar"},
+					IPFamilies: []v1.IPFamily{v1.IPv4Protocol},
+				},
+			},
+			externalWorkloads: []*ewv1alpha1.ExternalWorkload{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew0",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.1",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionTrue,
+							},
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew1",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.2",
+							},
+							{
+
+								Ip: "fd08::5678:0000:0000:9abc:def0",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionFalse,
+							},
+						},
+					},
+				},
+			},
+			expectedEndpointPorts: []discoveryv1.EndpointPort{
+				{
+					Name:     ptr.To("udp-example"),
+					Protocol: protoPtr(v1.ProtocolUDP),
+					Port:     ptr.To(int32(161)),
+				},
+				{
+					Name:     ptr.To("sctp-example"),
+					Protocol: protoPtr(v1.ProtocolSCTP),
+					Port:     ptr.To(int32(3456)),
+				},
+				{
+					Name:     ptr.To("tcp-example"),
+					Protocol: protoPtr(v1.ProtocolTCP),
+					Port:     ptr.To(int32(80)),
+				},
+			},
+			expectedEndpoints: []discoveryv1.Endpoint{
+				{
+					Conditions: discoveryv1.EndpointConditions{
+						Ready:       ptr.To(true),
+						Serving:     ptr.To(true),
+						Terminating: ptr.To(false),
+					},
+					Addresses: []string{"10.0.0.1"},
+					TargetRef: &v1.ObjectReference{Kind: "ExternalWorkload", Namespace: namespace, Name: "ew0"},
+				},
+				{
+					Conditions: discoveryv1.EndpointConditions{
+						Ready:       ptr.To(false),
+						Serving:     ptr.To(false),
+						Terminating: ptr.To(false),
+					},
+					Addresses: []string{"10.0.0.2"},
+					TargetRef: &v1.ObjectReference{Kind: "ExternalWorkload", Namespace: namespace, Name: "ew1"},
+				},
+			},
+		},
+		{
+			name: "Two Ready workloads with the same IPs",
+			service: &v1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "foobar",
+					Namespace:         namespace,
+					CreationTimestamp: creationTimestamp,
+				},
+				Spec: v1.ServiceSpec{
+					Ports: []v1.ServicePort{
+						{Name: "tcp-example", TargetPort: intstr.FromInt32(80), Protocol: v1.ProtocolTCP},
+						{Name: "udp-example", TargetPort: intstr.FromInt32(161), Protocol: v1.ProtocolUDP},
+						{Name: "sctp-example", TargetPort: intstr.FromInt32(3456), Protocol: v1.ProtocolSCTP},
+					},
+					Selector:   map[string]string{"foo": "bar"},
+					IPFamilies: []v1.IPFamily{v1.IPv4Protocol},
+				},
+			},
+			externalWorkloads: []*ewv1alpha1.ExternalWorkload{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew0",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.1",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionTrue,
+							},
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace:         namespace,
+						Name:              "ew1",
+						Labels:            map[string]string{"foo": "bar"},
+						DeletionTimestamp: nil,
+					},
+					Spec: ewv1alpha1.ExternalWorkloadSpec{
+						WorkloadIPs: []ewv1alpha1.WorkloadIP{
+							{
+								Ip: "10.0.0.1",
+							},
+						},
+						Ports: []ewv1alpha1.PortSpec{
+							{Name: "tcp-example", Port: 80, Protocol: v1.ProtocolTCP},
+							{Name: "udp-example", Port: 161, Protocol: v1.ProtocolUDP},
+							{Name: "sctp-example", Port: 3456, Protocol: v1.ProtocolSCTP},
+						},
+					},
+					Status: ewv1alpha1.ExternalWorkloadStatus{
+						Conditions: []ewv1alpha1.WorkloadCondition{
+							{
+								Type:   ewv1alpha1.WorkloadReady,
+								Status: ewv1alpha1.ConditionTrue,
+							},
+						},
+					},
+				},
+			},
+			expectedEndpointPorts: []discoveryv1.EndpointPort{
+				{
+					Name:     ptr.To("udp-example"),
+					Protocol: protoPtr(v1.ProtocolUDP),
+					Port:     ptr.To(int32(161)),
+				},
+				{
+					Name:     ptr.To("sctp-example"),
+					Protocol: protoPtr(v1.ProtocolSCTP),
+					Port:     ptr.To(int32(3456)),
+				},
+				{
+					Name:     ptr.To("tcp-example"),
+					Protocol: protoPtr(v1.ProtocolTCP),
+					Port:     ptr.To(int32(80)),
+				},
+			},
+			expectedEndpoints: []discoveryv1.Endpoint{
+				{
+					Conditions: discoveryv1.EndpointConditions{
+						Ready:       ptr.To(true),
+						Serving:     ptr.To(true),
+						Terminating: ptr.To(false),
+					},
+					Addresses: []string{"10.0.0.1"},
+					TargetRef: &v1.ObjectReference{Kind: "ExternalWorkload", Namespace: namespace, Name: "ew0"},
+				},
+				{
+					Conditions: discoveryv1.EndpointConditions{
+						Ready:       ptr.To(true),
+						Serving:     ptr.To(true),
+						Terminating: ptr.To(false),
+					},
+					Addresses: []string{"10.0.0.1"},
+					TargetRef: &v1.ObjectReference{Kind: "ExternalWorkload", Namespace: namespace, Name: "ew1"},
+				},
+			},
+		},
+	}
+
+	for _, testcase := range testcases {
+		t.Run(testcase.name, func(t *testing.T) {
+			client, actions, esController := newController(t)
+
+			for _, ew := range testcase.externalWorkloads {
+				esController.externalWorkloadsStore.Add(ew)
+			}
+			esController.serviceStore.Add(testcase.service)
+
+			_, err := esController.k8sAPI.Client.CoreV1().Services(testcase.service.Namespace).Create(context.TODO(), testcase.service, metav1.CreateOptions{})
+			if err != nil {
+				t.Errorf("Expected no error creating service, got: %s", err)
+			}
+			err = esController.syncService(fmt.Sprintf("%s/%s", testcase.service.Namespace, testcase.service.Name))
+			if err != nil {
+				t.Errorf("Expected no error, got: %s", err)
+			}
+
+			// last action should be to create endpoint slice
+			expectActions(t, actions(), 1, "create", "endpointslices")
+			sliceList, err := client.Client.DiscoveryV1().EndpointSlices(testcase.service.Namespace).List(context.TODO(), metav1.ListOptions{})
+			if err != nil {
+				t.Errorf("Expected no error fetching endpoint slices, got: %s", err)
+			}
+
+			if len(sliceList.Items) != 1 {
+				t.Errorf("Expected 1 endpoints slices")
+			}
+
+			// ensure all attributes of endpoint slice match expected state
+			slice := sliceList.Items[0]
+
+			// check expected ports
+
+			if !reflect.DeepEqual(testcase.expectedEndpointPorts, slice.Ports) {
+				t.Error("actual ports do not match expected ones")
+			}
+
+			// sort actual endpoints in terms of targetRef name, in ascending
+			// order. This will ensure reflection package doesn't give a
+			// spurious error.
+
+			sort.Slice(slice.Endpoints, func(i, j int) bool {
+				return slice.Endpoints[i].TargetRef.Name < slice.Endpoints[j].TargetRef.Name
+			})
+
+			if !reflect.DeepEqual(testcase.expectedEndpoints, slice.Endpoints) {
+				t.Error("actual endpoints do not match expected ones")
+			}
+		})
 	}
 }
 
 // Test diffing logic that determines if two workloads with the same name and
 // namespace have changed enough to warrant reconciliation
-func TestWorkloadSpecChanged(t *testing.T) {
+// TODO: Add tests for labels change
+func TestEwEndpointsChanged(t *testing.T) {
 	for _, tt := range []struct {
-		name          string
-		old           *ewv1alpha1.ExternalWorkload
-		updated       *ewv1alpha1.ExternalWorkload
-		expectChanged bool
+		name        string
+		old         *ewv1alpha1.ExternalWorkload
+		updated     *ewv1alpha1.ExternalWorkload
+		specChanged bool
 	}{
 		{
 			name: "no change",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
-			expectChanged: false,
+			specChanged: false,
 		},
 		{
 			name: "updated workload adds an IP address",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0", "192.0.3.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 		{
 			name: "updated workload removes an IP address",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0", "192.0.3.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 		{
 			name: "updated workload changes an IP address",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.3.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 		{
 			name: "updated workload adds new port",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1", 2: "port-2"},
 				[]string{"192.0.2.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 		{
 			name: "updated workload removes port",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1", 2: "port-2"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 		{
 			name: "updated workload changes port number",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{2: "port-1"},
 				[]string{"192.0.2.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 		{
 			name: "updated workload changes port name",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-foo"},
 				[]string{"192.0.2.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 		{
 			name: "updated workload removes port name",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				nil,
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				nil,
 				map[int32]string{1: ""},
 				[]string{"192.0.2.0"},
 			),
-			expectChanged: true,
+			specChanged: true,
 		},
 	} {
 		tt := tt // Pin
 		t.Run(tt.name, func(t *testing.T) {
-			changed := specChanged(tt.old, tt.updated)
-			if tt.expectChanged != changed {
-				t.Errorf("expected changed '%v', got '%v'", tt.expectChanged, changed)
+			specChanged, _ := ewEndpointsChanged(tt.old, tt.updated)
+			if tt.specChanged != specChanged {
+				t.Errorf("expected specChanged '%v', got '%v'", tt.specChanged, specChanged)
 			}
 		})
-
 	}
 }
 
@@ -202,11 +1075,15 @@ func TestWorkloadServicesToUpdate(t *testing.T) {
 		{
 			name: "no change",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				map[string]string{"app": "test"},
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				map[string]string{"app": "test"},
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
@@ -226,11 +1103,15 @@ func TestWorkloadServicesToUpdate(t *testing.T) {
 		{
 			name: "labels and spec have changed",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				map[string]string{"app": "test-1"},
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				map[string]string{"app": "test-2"},
 				map[int32]string{2: "port-1"},
 				[]string{"192.0.2.0"},
@@ -258,11 +1139,15 @@ func TestWorkloadServicesToUpdate(t *testing.T) {
 		{
 			name: "spec has changed",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				map[string]string{"app": "test-1"},
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				map[string]string{"app": "test-1"},
 				map[int32]string{2: "port-1"},
 				[]string{"192.0.2.0"},
@@ -282,11 +1167,15 @@ func TestWorkloadServicesToUpdate(t *testing.T) {
 		{
 			name: "labels have changed",
 			old: makeExternalWorkload(
+				"1",
+				"wlkd1",
 				map[string]string{"app": "test-1", "env": "staging"},
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
 			),
 			updated: makeExternalWorkload(
+				"2",
+				"wlkd1",
 				map[string]string{"app": "test-1", "env": "prod"},
 				map[int32]string{1: "port-1"},
 				[]string{"192.0.2.0"},
@@ -334,16 +1223,12 @@ func TestWorkloadServicesToUpdate(t *testing.T) {
 			ec.Start()
 			k8sAPI.Sync(nil)
 
-			services, err := ec.servicesToUpdate(tt.old, tt.updated)
-			if err != nil {
-				t.Fatalf("unexpected error %v", err)
-			}
-
+			services := ec.getServicesToUpdateOnExternalWorkloadChange(tt.old, tt.updated)
 			if len(services) != len(tt.expectServices) {
 				t.Fatalf("expected %d services to update, got %d services instead", len(tt.expectServices), len(services))
 			}
 
-			for _, svc := range services {
+			for svc := range services {
 				if _, ok := tt.expectServices[svc]; !ok {
 					t.Errorf("unexpected service key %s found in list of results", svc)
 				}
@@ -352,3 +1237,77 @@ func TestWorkloadServicesToUpdate(t *testing.T) {
 
 	}
 }
+
+// protoPtr takes a Protocol and returns a pointer to it.
+func protoPtr(proto v1.Protocol) *v1.Protocol {
+	return &proto
+}
+
+func newStatusCondition(ready bool) ewv1alpha1.WorkloadCondition {
+	var status ewv1alpha1.WorkloadConditionStatus
+	if ready {
+		status = ewv1alpha1.ConditionTrue
+	} else {
+		status = ewv1alpha1.ConditionFalse
+	}
+	return ewv1alpha1.WorkloadCondition{
+		Type:               ewv1alpha1.WorkloadReady,
+		Status:             status,
+		LastProbeTime:      metav1.Time{},
+		LastTransitionTime: metav1.NewTime(time.Now()),
+		Reason:             "test",
+		Message:            "test",
+	}
+}
+
+//nolint:all
+func expectActions(t *testing.T, actions []k8stesting.Action, num int, verb, resource string) {
+	t.Helper()
+	// if actions are less the below logic will panic
+	if num > len(actions) {
+		t.Fatalf("len of actions %v is unexpected. Expected to be at least %v", len(actions), num+1)
+	}
+
+	for i := 0; i < num; i++ {
+		relativePos := len(actions) - i - 1
+		if actions[relativePos].GetVerb() != verb {
+			t.Errorf("Expected action -%d verb to be %s, was: %s", i, verb, actions[relativePos].GetVerb())
+		}
+		if resource != actions[relativePos].GetResource().Resource {
+			t.Errorf("Expected action -%d resource to be %s, was: %s", i, resource, actions[relativePos].GetResource().Resource)
+		}
+	}
+}
+
+func createService(t *testing.T, esController *endpointSliceController, namespace, serviceName string) *v1.Service {
+	t.Helper()
+	service := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              serviceName,
+			Namespace:         namespace,
+			CreationTimestamp: metav1.NewTime(time.Now()),
+			UID:               types.UID(namespace + "-" + serviceName),
+		},
+		Spec: v1.ServiceSpec{
+			Ports:      []v1.ServicePort{{TargetPort: intstr.FromInt32(80)}},
+			Selector:   map[string]string{"foo": "bar"},
+			IPFamilies: []v1.IPFamily{v1.IPv4Protocol},
+		},
+	}
+	esController.serviceStore.Add(service)
+	_, err := esController.k8sAPI.Client.CoreV1().Services(namespace).Create(context.TODO(), service, metav1.CreateOptions{})
+	if err != nil {
+		t.Error("Expected no error creating service")
+	}
+	return service
+}
+
+func standardSyncService(t *testing.T, esController *endpointSliceController, namespace, serviceName string) {
+	t.Helper()
+	createService(t, esController, namespace, serviceName)
+
+	err := esController.syncService(fmt.Sprintf("%s/%s", namespace, serviceName))
+	if err != nil {
+		t.Error("Expected no error syncing service")
+	}
+}
diff --git a/controller/api/destination/external-workload/endpoints_reconciler_test.go b/controller/api/destination/external-workload/endpoints_reconciler_test.go
new file mode 100644
index 000000000..4d545d82c
--- /dev/null
+++ b/controller/api/destination/external-workload/endpoints_reconciler_test.go
@@ -0,0 +1,872 @@
+package externalworkload
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	ewv1alpha1 "github.com/linkerd/linkerd2/controller/gen/apis/externalworkload/v1alpha1"
+	"github.com/linkerd/linkerd2/controller/k8s"
+
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	k8stesting "k8s.io/client-go/testing"
+	epsliceutil "k8s.io/endpointslice/util"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/rand"
+	"sigs.k8s.io/yaml"
+)
+
+var (
+	httpUnnamedPort = corev1.ServicePort{
+		Port: 8080,
+		TargetPort: intstr.IntOrString{
+			Type:   intstr.Int,
+			IntVal: 8080,
+		},
+	}
+
+	httpNamedPort = corev1.ServicePort{
+		TargetPort: intstr.IntOrString{
+			Type:   intstr.String,
+			StrVal: "http",
+		},
+	}
+
+	defaultTestEndpointsQuota = 100
+
+	testControllerName = "test-controller"
+)
+
+// === Test create / update / delete ===
+
+// Test that when a service has no endpointslices written to the API Server, reconciling
+// with a workload will create a new endpointslice.
+func TestReconcilerCreatesNewEndpointSlice(t *testing.T) {
+	// We do not need to receive anything through the informers so
+	// create a client with no cached resources
+	k8sAPI, err := k8s.NewFakeAPI([]string{}...)
+	if err != nil {
+		t.Fatalf("unexpected error when creating Kubernetes clientset: %v", err)
+	}
+
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "")
+	ew := makeExternalWorkload("1", "wlkd-1", map[string]string{"app": ""}, map[int32]string{8080: ""}, []string{"192.0.2.0"})
+	ew.ObjectMeta.UID = types.UID(fmt.Sprintf("%s-%s", ew.Namespace, ew.Name))
+
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	err = r.reconcile(svc, []*ewv1alpha1.ExternalWorkload{ew}, nil)
+	if err != nil {
+		t.Fatalf("unexpected error when reconciling endpoints: %v", err)
+	}
+
+	expectedEndpoint := makeEndpoint([]string{"192.0.2.0"}, true, ew)
+	es := fetchEndpointSlices(t, k8sAPI, svc)
+	if len(es) != 1 {
+		t.Fatalf("expected %d endpointslices after reconciliation, got %d instead", 1, len(es))
+	}
+
+	if len(es[0].Endpoints) != 1 {
+		t.Fatalf("expected %d endpointslices after reconciliation, got %d instead", 1, len(es[0].Endpoints))
+	}
+
+	if es[0].AddressType != discoveryv1.AddressTypeIPv4 {
+		t.Fatalf("expected endpointslice to have AF %s, got %s instead", discoveryv1.AddressTypeIPv4, es[0].AddressType)
+	}
+	ep := es[0].Endpoints[0]
+	diffEndpoints(t, ep, expectedEndpoint)
+}
+
+// Test that when a service has no endpointslices written to the API Server, reconciling
+// with a workload will create a new endpointslice. Since it is a headless
+// service, we will also get a hostname
+func TestReconcilerCreatesNewEndpointSliceHeadless(t *testing.T) {
+	// We do not need to receive anything through the informers so
+	// create a client with no cached resources
+	k8sAPI, err := k8s.NewFakeAPI([]string{}...)
+	if err != nil {
+		t.Fatalf("unexpected error when creating Kubernetes clientset: %v", err)
+	}
+
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "")
+	svc.Spec.ClusterIP = corev1.ClusterIPNone
+	ew := makeExternalWorkload("1", "wlkd-1", map[string]string{"app": ""}, map[int32]string{8080: ""}, []string{"192.0.2.0"})
+	ew.Namespace = "default"
+	ew.ObjectMeta.UID = types.UID(fmt.Sprintf("%s-%s", ew.Namespace, ew.Name))
+
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	err = r.reconcile(svc, []*ewv1alpha1.ExternalWorkload{ew}, nil)
+	if err != nil {
+		t.Fatalf("unexpected error when reconciling endpoints: %v", err)
+	}
+
+	expectedEndpoint := makeEndpoint([]string{"192.0.2.0"}, true, ew)
+	es := fetchEndpointSlices(t, k8sAPI, svc)
+	if len(es) != 1 {
+		t.Fatalf("expected %d endpointslices after reconciliation, got %d instead", 1, len(es))
+	}
+
+	if len(es[0].Endpoints) != 1 {
+		t.Fatalf("expected %d endpointslices after reconciliation, got %d instead", 1, len(es[0].Endpoints))
+	}
+
+	if es[0].AddressType != discoveryv1.AddressTypeIPv4 {
+		t.Fatalf("expected endpointslice to have AF %s, got %s instead", discoveryv1.AddressTypeIPv4, es[0].AddressType)
+	}
+	ep := es[0].Endpoints[0]
+	diffEndpoints(t, ep, expectedEndpoint)
+
+	if _, ok := es[0].Labels[corev1.IsHeadlessService]; !ok {
+		t.Errorf("expected \"%s\" label to be present on the service", corev1.IsHeadlessService)
+	}
+
+	if ep.Hostname == nil {
+		t.Fatalf("expected endpoint to have a hostname")
+	}
+
+	if *ep.Hostname != ew.Name {
+		t.Errorf("expected \"%s\" as a hostname, got: %s", ew.Name, *ep.Hostname)
+	}
+
+}
+
+// Test that when a service has an endpointslice written to the API Server,
+// reconciling with the two workloads updates the endpointslice
+func TestReconcilerUpdatesEndpointSlice(t *testing.T) {
+	// Create a service
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "")
+
+	// Create our existing workload
+	ewCreated := makeExternalWorkload("1", "wlkd-1", map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{"192.0.2.1"})
+	ewCreated.ObjectMeta.UID = types.UID(fmt.Sprintf("%s-%s", ewCreated.Namespace, ewCreated.Name))
+
+	// Create an endpointslice
+	port := int32(8080)
+	ports := []discoveryv1.EndpointPort{{
+		Port: &port,
+	}}
+	es := makeEndpointSlice(svc, ports)
+	endpoints := []discoveryv1.Endpoint{}
+	endpoints = append(endpoints, externalWorkloadToEndpoint(discoveryv1.AddressTypeIPv4, ewCreated, svc))
+	es.Endpoints = endpoints
+	es.Generation = 1
+
+	// Create our "new" workload
+	ewUpdated := makeExternalWorkload("1", "wlkd-2", map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{"192.0.2.0"})
+	ewUpdated.ObjectMeta.UID = types.UID(fmt.Sprintf("%s-%s", ewUpdated.Namespace, ewUpdated.Name))
+
+	// Convert endpointslice to string and register with fake client
+	k8sAPI, err := k8s.NewFakeAPI(endpointSliceAsYaml(t, es))
+	if err != nil {
+		t.Fatalf("unexpected error when creating Kubernetes clientset: %v", err)
+	}
+
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	err = r.reconcile(svc, []*ewv1alpha1.ExternalWorkload{ewCreated, ewUpdated}, []*discoveryv1.EndpointSlice{es})
+	if err != nil {
+		t.Fatalf("unexpected error when reconciling endpoints: %v", err)
+	}
+
+	slice, err := k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Get(context.Background(), es.Name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error when retrieving endpointslice: %v", err)
+	}
+	if len(slice.Endpoints) != 2 {
+		t.Fatalf("expected %d endpointslices after reconciliation, got %d instead", 2, len(slice.Endpoints))
+	}
+
+	if slice.AddressType != discoveryv1.AddressTypeIPv4 {
+		t.Fatalf("expected endpointslice to have AF %s, got %s instead", discoveryv1.AddressTypeIPv4, slice.AddressType)
+	}
+
+	for _, ep := range slice.Endpoints {
+		if ep.TargetRef.Name == ewUpdated.Name {
+			expectedEndpoint := makeEndpoint([]string{"192.0.2.0"}, true, ewUpdated)
+			diffEndpoints(t, ep, expectedEndpoint)
+		} else if ep.TargetRef.Name == ewCreated.Name {
+			expectedEndpoint := makeEndpoint([]string{"192.0.2.1"}, true, ewCreated)
+			diffEndpoints(t, ep, expectedEndpoint)
+		} else {
+			t.Errorf("found unexpected targetRef name %s", ep.TargetRef.Name)
+		}
+	}
+}
+
+// When an endpoint has changed, we should see the endpointslice change its
+// endpoint
+func TestReconcilerUpdatesEndpointSliceInPlace(t *testing.T) {
+	// Create a service
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "")
+
+	// Create our existing workload
+	ewCreated := makeExternalWorkload("1", "wlkd-1", map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{"192.0.2.1"})
+	ewCreated.ObjectMeta.UID = types.UID(fmt.Sprintf("%s-%s", ewCreated.Namespace, ewCreated.Name))
+
+	// Create an endpointslice
+	port := int32(8080)
+	ports := []discoveryv1.EndpointPort{{
+		Port: &port,
+	}}
+	es := makeEndpointSlice(svc, ports)
+	endpoints := []discoveryv1.Endpoint{}
+	endpoints = append(endpoints, externalWorkloadToEndpoint(discoveryv1.AddressTypeIPv4, ewCreated, svc))
+	es.Endpoints = endpoints
+	es.Generation = 1
+
+	// Convert endpointslice to string and register with fake client
+	k8sAPI, err := k8s.NewFakeAPI(endpointSliceAsYaml(t, es))
+	if err != nil {
+		t.Fatalf("unexpected error when creating Kubernetes clientset: %v", err)
+	}
+
+	if err != nil {
+		t.Fatalf("unexpected error when retrieving endpointslice: %v", err)
+	}
+
+	// Change the workload
+	ewCreated.Labels = map[string]string{
+		corev1.LabelTopologyZone: "zone1",
+	}
+
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	err = r.reconcile(svc, []*ewv1alpha1.ExternalWorkload{ewCreated, ewCreated}, []*discoveryv1.EndpointSlice{es})
+	if err != nil {
+		t.Fatalf("unexpected error when reconciling endpoints: %v", err)
+	}
+
+	slice, err := k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Get(context.Background(), es.Name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error when retrieving endpointslice: %v", err)
+	}
+	if len(slice.Endpoints) != 1 {
+		t.Fatalf("expected %d endpointslices after reconciliation, got %d instead", 1, len(slice.Endpoints))
+	}
+
+	if slice.AddressType != discoveryv1.AddressTypeIPv4 {
+		t.Fatalf("expected endpointslice to have AF %s, got %s instead", discoveryv1.AddressTypeIPv4, slice.AddressType)
+	}
+
+	if slice.Generation == 1 {
+		t.Fatalf("expected endpointslice to have its generation bumped after update")
+	}
+
+	if *slice.Endpoints[0].Zone != "zone1" {
+		t.Fatalf("expected endpoint to be updated with new zone topology")
+	}
+}
+
+// === Test ports ===
+
+// A named port on a service can target a different port on a workload
+func TestReconcileEndpointSlicesNamedPorts(t *testing.T) {
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpNamedPort}, "192.0.2.1")
+	ews := []*ewv1alpha1.ExternalWorkload{}
+	// Generate a large number of external workloads
+	// randomise ports so that a named port maps to different target values
+	for i := 0; i < 300; i++ {
+		ready := !(i%3 == 0)
+		offset := i % 5
+		genIp := fmt.Sprintf("192.%d.%d.%d", i%5, i%3, i%2)
+		genPort := int32(8080 + offset)
+		ew := makeExternalWorkload("1", fmt.Sprintf("wlkd-%d", i), map[string]string{"app": "test"}, map[int32]string{genPort: "http"}, []string{genIp})
+		ew.Status.Conditions = []ewv1alpha1.WorkloadCondition{newStatusCondition(ready)}
+		ews = append(ews, ew)
+	}
+
+	k8sAPI, err := k8s.NewFakeAPI([]string{}...)
+	if err != nil {
+		t.Fatalf("unexpected error when initializing API client: %v", err)
+	}
+
+	// Start with 100 endpoints max quota. Since we have 5 possible ports
+	// mapping to name 'http' we will generate 5 slices
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{})
+	slices := fetchEndpointSlices(t, k8sAPI, svc)
+	expectedNumSlices := 5
+	if len(slices) != expectedNumSlices {
+		t.Fatalf("expected %d slices to be created, got %d instead", expectedNumSlices, len(slices))
+	}
+
+	// We should have 5 slices with 60 endpoints each
+	expectSlicesWithLengths(t, []int{60, 60, 60, 60, 60}, slices)
+	expectedSlices := []discoveryv1.EndpointSlice{}
+	for i := range slices {
+		port := int32(8080 + i)
+		expectedSlices = append(expectedSlices, discoveryv1.EndpointSlice{
+			Ports: []discoveryv1.EndpointPort{
+				{
+					Port: &port,
+				},
+			},
+			AddressType: discoveryv1.AddressTypeIPv4,
+		})
+	}
+
+	// Diff the ports
+	diffEndpointSlicePorts(t, expectedSlices, slices)
+}
+
+// === Test packing logic ===
+
+// a simple use case with 250 workloads matching a service and no existing slices
+// reconcile should create 3 slices, completely filling 2 of them
+func TestReconcileManyWorkloads(t *testing.T) {
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "10.0.2.1")
+	// start with 250 workloads
+	ews := []*ewv1alpha1.ExternalWorkload{}
+	for i := 0; i < 250; i++ {
+		ready := !(i%3 == 0)
+		genIp := fmt.Sprintf("192.%d.%d.%d", i%5, i%3, i%2)
+		ew := makeExternalWorkload("1", fmt.Sprintf("wlkd-%d", i), map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{genIp})
+
+		ew.Status.Conditions = []ewv1alpha1.WorkloadCondition{newStatusCondition(ready)}
+		ews = append(ews, ew)
+	}
+
+	k8sAPI, actions := newClientset(t, []string{})
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{})
+	expectActions(t, actions(), 3, "create", "endpointslices")
+
+	slices := fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+}
+
+// Test with preexisting slices. 250 pods matching a service:
+// * First es: 62 endpoints (all desired)
+// * Second es: 61 endpoints (all desired)
+// We have 127 leftover to add.
+//
+// We will drop 27 in the first slice closest to full
+func TestReconcileEndpointSlicesSomePreexisting(t *testing.T) {
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "10.0.2.1")
+	// start with 250 workloads
+	ews := []*ewv1alpha1.ExternalWorkload{}
+	for i := 0; i < 250; i++ {
+		ready := !(i%3 == 0)
+		genIp := fmt.Sprintf("192.%d.%d.%d", i%5, i%3, i%2)
+		ew := makeExternalWorkload("1", fmt.Sprintf("wlkd-%d", i), map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{genIp})
+
+		ew.Status.Conditions = []ewv1alpha1.WorkloadCondition{newStatusCondition(ready)}
+		ews = append(ews, ew)
+	}
+
+	// Create an endpointslice
+	port := int32(8080)
+	esPorts := []discoveryv1.EndpointPort{{
+		Port: &port,
+	}}
+
+	es1 := makeEndpointSlice(svc, esPorts)
+	// Take a quarter of workloads in the first slice
+	for i := 1; i < len(ews)-4; i += 4 {
+		addrs := []string{ews[i].Spec.WorkloadIPs[0].Ip}
+		isReady := IsEwReady(ews[i])
+		es1.Endpoints = append(es1.Endpoints, makeEndpoint(addrs, isReady, ews[i]))
+	}
+
+	es2 := makeEndpointSlice(svc, esPorts)
+	// Take a quarter of workloads in the second slice
+	for i := 3; i < len(ews)-4; i += 4 {
+		addrs := []string{ews[i].Spec.WorkloadIPs[0].Ip}
+		isReady := IsEwReady(ews[i])
+		es2.Endpoints = append(es2.Endpoints, makeEndpoint(addrs, isReady, ews[i]))
+	}
+
+	existingSlices := []*discoveryv1.EndpointSlice{es1, es2}
+	cmc := newCacheMutationCheck(existingSlices)
+	k8sAPI, actions := newClientset(t, []string{})
+	for _, slice := range existingSlices {
+		_, err := k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Create(context.TODO(), slice, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatalf("unexpected error when creating Kubernetes obj: %v", err)
+		}
+	}
+
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	r.reconcile(svc, ews, existingSlices)
+	expectActions(t, actions(), 2, "update", "endpointslices")
+
+	slices := fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+
+	// ensure cache mutation has not occurred
+	cmc.Check(t)
+}
+
+// Ensure reconciler updates everything in-place when a service requires a
+// change. That means we expect to only see updates, no creates.
+func TestReconcileEndpointSlicesUpdatingSvc(t *testing.T) {
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "10.0.2.1")
+	// start with 250 workloads
+	ews := []*ewv1alpha1.ExternalWorkload{}
+	for i := 0; i < 250; i++ {
+		ready := !(i%3 == 0)
+		genIp := fmt.Sprintf("192.%d.%d.%d", i%5, i%3, i%2)
+		ew := makeExternalWorkload("1", fmt.Sprintf("wlkd-%d", i), map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{genIp})
+
+		ew.Status.Conditions = []ewv1alpha1.WorkloadCondition{newStatusCondition(ready)}
+		ews = append(ews, ew)
+	}
+
+	k8sAPI, actions := newClientset(t, []string{})
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{})
+
+	slices := fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+	for _, ew := range ews {
+		ew.Spec.Ports[0].Port = int32(81)
+	}
+	svc.Spec.Ports[0].TargetPort.IntVal = 81
+
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{&slices[0], &slices[1], &slices[2]})
+	expectActions(t, actions(), 3, "update", "endpointslices")
+	slices = fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+	for _, slice := range slices {
+		if *slice.Ports[0].Port != 81 {
+			t.Errorf("expected targetPort value to be 81, got: %d", slice.Ports[0].Port)
+		}
+	}
+}
+
+// When service labels update, all slices will require a change.
+//
+// This test will ensure that we update slices with the appropriate labels when
+// a service has changed.
+func TestReconcileEndpointSlicesLabelsUpdatingSvc(t *testing.T) {
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "10.0.2.1")
+	// start with 250 workloads
+	ews := []*ewv1alpha1.ExternalWorkload{}
+	for i := 0; i < 250; i++ {
+		ready := !(i%3 == 0)
+		genIp := fmt.Sprintf("192.%d.%d.%d", i%5, i%3, i%2)
+		ew := makeExternalWorkload("1", fmt.Sprintf("wlkd-%d", i), map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{genIp})
+
+		ew.Status.Conditions = []ewv1alpha1.WorkloadCondition{newStatusCondition(ready)}
+		ews = append(ews, ew)
+	}
+
+	k8sAPI, actions := newClientset(t, []string{})
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{})
+
+	slices := fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+
+	// update service with new labels
+	svc.Labels = map[string]string{"foo": "bar"}
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{&slices[0], &slices[1], &slices[2]})
+	expectActions(t, actions(), 3, "update", "endpointslices")
+
+	slices = fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+	// check that the labels were updated
+	for _, slice := range slices {
+		w, ok := slice.Labels["foo"]
+		if !ok {
+			t.Errorf("expected label \"foo\" from parent service not found")
+		} else if "bar" != w {
+			t.Errorf("expected EndpointSlice to have parent service labels: have %s value, expected bar", w)
+		}
+	}
+}
+
+// In some cases, such as service labels updates, all slices for that service will require a change
+// However, this should not happen for reserved labels
+func TestReconcileEndpointSlicesReservedLabelsSvc(t *testing.T) {
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "10.0.2.1")
+	// start with 250 workloads
+	ews := []*ewv1alpha1.ExternalWorkload{}
+	for i := 0; i < 250; i++ {
+		ready := !(i%3 == 0)
+		genIp := fmt.Sprintf("192.%d.%d.%d", i%5, i%3, i%2)
+		ew := makeExternalWorkload("1", fmt.Sprintf("wlkd-%d", i), map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{genIp})
+
+		ew.Status.Conditions = []ewv1alpha1.WorkloadCondition{newStatusCondition(ready)}
+		ews = append(ews, ew)
+	}
+
+	k8sAPI, actions := newClientset(t, []string{})
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{})
+	numActionExpected := 3
+
+	slices := fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+	numActionExpected++
+
+	// update service with new labels
+	svc.Labels = map[string]string{discoveryv1.LabelServiceName: "bad", discoveryv1.LabelManagedBy: "actor", corev1.IsHeadlessService: "invalid"}
+	r.reconcile(svc, ews, []*discoveryv1.EndpointSlice{&slices[0], &slices[1], &slices[2]})
+	slices = fetchEndpointSlices(t, k8sAPI, svc)
+	numActionExpected++
+	if len(actions()) != numActionExpected {
+		t.Errorf("expected %d actions, got %d instead", numActionExpected, len(actions()))
+	}
+
+	expectSlicesWithLengths(t, []int{100, 100, 50}, slices)
+	// check that the labels were updated
+	for _, slice := range slices {
+		if v := slice.Labels[discoveryv1.LabelServiceName]; v == "bad" {
+			t.Errorf("unexpected label value \"%s\" from parent service found on slice", "bad")
+		}
+
+		if v := slice.Labels[discoveryv1.LabelManagedBy]; v == "actor" {
+			t.Errorf("unexpected label value \"%s\" from parent service found on slice", "actor")
+		}
+
+		if v := slice.Labels[corev1.IsHeadlessService]; v == "invalid" {
+			t.Errorf("unexpected label value \"%s\" from parent service found on slice", "invalid")
+		}
+	}
+}
+
+func TestEndpointSlicesAreRecycled(t *testing.T) {
+	svc := makeIPv4Service(map[string]string{"app": "test"}, []corev1.ServicePort{httpUnnamedPort}, "10.0.2.1")
+	// start with 250 workloads
+	ews := []*ewv1alpha1.ExternalWorkload{}
+	for i := 0; i < 300; i++ {
+		ready := !(i%3 == 0)
+		genIp := fmt.Sprintf("192.%d.%d.%d", i%5, i%3, i%2)
+		ew := makeExternalWorkload("1", fmt.Sprintf("wlkd-%d", i), map[string]string{"app": "test"}, map[int32]string{8080: ""}, []string{genIp})
+
+		ew.Status.Conditions = []ewv1alpha1.WorkloadCondition{newStatusCondition(ready)}
+		ews = append(ews, ew)
+	}
+
+	// Create an endpointslice
+	port := int32(8080)
+	esPorts := []discoveryv1.EndpointPort{{
+		Port: &port,
+	}}
+
+	// generate 10 existing slices with 30 endpoints each
+	existingSlices := []*discoveryv1.EndpointSlice{}
+	for i, ew := range ews {
+		sliceNum := i / 30
+		if i%30 == 0 {
+			existingSlices = append(existingSlices, makeEndpointSlice(svc, esPorts))
+		}
+
+		addrs := []string{ews[i].Spec.WorkloadIPs[0].Ip}
+		isReady := IsEwReady(ews[i])
+		existingSlices[sliceNum].Endpoints = append(existingSlices[sliceNum].Endpoints, makeEndpoint(addrs, isReady, ew))
+	}
+
+	cmc := newCacheMutationCheck(existingSlices)
+	k8sAPI, err := k8s.NewFakeAPI([]string{}...)
+	if err != nil {
+		t.Fatalf("unexpected error when creating Kubernetes clientset: %v", err)
+	}
+
+	for _, slice := range existingSlices {
+		_, err := k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).Create(context.TODO(), slice, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatalf("unexpected error when creating Kubernetes obj: %v", err)
+		}
+	}
+
+	for _, ew := range ews {
+		ew.Spec.Ports[0].Port = int32(81)
+	}
+
+	// changing a service port should require all slices to be updated, time for a repack
+	svc.Spec.Ports[0].TargetPort.IntVal = 81
+	r := newEndpointsReconciler(k8sAPI, testControllerName, defaultTestEndpointsQuota)
+	r.reconcile(svc, ews, existingSlices)
+
+	slices := fetchEndpointSlices(t, k8sAPI, svc)
+	expectSlicesWithLengths(t, []int{100, 100, 100}, slices)
+	// ensure cache mutation has not occurred
+	cmc.Check(t)
+}
+
+func newClientset(t *testing.T, k8sConfigs []string) (*k8s.API, func() []k8stesting.Action) {
+	k8sAPI, actions, err := k8s.NewFakeAPIWithActions(k8sConfigs...)
+
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	return k8sAPI, actions
+}
+
+func makeEndpointSlice(svc *corev1.Service, ports []discoveryv1.EndpointPort) *discoveryv1.EndpointSlice {
+	// We need an ownerRef to point to our service
+	ownerRef := metav1.NewControllerRef(svc, schema.GroupVersionKind{Version: "v1", Kind: "Service"})
+	slice := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            fmt.Sprintf("linkerd-external-%s-%s", svc.Name, rand.String(8)),
+			Namespace:       svc.Namespace,
+			Labels:          map[string]string{},
+			OwnerReferences: []metav1.OwnerReference{*ownerRef},
+		},
+		AddressType: discoveryv1.AddressTypeIPv4,
+		Endpoints:   []discoveryv1.Endpoint{},
+		Ports:       ports,
+	}
+	labels, _ := setEndpointSliceLabels(slice, svc, testControllerName)
+	slice.Labels = labels
+	return slice
+}
+
+// Helper function that tests a set of slices matches a list of expected lengths
+// for number of endpoints
+func expectSlicesWithLengths(t *testing.T, expectedLengths []int, es []discoveryv1.EndpointSlice) {
+	t.Helper()
+	noMatch := []string{}
+	for _, slice := range es {
+		epLen := len(slice.Endpoints)
+		matched := false
+		for i := 0; i < len(expectedLengths); i++ {
+			if epLen == expectedLengths[i] {
+				matched = true
+				expectedLengths = append(expectedLengths[:i], expectedLengths[i+1:]...)
+				break
+			}
+		}
+
+		if !matched {
+			noMatch = append(noMatch, fmt.Sprintf("%s/%s (%d)", slice.Namespace, slice.Name, len(slice.Endpoints)))
+		}
+	}
+
+	if len(noMatch) > 0 {
+		t.Fatalf("slices %s did not match the required lengths, unmatched lengths: %v", strings.Join(noMatch, ", "), expectedLengths)
+	}
+}
+
+func diffEndpointSlicePorts(t *testing.T, expected, actual []discoveryv1.EndpointSlice) {
+	t.Helper()
+	if len(expected) != len(actual) {
+		t.Fatalf("expected %d slices, got %d instead", len(expected), len(actual))
+	}
+
+	unmatched := []discoveryv1.EndpointSlice{}
+	for _, actualSlice := range actual {
+		matched := false
+		for i := 0; i < len(expected); i++ {
+			expectedSlice := expected[i]
+			expectedHash := epsliceutil.NewPortMapKey(expectedSlice.Ports)
+			actualHash := epsliceutil.NewPortMapKey(actualSlice.Ports)
+
+			if (actualSlice.AddressType == expectedSlice.AddressType) &&
+				(actualHash == expectedHash) {
+				matched = true
+				expected = append(expected[:i], expected[i+1:]...)
+				break
+			}
+		}
+
+		if !matched {
+			unmatched = append(unmatched, actualSlice)
+		}
+	}
+
+	if len(expected) != 0 {
+		t.Errorf("expected slices not found in actual list of EndpointSlices")
+	}
+
+	if len(unmatched) > 0 {
+		t.Errorf("found %d slices that do not match expected ports", len(unmatched))
+	}
+}
+
+// === Test utilities ===
+
+// Modify a slice's name in-place since the fake API server does not support
+// generated names
+func endpointSliceAsYaml(t *testing.T, es *discoveryv1.EndpointSlice) string {
+	if es.Name == "" {
+		es.Name = fmt.Sprintf("%s-%s", es.ObjectMeta.GenerateName, rand.String(5))
+		es.GenerateName = ""
+	}
+	es.TypeMeta = metav1.TypeMeta{
+		APIVersion: "discovery.k8s.io/v1",
+		Kind:       "EndpointSlice",
+	}
+
+	b, err := yaml.Marshal(es)
+	if err != nil {
+		t.Fatalf("unexpected error when serializing endpointslices to yaml")
+	}
+
+	return string(b)
+
+}
+
+func makeIPv4Service(selector map[string]string, ports []corev1.ServicePort, clusterIP string) *corev1.Service {
+	return &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-svc",
+			Namespace: "default",
+			UID:       "default-test-svc",
+		},
+		Spec: corev1.ServiceSpec{
+			Ports:      ports,
+			Selector:   selector,
+			ClusterIP:  clusterIP,
+			IPFamilies: []corev1.IPFamily{corev1.IPv4Protocol},
+		},
+		Status: corev1.ServiceStatus{},
+	}
+}
+
+func makeEndpoint(addrs []string, isReady bool, ew *ewv1alpha1.ExternalWorkload) discoveryv1.Endpoint {
+	rdy := &isReady
+	term := !isReady
+	ep := discoveryv1.Endpoint{
+		Addresses: addrs,
+		Conditions: discoveryv1.EndpointConditions{
+			Ready:       rdy,
+			Serving:     rdy,
+			Terminating: &term,
+		},
+		TargetRef: &corev1.ObjectReference{
+			Kind:      ew.Kind,
+			Namespace: ew.Namespace,
+			Name:      ew.Name,
+			UID:       ew.UID,
+		},
+	}
+	return ep
+}
+
+func fetchEndpointSlices(t *testing.T, k8sAPI *k8s.API, svc *corev1.Service) []discoveryv1.EndpointSlice {
+	t.Helper()
+	selector := labels.Set(map[string]string{
+		discoveryv1.LabelServiceName: svc.Name,
+		discoveryv1.LabelManagedBy:   testControllerName,
+	}).AsSelectorPreValidated()
+	fetchedSlices, err := k8sAPI.Client.DiscoveryV1().EndpointSlices(svc.Namespace).List(context.Background(), metav1.ListOptions{
+		LabelSelector: selector.String(),
+	})
+	if err != nil {
+		t.Fatalf("unexpected error when fetching endpointslices: %v", err)
+	}
+
+	return fetchedSlices.Items
+}
+
+func diffEndpoints(t *testing.T, actual, expected discoveryv1.Endpoint) {
+	t.Helper()
+	if len(actual.Addresses) != len(expected.Addresses) {
+		t.Errorf("expected %d addresses, got %d instead", len(expected.Addresses), len(actual.Addresses))
+	}
+
+	if actual.Conditions.Ready != nil && expected.Conditions.Ready != nil {
+		if *actual.Conditions.Ready != *expected.Conditions.Ready {
+			t.Errorf("expected \"ready\" condition to be %t, got %t instead", *expected.Conditions.Ready, *actual.Conditions.Ready)
+		}
+	}
+
+	if actual.Conditions.Serving != nil && expected.Conditions.Serving != nil {
+		if *actual.Conditions.Serving != *expected.Conditions.Serving {
+			t.Errorf("expected \"serving\" condition to be %t, got %t instead", *expected.Conditions.Serving, *actual.Conditions.Serving)
+		}
+	}
+
+	if actual.Conditions.Terminating != nil && expected.Conditions.Terminating != nil {
+		if *actual.Conditions.Terminating != *expected.Conditions.Terminating {
+			t.Errorf("expected \"terminating\" condition to be %t, got %t instead", *expected.Conditions.Terminating, *actual.Conditions.Terminating)
+		}
+	}
+
+	if actual.Zone != nil && expected.Zone != nil {
+		if *actual.Zone != *expected.Zone {
+			t.Errorf("expected \"zone=%s\", got \"zone=%s\" instead", *expected.Zone, *actual.Zone)
+		}
+	}
+
+	actualAddrs := toSet(actual.Addresses)
+	expAddrs := toSet(expected.Addresses)
+	for actualAddr := range actualAddrs {
+		if _, found := expAddrs[actualAddr]; !found {
+			t.Errorf("found unexpected address %s in the actual endpoint", actualAddr)
+		}
+	}
+
+	for expAddr := range expAddrs {
+		if _, found := actualAddrs[expAddr]; !found {
+			t.Errorf("expected to find address %s in the actual endpoint", expAddr)
+		}
+	}
+
+	expRef := expected.TargetRef
+	actRef := actual.TargetRef
+	if expRef.UID != actRef.UID {
+		t.Errorf("expected targetRef with UID %s; got %s instead", expRef.UID, actRef.UID)
+	}
+
+	if expRef.Name != actRef.Name {
+		t.Errorf("expected targetRef with name %s; got %s instead", expRef.Name, actRef.Name)
+	}
+
+}
+
+// === impl cache mutation check
+
+// Code originally forked from:
+//
+// https://github.com/kubernetes/endpointslice/commit/a09c1c9580d13f5020248d25c7fd11f5dde6dd9b
+
+// cacheMutationCheck helps ensure that cached objects have not been changed
+// in any way throughout a test run.
+type cacheMutationCheck struct {
+	objects []cacheObject
+}
+
+// cacheObject stores a reference to an original object as well as a deep copy
+// of that object to track any mutations in the original object.
+type cacheObject struct {
+	original runtime.Object
+	deepCopy runtime.Object
+}
+
+// newCacheMutationCheck initializes a cacheMutationCheck with EndpointSlices.
+func newCacheMutationCheck(endpointSlices []*discoveryv1.EndpointSlice) cacheMutationCheck {
+	cmc := cacheMutationCheck{}
+	for _, endpointSlice := range endpointSlices {
+		cmc.Add(endpointSlice)
+	}
+	return cmc
+}
+
+// Add appends a runtime.Object and a deep copy of that object into the
+// cacheMutationCheck.
+func (cmc *cacheMutationCheck) Add(o runtime.Object) {
+	cmc.objects = append(cmc.objects, cacheObject{
+		original: o,
+		deepCopy: o.DeepCopyObject(),
+	})
+}
+
+// Check verifies that no objects in the cacheMutationCheck have been mutated.
+func (cmc *cacheMutationCheck) Check(t *testing.T) {
+	for _, o := range cmc.objects {
+		if !reflect.DeepEqual(o.original, o.deepCopy) {
+			// Cached objects can't be safely mutated and instead should be deep
+			// copied before changed in any way.
+			t.Errorf("Cached object was unexpectedly mutated. Original: %+v, Mutated: %+v", o.deepCopy, o.original)
+		}
+	}
+}
+
+func toSet(s []string) map[string]struct{} {
+	set := map[string]struct{}{}
+	for _, k := range s {
+		set[k] = struct{}{}
+	}
+	return set
+}
diff --git a/controller/k8s/test_helper.go b/controller/k8s/test_helper.go
index 4145ccaaf..89e10def9 100644
--- a/controller/k8s/test_helper.go
+++ b/controller/k8s/test_helper.go
@@ -9,6 +9,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	clientsetscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/metadata/fake"
+	"k8s.io/client-go/testing"
 )
 
 // NewFakeAPI provides a mock Kubernetes API for testing.
@@ -21,6 +22,16 @@ func NewFakeAPI(configs ...string) (*API, error) {
 	return NewFakeClusterScopedAPI(clientSet, spClientSet), nil
 }
 
+// NewFakeAPI provides a mock Kubernetes API for testing.
+func NewFakeAPIWithActions(configs ...string) (*API, func() []testing.Action, error) {
+	clientSet, _, _, spClientSet, err := k8s.NewFakeClientSets(configs...)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return NewFakeClusterScopedAPI(clientSet, spClientSet), clientSet.Actions, nil
+}
+
 // NewFakeAPIWithL5dClient provides a mock Kubernetes API for testing like
 // NewFakeAPI, but it also returns the mock client for linkerd CRDs
 func NewFakeAPIWithL5dClient(configs ...string) (*API, *l5dcrdclient.Interface, error) {