Merge pull request #338 from shriramsharma/generate-ingress-dr

generating ingress DR for VS based routing

admiral/cmd/admiral/cmd/root.go

@@ -256,6 +256,7 @@ func GetRootCmd(args []string) *cobra.Command {
 	rootCmd.PersistentFlags().BoolVar(&params.EnableVSRouting, "enable_vs_routing", false, "Enable/Disable VS Based Routing")
 	rootCmd.PersistentFlags().StringArrayVar(&params.VSRoutingGateways, "vs_routing_gateways", []string{}, "The PASSTHROUGH gateways to use for VS based routing")
 	rootCmd.PersistentFlags().StringArrayVar(&params.IngressVSExportToNamespaces, "ingress_vs_export_to_namespaces", []string{"istio-system"}, "List of namespaces where the ingress VS should be exported")
+	rootCmd.PersistentFlags().StringVar(&params.IngressLBPolicy, "ingress_lb_policy", "round_robin", "loadbalancer policy for ingress destination rule (round_robin/random/passthrough/least_request)")
 
 	return rootCmd
 }

admiral/pkg/clusters/serviceentry.go

@@ -54,7 +54,6 @@ const (
 	gtpManagedByMeshAgent                        = "mesh-agent"
 	gtpManagerMeshAgentFieldValue                = "ewok-mesh-agent"
 	errorCluster                                 = "error-cluster"
-	ingressVSGenerationErrorMessage              = "skipped generating ingress virtual service on cluster %s due to error %w"
 )
 
 func createServiceEntryForDeployment(
@@ -163,6 +162,8 @@ func modifyServiceEntryForNewServiceOrPod(
 
 		// Holds the VS destinations for the TLSRoutes
 		sourceClusterToDestinations = make(map[string]map[string][]*networking.RouteDestination)
+		// Holds the DR hosts (*.svc.cluster.local) used for VS based routing
+		sourceClusterToDRHosts = make(map[string]map[string]string)
 	)
 
 	clusterName, ok := ctx.Value(common.ClusterName).(string)
@@ -772,8 +773,17 @@ func modifyServiceEntryForNewServiceOrPod(
 				ctxLogger.Errorf(common.CtxLogFormat, "getAllVSRouteDestinationsByCluster",
 					deploymentOrRolloutName, deploymentOrRolloutNS, sourceCluster, err)
 				modifySEerr = common.AppendError(modifySEerr, err)
+			} else if len(destinations) == 0 {
+				ctxLogger.Warnf(common.CtxLogFormat, "getAllVSRouteDestinationsByCluster",
+					deploymentOrRolloutName, deploymentOrRolloutNS, sourceCluster,
+					"No RouteDestinations generated for VS based routing ")
 			} else {
 				sourceClusterToDestinations[sourceCluster] = destinations
+				// Get the hosts to populate the DR
+				drHost := fmt.Sprintf("*.%s.%s", deploymentOrRolloutNS, common.DotLocalDomainSuffix)
+				sourceClusterToDRHosts[sourceCluster] = map[string]string{
+					deploymentOrRolloutNS + common.DotLocalDomainSuffix: drHost,
+				}
 			}
 		}
 	}
@@ -799,7 +809,21 @@ func modifyServiceEntryForNewServiceOrPod(
 		// gathered during the discovery phase and write them to the source cluster
 		err := addUpdateVirtualServicesForSourceIngress(ctx, ctxLogger, remoteRegistry, sourceClusterToDestinations)
 		if err != nil {
+			ctxLogger.Errorf(common.CtxLogFormat, "addUpdateVirtualServicesForSourceIngress",
+				deploymentOrRolloutName, deploymentOrRolloutNS, "", err)
 			modifySEerr = common.AppendError(modifySEerr, err)
+		} else {
+			err := addUpdateDestinationRuleForSourceIngress(
+				ctx,
+				ctxLogger,
+				remoteRegistry,
+				sourceClusterToDRHosts,
+				sourceIdentity)
+			if err != nil {
+				ctxLogger.Errorf(common.CtxLogFormat, "addUpdateDestinationRuleForSourceIngress",
+					deploymentOrRolloutName, deploymentOrRolloutNS, "", err)
+				modifySEerr = common.AppendError(modifySEerr, err)
+			}
 		}
 	}
 

admiral/pkg/clusters/virtualservice_routing.go

