feat: Enhance upstream and traffic-split handling in translator and tests

- Refactor `NewDefaultUpstream` initialization for streamlined field ordering.
- Update e2e framework's `GetServiceEndpoints` to accept `types.NamespacedName`.
- Incorporate upstream weights into traffic-split logic, including default handling and labels.
- Add comprehensive e2e tests for mixed backend and upstream scenarios.

Author: dspo

api/adc/types.go

@@ -533,15 +533,14 @@ func NewDefaultService() *Service {
 
 func NewDefaultUpstream() *Upstream {
 	return &Upstream{
-		Type:   Roundrobin,
-		Nodes:  make(UpstreamNodes, 0),
-		Scheme: SchemeHTTP,
 		Metadata: Metadata{
-			Desc: "Created by apisix-ingress-controller, DO NOT modify it manually",
 			Labels: map[string]string{
 				"managed-by": "apisix-ingress-controller",
 			},
 		},
+		Nodes:  make(UpstreamNodes, 0),
+		Scheme: SchemeHTTP,
+		Type:   Roundrobin,
 	}
 }
 

internal/provider/adc/translator/apisixroute.go

@@ -166,6 +166,9 @@ func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *a
 				t.Log.Error(err, "failed to translate ApisixUpstream", "ApisixUpstream", utils.NamespacedName(au))
 				continue
 			}
+			if upstreamRef.Weight != nil {
+				upstream.Labels["meta_weight"] = strconv.FormatInt(int64(*upstreamRef.Weight), 10)
+			}
 
 			upstreams = append(upstreams, upstream)
 		}
@@ -178,6 +181,17 @@ func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *a
 		}
 
 		var weightedUpstreams []adc.TrafficSplitConfigRuleWeightedUpstream
