Fix issue in ServiceEntry conversion

- I'm not sure if telemetry changed recently but when making requests
  of a service entry the telemetry may not include any cluster or namespace
  information, just the destination_service_name (i.e. the requested host)
- The subsequent "unknown" values for cluster and namespace can not be used
  to help fetch the ServiceEntry config for matching. Depending on
  random node ordering, sometimes we would "get lucky" and fetch the needed
  information, but the whole approach was inherently flawed.
- ServiceEntry definitions are really used by the source node, to properly
  route out of the mesh, and so the SE definition should really be fetched
  using the source node's cluster+namespace information (or so I currently
  think).  This PR makes that change.
- The PR also adds a useful optimization that can speed up this appender;
  or namespaces that only have injected service nodes we will no longer
  bother with loading SE definitions or doing the matching, because injected
  service nodes are not going to be converted to a ServiceEntry.

kiali#6962

graph/meta.go

@@ -36,6 +36,7 @@ const (
 	IsIngressGateway      MetadataKey = "isIngressGateway" // Identifies a node that is an Istio ingress gateway
 	IsIdle                MetadataKey = "isIdle"
 	IsInaccessible        MetadataKey = "isInaccessible"
+	IsInjected            MetadataKey = "isInjected"      // Identifies an injected service node (server-side use only)
 	IsK8sGatewayAPI       MetadataKey = "isK8sGatewayAPI" // true when config is autogenerated from K8s API Gateway
 	IsMTLS                MetadataKey = "isMTLS"
 	IsOutOfMesh           MetadataKey = "isOutOfMesh"

graph/telemetry/istio/appender/appender.go

@@ -306,11 +306,18 @@ func getServiceDefinition(cluster, namespace, serviceName string, gi *graph.Appe
 	return nil, false
 }
 
-func getServiceEntryHosts(gi *graph.AppenderGlobalInfo) (serviceEntryHosts, bool) {
-	if seHosts, ok := gi.Vendor[serviceEntryHostsKey]; ok {
+// getServiceEntryHosts returns ServiceEntryHost information cached for a specific cluster and namespace. If not
+// previously cached a new, empty cache entry is created and returned.
+func getServiceEntryHosts(cluster, namespace string, gi *graph.AppenderGlobalInfo) (serviceEntryHosts, bool) {
+	key := fmt.Sprintf("%s:%s:%s", serviceEntryHostsKey, cluster, namespace)
+	if seHosts, ok := gi.Vendor[key]; ok {
 		return seHosts.(serviceEntryHosts), true
 	}
-	return newServiceEntryHosts(), false
+
+	seHosts := newServiceEntryHosts()
+	gi.Vendor[key] = seHosts
+
+	return seHosts, false
 }
 
 // getWorkloadLists returns a map[clusterName]*models.WorkloadList for all clusters with traffic in the namespace, or if trafficMap is nil

graph/telemetry/istio/appender/service_entry.go

@@ -60,32 +60,100 @@ func (a ServiceEntryAppender) AppendGraph(trafficMap graph.TrafficMap, globalInf
 		return
 	}
 
-	a.applyServiceEntries(trafficMap, globalInfo, namespaceInfo)
-}
-
-func (a ServiceEntryAppender) applyServiceEntries(trafficMap graph.TrafficMap, globalInfo *graph.AppenderGlobalInfo, namespaceInfo *graph.AppenderNamespaceInfo) {
-	// a map of "se-service" nodes to the "se-aggregate" information
-	seMap := make(map[*serviceEntry][]*graph.Node)
-
+	// First, make sure we have loaded all of the relevant ServiceEntry definition information.  When a request is routed to
+	// a ServiceEntry, that ServiceEntry definition must be defined in the namespace of the "requester", meaning the source of the
+	// request. And so, we need to loop through the non-service nodes, looking for those with edges to a [non-injected] service, and
+	// then load the ServiceEntry definitions.
+	sourceNodes := []*graph.Node{}
 	for _, n := range trafficMap {
-		// only a service node can be a service entry
-		if n.NodeType != graph.NodeTypeService {
+		if n.NodeType == graph.NodeTypeService {
 			continue
 		}
-		// PassthroughCluster or BlackHoleCluster is not a service entry (nor a defined service)
-		if n.Metadata[graph.IsEgressCluster] == true {
-			continue
+		for _, e := range n.Edges {
+			// Must be a non-injected service node
+			candidate := e.Dest
+			isInjected := candidate.Metadata[graph.IsInjected] == true
+			if candidate.NodeType == graph.NodeTypeService && !isInjected {
+				sourceNodes = append(sourceNodes, n)
+			}
 		}
+	}
+	// If there are no sourceNodes with edges to an eligible service then we can return immediately.
+	if len(sourceNodes) == 0 {
+		return
+	}
+
+	// Otherwise, load accessible ServiceEntry information into globalInfo
+	for _, n := range sourceNodes {
+		a.loadServiceEntryHosts(n.Cluster, n.Namespace, globalInfo)
+	}
+
+	a.applyServiceEntries(trafficMap, sourceNodes, globalInfo, namespaceInfo)
+}
+
+func (a ServiceEntryAppender) loadServiceEntryHosts(cluster, namespace string, globalInfo *graph.AppenderGlobalInfo) {
+	isAccessible := false
+	for _, ns := range a.AccessibleNamespaces {
+		isAccessible = cluster == ns.Cluster && namespace == ns.Name
+		if isAccessible {
+			break
+		}
+	}
+	if !isAccessible {
+		return
+	}
+
+	serviceEntryHosts, found := getServiceEntryHosts(cluster, namespace, globalInfo)
+	if !found {
+		istioCfg, err := globalInfo.Business.IstioConfig.GetIstioConfigList(context.TODO(), business.IstioConfigCriteria{
+			Cluster:               cluster,
+			IncludeServiceEntries: true,
+			Namespace:             namespace,
+		})
+		graph.CheckError(err)
 
-		// To match, a service entry must be exported to the source namespace, and the requested
-		// service must match a defined host.  Note that the source namespace is assumed to be the
-		// the same as the appender namespace as all requests for the service entry should be coming
-		// from workloads in the current namespace being processed for the graph.
-		if se, ok := a.getServiceEntry(n.Cluster, n.Namespace, n.Service, globalInfo); ok {
-			if nodes, ok := seMap[se]; ok {
-				seMap[se] = append(nodes, n)
-			} else {
-				seMap[se] = []*graph.Node{n}
+		for _, entry := range istioCfg.ServiceEntries {
+			if entry.Spec.Hosts != nil {
+				location := "MESH_EXTERNAL"
+				if entry.Spec.Location.String() == "MESH_INTERNAL" {
+					location = "MESH_INTERNAL"
+				}
+				se := serviceEntry{
+					cluster:   cluster,
+					exportTo:  entry.Spec.ExportTo,
+					location:  location,
+					name:      entry.Name,
+					namespace: entry.Namespace,
+				}
+				for _, host := range entry.Spec.Hosts {
+					serviceEntryHosts.addHost(host, &se)
+				}
+			}
+		}
+	}
+}
+
+func (a ServiceEntryAppender) applyServiceEntries(trafficMap graph.TrafficMap, sourceNodes []*graph.Node, globalInfo *graph.AppenderGlobalInfo, namespaceInfo *graph.AppenderNamespaceInfo) {
+	// a map of "se-service" nodes to the "se-aggregate" information
+	seMap := make(map[*serviceEntry][]*graph.Node)
+
+	for _, n := range sourceNodes {
+		for _, e := range n.Edges {
+			// Must be a non-egress(PassthroughCluster or BlackHoleCluster) service node
+			candidate := e.Dest
+			isEgressCluster := candidate.Metadata[graph.IsEgressCluster] == true
+			if candidate.NodeType == graph.NodeTypeService && !isEgressCluster {
+				// To match, a service entry must be exported to the source namespace, and the requested
+				// service must match a defined host.  Note that the source namespace is assumed to be the
+				// the same as the appender namespace as all requests for the service entry should be coming
+				// from workloads in the current namespace being processed for the graph.
+				if se, ok := a.getServiceEntry(n.Cluster, n.Namespace, candidate.Service, globalInfo); ok {
+					if nodes, ok := seMap[se]; ok {
+						seMap[se] = append(nodes, candidate)
+					} else {
+						seMap[se] = []*graph.Node{candidate}
+					}
+				}
 			}
 		}
 	}
@@ -153,40 +221,7 @@ func (a ServiceEntryAppender) applyServiceEntries(trafficMap graph.TrafficMap, g
 // but exported to all namespaces (exportTo: *). It's possible that would allow traffic to flow from an
 // accessible workload through a serviceEntry whose definition we can't fetch.
 func (a ServiceEntryAppender) getServiceEntry(cluster, namespace, serviceName string, globalInfo *graph.AppenderGlobalInfo) (*serviceEntry, bool) {
-	serviceEntryHosts, found := getServiceEntryHosts(globalInfo)
-	if !found {
-		for _, ns := range a.AccessibleNamespaces {
-			// Narrow to only the cluster of this service node
-			if cluster == ns.Cluster {
-				istioCfg, err := globalInfo.Business.IstioConfig.GetIstioConfigList(context.TODO(), business.IstioConfigCriteria{
-					Cluster:               ns.Cluster,
-					IncludeServiceEntries: true,
-					Namespace:             ns.Name,
-				})
-				graph.CheckError(err)
-
-				for _, entry := range istioCfg.ServiceEntries {
-					if entry.Spec.Hosts != nil {
-						location := "MESH_EXTERNAL"
-						if entry.Spec.Location.String() == "MESH_INTERNAL" {
-							location = "MESH_INTERNAL"
-						}
-						se := serviceEntry{
-							cluster:   cluster,
-							exportTo:  entry.Spec.ExportTo,
-							location:  location,
-							name:      entry.Name,
-							namespace: entry.Namespace,
-						}
-						for _, host := range entry.Spec.Hosts {
-							serviceEntryHosts.addHost(host, &se)
-						}
-					}
-				}
-			}
-		}
-		globalInfo.Vendor[serviceEntryHostsKey] = serviceEntryHosts
-	}
+	serviceEntryHosts, _ := getServiceEntryHosts(cluster, namespace, globalInfo)
 
 	for host, serviceEntriesForHost := range serviceEntryHosts {
 		for _, se := range serviceEntriesForHost {

graph/telemetry/istio/istio.go

@@ -387,6 +387,7 @@ func addTraffic(trafficMap graph.TrafficMap, metric string, inject bool, val flo
 			log.Warningf("Skipping addTraffic (inject), %s", err)
 			return
 		}
+		injectedService.Metadata[graph.IsInjected] = true
 		if addEdgeTraffic(trafficMap, val, protocol, code, flags, host, source, injectedService, edgeTSHash, o) {
 			addToDestServices(injectedService.Metadata, destCluster, destSvcNs, destSvcName)