feat: Add full support for ApisixUpstream in translator and e2e tests

- Implement `translateApisixUpstream` to support ApisixUpstream translation.
- Add handling for externalNodes with types Domain and Service.
- Update ApisixRoute translator to integrate ApisixUpstream references.
- Introduce `SchemeToPort` and `MatchHostDef` utility functions for validation and defaults.
- Enable e2e tests for ApisixRoute referencing ApisixUpstream.

Author: dspo

api/v2/shared_types.go

@@ -125,3 +125,29 @@ const (
 	// PassHostPass represents rewrite option for pass_host Upstream settings.
 	PassHostRewrite = "rewrite"
 )
+
+const (
+	// ExternalTypeDomain type is a domain
+	// +k8s:deepcopy-gen=false
+	ExternalTypeDomain ApisixUpstreamExternalType = "Domain"
+
+	// ExternalTypeService type is a K8s ExternalName service
+	// +k8s:deepcopy-gen=false
+	ExternalTypeService ApisixUpstreamExternalType = "Service"
+)
+
+var schemeToPortMaps = map[string]int{
+	SchemeHTTP:  80,
+	SchemeHTTPS: 443,
+	SchemeGRPC:  80,
+	SchemeGRPCS: 443,
+}
+
+// SchemeToPort scheme converts to the default port
+// ref https://github.com/apache/apisix/blob/c5fc10d9355a0c177a7532f01c77745ff0639a7f/apisix/upstream.lua#L167-L172
+func SchemeToPort(schema string) int {
+	if val, ok := schemeToPortMaps[schema]; ok {
+		return val
+	}
+	return 80
+}

internal/provider/adc/translator/apisixroute.go

@@ -16,6 +16,7 @@ import (
 	"cmp"
 	"encoding/json"
 	"fmt"
+	"strconv"
 
 	"github.com/pkg/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -27,11 +28,13 @@ import (
 	apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
 	"github.com/apache/apisix-ingress-controller/internal/controller/label"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
+	types2 "github.com/apache/apisix-ingress-controller/internal/types"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
 	"github.com/apache/apisix-ingress-controller/pkg/id"
 	pkgutils "github.com/apache/apisix-ingress-controller/pkg/utils"
 )
 
