fix: use upstream id instead of inline upstream in traffic-split plugin (#2546)

Signed-off-by: Ashing Zheng <[email protected]>

Author: ronething

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

@@ -29,6 +29,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
@@ -807,3 +808,59 @@ func (c Config) MarshalJSON() ([]byte, error) {
 		TlsVerify:   c.TlsVerify,
 	})
 }
+
+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
+}
+
+// ComposeUpstreamNameForBackendRef composes upstream name using kind, namespace, name and port.
+func ComposeUpstreamNameForBackendRef(kind, namespace, name string, port int32) string {
+	return fmt.Sprintf("%s_%s_%s_%d", kind, namespace, name, port)
+}

api/adc/zz_generated.deepcopy.go

@@ -234,6 +234,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 := adc.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)
 	}
 
@@ -256,6 +260,9 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 			upstream.Labels["meta_weight"] = strconv.FormatInt(int64(*upstreamRef.Weight), 10)
 		}
 
+		upstreamName := adc.ComposeExternalUpstreamName(upsNN.Namespace, upsNN.Name)
+		upstream.Name = upstreamName
+		upstream.ID = id.GenID(upstreamName)
 		upstreams = append(upstreams, upstream)
 	}
 
@@ -267,8 +274,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"])
@@ -280,15 +295,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/apisixroute.go

@@ -543,6 +543,24 @@ func (t *Translator) TranslateHTTPRoute(tctx *provider.TranslateContext, httpRou
 
 			t.AttachBackendTrafficPolicyToUpstream(backend.BackendRef, tctx.BackendTrafficPolicies, upstream)
 			upstream.Nodes = upNodes
+
+			var (
+				kind string
+				port int32
+			)
+			if backend.Kind == nil {
+				kind = "Service"
+			} else {
+				kind = string(*backend.Kind)
+			}
+			if backend.Port != nil {
+				port = int32(*backend.Port)
+			}
+			namespace := string(*backend.Namespace)
+			name := string(backend.Name)
+			upstreamName := adctypes.ComposeUpstreamNameForBackendRef(kind, namespace, name, port)
+			upstream.Name = upstreamName
+			upstream.ID = id.GenID(upstreamName)
 			upstreams = append(upstreams, upstream)
 		}
 
@@ -554,11 +572,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 {
@@ -568,16 +597,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,
 				})
 			}