+
+		// set the default upstream's weight in traffic-split
+		weight, err := strconv.Atoi(upstream.Labels["meta_weight"])
+		if err != nil {
+			weight = apiv2.DefaultWeight
+		}
+		weightedUpstreams = append(weightedUpstreams, adc.TrafficSplitConfigRuleWeightedUpstream{
+			Weight: weight,
+		})
+
+		// set others upstreams in traffic-split
 		for _, item := range upstreams {
 			weight, err := strconv.Atoi(item.Labels["meta_weight"])
 			if err != nil {
@@ -189,7 +203,7 @@ func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *a
 			})
 		}
 		if len(weightedUpstreams) > 0 {
-			route.Plugins["traffic-split"] = &adc.TrafficSplitConfig{
+			service.Plugins["traffic-split"] = &adc.TrafficSplitConfig{
 				Rules: []adc.TrafficSplitConfigRule{
 					{
 						WeightedUpstreams: weightedUpstreams,
@@ -207,9 +221,6 @@ func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *a
 		service.Routes = []*adc.Route{route}
 
 		if backendErr != nil && len(upstream.Nodes) == 0 {
-			if service.Plugins == nil {
-				service.Plugins = make(map[string]any)
-			}
 			service.Plugins["fault-injection"] = map[string]any{
 				"abort": map[string]any{
 					"http_status": 500,

test/e2e/apisix/route.go

@@ -35,20 +35,20 @@ var _ = Describe("Test ApisixRoute", func() {
 		applier = framework.NewApplier(s.GinkgoT, s.K8sClient, s.CreateResourceFromString)
 	)
 
-	Context("Test ApisixRoute", func() {
-		BeforeEach(func() {
-			By("create GatewayProxy")
-			gatewayProxy := fmt.Sprintf(gatewayProxyYaml, s.Deployer.GetAdminEndpoint(), s.AdminKey())
-			err := s.CreateResourceFromStringWithNamespace(gatewayProxy, "default")
-			Expect(err).NotTo(HaveOccurred(), "creating GatewayProxy")
-			time.Sleep(5 * time.Second)
-
-			By("create IngressClass")
-			err = s.CreateResourceFromStringWithNamespace(ingressClassYaml, "")
-			Expect(err).NotTo(HaveOccurred(), "creating IngressClass")
-			time.Sleep(5 * time.Second)
-		})
+	BeforeEach(func() {
+		By("create GatewayProxy")
+		gatewayProxy := fmt.Sprintf(gatewayProxyYaml, s.Deployer.GetAdminEndpoint(), s.AdminKey())
+		err := s.CreateResourceFromStringWithNamespace(gatewayProxy, "default")
+		Expect(err).NotTo(HaveOccurred(), "creating GatewayProxy")
+		time.Sleep(5 * time.Second)
+
+		By("create IngressClass")
+		err = s.CreateResourceFromStringWithNamespace(ingressClassYaml, "")
+		Expect(err).NotTo(HaveOccurred(), "creating IngressClass")
+		time.Sleep(5 * time.Second)
+	})
 
+	Context("Test ApisixRoute", func() {
 		It("Basic tests", func() {
 			const apisixRouteSpec = `
 apiVersion: apisix.apache.org/v2
@@ -295,8 +295,11 @@ spec:
 			// ApisixUpstream is not implemented yet.
 			// So the case is pending for now
 		})
+	})
 
-		It("Test ApisixRoute reference ApisixUpstream", func() {
+	Context("Test ApisixRoute reference ApisixUpstream", func() {
+
+		It("Test reference ApisixUpstream", func() {
 			const apisixRouteSpec = `
 apiVersion: apisix.apache.org/v2
 kind: ApisixRoute
@@ -310,7 +313,7 @@ spec:
       paths:
       - /*
     upstreams:
-    - name: default-upstream 
+    - name: default-upstream
 `
 			const apisixUpstreamSpec0 = `
 apiVersion: apisix.apache.org/v2
@@ -362,7 +365,84 @@ spec:
 			err = s.CreateResourceFromString(apisixUpstreamSpec1)
 			Expect(err).ShouldNot(HaveOccurred(), "update apisixUpstream")
 			Eventually(request).WithArguments("/get").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+		})
+
+		It("Test a Mix of Backends and Upstreams", func() {
+			// apisixUpstreamSpec is an ApisixUpstream reference to the Service httpbin-service-e2e-test
+			const apisixUpstreamSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixUpstream
+metadata:
+  name: default-upstream
+spec:
+  ingressClassName: apisix
+  externalNodes:
+  - type: Domain
+    name: httpbin-service-e2e-test
+  passHost: node
+`
+			// apisixRouteSpec is an ApisixUpstream uses a backend and reference an upstream.
+			// It contains a plugin response-rewrite that lets us know what upstream the gateway forwards the request to.
+			const apisixRouteSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+  name: default
+spec:
+  ingressClassName: apisix
+  http:
+  - name: rule0
+    match:
+      paths:
+      - /*
+    backends:
+    - serviceName: httpbin-service-e2e-test
+      servicePort: 80
+    upstreams:
+    - name: default-upstream
+    plugins:
+    - name: response-rewrite
+      enable: true
+      config:
+        headers:
+          set:
+            "X-Upstream-Host": "$upstream_addr"
+`
+			By("apply ApisixUpstream")
+			err := s.CreateResourceFromString(apisixUpstreamSpec)
+			Expect(err).ShouldNot(HaveOccurred(), "apply ApisixUpstream")
+
+			By("apply ApisixRoute")
+			var apisixRoute apiv2.ApisixRoute
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, &apisixRoute, apisixRouteSpec)
+
+			By("verify ApisixRoute works")
+			request := func(path string) int {
+				return s.NewAPISIXClient().GET(path).Expect().Raw().StatusCode
+			}
+			Eventually(request).WithArguments("/get").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+
+			By("verify the backends and the upstreams work commonly")
+			// .backends -> service httpbin-service-e2e-test -> endpoints httpbin-service-e2e-test, so we will get the $upstream_addr as endpoint IP
+			// .upstreams -> service alias-httpbin-service-e2e-test -> service httpbin-service-e2e-test, so we will get the $upstream_addr as the service's ClusterIP
+			var upstreamAddrs = make(map[string]struct{})
+			for range 10 {
+				upstreamAddr := s.NewAPISIXClient().GET("/get").Expect().Raw().Header.Get("X-Upstream-Host")
+				upstreamAddrs[upstreamAddr] = struct{}{}
+			}
+
+			endpoints, err := s.GetServiceEndpoints(types.NamespacedName{Namespace: s.Namespace(), Name: "httpbin-service-e2e-test"})
+			Expect(err).ShouldNot(HaveOccurred(), "get endpoints")
+			Expect(endpoints).Should(HaveLen(1))
+			endpoint := net.JoinHostPort(endpoints[0], "80")
+
+			service, err := s.GetServiceByName("httpbin-service-e2e-test")
+			Expect(err).ShouldNot(HaveOccurred(), "get service")
+			clusterIP := net.JoinHostPort(service.Spec.ClusterIP, "80")
 
+			Expect(upstreamAddrs).Should(HaveLen(2))
+			Eventually(upstreamAddrs).Should(HaveKey(endpoint))
+			Eventually(upstreamAddrs).Should(HaveKey(clusterIP))
 		})
 	})
 })

test/e2e/framework/k8s.go

@@ -15,6 +15,7 @@ package framework
 import (
 	"bufio"
 	"bytes"
+	"cmp"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -33,6 +34,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	applycorev1 "k8s.io/client-go/applyconfigurations/core/v1"
 	applymetav1 "k8s.io/client-go/applyconfigurations/meta/v1"
@@ -108,8 +110,8 @@ func (f *Framework) ensureServiceWithTimeout(name, namespace string, desiredEndp
 	return nil
 }
 
-func (f *Framework) GetServiceEndpoints(name string) ([]string, error) {
-	ep, err := f.clientset.CoreV1().Endpoints(_namespace).Get(f.Context, name, metav1.GetOptions{})
+func (f *Framework) GetServiceEndpoints(nn types.NamespacedName) ([]string, error) {
+	ep, err := f.clientset.CoreV1().Endpoints(cmp.Or(nn.Namespace, _namespace)).Get(f.Context, nn.Name, metav1.GetOptions{})
 	if err != nil {
 		return nil, err
 	}