Fix destination rule issues (alibaba#282)

pkg/ingress/config/ingress_config.go

@@ -559,27 +559,36 @@ func (m *IngressConfig) convertDestinationRule(configs []common.WrapperConfig) [
 	IngressLog.Debugf("traffic policy number %d", len(convertOptions.Service2TrafficPolicy))
 
 	for _, wrapperTrafficPolicy := range convertOptions.Service2TrafficPolicy {
-		m.annotationHandler.ApplyTrafficPolicy(wrapperTrafficPolicy.TrafficPolicy, wrapperTrafficPolicy.WrapperConfig.AnnotationsConfig)
+		m.annotationHandler.ApplyTrafficPolicy(wrapperTrafficPolicy.TrafficPolicy, wrapperTrafficPolicy.PortTrafficPolicy, wrapperTrafficPolicy.WrapperConfig.AnnotationsConfig)
 	}
 
 	// Merge multi-port traffic policy per service into one destination rule.
 	destinationRules := map[string]*common.WrapperDestinationRule{}
 	for key, wrapperTrafficPolicy := range convertOptions.Service2TrafficPolicy {
-		serviceName := util.CreateServiceFQDN(key.Namespace, key.Name)
+		var serviceName string
+		if key.ServiceFQDN != "" {
+			serviceName = key.ServiceFQDN
+		} else {
+			serviceName = util.CreateServiceFQDN(key.Namespace, key.Name)
+		}
 		dr, exist := destinationRules[serviceName]
 		if !exist {
+			trafficPolicy := &networking.TrafficPolicy{}
+			if wrapperTrafficPolicy.PortTrafficPolicy != nil {
+				trafficPolicy.PortLevelSettings = []*networking.TrafficPolicy_PortTrafficPolicy{wrapperTrafficPolicy.PortTrafficPolicy}
+			} else if wrapperTrafficPolicy.TrafficPolicy != nil {
+				trafficPolicy = wrapperTrafficPolicy.TrafficPolicy
+			}
 			dr = &common.WrapperDestinationRule{
 				DestinationRule: &networking.DestinationRule{
-					Host: serviceName,
-					TrafficPolicy: &networking.TrafficPolicy{
-						PortLevelSettings: []*networking.TrafficPolicy_PortTrafficPolicy{wrapperTrafficPolicy.TrafficPolicy},
-					},
+					Host:          serviceName,
+					TrafficPolicy: trafficPolicy,
 				},
 				WrapperConfig: wrapperTrafficPolicy.WrapperConfig,
 				ServiceKey:    key,
 			}
-		} else {
-			dr.DestinationRule.TrafficPolicy.PortLevelSettings = append(dr.DestinationRule.TrafficPolicy.PortLevelSettings, wrapperTrafficPolicy.TrafficPolicy)
+		} else if wrapperTrafficPolicy.PortTrafficPolicy != nil {
+			dr.DestinationRule.TrafficPolicy.PortLevelSettings = append(dr.DestinationRule.TrafficPolicy.PortLevelSettings, wrapperTrafficPolicy.PortTrafficPolicy)
 		}
 
 		destinationRules[serviceName] = dr

pkg/ingress/kube/annotations/annotations.go

@@ -192,8 +192,8 @@ func (h *AnnotationHandlerManager) ApplyRoute(route *networking.HTTPRoute, confi
 	}
 }
 
-func (h *AnnotationHandlerManager) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress) {
+func (h *AnnotationHandlerManager) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy, portTrafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress) {
 	for _, handler := range h.trafficPolicyHandlers {
-		handler.ApplyTrafficPolicy(trafficPolicy, config)
+		handler.ApplyTrafficPolicy(trafficPolicy, portTrafficPolicy, config)
 	}
 }

pkg/ingress/kube/annotations/interface.go

@@ -38,5 +38,5 @@ type RouteHandler interface {
 
 type TrafficPolicyHandler interface {
 	// ApplyTrafficPolicy parsed ingress annotation config reflected on traffic policy
-	ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress)
+	ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy, portTrafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress)
 }

pkg/ingress/kube/annotations/loadbalance.go

