feat: add support for apisix backend mode (#181)

dspo · web-flow · commit 1d3f633d73f3 · 2025-06-25T23:10:45.000+08:00
diff --git a/.github/workflows/apisix-conformance-test.yml b/.github/workflows/apisix-conformance-test.yml
@@ -36,6 +36,11 @@ jobs:
     timeout-minutes: 60
     needs: 
       - prepare
+    strategy:
+      matrix:
+        provider_type:
+        - apisix-standalone
+        - apisix
     runs-on: buildjet-2vcpu-ubuntu-2204
     steps:
       - name: Checkout
@@ -79,6 +84,8 @@ jobs:
       - name: Run Conformance Test
         shell: bash
         continue-on-error: true
+        env:
+          PROVIDER_TYPE: ${{ matrix.provider_type }}
         run: |
           make conformance-test-standalone
       
diff --git a/.github/workflows/apisix-e2e-test.yml b/.github/workflows/apisix-e2e-test.yml
@@ -35,6 +35,11 @@ jobs:
   e2e-test:
     needs: 
       - prepare
+    strategy:
+      matrix:
+        provider_type:
+        - apisix-standalone
+        - apisix
     runs-on: buildjet-2vcpu-ubuntu-2204
     steps:
       - name: Checkout
@@ -83,5 +88,6 @@ jobs:
         shell: bash
         env:
           TEST_DIR: "./test/e2e/apisix/"
+          PROVIDER_TYPE: ${{ matrix.provider_type }}
         run: |
           make e2e-test
diff --git a/internal/controller/config/config.go b/internal/controller/config/config.go
@@ -113,7 +113,7 @@ func (c *Config) Validate() error {
 
 func validateProvider(config ProviderConfig) error {
 	switch config.Type {
-	case ProviderTypeStandalone:
+	case ProviderTypeStandalone, ProviderTypeAPISIX:
 		if config.SyncPeriod.Duration <= 0 {
 			return fmt.Errorf("sync_period must be greater than 0 for standalone provider")
 		}
diff --git a/internal/controller/config/types.go b/internal/controller/config/types.go
@@ -21,6 +21,7 @@ type ProviderType string
 const (
 	ProviderTypeStandalone ProviderType = "apisix-standalone"
 	ProviderTypeAPI7EE     ProviderType = "api7ee"
+	ProviderTypeAPISIX     ProviderType = "apisix"
 )
 
 const (
diff --git a/internal/provider/adc/adc.go b/internal/provider/adc/adc.go
@@ -15,12 +15,12 @@ package adc
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"os"
 	"sync"
 	"time"
 
 	"github.com/api7/gopkg/pkg/log"
+	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	networkingv1 "k8s.io/api/networking/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -48,6 +48,7 @@ type BackendMode string
 const (
 	BackendModeAPISIXStandalone string = "apisix-standalone"
 	BackendModeAPI7EE           string = "api7ee"
+	BackendModeAPISIX           string = "apisix"
 )
 
 type adcClient struct {
@@ -188,7 +189,7 @@ func (d *adcClient) Update(ctx context.Context, tctx *provider.TranslateContext,
 	// This mode is full synchronization,
 	// which only needs to be saved in cache
 	// and triggered by a timer for synchronization
-	if d.BackendMode == BackendModeAPISIXStandalone || apiv2.Is(obj) {
+	if d.BackendMode == BackendModeAPISIXStandalone || d.BackendMode == BackendModeAPISIX || apiv2.Is(obj) {
 		return nil
 	}
 
@@ -249,7 +250,7 @@ func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
 	log.Debugw("successfully deleted resources from store", zap.Any("object", obj))
 
 	switch d.BackendMode {
-	case BackendModeAPISIXStandalone:
+	case BackendModeAPISIXStandalone, BackendModeAPISIX:
 		// Full synchronization is performed on a gateway by gateway basis
 		// and it is not possible to perform scheduled synchronization
 		// on deleted gateway level resources
diff --git a/internal/provider/adc/executor.go b/internal/provider/adc/executor.go
@@ -46,15 +46,19 @@ func (e *DefaultADCExecutor) Execute(ctx context.Context, mode string, config ad
 
 func (e *DefaultADCExecutor) runADC(ctx context.Context, mode string, config adcConfig, args []string) error {
 	for _, addr := range config.ServerAddrs {
-		ctxWithTimeout, cancel := context.WithTimeout(ctx, 15*time.Second)
-		defer cancel()
-		if err := e.runForSingleServer(ctxWithTimeout, addr, mode, config, args); err != nil {
+		if err := e.runForSingleServerWithTimeout(ctx, addr, mode, config, args); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
+func (e *DefaultADCExecutor) runForSingleServerWithTimeout(ctx context.Context, serverAddr, mode string, config adcConfig, args []string) error {
+	ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
+	defer cancel()
+	return e.runForSingleServer(ctx, serverAddr, mode, config, args)
+}
+
 func (e *DefaultADCExecutor) runForSingleServer(ctx context.Context, serverAddr, mode string, config adcConfig, args []string) error {
 	cmdArgs := append([]string{}, args...)
 	if !config.TlsVerify {
@@ -108,6 +112,10 @@ func (e *DefaultADCExecutor) buildCmdError(runErr error, stdout, stderr []byte)
 
 func (e *DefaultADCExecutor) handleOutput(output []byte) error {
 	var result adctypes.SyncResult
+	if index := strings.IndexByte(string(output), '{'); index > 0 {
+		log.Warnf("extra output: %s", string(output[:index]))
+		output = output[index:]
+	}
 	if err := json.Unmarshal(output, &result); err != nil {
 		log.Errorw("failed to unmarshal adc output",
 			zap.Error(err),
diff --git a/test/conformance/apisix/suite_test.go b/test/conformance/apisix/suite_test.go
@@ -160,7 +160,7 @@ func TestMain(m *testing.M) {
 		Namespace:          namespace,
 		StatusAddress:      address,
 		InitSyncDelay:      1 * time.Minute,
-		ProviderType:       "apisix-standalone",
+		ProviderType:       framework.ProviderType,
 		ProviderSyncPeriod: 10 * time.Millisecond,
 	})
 
diff --git a/test/e2e/framework/apisix_consts.go b/test/e2e/framework/apisix_consts.go
@@ -13,16 +13,25 @@
 package framework
 
 import (
+	"cmp"
 	_ "embed"
+	"os"
 	"text/template"
 
 	"github.com/Masterminds/sprig/v3"
 )
 
 var (
-	//go:embed manifests/apisix-standalone.yaml
+	ProviderType = cmp.Or(os.Getenv("PROVIDER_TYPE"), "apisix-standalone")
+)
+
+var (
+	//go:embed manifests/apisix.yaml
 	apisixStandaloneTemplate string
 	APISIXStandaloneTpl      *template.Template
+
+	//go:embed manifests/etcd.yaml
+	EtcdSpec string
 )
 
 var (
diff --git a/test/e2e/framework/assertion.go b/test/e2e/framework/assertion.go
@@ -111,7 +111,7 @@ func APIv2MustHaveCondition(t testing.TestingT, cli client.Client, timeout time.
 	}
 	err := PollUntilAPIv2MustHaveStatus(cli, timeout, nn, obj, f)
 
-	require.NoError(t, err, "error waiting status to have a Condition matching %+v", nn, cond)
+	require.NoError(t, err, "error waiting %s status to have a Condition matching %+v", nn, cond)
 }
 
 func PollUntilAPIv2MustHaveStatus(cli client.Client, timeout time.Duration, nn types.NamespacedName, obj client.Object, f func(client.Object) bool) error {
diff --git a/test/e2e/framework/manifests/apisix.yaml b/test/e2e/framework/manifests/apisix.yaml
@@ -7,14 +7,21 @@ data:
     deployment:
       role: traditional
       role_traditional:
-          config_provider: yaml
+          # on backend mode apisix-standalone, config_provider is "yaml"
+          # on backend mode apisix, config_provider is "etcd"
+        config_provider: {{ .ConfigProvider | default "yaml" }}
       admin:
         allow_admin:
         - 0.0.0.0/0
         admin_key:
         - key: {{ .AdminKey }}
           name: admin
           role: admin
+      {{- if eq .ConfigProvider "etcd" }}
+      etcd:
+        host:
+        - "http://etcd:2379"
+      {{- end }}
     nginx_config:
       worker_processes: 2
       error_log_level: info
diff --git a/test/e2e/framework/manifests/etcd.yaml b/test/e2e/framework/manifests/etcd.yaml
@@ -0,0 +1,41 @@
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: etcd
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: etcd
+  template:
+    metadata:
+      labels:
+        app: etcd
+    spec:
+      containers:
+      - name: etcd
+        image: quay.io/coreos/etcd:v3.5.0
+        command:
+        - etcd
+        - --data-dir=/etcd-data
+        - --name=node1
+        - --initial-advertise-peer-urls=http://0.0.0.0:2380
+        - --listen-peer-urls=http://0.0.0.0:2380
+        - --advertise-client-urls=http://0.0.0.0:2379
+        - --listen-client-urls=http://0.0.0.0:2379
+        - --initial-cluster=node1=http://0.0.0.0:2380
+        ports:
+        - containerPort: 2379
+        - containerPort: 2380
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: etcd
+spec:
+  ports:
+  - port: 2379
+    targetPort: 2379
+  selector:
+    app: etcd
diff --git a/test/e2e/gatewayapi/httproute.go b/test/e2e/gatewayapi/httproute.go
@@ -524,7 +524,7 @@ spec:
 			By("create HTTPRoute")
 			ResourceApplied("HTTPRoute", "httpbin", exactRouteByGet, 1)
 
-			By("access daataplane to check the HTTPRoute")
+			By("access dataplane to check the HTTPRoute")
 			s.NewAPISIXClient().
 				GET("/get").
 				WithHost("httpbin.example").
@@ -534,13 +534,14 @@ spec:
 			By("delete Gateway")
 			err := s.DeleteResource("Gateway", "apisix")
 			Expect(err).NotTo(HaveOccurred(), "deleting Gateway")
-			time.Sleep(5 * time.Second)
 
-			s.NewAPISIXClient().
-				GET("/get").
-				WithHost("httpbin.example").
-				Expect().
-				Status(404)
+			Eventually(func() int {
+				return s.NewAPISIXClient().
+					GET("/get").
+					WithHost("httpbin.example").
+					Expect().
+					Raw().StatusCode
+			}).WithTimeout(5 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusNotFound))
 		})
 
 		It("Proxy External Service", func() {
diff --git a/test/e2e/scaffold/adc.go b/test/e2e/scaffold/adc.go
@@ -26,6 +26,7 @@ import (
 
 	adctypes "github.com/apache/apisix-ingress-controller/api/adc"
 	"github.com/apache/apisix-ingress-controller/internal/provider/adc/translator"
+	"github.com/apache/apisix-ingress-controller/test/e2e/framework"
 )
 
 // DataplaneResource defines the interface for accessing dataplane resources
@@ -131,6 +132,9 @@ func (a *adcDataplaneResource) dumpResources(ctx context.Context) (*translator.T
 		"ADC_SERVER=" + a.serverAddr,
 		"ADC_TOKEN=" + a.token,
 	}
+	if framework.ProviderType != "" {
+		adcEnv = append(adcEnv, "ADC_BACKEND="+framework.ProviderType)
+	}
 
 	var stdout, stderr bytes.Buffer
 	cmd := exec.CommandContext(ctxWithTimeout, "adc", args...)
diff --git a/test/e2e/scaffold/apisix_deployer.go b/test/e2e/scaffold/apisix_deployer.go
@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/apache/apisix-ingress-controller/internal/provider/adc"
 	"github.com/apache/apisix-ingress-controller/pkg/utils"
 	"github.com/apache/apisix-ingress-controller/test/e2e/framework"
 )
@@ -36,6 +37,8 @@ type APISIXDeployOptions struct {
 	ServiceType      string
 	ServiceHTTPPort  int
 	ServiceHTTPSPort int
+
+	ConfigProvider string
 }
 
 type APISIXDeployer struct {
@@ -176,7 +179,7 @@ func (s *APISIXDeployer) newAPISIXTunnels(serviceName string) error {
 
 func (s *APISIXDeployer) deployDataplane(opts *APISIXDeployOptions) *corev1.Service {
 	if opts.ServiceName == "" {
-		opts.ServiceName = "apisix-standalone"
+		opts.ServiceName = framework.ProviderType
 	}
 
 	if opts.ServiceHTTPPort == 0 {
@@ -186,12 +189,24 @@ func (s *APISIXDeployer) deployDataplane(opts *APISIXDeployOptions) *corev1.Serv
 	if opts.ServiceHTTPSPort == 0 {
 		opts.ServiceHTTPSPort = 443
 	}
+	opts.ConfigProvider = "yaml"
+
+	kubectlOpts := k8s.NewKubectlOptions("", "", opts.Namespace)
+
+	if framework.ProviderType == adc.BackendModeAPISIX {
+		opts.ConfigProvider = "etcd"
+		// deploy etcd
+		k8s.KubectlApplyFromString(s.GinkgoT, kubectlOpts, framework.EtcdSpec)
+		err := framework.WaitPodsAvailable(s.GinkgoT, kubectlOpts, metav1.ListOptions{
+			LabelSelector: "app=etcd",
+		})
+		Expect(err).ToNot(HaveOccurred(), "waiting for etcd pod ready")
+	}
 
 	buf := bytes.NewBuffer(nil)
 	err := framework.APISIXStandaloneTpl.Execute(buf, opts)
 	Expect(err).ToNot(HaveOccurred(), "executing template")
 
-	kubectlOpts := k8s.NewKubectlOptions("", "", opts.Namespace)
 	k8s.KubectlApplyFromString(s.GinkgoT, kubectlOpts, buf.String())
 
 	err = framework.WaitPodsAvailable(s.GinkgoT, kubectlOpts, metav1.ListOptions{
@@ -218,17 +233,19 @@ func (s *APISIXDeployer) deployDataplane(opts *APISIXDeployOptions) *corev1.Serv
 
 func (s *APISIXDeployer) DeployIngress() {
 	s.Framework.DeployIngress(framework.IngressDeployOpts{
-		ProviderSyncPeriod: 200 * time.Millisecond,
 		ControllerName:     s.opts.ControllerName,
+		ProviderType:       framework.ProviderType,
+		ProviderSyncPeriod: 200 * time.Millisecond,
 		Namespace:          s.namespace,
 		Replicas:           1,
 	})
 }
 
 func (s *APISIXDeployer) ScaleIngress(replicas int) {
 	s.Framework.DeployIngress(framework.IngressDeployOpts{
-		ProviderSyncPeriod: 200 * time.Millisecond,
 		ControllerName:     s.opts.ControllerName,
+		ProviderType:       framework.ProviderType,
+		ProviderSyncPeriod: 200 * time.Millisecond,
 		Namespace:          s.namespace,
 		Replicas:           replicas,
 	})

Original file line number	Diff line number	Diff line change
`@@ -113,7 +113,7 @@ func (c *Config) Validate() error {`
`113`	`113`
`114`	`114`	`func validateProvider(config ProviderConfig) error {`
`115`	`115`	`switch config.Type {`
`116`		`- case ProviderTypeStandalone:`
	`116`	`+ case ProviderTypeStandalone, ProviderTypeAPISIX:`
`117`	`117`	`if config.SyncPeriod.Duration <= 0 {`
`118`	`118`	`return fmt.Errorf("sync_period must be greater than 0 for standalone provider")`
`119`	`119`	`}`
Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@ type ProviderType string`
`21`	`21`	`const (`
`22`	`22`	`ProviderTypeStandalone ProviderType = "apisix-standalone"`
`23`	`23`	`ProviderTypeAPI7EE ProviderType = "api7ee"`
	`24`	`+ ProviderTypeAPISIX ProviderType = "apisix"`
`24`	`25`	`)`
`25`	`26`
`26`	`27`	`const (`
Original file line number	Diff line number	Diff line change
`@@ -111,7 +111,7 @@ func APIv2MustHaveCondition(t testing.TestingT, cli client.Client, timeout time.`
`111`	`111`	`}`
`112`	`112`	`err := PollUntilAPIv2MustHaveStatus(cli, timeout, nn, obj, f)`
`113`	`113`
`114`		`- require.NoError(t, err, "error waiting status to have a Condition matching %+v", nn, cond)`
	`114`	`+ require.NoError(t, err, "error waiting %s status to have a Condition matching %+v", nn, cond)`
`115`	`115`	`}`
`116`	`116`
`117`	`117`	`func PollUntilAPIv2MustHaveStatus(cli client.Client, timeout time.Duration, nn types.NamespacedName, obj client.Object, f func(client.Object) bool) error {`