feat: add support for apisix backend mode

.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
 

.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

internal/controller/config/config.go

@@ -118,7 +118,7 @@ func validateProvider(config ProviderConfig) error {
 			return fmt.Errorf("sync_period must be greater than 0 for standalone provider")
 		}
 		return nil
-	case ProviderTypeAPI7EE:
+	case ProviderTypeAPI7EE, ProviderTypeAPISIX:
 		return nil
 	default:
 		return fmt.Errorf("unsupported provider type: %s", config.Type)

internal/controller/config/types.go

@@ -21,6 +21,7 @@ type ProviderType string
 const (
 	ProviderTypeStandalone ProviderType = "apisix-standalone"
 	ProviderTypeAPI7EE     ProviderType = "api7ee"
+	ProviderTypeAPISIX     ProviderType = "apisix"
 )
 
 const (

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 {
@@ -185,11 +186,21 @@ 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) {
+	switch d.BackendMode {
+	case BackendModeAPI7EE:
+		if apiv2.Is(obj) {
+			return nil
+		}
+	case BackendModeAPISIX:
+		// do nothing
+	case BackendModeAPISIXStandalone:
+		// This mode is full synchronization,
+		// which only needs to be saved in cache
+		// and triggered by a timer for synchronization
 		return nil
+	default:
+		return errors.Errorf("unknown backend mode: %s", d.BackendMode)
+
 	}
 
 	return d.sync(ctx, Task{
@@ -260,7 +271,7 @@ func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
 			})
 		}
 		return nil
-	case BackendModeAPI7EE:
+	case BackendModeAPI7EE, BackendModeAPISIX:
 		return d.sync(ctx, Task{
 			Name:          obj.GetName(),
 			Labels:        labels,

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),

test/conformance/apisix/suite_test.go

@@ -12,6 +12,7 @@
 package conformance
 
 import (
+	"cmp"
 	"context"
 	"fmt"
 	"os"
@@ -160,7 +161,7 @@ func TestMain(m *testing.M) {
 		Namespace:          namespace,
 		StatusAddress:      address,
 		InitSyncDelay:      1 * time.Minute,
-		ProviderType:       "apisix-standalone",
+		ProviderType:       cmp.Or(os.Getenv(framework.EnvKeyProviderType), "apisix-standalone"),
 		ProviderSyncPeriod: 10 * time.Millisecond,
 	})
 

test/e2e/framework/apisix_consts.go

@@ -19,10 +19,17 @@ import (
 	"github.com/Masterminds/sprig/v3"
 )
 
+const (
+	EnvKeyProviderType = "PROVIDER_TYPE"
+)
+
 var (
 	//go:embed manifests/apisix-standalone.yaml
 	apisixStandaloneTemplate string
 	APISIXStandaloneTpl      *template.Template
+
+	//go:embed manifests/etcd.yaml
+	EtcdSpec string
 )
 
 var (

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 {

test/e2e/framework/manifests/apisix-standalone.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

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

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() {

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 providerType := os.Getenv(framework.EnvKeyProviderType); providerType != "" {
+		adcEnv = append(adcEnv, "ADC_BACKEND="+providerType)
+	}
 
 	var stdout, stderr bytes.Buffer
 	cmd := exec.CommandContext(ctxWithTimeout, "adc", args...)

test/e2e/scaffold/apisix_deployer.go

@@ -14,6 +14,7 @@ package scaffold
 
 import (
 	"bytes"
+	"cmp"
 	"fmt"
 	"os"
 	"time"
@@ -24,6 +25,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 +38,8 @@ type APISIXDeployOptions struct {
 	ServiceType      string
 	ServiceHTTPPort  int
 	ServiceHTTPSPort int
+
+	ConfigProvider string
 }
 
 type APISIXDeployer struct {
@@ -186,12 +190,25 @@ func (s *APISIXDeployer) deployDataplane(opts *APISIXDeployOptions) *corev1.Serv
 	if opts.ServiceHTTPSPort == 0 {
 		opts.ServiceHTTPSPort = 443
 	}
+	opts.ConfigProvider = "yaml"
+
+	kubectlOpts := k8s.NewKubectlOptions("", "", opts.Namespace)
+
+	if os.Getenv(framework.EnvKeyProviderType) == adc.BackendModeAPISIX {
+		opts.ServiceName = "apisix"
+		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 +235,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:       cmp.Or(os.Getenv(framework.EnvKeyProviderType), "apisix-standalone"),
+		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:       cmp.Or(os.Getenv(framework.EnvKeyProviderType), "apisix-standalone"),
+		ProviderSyncPeriod: 200 * time.Millisecond,
 		Namespace:          s.namespace,
 		Replicas:           replicas,
 	})