fix: use upstream id instead of inline upstream in traffic-split plugin

.github/workflows/apisix-e2e-test.yml

@@ -31,6 +31,9 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
 
+env:
+  ADC_VERSION: dev
+
 jobs:
   e2e-test:
     strategy:

api/adc/types.go

@@ -168,6 +168,7 @@ type Service struct {
 	StreamRoutes    []*StreamRoute `json:"stream_routes,omitempty" yaml:"stream_routes,omitempty"`
 	StripPathPrefix *bool          `json:"strip_path_prefix,omitempty" yaml:"strip_path_prefix,omitempty"`
 	Upstream        *Upstream      `json:"upstream,omitempty" yaml:"upstream,omitempty"`
+	Upstreams       []*Upstream    `json:"upstreams,omitempty" yaml:"upstreams,omitempty"`
 }
 
 // +k8s:deepcopy-gen=true

api/v2/shared_types.go

@@ -18,6 +18,8 @@
 package v2
 
 import (
+	"bytes"
+	"strconv"
 	"time"
 
 	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
@@ -172,3 +174,54 @@ func SchemeToPort(schema string) int {
 	}
 	return 80
 }
+
+var (
+	ResolveGranularity = struct {
+		Endpoint string
+		Service  string
+	}{
+		Endpoint: "endpoint",
+		Service:  "service",
+	}
+)
+
+// ComposeUpstreamName uses namespace, name, subset (optional), port, resolveGranularity info to compose
+// the upstream name.
+// the resolveGranularity is not composited in the upstream name when it is endpoint.
+// ref: https://github.com/apache/apisix-ingress-controller/blob/10059afe3e84b693cc61e6df7a0040890a9d16eb/pkg/types/apisix/v1/types.go#L595-L598
+func ComposeUpstreamName(namespace, name, subset string, port int32, resolveGranularity string) string {
+	pstr := strconv.Itoa(int(port))
+	// FIXME Use sync.Pool to reuse this buffer if the upstream
+	// name composing code path is hot.
+	var p []byte
+	plen := len(namespace) + len(name) + len(pstr) + 2
+	if subset != "" {
+		plen = plen + len(subset) + 1
+	}
+	if resolveGranularity == ResolveGranularity.Service {
+		plen = plen + len(resolveGranularity) + 1
+	}
+
+	p = make([]byte, 0, plen)
+	buf := bytes.NewBuffer(p)
+	buf.WriteString(namespace)
+	buf.WriteByte('_')
+	buf.WriteString(name)
+	buf.WriteByte('_')
+	if subset != "" {
+		buf.WriteString(subset)
+		buf.WriteByte('_')
+	}
+	buf.WriteString(pstr)
+	if resolveGranularity == ResolveGranularity.Service {
+		buf.WriteByte('_')
+		buf.WriteString(resolveGranularity)
+	}
+
+	return buf.String()
+}
+
+// ComposeExternalUpstreamName uses ApisixUpstream namespace, name to compose the upstream name.
+func ComposeExternalUpstreamName(namespace, name string) string {
+	return namespace + "_" + name
+}

internal/adc/translator/apisixroute.go

@@ -226,6 +226,10 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 		if backend.Weight != nil {
 			upstream.Labels["meta_weight"] = strconv.FormatInt(int64(*backend.Weight), 10)
 		}
+
+		upstreamName := apiv2.ComposeUpstreamName(ar.Namespace, backend.ServiceName, backend.Subset, int32(backend.ServicePort.IntValue()), backend.ResolveGranularity)
+		upstream.Name = upstreamName
+		upstream.ID = id.GenID(upstreamName)
 		upstreams = append(upstreams, upstream)
 	}
 
@@ -248,6 +252,9 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 			upstream.Labels["meta_weight"] = strconv.FormatInt(int64(*upstreamRef.Weight), 10)
 		}
 
+		upstreamName := apiv2.ComposeExternalUpstreamName(upsNN.Namespace, upsNN.Name)
+		upstream.Name = upstreamName
+		upstream.ID = id.GenID(upstreamName)
 		upstreams = append(upstreams, upstream)
 	}
 
@@ -259,8 +266,16 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 	// the first valid upstream is used as service.upstream;
 	// the others are configured in the traffic-split plugin
 	service.Upstream = upstreams[0]
+	// remove the id and name of the service.upstream, adc schema does not need id and name for it
+	service.Upstream.ID = ""
+	service.Upstream.Name = ""
+
 	upstreams = upstreams[1:]
 
