Repository: istio/istio (go)

Issue:
applying then deleting a ServiceEntry with the same host results in missing endpoint
### Is this the right place to submit this?

- [X] This is not a security vulnerability or a crashing bug
- [X] This is not a question about how to use Istio

### Bug Description

Apply se1. Apply se2, then delete se1. Only difference between the two is the exportTo field. The endpoint for `httpbin-vd.internal.com` will be removed and the service will become unreachable.

Restarting istiod or reapplying se1 will restore the endpoint.

```
apiVersion: networking.istio.io/v1beta1
kind: ServiceEntry
metadata:
  name: se1
  namespace: httpbin
spec:
  addresses:
  - 250.59.249.145
  exportTo:
  - default
  - httpbin
  - istio-system
  hosts:
  - httpbin-vd.internal.com
  location: MESH_INTERNAL
  ports:
  - name: http-8000
    number: 8000
    protocol: HTTP
    targetPort: 8080
  resolution: STATIC
  subjectAltNames:
  - spiffe://mgmt-cluster/ns/httpbin/sa/httpbin
  workloadSelector:
    labels:
      app: httpbin
```

```
apiVersion: networking.istio.io/v1beta1
kind: ServiceEntry
metadata:
  name: se2
  namespace: httpbin
spec:
  addresses:
  - 250.59.249.145
  exportTo:
  - istio-system
  - httpbin
  hosts:
  - httpbin-vd.internal.com
  location: MESH_INTERNAL
  ports:
  - name: http-8000
    number: 8000
    protocol: HTTP
    targetPort: 8080
  resolution: STATIC
  subjectAltNames:
  - spiffe://mgmt-cluster/ns/httpbin/sa/httpbin
  workloadSelector:
    labels:
      app: httpbin
```

### Version

```Text
reproduced on 1.19.10, 1.20.6, 1.22.0
```


### Additional Information

_No response_

Pull request description:
pilot: fix double SE deletion
Fixes #51212

In draft since I will probably go and add more comments, etc

Root cause is we key on `hostname+namespace` and delete instances. But we actually need to take into account which SE its from, otherwise if 2 SE share the same hostname, removal of one removes all instances

Patch:
diff --git a/istio.deps b/istio.deps
index 450a15dd20..e9df05b0b4 100644
--- a/istio.deps
+++ b/istio.deps
@@ -11,6 +11,6 @@
     "name": "ZTUNNEL_REPO_SHA",
     "repoName": "ztunnel",
     "file": "",
-    "lastStableSHA": "4ff9897189884b9f74abc1dc897cb4edb8c48732"
+    "lastStableSHA": "6532c553946856b4acc326f3b9ca6cc6abc718d0"
   }
 ]
diff --git a/pilot/pkg/serviceregistry/serviceentry/controller.go b/pilot/pkg/serviceregistry/serviceentry/controller.go
index 33832f3b01..eeba4d1273 100644
--- a/pilot/pkg/serviceregistry/serviceentry/controller.go
+++ b/pilot/pkg/serviceregistry/serviceentry/controller.go
@@ -81,6 +81,14 @@ const (
 	podConfigType
 )
 
