fix: scope ADC synchronization to controller-managed resources

docs/en/latest/FAQ.md

@@ -54,3 +54,28 @@ When both filters and a PluginConfig CRD are applied:
 
 * If filters are applied first, PluginConfig overrides any overlapping plugin settings.
 * If PluginConfig is applied first, filters merge with PluginConfig settings, and overlapping fields from filters take precedence.
+
+### Can Admin API resources coexist with controller-managed resources?
+
+With `provider.type: apisix` or `api7ee`, synchronization selects resources labeled
+`manager-by: apisix-ingress-controller`. Resources created outside the controller
+without this label are excluded from deletion, including when a Gateway or
+IngressClass is deleted. Do not reuse controller-generated resource IDs for
+externally managed resources: a matching ID can still be overwritten.
+
+ADC filters top-level resources. Routes nested under a selected Service are
+included with that Service, so externally managed routes should not share a
+controller-managed Service. Global rules and plugin metadata do not support this
+label-based isolation and remain subject to full reconciliation. The label also
+does not distinguish separate controller installations sharing one APISIX instance or API7 gateway group,
+or separate GatewayProxy resources that point to the same Admin API endpoint or gateway group: each
+GatewayProxy is synchronized on its own and can delete resources the other one created.
+Give each GatewayProxy its own APISIX instance or gateway group.
+This behavior does not apply to `apisix-standalone`, which manages a complete
+configuration.
+
+For the APISIX backend, unlabeled historical resources with IDs still present in
+the controller's desired configuration are updated and labeled during synchronization.
+Unlabeled orphaned resources are preserved. Review their ownership before deleting
+them or adding the controller label; adding the label allows the next synchronization
+to delete resources absent from the controller's desired configuration.

internal/adc/client/client.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"maps"
 	"os"
 	"slices"
 	"strings"
@@ -389,6 +390,21 @@ func (c *Client) push(ctx context.Context, config adctypes.Config, args []string
 	return alsoReport, err
 }
 
+func buildSyncArgs(filePath string, task Task, backend string) []string {
+	labels := task.Labels
+	if backend != backendAPISIXStandalone {
+		// Preserve per-resource constraints while scoping full sync and gateway cleanup;
+		// only standalone, which replaces the whole configuration, stays unscoped.
+		// ADC does not apply label selectors to global rules or plugin metadata.
+		labels = maps.Clone(labels)
+		if labels == nil {
+			labels = make(map[string]string)
+		}
+		labels[label.LabelManagedBy] = label.LabelManagedByValue
+	}
+	return BuildADCExecuteArgs(filePath, labels, task.ResourceTypes)
+}
+
 func (c *Client) sync(ctx context.Context, task Task) error {
 	c.log.V(1).Info("syncing resources", "task", task)
 
@@ -437,8 +453,6 @@ func (c *Client) sync(ctx context.Context, task Task) error {
 			pkgmetrics.RecordFileIODuration("prepare_sync_file", "success", time.Since(fileIOStart).Seconds())
 			defer cleanup()
 
-			args := BuildADCExecuteArgs(syncFilePath, task.Labels, task.ResourceTypes)
-
 			// Record sync duration for each config
 			startTime := time.Now()
 			resourceType := strings.Join(task.ResourceTypes, ",")
@@ -449,6 +463,7 @@ func (c *Client) sync(ctx context.Context, task Task) error {
 				config.BackendType = c.defaultMode
 			}
 
+			args := buildSyncArgs(syncFilePath, task, config.BackendType)
 			err = c.executor.Execute(ctx, config, args)
 			duration := time.Since(startTime).Seconds()
 
@@ -488,8 +503,6 @@ func (c *Client) sync(ctx context.Context, task Task) error {
 	defer cleanup()
 	c.log.V(1).Info("prepared sync file", "path", syncFilePath)
 
-	args := BuildADCExecuteArgs(syncFilePath, task.Labels, task.ResourceTypes)
-
 	for _, config := range task.Configs {
 		// Record sync duration for each config
 		startTime := time.Now()
@@ -501,6 +514,7 @@ func (c *Client) sync(ctx context.Context, task Task) error {
 			config.BackendType = c.defaultMode
 		}
 
+		args := buildSyncArgs(syncFilePath, task, config.BackendType)
 		alsoReport, err := c.push(ctx, config, args)
 		errs.Errors = append(errs.Errors, alsoReport...)
 

internal/adc/client/ownership_test.go

@@ -0,0 +1,109 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	adctypes "github.com/apache/apisix-ingress-controller/api/adc"
+	"github.com/apache/apisix-ingress-controller/internal/controller/label"
+	"github.com/apache/apisix-ingress-controller/internal/types"
+)
+
+func TestSyncOwnershipSelector(t *testing.T) {
+	for _, backend := range []string{"apisix", "api7ee", "custom", "apisix-standalone"} {
+		for _, explicit := range []bool{false, true} {
+			for _, operation := range []string{"full", "resource", "cleanup", "global-rule"} {
+				name := backend + "/default/" + operation
+				if explicit {
+					name = backend + "/explicit/" + operation
+				}
+				t.Run(name, func(t *testing.T) {
+					received := make(chan ADCServerRequest, 1)
+					server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+						var req ADCServerRequest
+						if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+							t.Error(err)
+						}
+						received <- req
+						_ = json.NewEncoder(w).Encode(adctypes.SyncResult{Status: adctypes.StatusSuccess})
+					}))
+					defer server.Close()
+					t.Setenv("ADC_SERVER_URL", server.URL)
+					c, err := New(logr.Discard(), backend, time.Second)
+					require.NoError(t, err)
+					config := adctypes.Config{Name: "proxy", ServerAddrs: []string{"http://apisix:9180"}}
+					if explicit {
+						config.BackendType = backend
+					}
+					owner := types.NamespacedNameKind{Kind: "Ingress", Namespace: "default", Name: "route"}
+					proxy := types.NamespacedNameKind{Kind: "GatewayProxy", Namespace: "default", Name: "proxy"}
+					task := Task{
+						Key:       owner,
+						Configs:   map[types.NamespacedNameKind]adctypes.Config{proxy: config},
+						Resources: &adctypes.Resources{},
+					}
+					expected := map[string]string{}
+					if backend != "apisix-standalone" {
+						expected[label.LabelManagedBy] = "apisix-ingress-controller"
+					}
+					switch operation {
+					case "full":
+						require.NoError(t, c.UpdateConfig(context.Background(), task))
+						_, err = c.Sync(context.Background())
+					case "cleanup":
+						require.NoError(t, c.UpdateConfig(context.Background(), task))
+						err = c.Delete(context.Background(), Task{Key: owner})
+					default:
+						task.Labels = map[string]string{label.LabelResourceKey: "Ingress/default/route", label.LabelNamespace: "default"}
+						if operation == "global-rule" {
+							task.ResourceTypes = []string{"global_rule"}
+						} else {
+							task.ResourceTypes = []string{"service"}
+						}
+						expected[label.LabelResourceKey] = "Ingress/default/route"
+						require.NoError(t, c.UpdateConfig(context.Background(), task))
+						err = c.sync(context.Background(), task)
+						assert.Equal(t, map[string]string{label.LabelResourceKey: "Ingress/default/route", label.LabelNamespace: "default"}, task.Labels)
+					}
+					require.NoError(t, err)
+					select {
+					case req := <-received:
+						assert.Equal(t, backend, req.Task.Opts.Backend)
+						if len(expected) == 0 {
+							assert.Empty(t, req.Task.Opts.LabelSelector)
+						} else {
+							assert.Equal(t, expected, req.Task.Opts.LabelSelector)
+						}
+					default:
+						t.Fatal("sync did not reach ADC")
+					}
+				})
+			}
+		}
+	}
+}

