fix: traffic-split weight distribution and add e2e tests (#2495)

Author: Revolyssup

internal/provider/adc/translator/apisixroute.go

@@ -198,10 +198,10 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 	var (
 		upstreams         = make([]*adc.Upstream, 0)
 		weightedUpstreams = make([]adc.TrafficSplitConfigRuleWeightedUpstream, 0)
-		backendErr        error
 	)
 
 	for _, backend := range rule.Backends {
+		var backendErr error
 		upstream := adc.NewDefaultUpstream()
 		// try to get the apisixupstream with the same name as the backend service to be upstream config.
 		// err is ignored because it does not care about the externalNodes of the apisixupstream.
@@ -223,7 +223,9 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 				continue
 			}
 		}
-
+		if backend.Weight != nil {
+			upstream.Labels["meta_weight"] = strconv.FormatInt(int64(*backend.Weight), 10)
+		}
 		upstreams = append(upstreams, upstream)
 	}
 
@@ -250,7 +252,7 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 	}
 
 	// no valid upstream
-	if backendErr != nil || len(upstreams) == 0 || len(upstreams[0].Nodes) == 0 {
+	if len(upstreams) == 0 || len(upstreams[0].Nodes) == 0 {
 		return
 	}
 

test/e2e/crds/v2/route.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"net/http"
 	"time"
@@ -651,6 +652,154 @@ spec:
 		})
 	})
 
+	Context("Test ApisixRoute Traffic Split", func() {
+		It("2:1 traffic split test", func() {
+			const apisixRouteSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+ name: default
+spec:
+ ingressClassName: apisix
+ http:
+ - name: rule1
+   match:
+     hosts:
+     - httpbin.org
+     paths:
+       - /get
+   backends:
+   - serviceName: httpbin-service-e2e-test
+     servicePort: 80
+     weight: 10
+   - serviceName: %s
+     servicePort: 9180
+     weight: 5
+`
+			By("apply ApisixRoute with traffic split")
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, new(apiv2.ApisixRoute),
+				fmt.Sprintf(apisixRouteSpec, s.Deployer.GetAdminServiceName()))
+			verifyRequest := func() int {
+				return s.NewAPISIXClient().GET("/get").WithHost("httpbin.org").Expect().Raw().StatusCode
+			}
+			By("send requests to verify traffic split")
+			var (
+				successCount int
+				failCount    int
+			)
+
+			s.RequestAssert(&scaffold.RequestAssert{
+				Method:  "GET",
+				Path:    "/get",
+				Host:    "httpbin.org",
+				Check:   scaffold.WithExpectedStatus(http.StatusOK),
+				Timeout: 10 * time.Second,
+			})
+			for range 90 {
+				code := verifyRequest()
+				if code == http.StatusOK {
+					successCount++
+				} else {
+					failCount++
+				}
+			}
+
+			By("verify traffic distribution ratio")
+			ratio := float64(successCount) / float64(failCount)
+			expectedRatio := 10.0 / 5.0 // 2:1 ratio
+			deviation := math.Abs(ratio - expectedRatio)
+			Expect(deviation).Should(BeNumerically("<", 0.5),
+				"traffic distribution deviation too large (got %.2f, expected %.2f)", ratio, expectedRatio)
+		})
+
+		It("zero-weight test", func() {
+			const apisixRouteSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+ name: default
+spec:
+ ingressClassName: apisix
+ http:
+ - name: rule1
+   match:
+     hosts:
+     - httpbin.org
+     paths:
+       - /get
+   backends:
+   - serviceName: httpbin-service-e2e-test
+     servicePort: 80
+     weight: 10
+   - serviceName: %s
+     servicePort: 9180
+     weight: 0
+`
+			By("apply ApisixRoute with zero-weight backend")
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, new(apiv2.ApisixRoute),
+				fmt.Sprintf(apisixRouteSpec, s.Deployer.GetAdminServiceName()))
+			verifyRequest := func() int {
+				return s.NewAPISIXClient().GET("/get").WithHost("httpbin.org").Expect().Raw().StatusCode
+			}
+
+			By("wait for route to be ready")
+			s.RequestAssert(&scaffold.RequestAssert{
+				Method:  "GET",
+				Path:    "/get",
+				Host:    "httpbin.org",
+				Check:   scaffold.WithExpectedStatus(http.StatusOK),
+				Timeout: 10 * time.Second,
+			})
+			By("send requests to verify zero-weight behavior")
+			for range 30 {
+				code := verifyRequest()
+				Expect(code).Should(Equal(200))
+			}
+		})
+		It("valid backend is set even if other backend is invalid", func() {
+			const apisixRouteSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+ name: default
+spec:
+ ingressClassName: apisix
+ http:
+ - name: rule1
+   match:
+     hosts:
+     - httpbin.org
+     paths:
+       - /get
+   backends:
+   - serviceName: httpbin-service-e2e-test
+     servicePort: 80
+     weight: 10
+   - serviceName: invalid-service
+     servicePort: 9180
+     weight: 5
+`
+			By("apply ApisixRoute with traffic split")
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, new(apiv2.ApisixRoute), apisixRouteSpec)
+			verifyRequest := func() int {
+				return s.NewAPISIXClient().GET("/get").WithHost("httpbin.org").Expect().Raw().StatusCode
+			}
+
+			By("wait for route to be ready")
+			s.RequestAssert(&scaffold.RequestAssert{
+				Method:  "GET",
+				Path:    "/get",
+				Host:    "httpbin.org",
+				Check:   scaffold.WithExpectedStatus(http.StatusOK),
+				Timeout: 10 * time.Second,
+			})
+			By("send requests to verify all requests routed to valid upstream")
+			for range 30 {
+				code := verifyRequest()
+				Expect(code).Should(Equal(200))
+			}
+		})
+	})
 	Context("Test ApisixRoute sync during startup", func() {
 		const route = `
 apiVersion: apisix.apache.org/v2

test/e2e/scaffold/apisix_deployer.go

@@ -409,6 +409,9 @@ func (s *APISIXDeployer) GetAdminEndpoint(svc ...*corev1.Service) string {
 	return fmt.Sprintf("http://%s.%s:9180", svc[0].Name, svc[0].Namespace)
 }
 
+func (s *APISIXDeployer) GetAdminServiceName() string {
+	return s.dataplaneService.Name
+}
 func (s *APISIXDeployer) DefaultDataplaneResource() DataplaneResource {
 	return newADCDataplaneResource(
 		framework.ProviderType,

test/e2e/scaffold/deployer.go

@@ -32,6 +32,7 @@ type Deployer interface {
 	CreateAdditionalGateway(namePrefix string) (string, *corev1.Service, error)
 	CleanupAdditionalGateway(identifier string) error
 	GetAdminEndpoint(...*corev1.Service) string
+	GetAdminServiceName() string
 	DefaultDataplaneResource() DataplaneResource
 }