feat: support new adc sync api (#235)

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

Author: ronething

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

@@ -71,7 +71,7 @@ jobs:
           ARCH: amd64
           ENABLE_PROXY: "false"
           BASE_IMAGE_TAG: "debug"
-          # ADC_VERSION: "dev"
+          ADC_VERSION: "dev"
         run: |
           echo "building images..."
           make build-image

Makefile

@@ -191,6 +191,7 @@ kind-load-images: pull-infra-images kind-load-ingress-image
 	@kind load docker-image kennethreitz/httpbin:latest --name $(KIND_NAME)
 	@kind load docker-image jmalloc/echo-server:latest --name $(KIND_NAME)
 	@kind load docker-image ghcr.io/api7/adc:dev --name $(KIND_NAME)
+	@kind load docker-image apache/apisix:dev --name $(KIND_NAME)
 
 .PHONY: kind-load-gateway-image
 kind-load-gateway-image:
@@ -217,6 +218,7 @@ pull-infra-images:
 	@docker pull kennethreitz/httpbin:latest
 	@docker pull jmalloc/echo-server:latest
 	@docker pull ghcr.io/api7/adc:dev
+	@docker pull apache/apisix:dev
 
 ##@ Build
 

internal/adc/client/executor.go

@@ -213,11 +213,12 @@ type ADCServerTask struct {
 // ADCServerOpts represents the options in ADC Server task
 type ADCServerOpts struct {
 	Backend             string            `json:"backend"`
-	Server              string            `json:"server"`
+	Server              []string          `json:"server"`
 	Token               string            `json:"token"`
 	LabelSelector       map[string]string `json:"labelSelector,omitempty"`
 	IncludeResourceType []string          `json:"includeResourceType,omitempty"`
 	TlsSkipVerify       *bool             `json:"tlsSkipVerify,omitempty"`
+	CacheKey            string            `json:"cacheKey"`
 }
 
 // HTTPADCExecutor implements ADCExecutor interface using HTTP calls to ADC Server
@@ -247,7 +248,15 @@ func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context, mode string, config a
 		Name: config.Name,
 	}
 
-	for _, addr := range config.ServerAddrs {
+	serverAddrs := func() []string {
+		if mode == "apisix-standalone" {
+			return []string{strings.Join(config.ServerAddrs, ",")}
+		}
+		return config.ServerAddrs
+	}()
+	log.Debugw("running http sync", zap.Strings("serverAddrs", serverAddrs), zap.String("mode", mode))
+
+	for _, addr := range serverAddrs {
 		if err := e.runHTTPSyncForSingleServer(ctx, addr, mode, config, args); err != nil {
 			log.Errorw("failed to run http sync for server", zap.String("server", addr), zap.Error(err))
 			var execErr types.ADCExecutionServerAddrError
@@ -365,11 +374,12 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr, mode
 		Task: ADCServerTask{
 			Opts: ADCServerOpts{
 				Backend:             mode,
-				Server:              serverAddr,
+				Server:              strings.Split(serverAddr, ","),
 				Token:               config.Token,
 				LabelSelector:       labels,
 				IncludeResourceType: types,
 				TlsSkipVerify:       ptr.To(!tlsVerify),
+				CacheKey:            config.Name,
 			},
 			Config: *resources,
 		},
@@ -386,6 +396,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr, mode
 		zap.String("url", e.serverURL+"/sync"),
 		zap.String("server", serverAddr),
 		zap.String("mode", mode),
+		zap.String("cacheKey", config.Name),
 		zap.Any("labelSelector", labels),
 		zap.Strings("includeResourceType", types),
 		zap.Bool("tlsSkipVerify", !tlsVerify),

internal/provider/apisix/status.go

@@ -205,6 +205,12 @@ func (d *apisixProvider) handleDetailedFailedStatuses(
 	statusUpdateMap map[types.NamespacedNameKind][]string,
 ) {
 	for _, status := range failedStatus.FailedStatuses {
+		// in the APISIX standalone mode, the related values in the sync failure event are empty.
+		if status.Event.ResourceType == "" {
+			d.handleEmptyFailedStatuses(configName, failedStatus, statusUpdateMap)
+			return
+		}
+
 		id := status.Event.ResourceID
 		labels, err := d.client.GetResourceLabel(configName, status.Event.ResourceType, id)
 		if err != nil {

test/e2e/framework/api7_dashboard.go

@@ -395,7 +395,7 @@ func (f *Framework) CreateNewGatewayGroupWithIngressE() (string, error) {
 }
 
 func (f *Framework) setDpManagerEndpoints() {
-	payload := []byte(fmt.Sprintf(`{"control_plane_address":["%s"]}`, DPManagerTLSEndpoint))
+	payload := []byte(fmt.Sprintf(`{"dp_manager_address":["%s"]}`, DPManagerTLSEndpoint))
 
 	respExp := f.DashboardHTTPClient().
 		PUT("/api/system_settings").
@@ -408,7 +408,7 @@ func (f *Framework) setDpManagerEndpoints() {
 	f.Logf("set dp manager endpoints response: %s", respExp.Body().Raw())
 
 	respExp.Status(200).
-		Body().Contains("control_plane_address")
+		Body().Contains("dp_manager_address")
 }
 
 func (f *Framework) GetDashboardEndpoint() string {

test/e2e/framework/manifests/ingress.yaml

@@ -437,6 +437,8 @@ spec:
           value: ingress
         - name: ADC_EXPERIMENTAL_FEATURE_FLAGS 
           value: remote-state-file,parallel-backend-request
+        - name: ADC_INGRESS_LOG_LEVEL
+          value: "debug"
         name: adc-server
         args:
         - "server"