@@ -8,6 +8,7 @@ import (
 
 	argo "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/common"
+	"github.com/istio-ecosystem/admiral/admiral/pkg/util"
 	log "github.com/sirupsen/logrus"
 	networkingV1Alpha3 "istio.io/api/networking/v1alpha3"
 	"istio.io/client-go/pkg/apis/networking/v1alpha3"
@@ -619,3 +620,101 @@ func getMeshHTTPPortForRollout(ports map[string]map[string]uint32) (uint32, erro
 func getMeshHTTPPortForDeployment(ports map[string]map[string]uint32) (uint32, error) {
 	return getMeshHTTPPort(common.Deployment, ports)
 }
+
+// addUpdateDestinationRuleForSourceIngress adds or updates the DestinationRule for the source ingress
+// This is where the DestinationRules are created for the cross-cluster VS based routing
+// The DestinationRule is created for the .svc.cluster.local hosts that were discovered during the discovery phase
+// on each source cluster
+func addUpdateDestinationRuleForSourceIngress(
+	ctx context.Context,
+	ctxLogger *log.Entry,
+	remoteRegistry *RemoteRegistry,
+	sourceClusterToDRHosts map[string]map[string]string,
+	sourceIdentity string) error {
+
+	if sourceIdentity == "" {
+		return fmt.Errorf("sourceIdentity is empty")
+	}
+
+	san := fmt.Sprintf("%s%s/%s", common.SpiffePrefix, common.GetSANPrefix(), sourceIdentity)
+
+	for sourceCluster, drHosts := range sourceClusterToDRHosts {
+
+		rc := remoteRegistry.GetRemoteController(sourceCluster)
+		if rc == nil {
+			ctxLogger.Warnf(common.CtxLogFormat, "addUpdateDestinationRuleForSourceIngress",
+				"", "", sourceCluster, "remote controller not initialized on this cluster")
+			continue
+		}
+
+		for name, drHost := range drHosts {
+			drObj := networkingV1Alpha3.DestinationRule{
+				Host:     drHost,
+				ExportTo: common.GetIngressVSExportToNamespace(),
+				TrafficPolicy: &networkingV1Alpha3.TrafficPolicy{
+					LoadBalancer: &networkingV1Alpha3.LoadBalancerSettings{
+						LbPolicy: &networkingV1Alpha3.LoadBalancerSettings_Simple{
+							Simple: getIngressDRLoadBalancerPolicy(),
+						},
+						LocalityLbSetting: &networkingV1Alpha3.LocalityLoadBalancerSetting{
+							Distribute: []*networkingV1Alpha3.LocalityLoadBalancerSetting_Distribute{
+								{
+									From: "*",
+									To:   map[string]uint32{"*": 100},
+								},
+							},
+						},
+					},
+					Tls: &networkingV1Alpha3.ClientTLSSettings{
+						SubjectAltNames: []string{san},
+					},
+				},
+			}
+			drName := fmt.Sprintf("%s-routing-dr", name)
+
+			newDR := createDestinationRuleSkeleton(drObj, drName, util.IstioSystemNamespace)
+
+			//Get existing DR
+			existingDR, err := rc.
+				DestinationRuleController.
+				IstioClient.
+				NetworkingV1alpha3().
+				DestinationRules(util.IstioSystemNamespace).Get(ctx, drName, metaV1.GetOptions{})
+			if err != nil {
+				ctxLogger.Warnf(common.CtxLogFormat,
+					"addUpdateDestinationRuleForSourceIngress",
+					drName,
+					util.IstioSystemNamespace,
+					sourceCluster, fmt.Sprintf("failed getting existing DR, error=%v", err))
+				existingDR = nil
+			}
+
+			err = addUpdateDestinationRule(ctxLogger, ctx, newDR, existingDR, util.IstioSystemNamespace, rc, remoteRegistry)
+			if err != nil {
+				return err
+			}
+
+		}
+
+	}
+	return nil
+}
+
+// getIngressDRLoadBalancerPolicy return the load balancer policy for the ingress destination rule
+// Default is networkingV1Alpha3.LoadBalancerSettings_ROUND_ROBIN
+func getIngressDRLoadBalancerPolicy() networkingV1Alpha3.LoadBalancerSettings_SimpleLB {
+
+	switch common.GetIngressLBPolicy() {
+	case "round_robin":
+		return networkingV1Alpha3.LoadBalancerSettings_ROUND_ROBIN
+	case "random":
+		return networkingV1Alpha3.LoadBalancerSettings_RANDOM
+	case "least_request":
+		return networkingV1Alpha3.LoadBalancerSettings_LEAST_REQUEST
+	case "passthrough":
+		return networkingV1Alpha3.LoadBalancerSettings_PASSTHROUGH
+	default:
+		return networkingV1Alpha3.LoadBalancerSettings_ROUND_ROBIN
+	}
+
+}