Repository: istio/istio
File: pilot/pkg/serviceregistry/kube/controller/ambient/services.go (Go)

Code before:

func (a *index) cleanUpServiceEntries(serviceEntries krt.Collection[*networkingclient.ServiceEntry],
	kubeServices krt.Collection[*v1.Service],
	opts krt.OptionsBuilder,
) krt.Collection[*networkingclient.ServiceEntry] {
	serviceEntriesByNamespacedHostname := krt.NewIndex(serviceEntries, "serviceEntriesByNamespacedHostname", func(se *networkingclient.ServiceEntry) []string {
		return slices.Map(se.Spec.Hosts, func(host string) string {
			return types.NamespacedName{
				Name:      host,
				Namespace: se.Namespace,
			}.String()
		})
	})

	kubeServicesByHostname := krt.NewIndex(kubeServices, "kubeServicesByNamespacedHostname", func(s *v1.Service) []string {
		hostname := kube.ServiceHostname(s.Name, s.Namespace, a.DomainSuffix).String()
		return []string{hostname}
	})

	return krt.NewCollection(serviceEntries, func(ctx krt.HandlerContext, se *networkingclient.ServiceEntry) **networkingclient.ServiceEntry {
		// TODO: should we just keep a pointer to the original SE but list the canonical hosts in some sort of wrapper type?
		newServiceEntry := se.DeepCopy()        // ugh... but we are about to mangle here :(
		newServiceEntry.Spec.Hosts = []string{} // reset hosts, we're pruning duplicate hosts
		for _, host := range se.Spec.Hosts {
			if ptr.Flatten(krt.FetchOne(ctx, kubeServices, krt.FilterIndex(kubeServicesByHostname, host))) != nil {
				log.Warnf("ServiceEntry %s/%s is in conflict with a Kubernetes Service hostname %s, skipping", se.Namespace, se.Name, host)
				continue // next host
			}

			key := types.NamespacedName{
				Name:      host,
				Namespace: se.Namespace,
			}.String()
			oldest := true
			for _, found := range krt.Fetch(ctx, serviceEntries, krt.FilterIndex(serviceEntriesByNamespacedHostname, key)) {
				if found.Namespace == se.Namespace && found.Name == se.Name {
					// this is me, just move on
					continue // next found
				}
				if found.CreationTimestamp.Time.Before(se.CreationTimestamp.Time) {
					log.Warnf("Hostname %s is already in use by a older ServiceEntry %s/%s", host, found.Namespace, found.Name)
					oldest = false
					// if we are already not the oldest, don't keep looking
					break // next host
				}
			}
			if oldest {
				// we are the canonical ServiceEntry for this hostname, keep the host
				newServiceEntry.Spec.Hosts = append(newServiceEntry.Spec.Hosts, host)
			}


Review comment (suggestion):
We're not skipping the entire SE right? Maybe this should be:
```suggestion
				log.Warnf("ServiceEntry %s/%s is in conflict with a Kubernetes Service hostname %s, skipping this host", se.Namespace, se.Name, host)
```

Code after:
	opts krt.OptionsBuilder,
) krt.Collection[krt.Collection[krt.ObjectWithCluster[model.ServiceInfo]]] {
	// This will contain the serviceinfos derived from Services AND ServiceEntries
	LocalServiceInfosWithCluster := krt.MapCollection(
		localServiceInfos,
		wrapObjectWithCluster[model.ServiceInfo](localCluster.ID),
		opts.WithName("LocalServiceInfosWithCluster")...)

	checkServiceScope := features.EnableAmbientMultiNetwork

	// This will contain the serviceinfos derived from ServiceEntries only
	return nestedCollectionFromLocalAndRemote(
		LocalServiceInfosWithCluster,
		clusters,
		func(ctx krt.HandlerContext, cluster *multicluster.Cluster) *krt.Collection[krt.ObjectWithCluster[model.ServiceInfo]] {
			opts := []krt.CollectionOption{
				krt.WithDebugging(opts.Debugger()),
				krt.WithStop(cluster.GetStop()),
			}
			services := cluster.Services()
			waypointsPtr := krt.FetchOne(ctx, globalWaypoints, krt.FilterIndex(waypointsByCluster, cluster.ID))
			if waypointsPtr == nil {
				log.Warnf("Cluster %s does not have waypoints assigned, skipping", cluster.ID)
				return nil
			}
			waypoints := *waypointsPtr
			namespaces := cluster.Namespaces()
			// N.B Never precompute the service info for remote clusters; the merge function will do that
			servicesInfo := krt.NewCollection(services, serviceServiceBuilder(
				waypoints,
				namespaces,
				meshConfig,
				domainSuffix,
				false,
				checkServiceScope,
				func(ctx krt.HandlerContext) network.ID {
					nw := krt.FetchOne(ctx, globalNetworks.RemoteSystemNamespaceNetworks, krt.FilterIndex(globalNetworks.SystemNamespaceNetworkByCluster, cluster.ID))
					if nw == nil {
						log.Warnf("Cluster %s does not have network assigned yet, skipping", cluster.ID)
						ctx.DiscardResult()
						return ""
					}
					return nw.Network
				}, false),
				append(
					opts,
					krt.WithName(fmt.Sprintf("ambient/ServiceServiceInfos[%s]", cluster.ID)),
					krt.WithMetadata(krt.Metadata{
						multicluster.ClusterKRTMetadataKey: cluster.ID,
					}),
				)...)