@@ -136,14 +136,16 @@ func (l loadBalance) Parse(annotations Annotations, config *Ingress, _ *GlobalCo
 	return nil
 }
 
-func (l loadBalance) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress) {
+func (l loadBalance) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy, portTrafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress) {
 	loadBalanceConfig := config.LoadBalance
 	if loadBalanceConfig == nil {
 		return
 	}
 
+	var loadBalancer *networking.LoadBalancerSettings
+
 	if loadBalanceConfig.cookie != nil {
-		trafficPolicy.LoadBalancer = &networking.LoadBalancerSettings{
+		loadBalancer = &networking.LoadBalancerSettings{
 			LbPolicy: &networking.LoadBalancerSettings_ConsistentHash{
 				ConsistentHash: &networking.LoadBalancerSettings_ConsistentHashLB{
 					HashKey: &networking.LoadBalancerSettings_ConsistentHashLB_HttpCookie{
@@ -171,18 +173,25 @@ func (l loadBalance) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy_
 				},
 			}
 		}
-		trafficPolicy.LoadBalancer = &networking.LoadBalancerSettings{
+		loadBalancer = &networking.LoadBalancerSettings{
 			LbPolicy: &networking.LoadBalancerSettings_ConsistentHash{
 				ConsistentHash: consistentHash,
 			},
 		}
 	} else {
-		trafficPolicy.LoadBalancer = &networking.LoadBalancerSettings{
+		loadBalancer = &networking.LoadBalancerSettings{
 			LbPolicy: &networking.LoadBalancerSettings_Simple{
 				Simple: loadBalanceConfig.simple,
 			},
 		}
 	}
+
+	if trafficPolicy != nil {
+		trafficPolicy.LoadBalancer = loadBalancer
+	}
+	if portTrafficPolicy != nil {
+		portTrafficPolicy.LoadBalancer = loadBalancer
+	}
 }
 
 func isCookieAffinity(annotations Annotations) bool {

pkg/ingress/kube/annotations/loadbalance_test.go

@@ -229,7 +229,7 @@ func TestLoadBalanceApplyTrafficPolicy(t *testing.T) {
 
 	for _, inputCase := range inputCases {
 		t.Run("", func(t *testing.T) {
-			loadBalance.ApplyTrafficPolicy(inputCase.input, inputCase.config)
+			loadBalance.ApplyTrafficPolicy(nil, inputCase.input, inputCase.config)
 			if !reflect.DeepEqual(inputCase.input, inputCase.expect) {
 				t.Fatal("Should be equal")
 			}

pkg/ingress/kube/annotations/upstreamtls.go

@@ -75,17 +75,6 @@ func (u upstreamTLS) Parse(annotations Annotations, config *Ingress, _ *GlobalCo
 		}
 	}
 
-	secretName, _ := annotations.ParseStringASAP(proxySSLSecret)
-	namespacedName := util.SplitNamespacedName(secretName)
-	if namespacedName.Name == "" {
-		return nil
-	}
-
-	if namespacedName.Namespace == "" {
-		namespacedName.Namespace = config.Namespace
-	}
-	upstreamTLSConfig.SecretName = namespacedName.String()
-
 	if sslVerify, err := annotations.ParseStringASAP(proxySSLVerify); err == nil {
 		if OnOffRegex.MatchString(sslVerify) {
 			upstreamTLSConfig.SSLVerify = onOffToBool(sslVerify)
@@ -96,22 +85,34 @@ func (u upstreamTLS) Parse(annotations Annotations, config *Ingress, _ *GlobalCo
 
 	if enableSNI, err := annotations.ParseStringASAP(proxySSLServerName); err == nil {
 		if OnOffRegex.MatchString(enableSNI) {
-			upstreamTLSConfig.SSLVerify = onOffToBool(enableSNI)
+			upstreamTLSConfig.EnableSNI = onOffToBool(enableSNI)
 		}
 	}
 
+	secretName, _ := annotations.ParseStringASAP(proxySSLSecret)
+	namespacedName := util.SplitNamespacedName(secretName)
+	if namespacedName.Name == "" {
+		return nil
+	}
+
+	if namespacedName.Namespace == "" {
+		namespacedName.Namespace = config.Namespace
+	}
+	upstreamTLSConfig.SecretName = namespacedName.String()
+
 	return nil
 }
 
-func (u upstreamTLS) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress) {
+func (u upstreamTLS) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy, portTrafficPolicy *networking.TrafficPolicy_PortTrafficPolicy, config *Ingress) {
 	if config.UpstreamTLS == nil {
 		return
 	}
 
 	upstreamTLSConfig := config.UpstreamTLS
 
+	var connectionPool *networking.ConnectionPoolSettings
 	if isH2(upstreamTLSConfig.BackendProtocol) {
-		trafficPolicy.ConnectionPool = &networking.ConnectionPoolSettings{
+		connectionPool = &networking.ConnectionPoolSettings{
 			Http: &networking.ConnectionPoolSettings_HTTPSettings{
 				H2UpgradePolicy: networking.ConnectionPoolSettings_HTTPSettings_UPGRADE,
 			},
@@ -125,8 +126,14 @@ func (u upstreamTLS) ApplyTrafficPolicy(trafficPolicy *networking.TrafficPolicy_
 	} else if isHTTPS(upstreamTLSConfig.BackendProtocol) {
 		tls = processSimple(config)
 	}
-
-	trafficPolicy.Tls = tls
+	if trafficPolicy != nil {
+		trafficPolicy.ConnectionPool = connectionPool
+		trafficPolicy.Tls = tls
+	}
+	if portTrafficPolicy != nil {
+		portTrafficPolicy.ConnectionPool = connectionPool
+		portTrafficPolicy.Tls = tls
+	}
 }
 
 func processMTLS(config *Ingress) *networking.ClientTLSSettings {

pkg/ingress/kube/annotations/upstreamtls_test.go

@@ -47,6 +47,7 @@ func TestUpstreamTLSParse(t *testing.T) {
 				SSLVerify:       true,
 				SNI:             "SSLName",
 				SecretName:      "namespace/SSLSecret",
+				EnableSNI:       true,
 			},
 		},
 		{
@@ -60,9 +61,10 @@ func TestUpstreamTLSParse(t *testing.T) {
 			},
 			expect: &UpstreamTLSConfig{
 				BackendProtocol: "HTTP2",
-				SSLVerify:       false,
-				SNI:             "",
+				SSLVerify:       true,
+				SNI:             "SSLName",
 				SecretName:      "",
+				EnableSNI:       true,
 			},
 		},
 	}
@@ -143,7 +145,7 @@ func TestApplyTrafficPolicy(t *testing.T) {
 
 	for _, testCase := range testCases {
 		t.Run("", func(t *testing.T) {
-			parser.ApplyTrafficPolicy(testCase.input, testCase.config)
+			parser.ApplyTrafficPolicy(nil, testCase.input, testCase.config)
 			if diff := cmp.Diff(testCase.expect, testCase.input); diff != "" {
 				t.Fatalf("TestApplyTrafficPolicy() mismatch (-want +got): \n%s", diff)
 			}

pkg/ingress/kube/common/controller.go

@@ -28,9 +28,10 @@ import (
 )
 
 type ServiceKey struct {
-	Namespace string
-	Name      string
-	Port      int32
+	Namespace   string
+	Name        string
+	ServiceFQDN string
+	Port        int32
 }
 
 type WrapperConfig struct {
@@ -98,8 +99,9 @@ type WrapperVirtualService struct {
 }
 
 type WrapperTrafficPolicy struct {
-	TrafficPolicy *networking.TrafficPolicy_PortTrafficPolicy
-	WrapperConfig *WrapperConfig
+	TrafficPolicy     *networking.TrafficPolicy
+	PortTrafficPolicy *networking.TrafficPolicy_PortTrafficPolicy
+	WrapperConfig     *WrapperConfig
 }
 
 type WrapperDestinationRule struct {

pkg/ingress/kube/ingress/controller.go

@@ -848,20 +848,9 @@ func (c *controller) ConvertTrafficPolicy(convertOptions *common.ConvertOptions,
 	}
 
 	if ingressV1Beta.Backend != nil {
-		serviceKey, err := c.createServiceKey(ingressV1Beta.Backend, cfg.Namespace)
+		err := c.storeBackendTrafficPolicy(wrapper, ingressV1Beta.Backend, convertOptions.Service2TrafficPolicy)
 		if err != nil {
-			IngressLog.Errorf("ignore default service %s within ingress %s/%s", serviceKey.Name, cfg.Namespace, cfg.Name)
-		} else {
-			if _, exist := convertOptions.Service2TrafficPolicy[serviceKey]; !exist {
-				convertOptions.Service2TrafficPolicy[serviceKey] = &common.WrapperTrafficPolicy{
-					TrafficPolicy: &networking.TrafficPolicy_PortTrafficPolicy{
-						Port: &networking.PortSelector{
-							Number: uint32(serviceKey.Port),
-						},
-					},
-					WrapperConfig: wrapper,
-				}
-			}
+			IngressLog.Errorf("ignore default service within ingress %s/%s, since error:%v", cfg.Namespace, cfg.Name, err)
 		}
 	}
 
@@ -871,22 +860,46 @@ func (c *controller) ConvertTrafficPolicy(convertOptions *common.ConvertOptions,
 		}
 
 		for _, httpPath := range rule.HTTP.Paths {
-			if httpPath.Backend.ServiceName == "" {
-				continue
-			}
-
-			serviceKey, err := c.createServiceKey(&httpPath.Backend, cfg.Namespace)
+			err := c.storeBackendTrafficPolicy(wrapper, &httpPath.Backend, convertOptions.Service2TrafficPolicy)
 			if err != nil {
-				IngressLog.Errorf("ignore service %s within ingress %s/%s", serviceKey.Name, cfg.Namespace, cfg.Name)
-				continue
+				IngressLog.Errorf("ignore service within ingress %s/%s, since error:%v", cfg.Namespace, cfg.Name, err)
 			}
+		}
+	}
 
-			if _, exist := convertOptions.Service2TrafficPolicy[serviceKey]; exist {
-				continue
+	return nil
+}
+
+func (c *controller) storeBackendTrafficPolicy(wrapper *common.WrapperConfig, backend *ingress.IngressBackend, store map[common.ServiceKey]*common.WrapperTrafficPolicy) error {
+	if backend == nil {
+		return errors.New("invalid empty backend")
+	}
+	if common.ValidateBackendResource(backend.Resource) && wrapper.AnnotationsConfig.Destination != nil {
+		for _, dest := range wrapper.AnnotationsConfig.Destination.McpDestination {
+			serviceKey := common.ServiceKey{
+				Namespace:   "mcp",
+				Name:        dest.Destination.Host,
+				ServiceFQDN: dest.Destination.Host,
 			}
+			if _, exist := store[serviceKey]; !exist {
+				store[serviceKey] = &common.WrapperTrafficPolicy{
+					TrafficPolicy: &networking.TrafficPolicy{},
+					WrapperConfig: wrapper,
+				}
+			}
+		}
+	} else {
+		if backend.ServiceName == "" {
+			return nil
+		}
+		serviceKey, err := c.createServiceKey(backend, wrapper.Config.Namespace)
+		if err != nil {
+			return fmt.Errorf("ignore service %s within ingress %s/%s", serviceKey.Name, wrapper.Config.Namespace, wrapper.Config.Name)
+		}
 
-			convertOptions.Service2TrafficPolicy[serviceKey] = &common.WrapperTrafficPolicy{
-				TrafficPolicy: &networking.TrafficPolicy_PortTrafficPolicy{
+		if _, exist := store[serviceKey]; !exist {
+			store[serviceKey] = &common.WrapperTrafficPolicy{
+				PortTrafficPolicy: &networking.TrafficPolicy_PortTrafficPolicy{
 					Port: &networking.PortSelector{
 						Number: uint32(serviceKey.Port),
 					},
@@ -895,7 +908,6 @@ func (c *controller) ConvertTrafficPolicy(convertOptions *common.ConvertOptions,
 			}
 		}
 	}
-
 	return nil
 }