+// configKeyWithParent is a superset of configKey that also encodes the parent resource. For instance, if something comes
+// from a ServiceEntry selector, the parent is the ServiceEntry
+// This is used to distinguish between 1 config (Pod/SE) selected by 2 different parents (ServiceEntry).
+type configKeyWithParent struct {
+	configKey
+	parent types.NamespacedName
+}
+
 // configKey unique identifies a config object managed by this registry (ServiceEntry and WorkloadEntry)
 type configKey struct {
 	kind      configType
@@ -173,7 +181,7 @@ func newController(store model.ConfigStore, xdsUpdater model.XDSUpdater, meshCon
 		meshWatcher: meshConfig,
 		serviceInstances: serviceInstancesStore{
 			ip2instance:            map[string][]*model.ServiceInstance{},
-			instances:              map[instancesKey]map[configKey][]*model.ServiceInstance{},
+			instances:              map[instancesKey]map[configKeyWithParent][]*model.ServiceInstance{},
 			instancesBySE:          map[types.NamespacedName]map[configKey][]*model.ServiceInstance{},
 			instancesByHostAndPort: sets.New[hostPort](),
 		},
@@ -273,9 +281,15 @@ func (s *Controller) workloadEntryHandler(old, curr config.Config, event model.E
 		}
 		instance := s.convertWorkloadEntryToServiceInstances(wle, services, se, &key, s.Cluster())
 		instancesUpdated = append(instancesUpdated, instance...)
+		parentKey := configKeyWithParent{
+			configKey: key,
+			parent:    namespacedName,
+		}
 		if event == model.EventDelete {
 			s.serviceInstances.deleteServiceEntryInstances(namespacedName, key)
+			s.serviceInstances.deleteInstanceKeys(parentKey, instancesUpdated)
 		} else {
+			s.serviceInstances.updateInstances(parentKey, instancesUpdated)
 			s.serviceInstances.updateServiceEntryInstancesPerConfig(namespacedName, key, instance)
 		}
 		addConfigs(se, services)
@@ -290,18 +304,20 @@ func (s *Controller) workloadEntryHandler(old, curr config.Config, event model.E
 			continue
 		}
 		instance := s.convertWorkloadEntryToServiceInstances(wle, services, se, &key, s.Cluster())
+		parentKey := configKeyWithParent{
+			configKey: key,
+			parent:    namespacedName,
+		}
 		instancesDeleted = append(instancesDeleted, instance...)
 		s.serviceInstances.deleteServiceEntryInstances(namespacedName, key)
+		s.serviceInstances.deleteInstanceKeys(parentKey, instance)
 		addConfigs(se, services)
 	}
 
-	s.serviceInstances.deleteInstanceKeys(key, instancesDeleted)
 	if event == model.EventDelete {
 		s.workloadInstances.Delete(wi)
-		s.serviceInstances.deleteInstanceKeys(key, instancesUpdated)
 	} else {
 		s.workloadInstances.Insert(wi)
-		s.serviceInstances.updateInstances(key, instancesUpdated)
 	}
 	s.mutex.Unlock()
 
@@ -385,14 +401,14 @@ func (s *Controller) serviceEntryHandler(old, curr config.Config, event model.Ev
 	serviceInstancesByConfig, serviceInstances := s.buildServiceInstances(curr, cs)
 	oldInstances := s.serviceInstances.getServiceEntryInstances(key)
 	for configKey, old := range oldInstances {
-		s.serviceInstances.deleteInstanceKeys(configKey, old)
+		s.serviceInstances.deleteInstanceKeys(configKeyWithParent{configKey: configKey, parent: key}, old)
 	}
 	if event == model.EventDelete {
 		s.serviceInstances.deleteAllServiceEntryInstances(key)
 	} else {
 		// Update the indexes with new instances.
 		for ckey, value := range serviceInstancesByConfig {
-			s.serviceInstances.addInstances(ckey, value)
+			s.serviceInstances.addInstances(configKeyWithParent{configKey: ckey, parent: key}, value)
 		}
 		s.serviceInstances.updateServiceEntryInstances(key, serviceInstancesByConfig)
 	}
@@ -530,6 +546,7 @@ func (s *Controller) WorkloadInstanceHandler(wi *model.WorkloadInstance, event m
 			// If the labels didn't change. And the new SE doesn't match then the old didn't match either and we can skip processing it.
 			continue
 		}
+		cpKey := configKeyWithParent{configKey: key, parent: config.NamespacedName(cfg)}
 
 		// If we are here, then there are 3 possible cases :
 		// Case 1 : The new wi is a subset of se
@@ -545,17 +562,24 @@ func (s *Controller) WorkloadInstanceHandler(wi *model.WorkloadInstance, event m
 			// If the workload instance still matches. We take care of the possible events.
 			instances = append(instances, currInstance...)
 			if addressToDelete != "" {
+				cfgInstancesDeleted := []*model.ServiceInstance{}
 				for _, i := range currInstance {
 					di := i.DeepCopy()
 					di.Endpoint.Address = addressToDelete
-					instancesDeleted = append(instancesDeleted, di)
+					cfgInstancesDeleted = append(cfgInstancesDeleted, di)
 				}
+				s.serviceInstances.deleteInstanceKeys(cpKey, cfgInstancesDeleted)
 				s.serviceInstances.deleteServiceEntryInstances(seNamespacedName, key)
 			} else if event == model.EventDelete {
 				s.serviceInstances.deleteServiceEntryInstances(seNamespacedName, key)
 			} else {
 				s.serviceInstances.updateServiceEntryInstancesPerConfig(seNamespacedName, key, currInstance)
 			}
+			if event == model.EventDelete {
+				s.serviceInstances.deleteInstanceKeys(cpKey, currInstance)
+			} else {
+				s.serviceInstances.updateInstances(cpKey, currInstance)
+			}
 			// If serviceentry's resolution is DNS, make a full push
 			// TODO: maybe cds?
 			if isDNSTypeServiceEntry(se) &&
@@ -576,19 +600,11 @@ func (s *Controller) WorkloadInstanceHandler(wi *model.WorkloadInstance, event m
 			// Since the instance doesn't match the SE anymore. We remove it from the list.
 			oldInstance := convertWorkloadInstanceToServiceInstance(oldWi, services, se)
 			instancesDeleted = append(instancesDeleted, oldInstance...)
+			s.serviceInstances.deleteInstanceKeys(cpKey, oldInstance)
 			s.serviceInstances.deleteServiceEntryInstances(seNamespacedName, key)
 		}
 	}
 
-	if len(instancesDeleted) > 0 {
-		s.serviceInstances.deleteInstanceKeys(key, instancesDeleted)
-	}
-
-	if event == model.EventDelete {
-		s.serviceInstances.deleteInstanceKeys(key, instances)
-	} else {
-		s.serviceInstances.updateInstances(key, instances)
-	}
 	s.mutex.Unlock()
 
 	s.edsUpdate(append(instances, instancesDeleted...))
diff --git a/pilot/pkg/serviceregistry/serviceentry/store.go b/pilot/pkg/serviceregistry/serviceentry/store.go
index 3898f895ce..355a64e778 100644
--- a/pilot/pkg/serviceregistry/serviceentry/store.go
+++ b/pilot/pkg/serviceregistry/serviceentry/store.go
@@ -30,7 +30,7 @@ type hostPort struct {
 type serviceInstancesStore struct {
 	ip2instance map[string][]*model.ServiceInstance
 	// service instances by hostname -> config
-	instances map[instancesKey]map[configKey][]*model.ServiceInstance
+	instances map[instancesKey]map[configKeyWithParent][]*model.ServiceInstance
 	// instances only for serviceentry
 	instancesBySE map[types.NamespacedName]map[configKey][]*model.ServiceInstance
 	// instancesByHostAndPort tells whether the host has instances.
@@ -61,7 +61,7 @@ func (s *serviceInstancesStore) getByKey(key instancesKey) []*model.ServiceInsta
 // deleteInstanceKeys deletes all instances with the given configKey and instanceKey
 // Note: as a convenience, this takes a []ServiceInstance instead of []instanceKey, as most callers have this format
 // However, this function only operates on the instance keys
-func (s *serviceInstancesStore) deleteInstanceKeys(key configKey, instances []*model.ServiceInstance) {
+func (s *serviceInstancesStore) deleteInstanceKeys(key configKeyWithParent, instances []*model.ServiceInstance) {
 	for _, i := range instances {
 		ikey := makeInstanceKey(i)
 		s.instancesByHostAndPort.Delete(hostPort{ikey, i.ServicePort.Port})
@@ -80,7 +80,7 @@ func (s *serviceInstancesStore) deleteInstanceKeys(key configKey, instances []*m
 }
 
 // addInstances add the instances to the store.
-func (s *serviceInstancesStore) addInstances(key configKey, instances []*model.ServiceInstance) {
+func (s *serviceInstancesStore) addInstances(key configKeyWithParent, instances []*model.ServiceInstance) {
 	for _, instance := range instances {
 		ikey := makeInstanceKey(instance)
 		hostPort := hostPort{ikey, instance.ServicePort.Port}
@@ -95,7 +95,7 @@ func (s *serviceInstancesStore) addInstances(key configKey, instances []*model.S
 			continue
 		}
 		if _, f := s.instances[ikey]; !f {
-			s.instances[ikey] = map[configKey][]*model.ServiceInstance{}
+			s.instances[ikey] = map[configKeyWithParent][]*model.ServiceInstance{}
 		}
 		s.instancesByHostAndPort.Insert(hostPort)
 		s.instances[ikey][key] = append(s.instances[ikey][key], instance)
@@ -105,7 +105,7 @@ func (s *serviceInstancesStore) addInstances(key configKey, instances []*model.S
 	}
 }
 
-func (s *serviceInstancesStore) updateInstances(key configKey, instances []*model.ServiceInstance) {
+func (s *serviceInstancesStore) updateInstances(key configKeyWithParent, instances []*model.ServiceInstance) {
 	// first delete
 	s.deleteInstanceKeys(key, instances)
 
diff --git a/pkg/config/model.go b/pkg/config/model.go
index e51e3a5e32..6e0f720789 100644
--- a/pkg/config/model.go
+++ b/pkg/config/model.go
@@ -445,7 +445,7 @@ type Namer interface {
 	GetNamespace() string
 }
 
-func NamespacedName(o metav1.Object) kubetypes.NamespacedName {
+func NamespacedName[T Namer](o T) kubetypes.NamespacedName {
 	return kubetypes.NamespacedName{
 		Namespace: o.GetNamespace(),
 		Name:      o.GetName(),


Test patch:
diff --git a/pilot/pkg/serviceregistry/serviceentry/controller_test.go b/pilot/pkg/serviceregistry/serviceentry/controller_test.go
index d128180e91..84abf0eb8c 100644
--- a/pilot/pkg/serviceregistry/serviceentry/controller_test.go
+++ b/pilot/pkg/serviceregistry/serviceentry/controller_test.go
@@ -180,6 +180,52 @@ func TestServiceDiscoveryGetService(t *testing.T) {
 	}
 }
 
+func TestServiceDiscoveryServiceDeleteOverlapping(t *testing.T) {
+	store, sd, events := initServiceDiscovery(t)
+	se1 := selector
+	se2 := func() *config.Config {
+		c := selector.DeepCopy()
+		c.Name = "alt"
+		return &c
+	}()
+	wle := createWorkloadEntry("wl", selector.Name,
+		&networking.WorkloadEntry{
+			Address:        "2.2.2.2",
+			Labels:         map[string]string{"app": "wle"},
+			ServiceAccount: "default",
+		})
+
+	expected := []*model.ServiceInstance{
+		makeInstanceWithServiceAccount(selector, "2.2.2.2", 444,
+			selector.Spec.(*networking.ServiceEntry).Ports[0],
+			map[string]string{"app": "wle"}, "default"),
+		makeInstanceWithServiceAccount(selector, "2.2.2.2", 445,
+			selector.Spec.(*networking.ServiceEntry).Ports[1],
+			map[string]string{"app": "wle"}, "default"),
+	}
+	for _, i := range expected {
+		i.Endpoint.WorkloadName = "wl"
+		i.Endpoint.Namespace = selector.Name
+	}
+
+	// Creating SE should give us instances
+	createConfigs([]*config.Config{wle, se1}, store, t)
+	events.WaitOrFail(t, "service")
+	expectServiceInstances(t, sd, se1, 0, expected)
+
+	// Create another identical SE (different name) gives us duplicate instances
+	// Arguable whether this is correct or not...
+	createConfigs([]*config.Config{se2}, store, t)
+	events.WaitOrFail(t, "service")
+	expectServiceInstances(t, sd, se1, 0, append(slices.Clone(expected), expected...))
+	events.Clear()
+
+	// When we delete, we should get back to the original state, not delete all instances
+	deleteConfigs([]*config.Config{se1}, store, t)
+	events.WaitOrFail(t, "xds full")
+	expectServiceInstances(t, sd, se1, 0, expected)
+}
+
 // TestServiceDiscoveryServiceUpdate test various add/update/delete events for ServiceEntry
 // nolint: lll
 func TestServiceDiscoveryServiceUpdate(t *testing.T) {
diff --git a/pilot/pkg/serviceregistry/serviceentry/store_test.go b/pilot/pkg/serviceregistry/serviceentry/store_test.go
index e2b1420dc2..814d52c92a 100644
--- a/pilot/pkg/serviceregistry/serviceentry/store_test.go
+++ b/pilot/pkg/serviceregistry/serviceentry/store_test.go
@@ -22,6 +22,7 @@ import (
 
 	networking "istio.io/api/networking/v1alpha3"
 	"istio.io/istio/pilot/pkg/model"
+	"istio.io/istio/pkg/config"
 	"istio.io/istio/pkg/config/constants"
 	"istio.io/istio/pkg/ptr"
 	"istio.io/istio/pkg/test/util/assert"
@@ -31,7 +32,7 @@ import (
 func TestServiceInstancesStore(t *testing.T) {
 	store := serviceInstancesStore{
 		ip2instance:            map[string][]*model.ServiceInstance{},
-		instances:              map[instancesKey]map[configKey][]*model.ServiceInstance{},
+		instances:              map[instancesKey]map[configKeyWithParent][]*model.ServiceInstance{},
 		instancesBySE:          map[types.NamespacedName]map[configKey][]*model.ServiceInstance{},
 		instancesByHostAndPort: sets.Set[hostPort]{},
 	}
@@ -44,7 +45,8 @@ func TestServiceInstancesStore(t *testing.T) {
 		namespace: "default",
 		name:      "test-wle",
 	}
-	store.addInstances(cKey, instances)
+	cpKey := configKeyWithParent{configKey: cKey, parent: config.NamespacedName(selector)}
+	store.addInstances(cpKey, instances)
 
 	// 1. test getByIP
 	gotInstances := store.getByIP("1.1.1.1")
@@ -99,7 +101,7 @@ func TestServiceInstancesStore(t *testing.T) {
 	}
 
 	// 7. test deleteInstanceKeys
-	store.deleteInstanceKeys(cKey, instances)
+	store.deleteInstanceKeys(cpKey, instances)
 	gotInstances = store.getAll()
 	if len(gotInstances) != 0 {
 		t.Errorf("got unexpected instances %v", gotSeInstances)
@@ -148,7 +150,7 @@ func TestServiceInstancesForDnsRoundRobinLB(t *testing.T) {
 	otherNs.Namespace = "other"
 	store := serviceInstancesStore{
 		ip2instance:            map[string][]*model.ServiceInstance{},
-		instances:              map[instancesKey]map[configKey][]*model.ServiceInstance{},
+		instances:              map[instancesKey]map[configKeyWithParent][]*model.ServiceInstance{},
 		instancesBySE:          map[types.NamespacedName]map[configKey][]*model.ServiceInstance{},
 		instancesByHostAndPort: sets.Set[hostPort]{},
 	}
@@ -160,11 +162,14 @@ func TestServiceInstancesForDnsRoundRobinLB(t *testing.T) {
 		namespace: "dns",
 		name:      "dns-round-robin-1",
 	}
+	cpKey := configKeyWithParent{configKey: cKey, parent: config.NamespacedName(selector)}
 	// Add instance related to first Service Entry and validate they are added correctly.
-	store.addInstances(cKey, instances)
+	store.addInstances(cpKey, instances)
 
 	store.addInstances(
-		configKey{namespace: otherNs.Namespace, name: otherNs.Name},
+		configKeyWithParent{
+			configKey: configKey{namespace: otherNs.Namespace, name: otherNs.Name},
+		},
 		[]*model.ServiceInstance{
 			makeInstance(otherNs, "1.1.1.1", 444, otherNs.Spec.(*networking.ServiceEntry).Ports[0], nil, PlainText),
 			makeInstance(otherNs, "1.1.1.1", 445, otherNs.Spec.(*networking.ServiceEntry).Ports[1], nil, PlainText),
@@ -193,11 +198,13 @@ func TestServiceInstancesForDnsRoundRobinLB(t *testing.T) {
 	instances = []*model.ServiceInstance{
 		makeInstance(dnsRoundRobinLBSE2, "2.2.2.2", 444, dnsRoundRobinLBSE2.Spec.(*networking.ServiceEntry).Ports[0], nil, PlainText),
 	}
-	cKey = configKey{
-		namespace: "dns",
-		name:      "dns-round-robin-2",
+	cpKey = configKeyWithParent{
+		configKey: configKey{
+			namespace: "dns",
+			name:      "dns-round-robin-2",
+		},
 	}
-	store.addInstances(cKey, instances)
+	store.addInstances(cpKey, instances)
 
 	assert.Equal(t, store.getByKey(instancesKey{
 		hostname:  "example.com",