+	if len(upstreams) > 0 {
+		service.Upstreams = upstreams
+	}
+
 	// set weight in traffic-split for the default upstream
 	if len(upstreams) > 0 {
 		weight, err := strconv.Atoi(service.Upstream.Labels["meta_weight"])
@@ -272,15 +287,15 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 		})
 	}
 
-	// set others upstreams in traffic-split
+	// set others upstreams in traffic-split using upstream_id
 	for _, item := range upstreams {
 		weight, err := strconv.Atoi(item.Labels["meta_weight"])
 		if err != nil {
 			weight = apiv2.DefaultWeight
 		}
 		weightedUpstreams = append(weightedUpstreams, adc.TrafficSplitConfigRuleWeightedUpstream{
-			Upstream: item,
-			Weight:   weight,
+			UpstreamID: item.ID,
+			Weight:     weight,
 		})
 	}
 

internal/adc/translator/httproute.go

@@ -329,6 +329,25 @@ func (t *Translator) TranslateBackendRefWithFilter(tctx *provider.TranslateConte
 	return t.translateBackendRef(tctx, ref, endpointFilter)
 }
 
+func (t *Translator) composeUpstreamNameForBackendRef(ref gatewayv1.BackendRef) string {
+	var (
+		kind      string
+		namespace = string(*ref.Namespace)
+		name      = string(ref.Name)
+		port      int32
+	)
+	if ref.Kind == nil {
+		kind = "Service"
+	} else {
+		kind = string(*ref.Kind)
+	}
+	if ref.Port != nil {
+		port = int32(*ref.Port)
+	}
+
+	return fmt.Sprintf("%s_%s_%s_%d", kind, namespace, name, port)
+}
+
 func (t *Translator) translateBackendRef(tctx *provider.TranslateContext, ref gatewayv1.BackendRef, endpointFilter func(*discoveryv1.Endpoint) bool) (adctypes.UpstreamNodes, error) {
 	if ref.Kind != nil && *ref.Kind != "Service" {
 		return adctypes.UpstreamNodes{}, fmt.Errorf("kind %s is not supported", *ref.Kind)
@@ -497,6 +516,10 @@ func (t *Translator) TranslateHTTPRoute(tctx *provider.TranslateContext, httpRou
 
 			t.AttachBackendTrafficPolicyToUpstream(backend.BackendRef, tctx.BackendTrafficPolicies, upstream)
 			upstream.Nodes = upNodes
+
+			upstreamName := t.composeUpstreamNameForBackendRef(backend.BackendRef)
+			upstream.Name = upstreamName
+			upstream.ID = id.GenID(upstreamName)
 			upstreams = append(upstreams, upstream)
 		}
 
@@ -508,11 +531,22 @@ func (t *Translator) TranslateHTTPRoute(tctx *provider.TranslateContext, httpRou
 		} else if len(upstreams) == 1 {
 			// Single backend - use directly as service upstream
 			service.Upstream = upstreams[0]
+			// remove the id and name of the service.upstream, adc schema does not need id and name for it
+			service.Upstream.ID = ""
+			service.Upstream.Name = ""
 		} else {
 			// Multiple backends - use traffic-split plugin
 			service.Upstream = upstreams[0]
+			// remove the id and name of the service.upstream, adc schema does not need id and name for it
+			service.Upstream.ID = ""
+			service.Upstream.Name = ""
+
 			upstreams = upstreams[1:]
 
+			if len(upstreams) > 0 {
+				service.Upstreams = upstreams
+			}
+
 			// Set weight in traffic-split for the default upstream
 			weight := apiv2.DefaultWeight
 			if rule.BackendRefs[0].Weight != nil {
@@ -522,16 +556,16 @@ func (t *Translator) TranslateHTTPRoute(tctx *provider.TranslateContext, httpRou
 				Weight: weight,
 			})
 
-			// Set other upstreams in traffic-split
+			// Set other upstreams in traffic-split using upstream_id
 			for i, upstream := range upstreams {
 				weight := apiv2.DefaultWeight
 				// get weight from the backend refs starting from the second backend
 				if i+1 < len(rule.BackendRefs) && rule.BackendRefs[i+1].Weight != nil {
 					weight = int(*rule.BackendRefs[i+1].Weight)
 				}
 				weightedUpstreams = append(weightedUpstreams, adctypes.TrafficSplitConfigRuleWeightedUpstream{
-					Upstream: upstream,
-					Weight:   weight,
+					UpstreamID: upstream.ID,
+					Weight:     weight,
 				})
 			}