internal/controller/label/label.go

@@ -33,6 +33,7 @@ const (
 	LabelNamespace      = "k8s/namespace"
 	LabelControllerName = "k8s/controller-name"
 	LabelManagedBy      = "manager-by"
+	LabelManagedByValue = "apisix-ingress-controller"
 	LabelResourceKey    = "k8s/resource-key"
 )
 
@@ -42,7 +43,7 @@ func GenLabel(obj client.Object, args ...string) Label {
 	label[LabelNamespace] = obj.GetNamespace()
 	label[LabelName] = obj.GetName()
 	label[LabelControllerName] = config.ControllerConfig.ControllerName
-	label[LabelManagedBy] = "apisix-ingress-controller"
+	label[LabelManagedBy] = LabelManagedByValue
 	label[LabelResourceKey] = fmt.Sprintf("%s/%s/%s", label[LabelKind], label[LabelNamespace], label[LabelName])
 	for i := 0; i+1 < len(args); i += 2 {
 		label[args[i]] = args[i+1]

internal/provider/common/retrier.go

@@ -70,25 +70,38 @@ func (r *Retrier) Reset() {
 		r.timer.Stop()
 		r.timer = nil
 	}
+	select {
+	case <-r.ch:
+	default:
+	}
 	r.backoff.Reset()
 }
 
 func (r *Retrier) Next() {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
-	if r.timer != nil {
-		r.timer.Stop()
-		r.timer = nil
+	// Event-triggered failures must not postpone an already scheduled retry.
+	if r.timer != nil || len(r.ch) > 0 {
+		return
 	}
 
 	delay := r.backoff.Next()
-	r.timer = time.AfterFunc(delay, func() {
+	var timer *time.Timer
+	timer = time.AfterFunc(delay, func() {
+		r.mu.Lock()
+		defer r.mu.Unlock()
+		// Reset may have canceled this callback while it waited for the lock.
+		if r.timer != timer {
+			return
+		}
+		r.timer = nil
 		select {
 		case r.ch <- struct{}{}:
 		default:
 		}
 	})
+	r.timer = timer
 }
 
 func (r *Retrier) C() <-chan struct{} {

internal/provider/common/retrier_test.go

@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package common
+
+import (
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+func TestRetrierPreservesPendingDeadline(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		r := NewRetrier(NewExponentialBackoff(10*time.Second, time.Minute))
+		defer r.Reset()
+		r.Next()
+		for range 9 {
+			time.Sleep(time.Second)
+			r.Next()
+		}
+		time.Sleep(time.Second)
+		synctest.Wait()
+		select {
+		case <-r.C():
+		default:
+			t.Fatal("event-triggered failures postponed the pending retry")
+		}
+		r.Next()
+		time.Sleep(20*time.Second - time.Millisecond)
+		synctest.Wait()
+		select {
+		case <-r.C():
+			t.Fatal("the next retry fired before its backoff delay")
+		default:
+		}
+		time.Sleep(time.Millisecond)
+		synctest.Wait()
+		select {
+		case <-r.C():
+		default:
+			t.Fatal("the next retry must advance backoff only once")
+		}
+	})
+}
+
+func TestRetrierResetClearsPendingRetry(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		r := NewRetrier(NewExponentialBackoff(time.Second, time.Minute))
+		defer r.Reset()
+		r.Next()
+		time.Sleep(time.Second)
+		synctest.Wait()
+		r.Reset()
+		select {
+		case <-r.C():
+			t.Fatal("reset left a stale retry notification")
+		default:
+		}
+		r.Next()
+		time.Sleep(time.Second)
+		synctest.Wait()
+		select {
+		case <-r.C():
+		default:
+			t.Fatal("reset did not restore the base delay")
+		}
+	})
+}