Support for GlobalTrafficPolicy priority processing (#249)

* Support for GlobalTrafficPolicy priority processing
Signed-off-by: vjoshi3 <vrushali_joshi@intuit.com>
Co-authored-by: vjoshi3 <vrushali_joshi@intuit.com>

admiral/cmd/admiral/cmd/root.go

@@ -115,6 +115,8 @@ func GetRootCmd(args []string) *cobra.Command {
 		"The label value, on a namespace or service, which tells Istio to perform sidecar injection")
 	rootCmd.PersistentFlags().StringVar(&params.LabelSet.AdmiralIgnoreLabel, "admiral_ignore_label", "admiral-ignore",
 		"The label value, on a namespace, which tells Istio to perform sidecar injection")
+	rootCmd.PersistentFlags().StringVar(&params.LabelSet.PriorityKey, "priority_key", "priority",
+		"The label value, on admiral resources, which tells admiral to give higher priority while processing admiral resource. Currently, this will be used for GlobalTrafficPolicy processing.")
 	rootCmd.PersistentFlags().StringVar(&params.HostnameSuffix, "hostname_suffix", "global",
 		"The hostname suffix to customize the cname generated by admiral. Default suffix value will be \"global\"")
 	rootCmd.PersistentFlags().StringVar(&params.LabelSet.WorkloadIdentityKey, "workload_identity_key", "identity",
@@ -131,10 +133,9 @@ func GetRootCmd(args []string) *cobra.Command {
 	rootCmd.PersistentFlags().StringVar(&params.LabelSet.GatewayApp, "gateway_app", "istio-ingressgateway",
 		"The the value of the `app` label to use to match and find the service that represents the ingress for cross cluster traffic (AUTO_PASSTHROUGH mode)")
 	rootCmd.PersistentFlags().BoolVar(&params.MetricsEnabled, "metrics", true, "Enable prometheus metrics collections")
-	rootCmd.PersistentFlags().StringVar(&params.AdmiralStateCheckerName,"admiral_state_checker_name","NoOPStateChecker","The value of the admiral_state_checker_name label to configure the DR Strategy for Admiral")
-	rootCmd.PersistentFlags().StringVar(&params.DRStateStoreConfigPath,"dr_state_store_config_path","","Location of config file which has details for data store. Ex:- Dynamo DB connection details")
-	rootCmd.PersistentFlags().StringVar(&params.ServiceEntryIPPrefix,"se_ip_prefix","240.0","IP prefix for the auto generated IPs for service entries. Only the first two octets:  Eg- 240.0")
-
+	rootCmd.PersistentFlags().StringVar(&params.AdmiralStateCheckerName, "admiral_state_checker_name", "NoOPStateChecker", "The value of the admiral_state_checker_name label to configure the DR Strategy for Admiral")
+	rootCmd.PersistentFlags().StringVar(&params.DRStateStoreConfigPath, "dr_state_store_config_path", "", "Location of config file which has details for data store. Ex:- Dynamo DB connection details")
+	rootCmd.PersistentFlags().StringVar(&params.ServiceEntryIPPrefix, "se_ip_prefix", "240.0", "IP prefix for the auto generated IPs for service entries. Only the first two octets:  Eg- 240.0")
 
 	return rootCmd
 }

admiral/pkg/clusters/registry_test.go

@@ -40,6 +40,7 @@ func init() {
 
 	p.LabelSet.WorkloadIdentityKey = "identity"
 	p.LabelSet.GlobalTrafficDeploymentLabel = "identity"
+	p.LabelSet.PriorityKey = "priority"
 
 	common.InitializeConfig(p)
 }

admiral/pkg/clusters/serviceentry.go

@@ -242,7 +242,7 @@ func modifyServiceEntryForNewServiceOrPod(event admiral.EventType, env string, s
 }
 
 //Does two things;
-//i)  Picks the GTP that was created most recently from the passed in GTP list (GTPs from all clusters)
+//i)  Picks the GTP that was created most recently from the passed in GTP list based on GTP priority label (GTPs from all clusters)
 //ii) Updates the global GTP cache with the selected GTP in i)
 func updateGlobalGtpCache(cache *AdmiralCache, identity, env string, gtps map[string][]*v1.GlobalTrafficPolicy) {
 	defer util.LogElapsedTime("updateGlobalGtpCache", identity, env, "")()
@@ -256,13 +256,8 @@ func updateGlobalGtpCache(cache *AdmiralCache, identity, env string, gtps map[st
 		return
 	} else if len(gtpsOrdered) > 1 {
 		log.Debugf("More than one GTP found for identity=%s in env=%s.", identity, env)
-		//sort by creation time with most recent at the beginning
-		sort.Slice(gtpsOrdered, func(i, j int) bool {
-			iTime := gtpsOrdered[i].CreationTimestamp
-			jTime := gtpsOrdered[j].CreationTimestamp
-			log.Debugf("GTP sorting identity=%s env=%s name1=%s creationTime1=%v name2=%s creationTime2=%v", identity, env, gtpsOrdered[i].Name, iTime, gtpsOrdered[j].Name, jTime)
-			return iTime.After(jTime.Time)
-		})
+		//sort by creation time and priority, gtp with highest priority and most recent at the beginning
+		sortGtpsByPriorityAndCreationTime(gtpsOrdered, identity, env)
 	}
 
 	mostRecentGtp := gtpsOrdered[0]
@@ -276,6 +271,30 @@ func updateGlobalGtpCache(cache *AdmiralCache, identity, env string, gtps map[st
 	}
 }
 
+func sortGtpsByPriorityAndCreationTime(gtpsToOrder []*v1.GlobalTrafficPolicy, identity string, env string) {
+	sort.Slice(gtpsToOrder, func(i, j int) bool {
+		iPriority := getGtpPriority(gtpsToOrder[i])
+		jPriority := getGtpPriority(gtpsToOrder[j])
+
+		iTime := gtpsToOrder[i].CreationTimestamp
+		jTime := gtpsToOrder[j].CreationTimestamp
+
+		if iPriority != jPriority {
+			log.Debugf("GTP sorting identity=%s env=%s name1=%s creationTime1=%v priority1=%d name2=%s creationTime2=%v priority2=%d", identity, env, gtpsToOrder[i].Name, iTime, iPriority, gtpsToOrder[j].Name, jTime, jPriority)
+			return iPriority > jPriority
+		}
+		log.Debugf("GTP sorting identity=%s env=%s name1=%s creationTime1=%v priority1=%d name2=%s creationTime2=%v priority2=%d", identity, env, gtpsToOrder[i].Name, iTime, iPriority, gtpsToOrder[j].Name, jTime, jPriority)
+		return iTime.After(jTime.Time)
+	})
+}
+func getGtpPriority(gtp *v1.GlobalTrafficPolicy) int {
+	if val, ok := gtp.ObjectMeta.Labels[common.GetAdmiralParams().LabelSet.PriorityKey]; ok {
+		if convertedValue, err := strconv.Atoi(strings.TrimSpace(val)); err == nil {
+			return convertedValue
+		}
+	}
+	return 0
+}
 func updateEndpointsForBlueGreen(rollout *argo.Rollout, weightedServices map[string]*WeightedService, cnames map[string]string,
 	ep *networking.ServiceEntry_Endpoint, sourceCluster string, meshHost string) {
 	activeServiceName := rollout.Spec.Strategy.BlueGreen.ActiveService

admiral/pkg/clusters/serviceentry_test.go

@@ -31,6 +31,27 @@ import (
 	"k8s.io/client-go/rest"
 )
 
+func init() {
+	p := common.AdmiralParams{
+		KubeconfigPath:             "testdata/fake.config",
+		LabelSet:                   &common.LabelSet{},
+		EnableSAN:                  true,
+		SANPrefix:                  "prefix",
+		HostnameSuffix:             "mesh",
+		SyncNamespace:              "ns",
+		CacheRefreshDuration:       time.Minute,
+		ClusterRegistriesNamespace: "default",
+		DependenciesNamespace:      "default",
+		SecretResolver:             "",
+	}
+
+	p.LabelSet.WorkloadIdentityKey = "identity"
+	p.LabelSet.GlobalTrafficDeploymentLabel = "identity"
+	p.LabelSet.PriorityKey = "priority"
+
+	common.InitializeConfig(p)
+}
+
 func TestAddServiceEntriesWithDr(t *testing.T) {
 	admiralCache := AdmiralCache{}
 
@@ -1310,9 +1331,25 @@ func TestUpdateGlobalGtpCache(t *testing.T) {
 			Policy: []*model.TrafficPolicy{{DnsPrefix: "hellogtp2"}},
 		}}
 
+		gtp7 = &v13.GlobalTrafficPolicy{ObjectMeta: v12.ObjectMeta{Name: "gtp7", Namespace: "namespace1", CreationTimestamp: v12.NewTime(time.Now().Add(time.Duration(-45))), Labels: map[string]string{"identity": identity1, "env": env_stage, "priority": "2"}}, Spec: model.GlobalTrafficPolicy{
+			Policy: []*model.TrafficPolicy{{DnsPrefix: "hellogtp7"}},
+		}}
+
 		gtp3 = &v13.GlobalTrafficPolicy{ObjectMeta: v12.ObjectMeta{Name: "gtp3", Namespace: "namespace2", CreationTimestamp: v12.NewTime(time.Now()), Labels: map[string]string{"identity": identity1, "env": env_stage}}, Spec: model.GlobalTrafficPolicy{
 			Policy: []*model.TrafficPolicy{{DnsPrefix: "hellogtp3"}},
 		}}
+
+		gtp4 = &v13.GlobalTrafficPolicy{ObjectMeta: v12.ObjectMeta{Name: "gtp4", Namespace: "namespace1", CreationTimestamp: v12.NewTime(time.Now().Add(time.Duration(-30))), Labels: map[string]string{"identity": identity1, "env": env_stage, "priority": "10"}}, Spec: model.GlobalTrafficPolicy{
+			Policy: []*model.TrafficPolicy{{DnsPrefix: "hellogtp4"}},
+		}}
+
+		gtp5 = &v13.GlobalTrafficPolicy{ObjectMeta: v12.ObjectMeta{Name: "gtp5", Namespace: "namespace1", CreationTimestamp: v12.NewTime(time.Now().Add(time.Duration(-15))), Labels: map[string]string{"identity": identity1, "env": env_stage, "priority": "2"}}, Spec: model.GlobalTrafficPolicy{
+			Policy: []*model.TrafficPolicy{{DnsPrefix: "hellogtp5"}},
+		}}
+
+		gtp6 = &v13.GlobalTrafficPolicy{ObjectMeta: v12.ObjectMeta{Name: "gtp6", Namespace: "namespace3", CreationTimestamp: v12.NewTime(time.Now()), Labels: map[string]string{"identity": identity1, "env": env_stage, "priority": "1000"}}, Spec: model.GlobalTrafficPolicy{
+			Policy: []*model.TrafficPolicy{{DnsPrefix: "hellogtp6"}},
+		}}
 	)
 
 	testCases := []struct {
@@ -1321,13 +1358,14 @@ func TestUpdateGlobalGtpCache(t *testing.T) {
 		env         string
 		gtps        map[string][]*v13.GlobalTrafficPolicy
 		expectedGtp *v13.GlobalTrafficPolicy
-	}{{
-		name:        "Should return nil when no GTP present",
-		gtps:        map[string][]*v13.GlobalTrafficPolicy{},
-		identity:    identity1,
-		env:         env_stage,
-		expectedGtp: nil,
-	},
+	}{
+		{
+			name:        "Should return nil when no GTP present",
+			gtps:        map[string][]*v13.GlobalTrafficPolicy{},
+			identity:    identity1,
+			env:         env_stage,
+			expectedGtp: nil,
+		},
 		{
 			name:        "Should return the only existing gtp",
 			gtps:        map[string][]*v13.GlobalTrafficPolicy{"c1": {gtp}},
@@ -1349,6 +1387,27 @@ func TestUpdateGlobalGtpCache(t *testing.T) {
 			env:         env_stage,
 			expectedGtp: gtp3,
 		},
+		{
+			name:        "Should return the existing priority gtp within the cluster",
+			gtps:        map[string][]*v13.GlobalTrafficPolicy{"c1": {gtp, gtp2, gtp7}},
+			identity:    identity1,
+			env:         env_stage,
+			expectedGtp: gtp7,
+		},
+		{
+			name:        "Should return the recently created priority gtp within the cluster",
+			gtps:        map[string][]*v13.GlobalTrafficPolicy{"c1": {gtp5, gtp4, gtp, gtp2}},
+			identity:    identity1,
+			env:         env_stage,
+			expectedGtp: gtp4,
+		},
+		{
+			name:        "Should return the recently created priority gtp from another cluster",
+			gtps:        map[string][]*v13.GlobalTrafficPolicy{"c1": {gtp, gtp2, gtp4, gtp5, gtp7}, "c2": {gtp6}, "c3": {gtp3}},
+			identity:    identity1,
+			env:         env_stage,
+			expectedGtp: gtp6,
+		},
 	}
 
 	for _, c := range testCases {

admiral/pkg/clusters/types_test.go

@@ -36,6 +36,7 @@ func init() {
 
 	p.LabelSet.WorkloadIdentityKey = "identity"
 	p.LabelSet.GlobalTrafficDeploymentLabel = "identity"
+	p.LabelSet.PriorityKey = "priority"
 
 	common.InitializeConfig(p)
 }

admiral/pkg/controller/common/types.go

@@ -36,18 +36,18 @@ type AdmiralParams struct {
 	DependenciesNamespace      string
 	SyncNamespace              string
 	EnableSAN                  bool
-	SANPrefix               string
-	SecretResolver          string
-	LabelSet                *LabelSet
-	LogLevel                int
-	HostnameSuffix          string
-	PreviewHostnamePrefix   string
-	MetricsEnabled          bool
-	WorkloadSidecarUpdate   string
-	WorkloadSidecarName     string
-	AdmiralStateCheckerName string
-	DRStateStoreConfigPath  string
-	ServiceEntryIPPrefix    string
+	SANPrefix                  string
+	SecretResolver             string
+	LabelSet                   *LabelSet
+	LogLevel                   int
+	HostnameSuffix             string
+	PreviewHostnamePrefix      string
+	MetricsEnabled             bool
+	WorkloadSidecarUpdate      string
+	WorkloadSidecarName        string
+	AdmiralStateCheckerName    string
+	DRStateStoreConfigPath     string
+	ServiceEntryIPPrefix       string
 }
 
 func (b AdmiralParams) String() string {
@@ -58,9 +58,9 @@ func (b AdmiralParams) String() string {
 		fmt.Sprintf("EnableSAN=%v ", b.EnableSAN) +
 		fmt.Sprintf("SANPrefix=%v ", b.SANPrefix) +
 		fmt.Sprintf("LabelSet=%v ", b.LabelSet) +
-		fmt.Sprintf("SecretResolver=%v ", b.SecretResolver)+
-		fmt.Sprintf("AdmiralStateCheckername=%v ", b.AdmiralStateCheckerName)+
-		fmt.Sprintf("DRStateStoreConfigPath=%v ", b.DRStateStoreConfigPath)+
+		fmt.Sprintf("SecretResolver=%v ", b.SecretResolver) +
+		fmt.Sprintf("AdmiralStateCheckername=%v ", b.AdmiralStateCheckerName) +
+		fmt.Sprintf("DRStateStoreConfigPath=%v ", b.DRStateStoreConfigPath) +
 		fmt.Sprintf("ServiceEntryIPPrefix=%v ", b.ServiceEntryIPPrefix)
 }
 
@@ -70,6 +70,7 @@ type LabelSet struct {
 	NamespaceSidecarInjectionLabel      string
 	NamespaceSidecarInjectionLabelValue string
 	AdmiralIgnoreLabel                  string
+	PriorityKey                         string
 	WorkloadIdentityKey                 string //Should always be used for both label and annotation (using label as the primary, and falling back to annotation if the label is not found)
 	GlobalTrafficDeploymentLabel        string //label used to tie together deployments and globaltrafficpolicy objects. Configured separately from the identity key because this one _must_ be a label
 	EnvKey                              string //key used to group deployments by env. The order would be to use annotation `EnvKey` and then label `EnvKey` and then fallback to label `env` label

install/sample/gtp-priority.yaml

@@ -0,0 +1,19 @@
+apiVersion: admiral.io/v1alpha1
+kind: GlobalTrafficPolicy
+metadata:
+  name: gtp-service1
+  namespace: sample
+  annotations:
+    admiral.io/env: stage
+  labels:
+    identity: greeting
+    priority: "1" #0 value or missing label represents no priority is set, any other value determines the processing priority
+spec:
+  policy:
+    - dnsPrefix: default # default is a keyword, alternatively you can use `env` (ex: stage)
+      lbType: 1 #0 represents TOPOLOGY, 1 represents FAILOVER
+      target:
+        - region: us-west-2
+          weight: 80
+        - region: us-east-2
+          weight: 20