+//nolint:gocyclo
 func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *apiv2.ApisixRoute) (result *TranslateResult, err error) {
 	result = &TranslateResult{}
 	for ruleIndex, rule := range ar.Spec.HTTP {
@@ -113,8 +116,16 @@ func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *a
 		route.Timeout = timeout
 		route.Uris = rule.Match.Paths
 		route.Vars = vars
+		for key, value := range ar.GetObjectMeta().GetLabels() {
+			route.Labels[key] = value
+		}
+
+		//nolint:staticcheck
+		if rule.PluginConfigName != "" {
+			// FIXME: handle PluginConfig
+		}
 
-		// translate to adc.Upstream
+		// translate backends
 		var backendErr error
 		for _, backend := range rule.Backends {
 			var (
@@ -137,9 +148,61 @@ func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *a
 			upstream.Nodes = append(upstream.Nodes, upNodes...)
 		}
 
-		//nolint:staticcheck
-		if len(rule.Backends) == 0 && len(rule.Upstreams) > 0 {
-			// FIXME: when the API ApisixUpstream is supported
+		var (
+			apisixUpstreams []*apiv2.ApisixUpstream
+			adcUpstreams    []*adc.Upstream
+		)
+		for _, upstreamRef := range rule.Upstreams {
+			refKey := types2.NamespacedNameKind{
+				Namespace: ar.GetNamespace(),
+				Name:      upstreamRef.Name,
+				Kind:      "ApisixUpstream",
+			}
+			apisixUpstream, ok := tctx.Upstreams[refKey]
+			if !ok {
+				continue
+			}
+
+			// todo: translate external upstream
+			adcUpstream, err := t.translateApisixUpstream(tctx, apisixUpstream)
+			if err != nil {
+				t.Log.Error(err, "failed to translate ApisixUpstream", "ApisixUpstream", utils.NamespacedName(apisixUpstream))
+				continue
+			}
+
+			apisixUpstreams = append(apisixUpstreams, apisixUpstream)
+			adcUpstreams = append(adcUpstreams, adcUpstream)
+		}
+
+		_ = apisixUpstreams
+
+		// If no .http[].backends is used and only .http[].upstreams is used, the first valid upstream is used as service.upstream;
+		// Other upstreams are configured in the traffic-split plugin
+		if len(rule.Backends) == 0 && len(adcUpstreams) > 0 {
+			service.Upstream = adcUpstreams[0]
+			adcUpstreams = adcUpstreams[1:]
+		}
+
+		var wups []adc.TrafficSplitConfigRuleWeightedUpstream
+		for _, adcUpstream := range adcUpstreams {
+			weight, err := strconv.Atoi(adcUpstream.Labels["meta_weight"])
+			if err != nil {
+				t.Log.Error(err, "failed to parse meta_weight from upstream labels", "labels", adcUpstream.GetLabels())
+				weight = apiv2.DefaultWeight
+			}
+			wups = append(wups, adc.TrafficSplitConfigRuleWeightedUpstream{
+				Upstream: adcUpstream,
+				Weight:   weight,
+			})
+		}
+		if len(wups) > 0 {
+			route.Plugins["traffic-split"] = &adc.TrafficSplitConfig{
+				Rules: []adc.TrafficSplitConfigRule{
+					{
+						WeightedUpstreams: wups,
+					},
+				},
+			}
 		}
 
 		// translate to adc.Service

internal/provider/adc/translator/apisixupstream.go

@@ -14,13 +14,23 @@ package translator
 
 import (
 	"cmp"
-	"errors"
+	"fmt"
+
+	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 
 	"github.com/apache/apisix-ingress-controller/api/adc"
 	apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
+	"github.com/apache/apisix-ingress-controller/internal/provider"
+	"github.com/apache/apisix-ingress-controller/internal/utils"
 )
 
-func (t *Translator) TranslateApisixUpstream(au *apiv2.ApisixUpstream) (ups *adc.Upstream, err error) {
+func (t *Translator) translateApisixUpstream(tctx *provider.TranslateContext, au *apiv2.ApisixUpstream) (ups *adc.Upstream, err error) {
+	if len(au.Spec.ExternalNodes) == 0 && au.Spec.Discovery == nil {
+		return nil, errors.Errorf("%s has empty externalNodes or discovery configuration", utils.NamespacedName(au))
+	}
+
 	ups = adc.NewDefaultUpstream()
 	for _, f := range []func(*apiv2.ApisixUpstream, *adc.Upstream) error{
 		translateApisixUpstreamScheme,
@@ -35,6 +45,13 @@ func (t *Translator) TranslateApisixUpstream(au *apiv2.ApisixUpstream) (ups *adc
 			return
 		}
 	}
+	for _, f := range []func(*provider.TranslateContext, *apiv2.ApisixUpstream, *adc.Upstream) error{
+		translateApisixUpstreamExternalNodes,
+	} {
+		if err = f(tctx, au, ups); err != nil {
+			return
+		}
+	}
 
 	return
 }
@@ -145,3 +162,78 @@ func translateApisixUpstreamDiscovery(upstream *apiv2.ApisixUpstream, upstream2
 	// todo: no filed `.Discovery*` in adc.Upstream
 	return nil
 }
+
+func translateApisixUpstreamExternalNodes(tctx *provider.TranslateContext, au *apiv2.ApisixUpstream, ups *adc.Upstream) error {
+	for _, node := range au.Spec.ExternalNodes {
+		switch node.Type {
+		case apiv2.ExternalTypeDomain:
+			if err := translateApisixUpstreamExternalNodesDomain(au, ups, node); err != nil {
+				return err
+			}
+		default: // apiv2.ExternalTypeService or default
+			if err := translateApisixUpstreamExternalNodesExternalName(tctx, au, ups, node); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+func translateApisixUpstreamExternalNodesDomain(au *apiv2.ApisixUpstream, ups *adc.Upstream, node apiv2.ApisixUpstreamExternalNode) error {
+	weight := apiv2.DefaultWeight
+	if node.Weight != nil {
+		weight = *node.Weight
+	}
+
+	if !utils.MatchHostDef(node.Name) {
+		return fmt.Errorf("ApisixUpstream %s/%s ExternalNodes[]'s name %s as Domain must match lowercase RFC 1123 subdomain.  "+
+			"a lowercase RFC 1123 subdomain must consist of lower case alphanumeric characters, '-' or '.', and must start and end with an alphanumeric character",
+			au.Namespace, au.Name, node.Name)
+	}
+
+	n := adc.UpstreamNode{
+		Host:   node.Name,
+		Weight: weight,
+	}
+
+	if node.Port != nil {
+		n.Port = *node.Port
+	} else {
+		n.Port = apiv2.SchemeToPort(au.Spec.Scheme)
+	}
+
+	ups.Nodes = append(ups.Nodes, n)
+
+	return nil
+}
+
+func translateApisixUpstreamExternalNodesExternalName(tctx *provider.TranslateContext, au *apiv2.ApisixUpstream, ups *adc.Upstream, node apiv2.ApisixUpstreamExternalNode) error {
+	serviceNN := types.NamespacedName{Namespace: au.GetNamespace(), Name: node.Name}
+	svc, ok := tctx.Services[serviceNN]
+	if !ok {
+		return errors.Errorf("service not found, service: %s", serviceNN)
+	}
+
+	if svc.Spec.Type != corev1.ServiceTypeExternalName {
+		return errors.Errorf("ApisixUpstream %s ExternalNodes[] must refers to a ExternalName service: %s", utils.NamespacedName(au), node.Name)
+	}
+
+	weight := apiv2.DefaultWeight
+	if node.Weight != nil {
+		weight = *node.Weight
+	}
+	n := adc.UpstreamNode{
+		Host:   svc.Spec.ExternalName,
+		Weight: weight,
+	}
+
+	if node.Port != nil {
+		n.Port = *node.Port
+	} else {
+		n.Port = apiv2.SchemeToPort(au.Spec.Scheme)
+	}
+
+	ups.Nodes = append(ups.Nodes, n)
+
+	return nil
+}

internal/utils/k8s.go

@@ -14,6 +14,7 @@ package utils
 
 import (
 	"net"
+	"regexp"
 
 	k8stypes "k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -47,3 +48,14 @@ func ValidateRemoteAddrs(remoteAddrs []string) error {
 	}
 	return nil
 }
+
+var hostDef = "^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
+var hostDefRegex = regexp.MustCompile(hostDef)
+
+// MatchHostDef checks that host matches host's shcema
+// ref to : https://github.com/apache/apisix/blob/c5fc10d9355a0c177a7532f01c77745ff0639a7f/apisix/schema_def.lua#L40
+// ref to : https://github.com/kubernetes/kubernetes/blob/976a940f4a4e84fe814583848f97b9aafcdb083f/staging/src/k8s.io/apimachinery/pkg/util/validation/validation.go#L205
+// They define regex differently, but k8s's dns is more accurate
+func MatchHostDef(host string) bool {
+	return hostDefRegex.MatchString(host)
+}

test/e2e/apisix/route.go

@@ -296,10 +296,45 @@ spec:
 			// So the case is pending for now
 		})
 
-		PIt("Test ApisixRoute reference ApisixUpstream", func() {
-			// This case depends on ApisixUpstream.
-			// ApisixUpstream is not implemented yet.
-			// So the case is pending for now.
+		It("Test ApisixRoute reference ApisixUpstream", func() {
+			const apisixRouteSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+  name: default
+spec:
+  ingressClassName: apisix
+  http:
+  - name: rule0
+    match:
+      paths:
+      - /*
+    upstreams:
+    - name: default-upstream 
+`
+			const apisixUpstreamSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixUpstream
+metadata:
+  name: default-upstream
+spec:
+  ingressClassName: apisix
+  externalNodes:
+  - type: Service
+    name: httpbin-service-e2e-test
+`
+			By("create ApisixUpstream and ApisixRoute")
+			err := s.CreateResourceFromString(apisixUpstreamSpec)
+			Expect(err).ShouldNot(HaveOccurred(), "apply apisixUpstreamSpec")
+
+			var apisxiRoute apiv2.ApisixRoute
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, &apisxiRoute, apisixRouteSpec)
+
+			By("verify ApisixRoute and ApisixUpstream works")
+			request := func(path string) int {
+				return s.NewAPISIXClient().GET(path).WithHost("httpbin").Expect().Raw().StatusCode
+			}
+			Eventually(request).WithArguments("/get").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
 		})
 	})
 })