This is an automated email from the ASF dual-hosted git repository.
AlinsRan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix-ingress-controller.git
The following commit(s) were added to refs/heads/master by this push:
new 6fc49d34 fix: rebuild the ADC baseline on leader acquisition (#2785)
6fc49d34 is described below
commit 6fc49d340cf399cdb3e2b300647b68c7da2bfaa4
Author: AlinsRan <[email protected]>
AuthorDate: Sat Jul 18 00:30:00 2026 +0800
fix: rebuild the ADC baseline on leader acquisition (#2785)
---
Makefile | 2 +-
api/adc/types.go | 5 +
config/manager/kustomization.yaml | 2 +-
docs/en/latest/developer-guide.md | 2 +-
internal/adc/client/client.go | 110 ++++++++++++++-
internal/adc/client/executor.go | 19 ++-
internal/adc/client/executor_test.go | 267 +++++++++++++++++++++++++++++++++++
internal/provider/apisix/provider.go | 6 +
test/e2e/apisix/conf_version.go | 215 ++++++++++++++++++++++++++++
test/e2e/scaffold/apisix_deployer.go | 12 +-
test/e2e/scaffold/deployer.go | 2 +
11 files changed, 633 insertions(+), 9 deletions(-)
diff --git a/Makefile b/Makefile
index 2926d6a7..b2668923 100644
--- a/Makefile
+++ b/Makefile
@@ -27,7 +27,7 @@ IMG ?= apache/apisix-ingress-controller:$(IMAGE_TAG)
ENVTEST_K8S_VERSION = 1.30.0
KIND_NAME ?= apisix-ingress-cluster
-ADC_VERSION ?= 0.23.1
+ADC_VERSION ?= 0.27.1
DIR := $(shell pwd)
diff --git a/api/adc/types.go b/api/adc/types.go
index 130d63d0..d4a04794 100644
--- a/api/adc/types.go
+++ b/api/adc/types.go
@@ -789,6 +789,11 @@ type Config struct {
Token string
TlsVerify bool
BackendType string
+
+ // BypassCache makes the ADC server drop the in-memory baseline it
holds for this
+ // cacheKey and re-derive it from the data plane before computing the
diff. It is a
+ // per-request flag set on the sync path, not part of the translated
configuration.
+ BypassCache bool
}
// MarshalJSON implements custom JSON marshaling for adcConfig
diff --git a/config/manager/kustomization.yaml
b/config/manager/kustomization.yaml
index 83a270e3..e8a9cbf9 100644
--- a/config/manager/kustomization.yaml
+++ b/config/manager/kustomization.yaml
@@ -8,4 +8,4 @@ images:
newTag: dev
- name: sidecar
newName: ghcr.io/api7/adc
- newTag: 0.23.1
+ newTag: 0.27.1
diff --git a/docs/en/latest/developer-guide.md
b/docs/en/latest/developer-guide.md
index ad3b3906..9aa6e105 100644
--- a/docs/en/latest/developer-guide.md
+++ b/docs/en/latest/developer-guide.md
@@ -36,7 +36,7 @@ Before you get started make sure you have:
1. Installed [Go 1.23](https://golang.org/dl/) or later
2. A Kubernetes cluster available. We recommend using
[kind](https://kind.sigs.k8s.io/).
3. Installed APISIX in Kubernetes using
[Helm](https://github.com/apache/apisix-helm-chart).
-4. Installed [ADC v0.20.0+](https://github.com/api7/adc/releases)
+4. Installed [ADC v0.27.0+](https://github.com/api7/adc/releases)
## Fork and clone
diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go
index 203d9e14..9c0b8b73 100644
--- a/internal/adc/client/client.go
+++ b/internal/adc/client/client.go
@@ -48,6 +48,13 @@ 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
}
@@ -64,6 +71,7 @@ func New(log logr.Logger, defaultMode string, timeout
time.Duration) (*Client, e
return &Client{
Store: store,
+ rebuiltBaselines: make(map[string]struct{}),
executor: NewHTTPADCExecutor(log, serverURL, timeout),
ConfigManager: configManager,
ADCDebugProvider: common.NewADCDebugProvider(store,
configManager),
@@ -72,6 +80,54 @@ func New(log logr.Logger, defaultMode string, timeout
time.Duration) (*Client, e
}, 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.
+//
+// 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 {
+ return err != nil && strings.Contains(err.Error(), confVersionField)
+}
+
+// confVersionField names the monotonic version APISIX standalone keeps per
resource type
+// (routes_conf_version, upstreams_conf_version, ...) and refuses a push that
moves back.
+const confVersionField = "conf_version"
+
type Task struct {
Key types.NamespacedNameKind
Name string
@@ -262,6 +318,56 @@ func (c *Client) Sync(ctx context.Context)
(map[string]types.ADCExecutionErrors,
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.
+//
+// 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, args
[]string) ([]types.ADCExecutionError, error) {
+ standalone := config.BackendType == backendAPISIXStandalone
+ config.BypassCache = standalone && !c.baselineIsCurrent(config.Name)
+
+ err := c.executor.Execute(ctx, config, args)
+
+ 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, args)
+
+ // 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)
+ }
+ return alsoReport, err
+}
+
func (c *Client) sync(ctx context.Context, task Task) error {
c.log.V(1).Info("syncing resources", "task", task)
@@ -297,7 +403,9 @@ func (c *Client) sync(ctx context.Context, task Task) error
{
config.BackendType = c.defaultMode
}
- err := c.executor.Execute(ctx, config, args)
+ alsoReport, err := c.push(ctx, config, args)
+ errs.Errors = append(errs.Errors, alsoReport...)
+
duration := time.Since(startTime).Seconds()
status := adctypes.StatusSuccess
diff --git a/internal/adc/client/executor.go b/internal/adc/client/executor.go
index 5664b4f2..c9add15b 100644
--- a/internal/adc/client/executor.go
+++ b/internal/adc/client/executor.go
@@ -39,6 +39,11 @@ import (
const (
defaultHTTPADCExecutorAddr = "http://127.0.0.1:3000";
+
+ pathSync = "/sync"
+ pathValidate = "/validate"
+
+ backendAPISIXStandalone = "apisix-standalone"
)
type ADCExecutor interface {
@@ -80,6 +85,11 @@ type ADCServerOpts struct {
IncludeResourceType []string
`json:"includeResourceType,omitempty"`
TlsSkipVerify *bool `json:"tlsSkipVerify,omitempty"`
CacheKey string `json:"cacheKey"`
+ // BypassCache is only accepted by the /sync task of ADC >= 0.27.0.
Both ADC task
+ // schemas reject unknown fields, so omitempty is what keeps every
other request --
+ // /validate, and every sync that is not recovering from a rejection --
byte for byte
+ // what an older ADC server already accepts.
+ BypassCache bool `json:"bypassCache,omitempty"`
}
type ADCValidateResult struct {
@@ -141,7 +151,7 @@ func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context,
config adctypes.Confi
}
serverAddrs := func() []string {
- if config.BackendType == "apisix-standalone" {
+ if config.BackendType == backendAPISIXStandalone {
return []string{strings.Join(config.ServerAddrs, ",")}
}
return config.ServerAddrs
@@ -218,7 +228,7 @@ func (e *HTTPADCExecutor) runHTTPSyncForSingleServer(ctx
context.Context, server
}
// Build HTTP request
- req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types,
resources, http.MethodPut, "/sync")
+ req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types,
resources, http.MethodPut, pathSync)
if err != nil {
return fmt.Errorf("failed to build HTTP request: %w", err)
}
@@ -252,7 +262,7 @@ func (e *HTTPADCExecutor)
runHTTPValidateForSingleServer(ctx context.Context, se
return fmt.Errorf("failed to load resources from file %s: %w",
filePath, err)
}
- req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types,
resources, http.MethodPut, "/validate")
+ req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types,
resources, http.MethodPut, pathValidate)
if err != nil {
return fmt.Errorf("failed to build validate request: %w", err)
}
@@ -326,6 +336,7 @@ func (e *HTTPADCExecutor) loadResourcesFromFile(filePath
string) (*adctypes.Reso
func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr
string, config adctypes.Config, labels map[string]string, types []string,
resources *adctypes.Resources, method string, path string) (*http.Request,
error) {
// Prepare request body
tlsVerify := config.TlsVerify
+ bypassCache := path == pathSync && config.BypassCache
reqBody := ADCServerRequest{
Task: ADCServerTask{
Opts: ADCServerOpts{
@@ -336,6 +347,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx
context.Context, serverAddr strin
IncludeResourceType: types,
TlsSkipVerify: ptr.To(!tlsVerify),
CacheKey: config.Name,
+ BypassCache: bypassCache,
},
Config: *resources,
},
@@ -353,6 +365,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx
context.Context, serverAddr strin
"server", serverAddr,
"mode", config.BackendType,
"cacheKey", config.Name,
+ "bypassCache", bypassCache,
"labelSelector", labels,
"includeResourceType", types,
"tlsSkipVerify", !tlsVerify,
diff --git a/internal/adc/client/executor_test.go
b/internal/adc/client/executor_test.go
new file mode 100644
index 00000000..9e7ee71c
--- /dev/null
+++ b/internal/adc/client/executor_test.go
@@ -0,0 +1,267 @@
+// 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"
+ "errors"
+ "io"
+ "net/http"
+ "testing"
+
+ "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/types"
+)
+
+func TestHTTPADCExecutorBuildHTTPRequestBypassCache(t *testing.T) {
+ e := &HTTPADCExecutor{
+ serverURL: "http://127.0.0.1:3000";,
+ log: logr.Discard(),
+ }
+
+ build := func(config adctypes.Config, path string) (ADCServerOpts,
string) {
+ req, err := e.buildHTTPRequest(context.Background(),
"http://apisix:9180";, config, nil, nil,
+ &adctypes.Resources{}, http.MethodPut, path)
+ require.NoError(t, err)
+ body, err := io.ReadAll(req.Body)
+ require.NoError(t, err)
+ var parsed ADCServerRequest
+ require.NoError(t, json.Unmarshal(body, &parsed))
+ return parsed.Task.Opts, string(body)
+ }
+
+ config := adctypes.Config{Name: "GatewayProxy/ns/name"}
+
+ // A sync that is not recovering from a rejection must stay byte for
byte the request an
+ // ADC server older than 0.27.0 -- which rejects unknown fields --
already accepts.
+ opts, raw := build(config, pathSync)
+ assert.Equal(t, "GatewayProxy/ns/name", opts.CacheKey)
+ assert.NotContains(t, raw, "bypassCache")
+
+ config.BypassCache = true
+ opts, raw = build(config, pathSync)
+ assert.Equal(t, "GatewayProxy/ns/name", opts.CacheKey, "the cacheKey
itself must stay stable")
+ assert.True(t, opts.BypassCache)
+ assert.Contains(t, raw, "bypassCache")
+
+ // The ADC validate task schema rejects unknown fields too, so
bypassCache must never
+ // reach it, even when the config carries the flag.
+ _, raw = build(config, pathValidate)
+ assert.NotContains(t, raw, "bypassCache")
+}
+
+// confVersionError is what a push carrying a conf_version older than the data
plane's
+// comes back as, once the ADC server has relayed the rejection to us.
+func confVersionError() error {
+ return rejection("upstreams_conf_version must be greater than or equal
to (1779434128737)")
+}
+
+func rejection(reason string) error {
+ return types.ADCExecutionError{
+ Name: "GatewayProxy/ns/name",
+ FailedErrors: []types.ADCExecutionServerAddrError{{
+ ServerAddr: "http://apisix:9180";,
+ Err: reason,
+ }},
+ }
+}
+
+// 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, _
[]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, []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 newSyncTask() Task {
+ return Task{
+ Name: "GatewayProxy/ns/name-sync",
+ Configs: map[types.NamespacedNameKind]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.sync(context.Background(), newSyncTask()))
+ require.NoError(t, c.sync(context.Background(), newSyncTask()))
+ assert.Equal(t, []bool{true, false}, exec.bypassSeq)
+
+ // Winning the election again puts every baseline back in doubt.
+ c.InvalidateADCCache()
+ require.NoError(t, c.sync(context.Background(), newSyncTask()))
+ 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.sync(context.Background(), newSyncTask()))
+ require.NoError(t, c.sync(context.Background(), newSyncTask()))
+
+ 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.
+ task := newSyncTask()
+ require.NoError(t, c.sync(context.Background(), task))
+
+ 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 task, it would reach the config the ConfigManager
holds and turn a
+ // one-off rebuild into a data plane fetch on every later sync.
+ assert.False(t, task.Configs[types.NamespacedNameKind{}].BypassCache,
+ "the rebuild must not write BypassCache back into the task
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.sync(context.Background(),
newSyncTask()))
+
+ 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.sync(context.Background(), newSyncTask()))
+
+ 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)
+
+ task := newSyncTask()
+ task.Configs[types.NamespacedNameKind{}] = adctypes.Config{Name:
"GatewayProxy/ns/name", BackendType: "apisix"}
+ require.Error(t, c.sync(context.Background(), task))
+
+ 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.sync(context.Background(), newSyncTask())
+
+ 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.sync(context.Background(), newSyncTask())
+
+ var execErrs types.ADCExecutionErrors
+ require.ErrorAs(t, err, &execErrs)
+ assert.Len(t, execErrs.Errors, 1)
+}
+
+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")),
+ "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 ef6e3fc8..c34f3c70 100644
--- a/internal/provider/apisix/provider.go
+++ b/internal/provider/apisix/provider.go
@@ -251,6 +251,12 @@ func (d *apisixProvider) buildConfig(tctx
*provider.TranslateContext, nnk types.
}
func (d *apisixProvider) Start(ctx context.Context) error {
+ // Start only runs once this pod has won the election, and a leadership
change is the
+ // 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.log.Info("starting provider, waiting for readiness")
d.readier.WaitReady(ctx, 5*time.Minute)
d.log.Info("Ready detected, starting sync loop")
diff --git a/test/e2e/apisix/conf_version.go b/test/e2e/apisix/conf_version.go
new file mode 100644
index 00000000..c28a3148
--- /dev/null
+++ b/test/e2e/apisix/conf_version.go
@@ -0,0 +1,215 @@
+// 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 (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/gavv/httpexpect/v2"
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+
+ "github.com/apache/apisix-ingress-controller/test/e2e/framework"
+ "github.com/apache/apisix-ingress-controller/test/e2e/scaffold"
+)
+
+var _ = Describe("Test ADC cache alignment", Label("networking.k8s.io",
"ingress"), func() {
+ // conf_version only exists in APISIX standalone mode. Bail out while
the tree is
+ // built rather than skipping in a BeforeEach: the scaffold registers
its own
+ // BeforeEach first, so a skip would already have paid for a data plane.
+ if framework.ProviderType != framework.ProviderTypeAPISIXStandalone {
+ return
+ }
+
+ s := scaffold.NewDefaultScaffold()
+
+ const gatewayProxyYaml = `
+apiVersion: apisix.apache.org/v1alpha1
+kind: GatewayProxy
+metadata:
+ name: apisix-proxy-config
+spec:
+ provider:
+ type: ControlPlane
+ controlPlane:
+ endpoints:
+ - %s
+ auth:
+ type: AdminKey
+ adminKey:
+ value: "%s"
+`
+
+ const ingressClassYaml = `
+apiVersion: networking.k8s.io/v1
+kind: IngressClass
+metadata:
+ name: %s
+spec:
+ controller: "%s"
+ parameters:
+ apiGroup: "apisix.apache.org"
+ kind: "GatewayProxy"
+ name: "apisix-proxy-config"
+ namespace: "%s"
+ scope: "Namespace"
+`
+
+ const ingressYaml = `
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+ name: httpbin
+spec:
+ ingressClassName: %s
+ rules:
+ - host: conf-version.example
+ http:
+ paths:
+ - path: %s
+ pathType: Exact
+ backend:
+ service:
+ name: httpbin-service-e2e-test
+ port:
+ number: 80
+`
+
+ // confVersions returns every *_conf_version the data plane currently
holds, read
+ // straight from its Admin API rather than through the controller.
+ confVersions := func(admin *httpexpect.Expect) map[string]int64 {
+ body := admin.GET("/apisix/admin/configs").
+ WithHeader("X-API-KEY", s.AdminKey()).
+ Expect().Status(http.StatusOK).Body().Raw()
+
+ var config map[string]any
+ Expect(json.Unmarshal([]byte(body),
&config)).NotTo(HaveOccurred(), "decoding standalone config")
+
+ versions := map[string]int64{}
+ for key, value := range config {
+ if number, ok := value.(float64); ok &&
strings.HasSuffix(key, "_conf_version") {
+ versions[key] = int64(number)
+ }
+ }
+ return versions
+ }
+
+ // bumpConfVersions rewrites every *_conf_version the data plane holds
to a timestamp
+ // an hour ahead and pushes the configuration back, without going
through the
+ // controller. This is what the leader in between would have left
behind: it pushed
+ // while this pod was on standby, so APISIX now holds versions newer
than the ones the
+ // ADC sidecar generated during this pod's previous term -- and the
sidecar survives a
+ // leadership change, so it still diffs against that older baseline.
+ bumpConfVersions := func(admin *httpexpect.Expect) int64 {
+ body := admin.GET("/apisix/admin/configs").
+ WithHeader("X-API-KEY", s.AdminKey()).
+ Expect().Status(http.StatusOK).Body().Raw()
+
+ var config map[string]any
+ Expect(json.Unmarshal([]byte(body),
&config)).NotTo(HaveOccurred(), "decoding standalone config")
+
+ future := time.Now().Add(time.Hour).UnixMilli()
+ bumped := 0
+ for key := range config {
+ // APISIX echoes its own metadata back, but rejects it
on write.
+ if strings.HasPrefix(strings.ToUpper(key), "X-") {
+ delete(config, key)
+ continue
+ }
+ if strings.HasSuffix(key, "_conf_version") {
+ config[key] = future
+ bumped++
+ }
+ }
+ Expect(bumped).NotTo(BeZero(), "the data plane should already
hold conf_versions")
+
+ admin.PUT("/apisix/admin/configs").
+ WithHeader("X-API-KEY", s.AdminKey()).
+ // APISIX requires a digest and compares it with the
stored one to skip an
+ // update that repeats it; it never recomputes it from
the body.
+ WithHeader("X-Digest",
fmt.Sprintf("e2e-conf-version-bump-%d", future)).
+ WithJSON(config).
+ Expect().Status(http.StatusAccepted)
+
+ return future
+ }
+
+ It("recovers when the data plane holds newer conf_versions than the ADC
cache", func() {
+ By("create GatewayProxy, IngressClass and Ingress")
+ Expect(s.CreateResourceFromString(fmt.Sprintf(gatewayProxyYaml,
s.Deployer.GetAdminEndpoint(), s.AdminKey()))).
+ NotTo(HaveOccurred(), "creating GatewayProxy")
+ Expect(s.CreateResourceFromStringWithNamespace(
+ fmt.Sprintf(ingressClassYaml, s.Namespace(),
s.GetControllerName(), s.Namespace()), "")).
+ NotTo(HaveOccurred(), "creating IngressClass")
+ Expect(s.CreateResourceFromString(fmt.Sprintf(ingressYaml,
s.Namespace(), "/get"))).
+ NotTo(HaveOccurred(), "creating Ingress")
+
+ s.RequestAssert(&scaffold.RequestAssert{
+ Method: "GET",
+ Path: "/get",
+ Host: "conf-version.example",
+ Check: scaffold.WithExpectedStatus(http.StatusOK),
+ })
+
+ admin := s.Deployer.AdminAPIClient()
+
+ By("push newer conf_versions straight into the data plane")
+ future := bumpConfVersions(admin)
+
+ By("update the Ingress and expect the change to reach the data
plane")
+ // The ADC sidecar still diffs against its own baseline, whose
conf_versions are
+ // now older than the ones the data plane holds, and APISIX
rejects the whole
+ // configuration with "<resource>_conf_version must be greater
than or equal to".
+ // The controller has to rebuild that baseline from the data
plane before anything
+ // can land again.
+ Expect(s.CreateResourceFromString(fmt.Sprintf(ingressYaml,
s.Namespace(), "/headers"))).
+ NotTo(HaveOccurred(), "updating Ingress")
+
+ s.RequestAssert(&scaffold.RequestAssert{
+ Method: "GET",
+ Path: "/headers",
+ Host: "conf-version.example",
+ Check: scaffold.WithExpectedStatus(http.StatusOK),
+ })
+ s.RequestAssert(&scaffold.RequestAssert{
+ Method: "GET",
+ Path: "/get",
+ Host: "conf-version.example",
+ Check:
scaffold.WithExpectedStatus(http.StatusNotFound),
+ })
+
+ By("the versions the data plane holds moved past the injected
ones")
+ s.RetryAssertion(func() error {
+ versions := confVersions(admin)
+ if len(versions) == 0 {
+ return errors.New("the data plane reported no
conf_version at all")
+ }
+ for key, version := range versions {
+ if version < future {
+ return fmt.Errorf("%s is %d, expected
it to be at least %d", key, version, future)
+ }
+ }
+ return nil
+ }).ShouldNot(HaveOccurred(), "checking conf_versions")
+ })
+})
diff --git a/test/e2e/scaffold/apisix_deployer.go
b/test/e2e/scaffold/apisix_deployer.go
index 7c4b1954..19f30f36 100644
--- a/test/e2e/scaffold/apisix_deployer.go
+++ b/test/e2e/scaffold/apisix_deployer.go
@@ -23,6 +23,7 @@ import (
"os"
"time"
+ "github.com/gavv/httpexpect/v2"
"github.com/gruntwork-io/terratest/modules/k8s"
. "github.com/onsi/ginkgo/v2" //nolint:staticcheck
. "github.com/onsi/gomega" //nolint:staticcheck
@@ -146,9 +147,8 @@ func (s *APISIXDeployer) AfterEach() {
_, _ = fmt.Fprintln(GinkgoWriter, output)
}
if framework.ProviderType ==
framework.ProviderTypeAPISIXStandalone && s.adminTunnel != nil {
- client := NewClient("http", s.adminTunnel.Endpoint())
reporter := &ErrorReporter{}
- body :=
client.GET("/apisix/admin/configs").WithHeader("X-API-KEY",
s.AdminKey()).WithReporter(reporter).Expect().Body().Raw()
+ body :=
s.AdminAPIClient().GET("/apisix/admin/configs").WithHeader("X-API-KEY",
s.AdminKey()).WithReporter(reporter).Expect().Body().Raw()
_, _ = fmt.Fprintln(GinkgoWriter, "Dumping APISIX
configs:")
_, _ = fmt.Fprintln(GinkgoWriter, body)
}
@@ -359,6 +359,14 @@ func (s *APISIXDeployer) createAdminTunnel(svc
*corev1.Service) (*k8s.Tunnel, er
return adminTunnel, nil
}
+// AdminAPIClient returns a client for the data plane Admin API, reachable
through the
+// tunnel the scaffold forwards to it. In standalone mode it is the only way
to read or
+// write the configuration APISIX actually holds, bypassing the controller.
+func (s *APISIXDeployer) AdminAPIClient() *httpexpect.Expect {
+ Expect(s.adminTunnel).NotTo(BeNil(), "admin tunnel should be created")
+ return NewClient("http", s.adminTunnel.Endpoint())
+}
+
func (s *APISIXDeployer) closeAdminTunnel() {
if s.adminTunnel != nil {
s.adminTunnel.Close()
diff --git a/test/e2e/scaffold/deployer.go b/test/e2e/scaffold/deployer.go
index a1b2e9b0..77557828 100644
--- a/test/e2e/scaffold/deployer.go
+++ b/test/e2e/scaffold/deployer.go
@@ -18,6 +18,7 @@
package scaffold
import (
+ "github.com/gavv/httpexpect/v2"
corev1 "k8s.io/api/core/v1"
)
@@ -34,6 +35,7 @@ type Deployer interface {
CleanupAdditionalGateway(identifier string) error
GetAdminEndpoint(...*corev1.Service) string
GetAdminServiceName() string
+ AdminAPIClient() *httpexpect.Expect
DefaultDataplaneResource() DataplaneResource
}
