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
commit 4da04fac74eefc3f7af96b3e1abb26ad40ea869a Author: bzp2010 <[email protected]> AuthorDate: Thu Sep 10 19:53:31 2026 +0800 refactor: clarify module boundaries between provider and ADC client --- internal/adc/client/client.go | 193 ++++----------------- internal/adc/client/executor.go | 62 ++++--- internal/adc/client/executor_test.go | 181 +------------------- internal/provider/apisix/provider.go | 132 +++++++++++++-- internal/provider/apisix/provider_test.go | 13 +- internal/provider/apisix/sync_baseline_test.go | 223 +++++++++++++++++++++++++ 6 files changed, 422 insertions(+), 382 deletions(-) diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go index 0e9e4249..0bec1602 100644 --- a/internal/adc/client/client.go +++ b/internal/adc/client/client.go @@ -16,18 +16,17 @@ // under the License. // Package client talks to the ADC server: given a fully-prepared sync or validate -// request, it translates it to ADC's wire format, sends it, and interprets the response. -// It holds no bookkeeping of its own about which Kubernetes resource maps to which -// GatewayProxy, or what a GatewayProxy's current resource snapshot is -- that is AIC's own -// state, owned by the caller and handed in as input on every call. +// request, it translates it to ADC's wire format, sends it once, and interprets the +// response into a typed error. It holds no bookkeeping of its own: not which Kubernetes +// resource maps to which GatewayProxy, not a GatewayProxy's current resource snapshot, +// and not whether a data plane's diff baseline can be trusted. All of that is AIC's own +// state, owned by the caller, which also owns every decision to retry. package client import ( "context" - "fmt" "os" "strings" - "sync" "time" "github.com/go-logr/logr" @@ -43,13 +42,6 @@ type Client struct { defaultMode string - // rebuiltMu guards rebuiltBaselines. - rebuiltMu sync.Mutex - // rebuiltBaselines holds the cacheKeys whose ADC baseline this leadership term has - // already re-derived from the data plane. A key missing from it is synced with - // bypassCache first. See InvalidateADCCache. - rebuiltBaselines map[string]struct{} - log logr.Logger } @@ -63,54 +55,23 @@ func New(log logr.Logger, defaultMode string, timeout time.Duration) (*Client, e logger.Info("ADC client initialized") return &Client{ - rebuiltBaselines: make(map[string]struct{}), - executor: NewHTTPADCExecutor(log, serverURL, timeout), - log: logger, - defaultMode: defaultMode, + executor: NewHTTPADCExecutor(log, serverURL, timeout), + log: logger, + defaultMode: defaultMode, }, nil } -// InvalidateADCCache forgets which ADC baselines are known to be current, so that the -// next sync of each cacheKey re-derives its baseline from the data plane. -// -// It is called on leader acquisition, which is 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 (c *Client) InvalidateADCCache() { - c.rebuiltMu.Lock() - defer c.rebuiltMu.Unlock() - clear(c.rebuiltBaselines) -} - -func (c *Client) baselineIsCurrent(cacheKey string) bool { - c.rebuiltMu.Lock() - defer c.rebuiltMu.Unlock() - _, ok := c.rebuiltBaselines[cacheKey] - return ok -} - -func (c *Client) markBaselineCurrent(cacheKey string) { - c.rebuiltMu.Lock() - defer c.rebuiltMu.Unlock() - c.rebuiltBaselines[cacheKey] = struct{}{} -} - -// isConfVersionRejection reports whether the data plane refused the push because of a -// conf_version, which is the one rejection re-deriving the baseline can answer. +// IsConfVersionRejection reports whether err is the data plane refusing a push because +// its conf_version is behind. That is the one rejection a caller can answer, by asking +// ADC to rebuild its diff baseline from the data plane (SyncInput.Config.BypassCache) and +// syncing again. This package never makes that decision; it only lets a caller recognize +// the case. // // It matches the field name, not the sentence. conf_version is part of the standalone -// admin API -- we send those keys ourselves -- so any rejection that concerns it names it, -// whatever prose APISIX wraps it in. Matching the sentence would tie us to prose APISIX is -// free to reword; matching the field only breaks if it renames the API. -// -// This backs the safety net, not the fix. A baseline is rebuilt on leader acquisition, -// which is where staleness comes from, so if this ever stopped firing the reported bug -// would not come back with it. -func isConfVersionRejection(err error) bool { +// admin API, callers send those keys themselves, so any rejection that concerns it names +// the field whatever prose APISIX wraps it in. Matching the sentence would tie this to +// prose APISIX is free to reword; matching the field only breaks if it renames the API. +func IsConfVersionRejection(err error) bool { return err != nil && strings.Contains(err.Error(), confVersionField) } @@ -196,97 +157,20 @@ func (in SyncInput) MarshalLog() any { } } -// Sync pushes every given SyncInput to its data plane in one sweep, and reports the -// parsed, typed error for each one that failed, keyed by its Name -- an input whose name -// is absent from the returned map genuinely succeeded. It never returns a raw HTTP status -// or body; every response ADC can send back is already interpreted by the time it gets -// here. -func (c *Client) Sync(ctx context.Context, inputs []SyncInput) (map[string]types.ADCExecutionErrors, error) { - if len(inputs) == 0 { - return nil, nil - } - c.log.V(1).Info("syncing resources", "inputs", inputs) - - failedMap := map[string]types.ADCExecutionErrors{} - var failedNames []string - for _, in := range inputs { - if in.Resources == nil { - continue - } - if err := c.syncOne(ctx, in); err != nil { - c.log.Error(err, "failed to sync resources", "name", in.Name) - failedNames = append(failedNames, in.Name) - var execErrs types.ADCExecutionErrors - if errors.As(err, &execErrs) { - failedMap[in.Name] = execErrs - } - } - } - - var err error - if len(failedNames) > 0 { - err = fmt.Errorf("failed to sync %d configs: %s", - len(failedNames), - strings.Join(failedNames, ", ")) - } - return failedMap, err -} - -// push syncs one config through the ADC server, re-deriving the baseline ADC diffs against -// whenever that baseline cannot be trusted. Beside the error to report it returns the ones -// to report next to it, which a rebuild that failed leaves behind. +// Sync sends in to its data plane once and returns the parsed, typed error if the push +// failed, or nil if it succeeded. It never returns a raw HTTP status or body: every +// response ADC can send back is already interpreted by the time it gets here, into a +// types.ADCExecutionServerAddrError. // -// The ADC sidecar outlives the controller process, so the baseline it holds for a cacheKey -// may be one an earlier leadership term left behind. It is re-derived from the data plane -// the first time this term syncs the key, before anything can be pushed from it, and only -// a sync ADC accepts settles the question. -// -// Rebuilding on leader acquisition covers where staleness comes from. The safety net covers -// what it cannot foresee -- another writer on this data plane, a desync no leadership change -// explains -- and a conf_version the data plane refuses is the only way any of that shows -// itself. Re-read the data plane and push again. -func (c *Client) push(ctx context.Context, config adctypes.Config, resources *adctypes.Resources, labels map[string]string, resourceTypes []string) ([]types.ADCExecutionError, error) { - standalone := config.BackendType == backendAPISIXStandalone - config.BypassCache = standalone && !c.baselineIsCurrent(config.Name) - - err := c.executor.Execute(ctx, config, resources, labels, resourceTypes) - - var alsoReport []types.ADCExecutionError - if standalone && !config.BypassCache && isConfVersionRejection(err) { - c.log.Info("data plane rejected a stale conf_version, rebuilding the ADC baseline", - "config", config.Name, "error", err.Error()) - // Keep the rejection visible even when the sync recovers. The rebuild is not rate - // limited, so a rejection on every sync -- someone else writing to this data plane -- - // turns every sync into a full fetch and diff, and this counter is what says so. - pkgmetrics.RecordExecutionError(config.Name, "conf_version_conflict") - - config.BypassCache = true - retryErr := c.executor.Execute(ctx, config, resources, labels, resourceTypes) - - // Report the rejection as well. On its own a failed rebuild says nothing about what it - // was rebuilding for, and it is the rejection that names the cause -- an ADC server too - // old to know bypassCache, say, answers with a schema error that points nowhere near - // it. Unless the rebuild was rejected the same way, in which case saying it twice only - // pads the status message. - var rejected types.ADCExecutionError - if retryErr != nil && retryErr.Error() != err.Error() && errors.As(err, &rejected) { - alsoReport = append(alsoReport, rejected) - } - err = retryErr - } - - // Only a sync ADC accepted proves its baseline is now derived from the data plane. - if err == nil && config.BypassCache { - c.markBaselineCurrent(config.Name) +// It never retries. A caller that wants ADC to rebuild its diff baseline from the data +// plane (see IsConfVersionRejection) sets in.Config.BypassCache and calls again itself, +// this package keeps no state across calls to base that decision on. +func (c *Client) Sync(ctx context.Context, in SyncInput) error { + if in.Resources == nil { + return nil } - return alsoReport, err -} - -func (c *Client) syncOne(ctx context.Context, in SyncInput) error { c.log.V(1).Info("syncing resources", "input", in) - var errs types.ADCExecutionErrors - config := in.Config if config.BackendType == "" { config.BackendType = c.defaultMode @@ -298,33 +182,22 @@ func (c *Client) syncOne(ctx context.Context, in SyncInput) error { resourceType = "all" } - alsoReport, err := c.push(ctx, config, in.Resources, in.Labels, in.ResourceTypes) - errs.Errors = append(errs.Errors, alsoReport...) + err := c.executor.Execute(ctx, config, in.Resources, in.Labels, in.ResourceTypes) duration := time.Since(startTime).Seconds() - status := adctypes.StatusSuccess if err != nil { status = "failure" c.log.Error(err, "failed to sync with ADC", "config", config) - var execErr types.ADCExecutionError + errorType := "unknown" + var execErr types.ADCExecutionServerAddrError if errors.As(err, &execErr) { - errs.Errors = append(errs.Errors, execErr) - pkgmetrics.RecordExecutionError(config.Name, execErr.Name) - } else { - errs.Errors = append(errs.Errors, types.ADCExecutionError{ - Name: config.Name, - FailedErrors: []types.ADCExecutionServerAddrError{{Err: err.Error()}}, - }) - pkgmetrics.RecordExecutionError(config.Name, "unknown") + errorType = "sync_failed" } + pkgmetrics.RecordExecutionError(config.Name, errorType) } - pkgmetrics.RecordSyncDuration(config.Name, resourceType, status, duration) - if len(errs.Errors) > 0 { - return errs - } - return nil + return err } diff --git a/internal/adc/client/executor.go b/internal/adc/client/executor.go index 6f02433b..0b0aa8a7 100644 --- a/internal/adc/client/executor.go +++ b/internal/adc/client/executor.go @@ -42,7 +42,11 @@ const ( pathSync = "/sync" pathValidate = "/validate" - backendAPISIXStandalone = "apisix-standalone" + // BackendAPISIXStandalone is the one backend type this package resolves a + // multi-address ServerAddrs into a single joined sync target for. It is exported so + // apisixProvider, which owns the conf_version rebuild decision, can recognize the + // same backend type without repeating the string. + BackendAPISIXStandalone = "apisix-standalone" ) type ADCExecutor interface { @@ -150,38 +154,44 @@ func (e *HTTPADCExecutor) Validate(ctx context.Context, config adctypes.Config, return e.runHTTPValidate(ctx, config, resources, labels, resourceTypes) } -// runHTTPSync performs HTTP sync to ADC Server for each server address +// runHTTPSync sends config's sync to the single address it targets. Deciding how many +// addresses a GatewayProxy has (and so how many syncs to send) belongs to the caller +// that built config.ServerAddrs; this only resolves the list it was handed: for +// apisix-standalone, ADC treats every entry as one logical destination, so they are +// joined with commas into one target; every other backend type uses the first entry +// only, since a GatewayProxy is expected to resolve to one address there even though +// nothing enforces it yet. +// +// The error, if any, is the parsed per-address failure. This package never decides +// whether to retry it, callers interpret it and ask again if they choose to. func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context, config adctypes.Config, resources *adctypes.Resources, labels map[string]string, resourceTypes []string) error { - var execErrs = types.ADCExecutionError{ - Name: config.Name, + addr := syncTargetAddr(config) + if addr == "" { + return nil } + e.log.V(1).Info("running http sync", "serverAddr", addr) - serverAddrs := func() []string { - if config.BackendType == backendAPISIXStandalone { - return []string{strings.Join(config.ServerAddrs, ",")} + if err := e.runHTTPSyncForSingleServer(ctx, addr, config, resources, labels, resourceTypes); err != nil { + e.log.Error(err, "failed to run http sync for server", "server", addr) + var execErr types.ADCExecutionServerAddrError + if errors.As(err, &execErr) { + return execErr } - return config.ServerAddrs - }() - e.log.V(1).Info("running http sync", "serverAddrs", serverAddrs) + return types.ADCExecutionServerAddrError{ServerAddr: addr, Err: err.Error()} + } + return nil +} - for _, addr := range serverAddrs { - if err := e.runHTTPSyncForSingleServer(ctx, addr, config, resources, labels, resourceTypes); err != nil { - e.log.Error(err, "failed to run http sync for server", "server", addr) - var execErr types.ADCExecutionServerAddrError - if errors.As(err, &execErr) { - execErrs.FailedErrors = append(execErrs.FailedErrors, execErr) - } else { - execErrs.FailedErrors = append(execErrs.FailedErrors, types.ADCExecutionServerAddrError{ - ServerAddr: addr, - Err: err.Error(), - }) - } - } +// syncTargetAddr resolves config.ServerAddrs into the one address a sync request targets. +// See runHTTPSync. +func syncTargetAddr(config adctypes.Config) string { + if config.BackendType == BackendAPISIXStandalone { + return strings.Join(config.ServerAddrs, ",") } - if len(execErrs.FailedErrors) > 0 { - return execErrs + if len(config.ServerAddrs) == 0 { + return "" } - return nil + return config.ServerAddrs[0] } func (e *HTTPADCExecutor) runHTTPValidate(ctx context.Context, config adctypes.Config, resources *adctypes.Resources, labels map[string]string, resourceTypes []string) error { diff --git a/internal/adc/client/executor_test.go b/internal/adc/client/executor_test.go index 609e0ee1..f5935cfe 100644 --- a/internal/adc/client/executor_test.go +++ b/internal/adc/client/executor_test.go @@ -118,177 +118,6 @@ func rejection(reason string) error { } } -// fakeExecutor answers each Execute call with the next error in errs, and records the -// BypassCache flag it was called with. -type fakeExecutor struct { - errs []error - bypassSeq []bool -} - -func (f *fakeExecutor) Execute(_ context.Context, config adctypes.Config, _ *adctypes.Resources, _ map[string]string, _ []string) error { - f.bypassSeq = append(f.bypassSeq, config.BypassCache) - if len(f.errs) == 0 { - return nil - } - err := f.errs[0] - f.errs = f.errs[1:] - return err -} - -func (f *fakeExecutor) Validate(context.Context, adctypes.Config, *adctypes.Resources, map[string]string, []string) error { - return nil -} - -// newTestClient starts out as a controller that has just been elected: no ADC baseline is -// known to be current, so the first sync of a cacheKey rebuilds it. -func newTestClient(exec ADCExecutor) *Client { - return &Client{ - executor: exec, - rebuiltBaselines: make(map[string]struct{}), - log: logr.Discard(), - } -} - -// afterFirstSync is the state a controller settles into once the first sync of its term -// has landed: the ADC baseline for this cacheKey is known to be derived from the data -// plane, so nothing rebuilds it again unless the data plane says otherwise. -func afterFirstSync(exec ADCExecutor) *Client { - c := newTestClient(exec) - c.markBaselineCurrent(syncTaskCacheKey) - return c -} - -const syncTaskCacheKey = "GatewayProxy/ns/name" - -func newSyncInput() SyncInput { - return SyncInput{ - Name: "GatewayProxy/ns/name-sync", - Config: adctypes.Config{Name: "GatewayProxy/ns/name", BackendType: "apisix-standalone"}, - Resources: &adctypes.Resources{}, - } -} - -func TestClientSyncRebuildsOnceAfterElectionThenReusesTheADCCache(t *testing.T) { - exec := &fakeExecutor{} - c := newTestClient(exec) - - // The sidecar may still hold a baseline from an earlier term, so the first sync of a - // cacheKey re-derives it from the data plane. Once ADC has accepted that sync, its - // baseline is current and later syncs diff against it. - require.NoError(t, c.syncOne(context.Background(), newSyncInput())) - require.NoError(t, c.syncOne(context.Background(), newSyncInput())) - assert.Equal(t, []bool{true, false}, exec.bypassSeq) - - // Winning the election again puts every baseline back in doubt. - c.InvalidateADCCache() - require.NoError(t, c.syncOne(context.Background(), newSyncInput())) - assert.Equal(t, []bool{true, false, true}, exec.bypassSeq) -} - -func TestClientSyncRebuildsAgainWhenTheRebuildWasNotAccepted(t *testing.T) { - // Nothing proves the baseline is current except ADC accepting the sync that rebuilt it. - exec := &fakeExecutor{errs: []error{types.ADCExecutionError{ - Name: "GatewayProxy/ns/name", - FailedErrors: []types.ADCExecutionServerAddrError{{Err: "connection refused"}}, - }}} - c := newTestClient(exec) - - require.Error(t, c.syncOne(context.Background(), newSyncInput())) - require.NoError(t, c.syncOne(context.Background(), newSyncInput())) - - assert.Equal(t, []bool{true, true}, exec.bypassSeq) -} - -func TestClientSyncRebuildsADCBaselineWhenTheDataPlaneRejectsThePush(t *testing.T) { - exec := &fakeExecutor{errs: []error{confVersionError()}} - c := afterFirstSync(exec) - - // The data plane holds a conf_version newer than the one the ADC baseline carries, so - // the push is rejected. The retry rebuilds that baseline from the data plane. - in := newSyncInput() - require.NoError(t, c.syncOne(context.Background(), in)) - - assert.Equal(t, []bool{false, true}, exec.bypassSeq) - - // BypassCache is scoped to the request that recovers from the rejection. Were it to - // survive in the input, it would reach the config ConfigManager holds and turn a - // one-off rebuild into a data plane fetch on every later sync. - assert.False(t, in.Config.BypassCache, - "the rebuild must not write BypassCache back into the input's config") -} - -func TestClientSyncDoesNotRebuildOnUnrelatedFailures(t *testing.T) { - // Re-deriving the baseline answers a stale conf_version and nothing else. A data plane - // that cannot be reached, or one that refuses the configuration on its merits, is not a - // question the baseline can answer, and a rebuild would only cost a fetch. - for name, err := range map[string]error{ - "unreachable": rejection("connection refused"), - "invalid plugins": rejection(`failed to check the configuration of plugin limit-count: value should match only one schema`), - } { - t.Run(name, func(t *testing.T) { - exec := &fakeExecutor{errs: []error{err}} - c := afterFirstSync(exec) - - require.Error(t, c.syncOne(context.Background(), newSyncInput())) - - assert.Equal(t, []bool{false}, exec.bypassSeq) - }) - } -} - -func TestClientSyncRebuildsHoweverTheRejectionIsWorded(t *testing.T) { - // The rejection is recognised by the field it names, not by the sentence around it: - // conf_version is part of the standalone admin API, the wording is APISIX's to change. - exec := &fakeExecutor{errs: []error{rejection("upstreams_conf_version has moved backwards")}} - c := afterFirstSync(exec) - - require.NoError(t, c.syncOne(context.Background(), newSyncInput())) - - assert.Equal(t, []bool{false, true}, exec.bypassSeq) -} - -func TestClientSyncDoesNotRebuildOutsideStandalone(t *testing.T) { - // conf_version, and the whole notion of a version the data plane can refuse, only - // exists in standalone mode. - exec := &fakeExecutor{errs: []error{confVersionError()}} - c := afterFirstSync(exec) - - in := newSyncInput() - in.Config = adctypes.Config{Name: "GatewayProxy/ns/name", BackendType: "apisix"} - require.Error(t, c.syncOne(context.Background(), in)) - - assert.Equal(t, []bool{false}, exec.bypassSeq) -} - -func TestClientSyncSurfacesErrorWhenRebuildFails(t *testing.T) { - // An ADC server older than 0.27.0 answers the rebuild with a schema error, which on - // its own points nowhere near the cause. - exec := &fakeExecutor{errs: []error{confVersionError(), rejection(`unrecognized key "bypassCache"`)}} - c := afterFirstSync(exec) - - err := c.syncOne(context.Background(), newSyncInput()) - - require.Error(t, err, "a rebuild that still fails must not be swallowed") - assert.Equal(t, []bool{false, true}, exec.bypassSeq, "the rebuild is attempted once, not in a loop") - assert.Contains(t, err.Error(), "conf_version must be greater than or equal to", - "the rejection that triggered the rebuild must stay in the reported error") - assert.Contains(t, err.Error(), `unrecognized key "bypassCache"`, - "so must the reason the rebuild itself failed") -} - -func TestClientSyncDoesNotReportTheSameRejectionTwice(t *testing.T) { - // Someone else keeps writing to this data plane, so the rebuilt baseline is stale again - // by the time it is pushed. Reporting that one rejection twice only pads the status. - exec := &fakeExecutor{errs: []error{confVersionError(), confVersionError()}} - c := afterFirstSync(exec) - - err := c.syncOne(context.Background(), newSyncInput()) - - var execErrs types.ADCExecutionErrors - require.ErrorAs(t, err, &execErrs) - assert.Len(t, execErrs.Errors, 1) -} - func httpResponse(status int, body string) *http.Response { return &http.Response{StatusCode: status, Body: io.NopCloser(strings.NewReader(body))} } @@ -510,10 +339,10 @@ func TestHandleHTTPResponseAppliedSucceeds(t *testing.T) { } func TestIsConfVersionRejection(t *testing.T) { - assert.False(t, isConfVersionRejection(nil)) - assert.False(t, isConfVersionRejection(errors.New("context deadline exceeded"))) - assert.False(t, isConfVersionRejection(rejection("connection refused"))) - assert.True(t, isConfVersionRejection(confVersionError())) - assert.True(t, isConfVersionRejection(rejection("routes_conf_version has moved backwards")), + assert.False(t, IsConfVersionRejection(nil)) + assert.False(t, IsConfVersionRejection(errors.New("context deadline exceeded"))) + assert.False(t, IsConfVersionRejection(rejection("connection refused"))) + assert.True(t, IsConfVersionRejection(confVersionError())) + assert.True(t, IsConfVersionRejection(rejection("routes_conf_version has moved backwards")), "the field is what names the rejection, not the sentence") } diff --git a/internal/provider/apisix/provider.go b/internal/provider/apisix/provider.go index 45a0179a..493aa2c4 100644 --- a/internal/provider/apisix/provider.go +++ b/internal/provider/apisix/provider.go @@ -43,6 +43,7 @@ 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 ( @@ -72,6 +73,13 @@ 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{} + updater status.Updater statusUpdateMap map[types.NamespacedNameKind][]string @@ -101,17 +109,18 @@ func New(log logr.Logger, updater status.Updater, readier readiness.ReadinessMan configManager := common.NewConfigManager[types.NamespacedNameKind, adctypes.Config]() return &apisixProvider{ - client: cli, - store: store, - configManager: configManager, - debugProvider: common.NewADCDebugProvider(store, configManager), - syncLocks: newKeyedMutex(), - Options: o, - translator: translator.NewTranslator(log, o.ListenerPortMatchMode), - updater: updater, - readier: readier, - syncCh: make(chan struct{}, 1), - log: logger, + client: cli, + store: store, + configManager: configManager, + debugProvider: common.NewADCDebugProvider(store, configManager), + syncLocks: newKeyedMutex(), + rebuiltBaselines: make(map[string]struct{}), + Options: o, + translator: translator.NewTranslator(log, o.ListenerPortMatchMode), + updater: updater, + readier: readier, + syncCh: make(chan struct{}, 1), + log: logger, }, nil } @@ -318,6 +327,35 @@ 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 @@ -335,8 +373,74 @@ func (d *apisixProvider) syncConfigNow( if err != nil { return types.ADCExecutionErrors{}, err } - failedMap, err := d.client.Sync(ctx, []adcclient.SyncInput{input}) - return failedMap[name], err + execErrs := d.pushConfig(ctx, input) + if len(execErrs.Errors) > 0 { + return execErrs, execErrs + } + 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. +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)) + } + } + + // 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 { + execErrs.Errors = append(execErrs.Errors, toADCExecutionError(input.Name, err)) + } + return execErrs +} + +// toADCExecutionError shapes one sync error into the per-config form status reporting +// consumes. A parsed per-server error travels through with its structured detail intact; +// anything else becomes a bare message. +func toADCExecutionError(name string, err error) types.ADCExecutionError { + var addrErr types.ADCExecutionServerAddrError + if errors.As(err, &addrErr) { + return types.ADCExecutionError{Name: name, FailedErrors: []types.ADCExecutionServerAddrError{addrErr}} + } + return types.ADCExecutionError{Name: name, FailedErrors: []types.ADCExecutionServerAddrError{{Err: err.Error()}}} } // syncEvictedConfigsNow pushes an empty resource set for each of the given configs @@ -386,7 +490,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.client.InvalidateADCCache() + d.invalidateBaselineCache() 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 edd182a8..599de304 100644 --- a/internal/provider/apisix/provider_test.go +++ b/internal/provider/apisix/provider_test.go @@ -56,12 +56,13 @@ func newTestProvider(t *testing.T) *apisixProvider { cli, err := adcclient.New(logr.Discard(), ProviderTypeAPISIX, time.Second) require.NoError(t, err) return &apisixProvider{ - client: cli, - store: cache.NewStore(logr.Discard()), - configManager: common.NewConfigManager[types.NamespacedNameKind, adctypes.Config](), - syncLocks: newKeyedMutex(), - syncCh: make(chan struct{}, 1), - log: logr.Discard(), + client: cli, + store: cache.NewStore(logr.Discard()), + configManager: common.NewConfigManager[types.NamespacedNameKind, adctypes.Config](), + syncLocks: newKeyedMutex(), + rebuiltBaselines: make(map[string]struct{}), + 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 new file mode 100644 index 00000000..f53bc120 --- /dev/null +++ b/internal/provider/apisix/sync_baseline_test.go @@ -0,0 +1,223 @@ +// 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 apisix + +import ( + "context" + "encoding/json" + "net/http" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + adctypes "github.com/apache/apisix-ingress-controller/api/adc" + adcclient "github.com/apache/apisix-ingress-controller/internal/adc/client" +) + +// The ADC diff baseline lives on apisixProvider now: it decides when ADC's cached view of +// a data plane cannot be trusted (BypassCache), retries a stale-conf_version rejection +// once against a rebuilt baseline, and records which cacheKeys this leadership term has +// already rebuilt. The adc client only sends one request and parses the reply. These +// exercise that decision through the real HTTP path. + +type adcResp struct { + status int + body any +} + +func respOK() adcResp { + return adcResp{status: http.StatusOK, body: adctypes.SyncResult{Status: adctypes.StatusSuccess}} +} + +func respRejected(reason string) adcResp { + return adcResp{ + status: http.StatusUnprocessableEntity, + body: adctypes.SyncResult{ + Status: "all_failed", + Failed: []adctypes.SyncStatus{{Reason: reason}}, + }, + } +} + +func respConfVersionRejected() adcResp { + return respRejected("upstreams_conf_version must be greater than or equal to (1779434128737)") +} + +// scriptedADC stands up a mock ADC server that answers each request with the next +// response in the script (repeating the last once the script runs out), and returns a +// snapshot func for the requests it received. +func scriptedADC(t *testing.T, script ...adcResp) func() []adcclient.ADCServerRequest { + t.Helper() + var mu sync.Mutex + var got []adcclient.ADCServerRequest + withMockADCServer(t, func(w http.ResponseWriter, r *http.Request) { + var req adcclient.ADCServerRequest + require.NoError(t, json.NewDecoder(r.Body).Decode(&req)) + mu.Lock() + i := len(got) + got = append(got, req) + mu.Unlock() + resp := script[min(i, len(script)-1)] + w.WriteHeader(resp.status) + if resp.body != nil { + _ = json.NewEncoder(w).Encode(resp.body) + } + }) + return func() []adcclient.ADCServerRequest { + mu.Lock() + defer mu.Unlock() + return append([]adcclient.ADCServerRequest(nil), got...) + } +} + +func standaloneInput(name string) adcclient.SyncInput { + return adcclient.SyncInput{ + Name: name, + Config: adctypes.Config{ + Name: name, + BackendType: adcclient.BackendAPISIXStandalone, + ServerAddrs: []string{"http://apisix:9180"}, + }, + Resources: &adctypes.Resources{}, + } +} + +func bypassSeq(reqs []adcclient.ADCServerRequest) []bool { + seq := make([]bool, len(reqs)) + for i, req := range reqs { + seq[i] = req.Task.Opts.BypassCache + } + return seq +} + +func TestPushRebuildsBaselineOncePerTermThenReusesIt(t *testing.T) { + reqs := scriptedADC(t, respOK()) + d := newTestProvider(t) + in := standaloneInput("proxy") + + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + d.invalidateBaselineCache() + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + + assert.Equal(t, []bool{true, false, true}, bypassSeq(reqs()), + "the first push of a term rebuilds the baseline, later ones reuse it, a new term rebuilds again") +} + +func TestPushRebuildsAgainWhenTheRebuildWasNotAccepted(t *testing.T) { + // Nothing proves the baseline current except ADC accepting the push that rebuilt it. + reqs := scriptedADC(t, respRejected("connection refused"), respOK()) + d := newTestProvider(t) + in := standaloneInput("proxy") + + require.NotEmpty(t, d.pushConfig(context.Background(), in).Errors) + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + + assert.Equal(t, []bool{true, true}, bypassSeq(reqs())) +} + +func TestPushRebuildsBaselineWhenTheDataPlaneRejectsAStaleConfVersion(t *testing.T) { + reqs := scriptedADC(t, respOK(), respConfVersionRejected(), respOK()) + d := newTestProvider(t) + in := standaloneInput("proxy") + + require.Empty(t, d.pushConfig(context.Background(), in).Errors) // settles the baseline + require.Empty(t, d.pushConfig(context.Background(), in).Errors) // rejected, then retried with a rebuild + + assert.Equal(t, []bool{true, false, true}, bypassSeq(reqs())) + assert.False(t, in.Config.BypassCache, + "the rebuild must not write BypassCache back into the caller's input") +} + +func TestPushDoesNotRebuildOnUnrelatedFailures(t *testing.T) { + // Re-deriving the baseline answers a stale conf_version and nothing else. An + // unreachable data plane, or one refusing the configuration on its merits, is not a + // question a rebuild can answer. + for name, reason := range map[string]string{ + "unreachable": "connection refused", + "invalid plugins": `failed to check the configuration of plugin limit-count: value should match only one schema`, + } { + t.Run(name, func(t *testing.T) { + reqs := scriptedADC(t, respOK(), respRejected(reason)) + d := newTestProvider(t) + in := standaloneInput("proxy") + + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + require.NotEmpty(t, d.pushConfig(context.Background(), in).Errors) + + assert.Equal(t, []bool{true, false}, bypassSeq(reqs())) + }) + } +} + +func TestPushDoesNotRebuildOutsideStandalone(t *testing.T) { + // conf_version, and the whole notion of a version the data plane can refuse, only + // exists in standalone mode. + reqs := scriptedADC(t, respConfVersionRejected()) + d := newTestProvider(t) + in := standaloneInput("proxy") + in.Config.BackendType = "apisix" + + require.NotEmpty(t, d.pushConfig(context.Background(), in).Errors) + + assert.Equal(t, []bool{false}, bypassSeq(reqs())) +} + +func TestPushSurfacesBothReasonsWhenTheRebuildAlsoFails(t *testing.T) { + // An ADC server older than 0.27.0 answers the rebuild with a schema error, which on + // its own points nowhere near the cause. The rejection that triggered it must stay. + reqs := scriptedADC(t, respOK(), respConfVersionRejected(), respRejected(`unrecognized key "bypassCache"`)) + d := newTestProvider(t) + in := standaloneInput("proxy") + + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + execErrs := d.pushConfig(context.Background(), in) + + require.NotEmpty(t, execErrs.Errors) + msg := execErrs.Error() + assert.Contains(t, msg, "conf_version must be greater than or equal to") + assert.Contains(t, msg, `unrecognized key "bypassCache"`) + assert.Len(t, reqs(), 3, "the rebuild is attempted once, not in a loop") +} + +func TestPushDoesNotReportTheSameRejectionTwice(t *testing.T) { + // Someone else keeps writing to this data plane, so the rebuilt baseline is stale + // again by the time it is pushed. Reporting that one rejection twice only pads status. + scriptedADC(t, respOK(), respConfVersionRejected(), respConfVersionRejected()) + d := newTestProvider(t) + in := standaloneInput("proxy") + + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + execErrs := d.pushConfig(context.Background(), in) + + assert.Len(t, execErrs.Errors, 1) +} + +func TestPushRebuildsHoweverTheRejectionIsWorded(t *testing.T) { + // The rejection is recognised by the field it names, not the sentence around it. + reqs := scriptedADC(t, respOK(), respRejected("upstreams_conf_version has moved backwards"), respOK()) + d := newTestProvider(t) + in := standaloneInput("proxy") + + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + require.Empty(t, d.pushConfig(context.Background(), in).Errors) + + assert.Equal(t, []bool{true, false, true}, bypassSeq(reqs())) +}
