feat: enhance subset support for ApisixUpstream and update validations, RBAC, and e2e tests

Author: dspo

api/v2/apisixupstream_types.go

@@ -19,7 +19,7 @@ import (
 )
 
 // ApisixUpstreamSpec describes the specification of ApisixUpstream.
-// +kubebuilder:validation:XValidation:rule="has(self.externalNodes)!=has(self.discovery)"
+// +kubebuilder:validation:XValidation:rule="has(self.subsets) || (has(self.externalNodes)!=has(self.discovery))"
 type ApisixUpstreamSpec struct {
 	// IngressClassName is the name of an IngressClass cluster resource.
 	// controller implementations use this field to know whether they should be

config/crd/bases/apisix.apache.org_apisixupstreams.yaml

@@ -562,7 +562,7 @@ spec:
                 type: string
             type: object
             x-kubernetes-validations:
-            - rule: has(self.externalNodes)!=has(self.discovery)
+            - rule: has(self.subsets) || (has(self.externalNodes)!=has(self.discovery))
           status:
             description: ApisixStatus is the status report for Apisix ingress Resources
             properties:

config/rbac/role.yaml

@@ -15,6 +15,7 @@ rules:
   - ""
   resources:
   - namespaces
+  - pods
   - secrets
   - services
   verbs:

internal/controller/apisixroute_controller.go

@@ -350,9 +350,8 @@ func (r *ApisixRouteReconciler) validateBackends(ctx context.Context, tc *provid
 		if err != nil {
 			return err
 		}
-		endpoints.Items = r.filterEndpointSlicesBySubsetLabels(ctx, endpoints.Items, subsetLabels)
 
-		tc.EndpointSlices[serviceNN] = endpoints.Items
+		tc.EndpointSlices[serviceNN] = r.filterEndpointSlicesBySubsetLabels(ctx, endpoints.Items, subsetLabels)
 	}
 
 	return nil
@@ -714,7 +713,7 @@ func (r *ApisixRouteReconciler) getSubsetLabels(ctx context.Context, ar *apiv2.A
 		return nil, err
 	}
 
-	// try tro get the subset labels from the ApisixUpstream subsets
+	// try to get the subset labels from the ApisixUpstream subsets
 	for _, subset := range au.Spec.Subsets {
 		if backend.Subset == subset.Name {
 			return subset.Labels, nil

internal/controller/indexer/indexer.go

@@ -527,6 +527,11 @@ func ApisixRouteSecretIndexFunc(cli client.Client) func(client.Object) []string
 func ApisixRouteApisixUpstreamIndexFunc(obj client.Object) (keys []string) {
 	ar := obj.(*apiv2.ApisixRoute)
 	for _, rule := range ar.Spec.HTTP {
+		for _, backend := range rule.Backends {
+			if backend.Subset != "" && backend.ServiceName != "" {
+				keys = append(keys, GenIndexKey(ar.GetNamespace(), backend.ServiceName))
+			}
+		}
 		for _, upstream := range rule.Upstreams {
 			if upstream.Name != "" {
 				keys = append(keys, GenIndexKey(ar.GetNamespace(), upstream.Name))

internal/manager/controllers.go

@@ -29,6 +29,7 @@ import (
 // +kubebuilder:rbac:groups=coordination.k8s.io,resources=leases,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups="discovery.k8s.io",resources=endpointslices,verbs=get;list;watch
 // +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch
+// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch
 // +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch
 // +kubebuilder:rbac:groups="",resources=namespaces,verbs=get;list;watch
 

test/e2e/apisix/route.go

@@ -291,10 +291,67 @@ spec:
 			s.NewAPISIXClient().GET("/get").Expect().Header("X-Upstream-IP").IsEqual(clusterIP)
 		})
 
-		PIt("Test ApisixRoute subset", func() {
-			// route.Spec.HTTP[].Backends[].Subset depends on ApisixUpstream.
-			// ApisixUpstream is not implemented yet.
-			// So the case is pending for now
+		It("Test ApisixRoute subset", func() {
+			const apisixRouteSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+  name: default
+spec:
+  ingressClassName: apisix
+  http:
+  - name: rule0
+    match:
+      hosts:
+      - httpbin
+      paths:
+      - /*
+    backends:
+    - serviceName: httpbin-service-e2e-test
+      servicePort: 80
+      subset: test-subset
+`
+			const apisixUpstreamSpec0 = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixUpstream
+metadata:
+  name: httpbin-service-e2e-test
+spec:
+  ingressClassName: apisix
+  subsets:
+  - name: test-subset
+    labels:
+      unknown-key: unknown-value
+`
+			const apisixUpstreamSpec1 = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixUpstream
+metadata:
+  name: httpbin-service-e2e-test
+spec:
+  ingressClassName: apisix
+  subsets:
+  - name: test-subset
+    labels:
+      app: httpbin-deployment-e2e-test
+`
+			request := func() int {
+				return s.NewAPISIXClient().GET("/get").WithHost("httpbin").Expect().Raw().StatusCode
+			}
+			By("apply ApisixRoute")
+			var apisixRoute apiv2.ApisixRoute
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, &apisixRoute, apisixRouteSpec)
+			Eventually(request).WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+
+			// no pod matches the subset label "unknown-key: unknown-value" so there will be no node in the upstream,
+			// to request the route will get http.StatusServiceUnavailable
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "httpbin-service-e2e-test"}, new(apiv2.ApisixUpstream), apisixUpstreamSpec0)
+			Eventually(request).WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusServiceUnavailable))
+
+			// the pod matches the subset label "app: httpbin-deployment-e2e-test",
+			// to request the route will be OK
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "httpbin-service-e2e-test"}, new(apiv2.ApisixUpstream), apisixUpstreamSpec1)
+			Eventually(request).WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
 		})
 	})
 

test/e2e/apisix/upstream.go

@@ -53,22 +53,12 @@ spec:
 `
 			err = s.CreateResourceFromString(apisixUpstreamSpec0)
 			Expect(err).Should(HaveOccurred())
-			Expect(err.Error()).Should(ContainSubstring("failed rule: has(self.externalNodes)!=has(self.discovery)"))
+			Expect(err.Error()).Should(ContainSubstring("has(self.externalNodes)!=has(self.discovery)"))
 
 			err = s.CreateResourceFromString(apisixUpstreamSpec1)
 			Expect(err).Should(HaveOccurred())
-			Expect(err.Error()).Should(ContainSubstring("failed rule: has(self.externalNodes)!=has(self.discovery)"))
+			Expect(err.Error()).Should(ContainSubstring("has(self.externalNodes)!=has(self.discovery)"))
 
 		})
-
-		It("", func() {})
-
-		It("", func() {})
-
-		It("", func() {})
-
-		It("", func() {})
-
-		It("", func() {})
 	})
 })

test/e2e/framework/manifests/ingress.yaml

@@ -82,6 +82,7 @@ rules:
   - ""
   resources:
   - services
+  - pods
   verbs:
   - get
   - list