This is an automated email from the ASF dual-hosted git repository.
bzp2010 pushed a commit to branch bzp/feat-refactor-adc-api
in repository https://gitbox.apache.org/repos/asf/apisix-ingress-controller.git
The following commit(s) were added to refs/heads/bzp/feat-refactor-adc-api by
this push:
new 4ec7f5da sperate standalone syncer policy layer
4ec7f5da is described below
commit 4ec7f5da997ade78e4821cb872d85959f93d56a9
Author: bzp2010 <[email protected]>
AuthorDate: Thu Sep 10 23:01:47 2026 +0800
sperate standalone syncer policy layer
---
internal/adc/client/standalone_syncer.go | 120 +++++++++++++++++++++++++
internal/provider/apisix/provider.go | 94 ++++---------------
internal/provider/apisix/provider_test.go | 2 +-
internal/provider/apisix/sync_baseline_test.go | 2 +-
4 files changed, 139 insertions(+), 79 deletions(-)
diff --git a/internal/adc/client/standalone_syncer.go
b/internal/adc/client/standalone_syncer.go
new file mode 100644
index 00000000..87cd0291
--- /dev/null
+++ b/internal/adc/client/standalone_syncer.go
@@ -0,0 +1,120 @@
+// 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"
+ "sync"
+
+ "github.com/go-logr/logr"
+
+ pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
+)
+
+// StandaloneSyncer drives apisix-standalone's diff-baseline recovery on top
of the
+// one-shot Client. It is only for apisix-standalone: no other backend type
keeps a
+// conf_version, so no other backend type needs any of this.
+//
+// APISIX standalone keeps a monotonic conf_version per resource type and
refuses a whole
+// push whose version is behind the data plane's. ADC diffs against a cached
baseline to
+// build that push, and the baseline can go stale two ways:
+//
+// - Across a leadership change. The ADC server is a sidecar that outlives
the manager
+// container, so what it holds for a cacheKey can be the snapshot this pod
left behind
+// in an earlier term, while the leader in between moved the data plane's
conf_version
+// past it. InvalidateBaselines on leader acquisition forces the first
push of every
+// cacheKey this term to re-derive its baseline from the data plane.
+// - Within a term, from a desync no leadership change explains, e.g.
another writer on
+// the same data plane. A conf_version the data plane refuses is the only
way that
+// shows itself; Sync answers it with one rebuild-and-retry.
+type StandaloneSyncer struct {
+ client *Client
+ log logr.Logger
+
+ mu sync.Mutex
+ rebuilt map[string]struct{}
+}
+
+func NewStandaloneSyncer(c *Client, log logr.Logger) *StandaloneSyncer {
+ return &StandaloneSyncer{
+ client: c,
+ log: log.WithName("standalone-syncer"),
+ rebuilt: make(map[string]struct{}),
+ }
+}
+
+// InvalidateBaselines forgets every rebuilt-this-term record, so the next
push of each
+// cacheKey re-derives its baseline. Call on leader acquisition.
+func (s *StandaloneSyncer) InvalidateBaselines() {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ clear(s.rebuilt)
+}
+
+func (s *StandaloneSyncer) isCurrent(cacheKey string) bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ _, ok := s.rebuilt[cacheKey]
+ return ok
+}
+
+func (s *StandaloneSyncer) markCurrent(cacheKey string) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.rebuilt[cacheKey] = struct{}{}
+}
+
+// Sync pushes in, rebuilding ADC's baseline first if this term has not yet
pushed this
+// cacheKey, and once more if the data plane rejects the push over a stale
conf_version.
+//
+// It returns every error worth reporting, so 0, 1, or 2 of them: the final
failure, plus
+// the conf_version rejection that triggered a rebuild which then failed for a
different
+// reason (on its own that failure points nowhere near its cause, e.g. an ADC
server too
+// old to know bypassCache answers with a schema error). BypassCache is scoped
to the call
+// that recovers from a rejection and never written back into in.
+func (s *StandaloneSyncer) Sync(ctx context.Context, in SyncInput) []error {
+ in.Config.BypassCache = !s.isCurrent(in.Name)
+ err := s.client.Sync(ctx, in)
+
+ var report []error
+ if !in.Config.BypassCache && IsConfVersionRejection(err) {
+ s.log.Info("data plane rejected a stale conf_version,
rebuilding the ADC baseline",
+ "config", in.Name, "error", err.Error())
+ // The rebuild is not rate limited, so a rejection on every
push (someone else
+ // writing to this data plane) turns every push into a full
fetch and diff, and
+ // this counter is what says so.
+ pkgmetrics.RecordExecutionError(in.Name,
"conf_version_conflict")
+
+ rejection := err
+ in.Config.BypassCache = true
+ err = s.client.Sync(ctx, in)
+
+ if err != nil && err.Error() != rejection.Error() {
+ report = append(report, rejection)
+ }
+ }
+
+ // Only a push ADC accepted proves its baseline is now derived from the
data plane.
+ if err == nil && in.Config.BypassCache {
+ s.markCurrent(in.Name)
+ }
+ if err != nil {
+ report = append(report, err)
+ }
+ return report
+}
diff --git a/internal/provider/apisix/provider.go
b/internal/provider/apisix/provider.go
index 493aa2c4..f5ffa08c 100644
--- a/internal/provider/apisix/provider.go
+++ b/internal/provider/apisix/provider.go
@@ -43,7 +43,6 @@ import (
"github.com/apache/apisix-ingress-controller/internal/provider/common"
"github.com/apache/apisix-ingress-controller/internal/types"
"github.com/apache/apisix-ingress-controller/internal/utils"
- pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
)
const (
@@ -73,12 +72,10 @@ type apisixProvider struct {
// snapshot together with pushing it
syncLocks *keyedMutex
- // rebuiltMu guards rebuiltBaselines.
- rebuiltMu sync.Mutex
- // rebuiltBaselines holds the cacheKeys whose ADC diff baseline this
leadership term
- // has already re-derived from the data plane. A key absent from it is
pushed with
- // BypassCache first. See invalidateBaselineCache.
- rebuiltBaselines map[string]struct{}
+ // standaloneSyncer owns apisix-standalone's ADC diff-baseline
recovery: the
+ // BypassCache decision, the one retry for a stale conf_version, and
the per-term
+ // record of which cacheKeys it has rebuilt. Unused for every other
backend type.
+ standaloneSyncer *adcclient.StandaloneSyncer
updater status.Updater
statusUpdateMap map[types.NamespacedNameKind][]string
@@ -114,7 +111,7 @@ func New(log logr.Logger, updater status.Updater, readier
readiness.ReadinessMan
configManager: configManager,
debugProvider: common.NewADCDebugProvider(store,
configManager),
syncLocks: newKeyedMutex(),
- rebuiltBaselines: make(map[string]struct{}),
+ standaloneSyncer: adcclient.NewStandaloneSyncer(cli, logger),
Options: o,
translator: translator.NewTranslator(log,
o.ListenerPortMatchMode),
updater: updater,
@@ -327,35 +324,6 @@ func (d *apisixProvider) evictFromStore(
return nil
}
-// invalidateBaselineCache forgets which ADC diff baselines are known to be
current, so
-// the next push of each cacheKey re-derives its baseline from the data plane.
-//
-// Called on leader acquisition, the one moment a stale baseline can enter the
picture.
-// The ADC server is a sidecar that outlives the controller process: losing
the lease
-// terminates the manager container but not the sidecar, so what ADC holds for
a cacheKey
-// (the last synced content plus the conf_version it generated) can still be
the snapshot
-// this pod left behind in an earlier term, while the leader in between kept
pushing and
-// moved the data plane's conf_version past it. APISIX standalone requires
those versions
-// to be monotonic and refuses the whole configuration otherwise.
-func (d *apisixProvider) invalidateBaselineCache() {
- d.rebuiltMu.Lock()
- defer d.rebuiltMu.Unlock()
- clear(d.rebuiltBaselines)
-}
-
-func (d *apisixProvider) baselineIsCurrent(cacheKey string) bool {
- d.rebuiltMu.Lock()
- defer d.rebuiltMu.Unlock()
- _, ok := d.rebuiltBaselines[cacheKey]
- return ok
-}
-
-func (d *apisixProvider) markBaselineCurrent(cacheKey string) {
- d.rebuiltMu.Lock()
- defer d.rebuiltMu.Unlock()
- d.rebuiltBaselines[cacheKey] = struct{}{}
-}
-
// syncConfigNow reads name's current data (via build, called only once this
cacheKey's
// lock is actually held) and pushes it -- one atomic read-then-push step per
cacheKey, so
// whichever caller is granted the lock decides what to push only once it
holds it: nothing
@@ -380,53 +348,25 @@ func (d *apisixProvider) syncConfigNow(
return execErrs, nil
}
-// pushConfig syncs input through the adc client once, and when
apisix-standalone rejects
-// it over a stale conf_version, asks ADC to rebuild its diff baseline from
the data plane
-// (SyncInput.Config.BypassCache) and syncs again. The adc client never
retries on its
-// own: this is the one rejection AIC knows how to answer, so AIC owns both
the decision
-// and the record of which baselines this leadership term has already rebuilt.
-//
-// invalidateBaselineCache on leader acquisition forces the first push of
every cacheKey
-// this term to rebuild, which covers where staleness comes from. This retry
is the safety
-// net for a desync no leadership change explains, such as another writer on
the same data
-// plane, and a conf_version the data plane refuses is the only way that shows
itself.
+// pushConfig sends input to its data plane and shapes whatever failed into
the form
+// status reporting consumes. apisix-standalone goes through standaloneSyncer,
which may
+// rebuild ADC's diff baseline and retry once; every other backend type is a
single
+// one-shot push through the adc client, which never retries.
func (d *apisixProvider) pushConfig(ctx context.Context, input
adcclient.SyncInput) types.ADCExecutionErrors {
backend := input.Config.BackendType
if backend == "" {
backend = d.DefaultBackendMode
}
- standalone := backend == adcclient.BackendAPISIXStandalone
- input.Config.BypassCache = standalone &&
!d.baselineIsCurrent(input.Name)
- err := d.client.Sync(ctx, input)
-
- var execErrs types.ADCExecutionErrors
- if standalone && !input.Config.BypassCache &&
adcclient.IsConfVersionRejection(err) {
- d.log.Info("data plane rejected a stale conf_version,
rebuilding the ADC baseline",
- "config", input.Name, "error", err.Error())
- // The rebuild is not rate limited, so a rejection on every
push (someone else writing
- // to this data plane) turns every push into a full fetch and
diff, and this counter
- // is what says so.
- pkgmetrics.RecordExecutionError(input.Name,
"conf_version_conflict")
-
- rejection := err
- input.Config.BypassCache = true
- err = d.client.Sync(ctx, input)
-
- // Keep the rejection visible when the rebuild itself fails: on
its own a failed
- // rebuild points nowhere near what it was rebuilding for (an
ADC server too old for
- // bypassCache answers with a schema error). Unless the rebuild
hit the very same
- // rejection, where repeating it only pads the status message.
- if err != nil && err.Error() != rejection.Error() {
- execErrs.Errors = append(execErrs.Errors,
toADCExecutionError(input.Name, rejection))
- }
+ var errs []error
+ if backend == adcclient.BackendAPISIXStandalone {
+ errs = d.standaloneSyncer.Sync(ctx, input)
+ } else if err := d.client.Sync(ctx, input); err != nil {
+ errs = []error{err}
}
- // Only a push ADC accepted proves its baseline is now derived from the
data plane.
- if err == nil && input.Config.BypassCache {
- d.markBaselineCurrent(input.Name)
- }
- if err != nil {
+ var execErrs types.ADCExecutionErrors
+ for _, err := range errs {
execErrs.Errors = append(execErrs.Errors,
toADCExecutionError(input.Name, err))
}
return execErrs
@@ -490,7 +430,7 @@ func (d *apisixProvider) Start(ctx context.Context) error {
// one thing that leaves the ADC sidecar holding a baseline from an
earlier term: it
// survives the manager container, the configuration it was derived
from does not.
// Rebuild every baseline from the data plane before syncing from it.
- d.invalidateBaselineCache()
+ d.standaloneSyncer.InvalidateBaselines()
d.log.Info("starting provider, waiting for readiness")
d.readier.WaitReady(ctx, 5*time.Minute)
diff --git a/internal/provider/apisix/provider_test.go
b/internal/provider/apisix/provider_test.go
index 599de304..66a91b7f 100644
--- a/internal/provider/apisix/provider_test.go
+++ b/internal/provider/apisix/provider_test.go
@@ -60,7 +60,7 @@ func newTestProvider(t *testing.T) *apisixProvider {
store: cache.NewStore(logr.Discard()),
configManager:
common.NewConfigManager[types.NamespacedNameKind, adctypes.Config](),
syncLocks: newKeyedMutex(),
- rebuiltBaselines: make(map[string]struct{}),
+ standaloneSyncer: adcclient.NewStandaloneSyncer(cli,
logr.Discard()),
syncCh: make(chan struct{}, 1),
log: logr.Discard(),
}
diff --git a/internal/provider/apisix/sync_baseline_test.go
b/internal/provider/apisix/sync_baseline_test.go
index f53bc120..093c9a71 100644
--- a/internal/provider/apisix/sync_baseline_test.go
+++ b/internal/provider/apisix/sync_baseline_test.go
@@ -114,7 +114,7 @@ func TestPushRebuildsBaselineOncePerTermThenReusesIt(t
*testing.T) {
require.Empty(t, d.pushConfig(context.Background(), in).Errors)
require.Empty(t, d.pushConfig(context.Background(), in).Errors)
- d.invalidateBaselineCache()
+ d.standaloneSyncer.InvalidateBaselines()
require.Empty(t, d.pushConfig(context.Background(), in).Errors)
assert.Equal(t, []bool{true, false, true}, bypassSeq(reqs()),
