From 817216d92a6b99e143d9107a176d0c7b8926db24 Mon Sep 17 00:00:00 2001 From: AlinsRan Date: Mon, 13 Jul 2026 16:30:59 +0800 Subject: [PATCH] fix: rebuild the ADC baseline on leader acquisition The ADC server runs as a sidecar that outlives the controller process. Losing the lease terminates the manager container but not the sidecar, so the baseline ADC holds per cacheKey -- the last synced content plus the conf_version it generated -- survives a leadership change. Once the pod is elected again, ADC diffs against that frozen snapshot and carries over conf_versions older than the ones the leader in between has already pushed. APISIX standalone requires *_conf_version to be monotonic and rejects the request: upstreams_conf_version must be greater than or equal to (1779434128737) The data plane validates the configuration as a whole, so every later sync keeps failing and no update lands at all, endpoint updates included. Backends running a single pod then return 502/504 once that pod is rescheduled, while multi-pod backends keep serving on their remaining endpoints. Winning the election is the one moment a baseline from an earlier term can enter the picture, so that is where to answer it: the provider, which only runs once elected, puts every baseline in doubt, and the first sync of each cacheKey re-derives it from the data plane through ADC's bypassCache before anything is pushed from it. Only a sync ADC accepts settles the question, so a rebuild that never landed is attempted again rather than assumed. Prevention cannot cover a desync no leadership change explains -- another writer on this data plane, something we did not foresee -- and the only way any of that shows itself is the data plane refusing a conf_version. So that rejection rebuilds the baseline and pushes once more. It is recognised by the field it names rather than by the sentence around it: conf_version belongs to the standalone admin API, we send those keys ourselves, while the prose around them is APISIX's to reword. This is a net, not the fix: nothing else rebuilds on it, and the reported bug does not come back if it ever stops firing. bypassCache needs ADC >= 0.27.0, so bump ADC_VERSION to 0.27.1. The sidecar image the chart deploys has to move with it, or the rebuild is answered with a schema error instead of healing. The e2e bumps every *_conf_version the data plane holds straight through its Admin API -- what the leader in between would have left behind -- and then expects an Ingress update to still reach the data plane. Nothing restarts during the spec, so what it exercises is the safety net. It reproduces the bug without the fix. --- Makefile | 2 +- api/adc/types.go | 5 + 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 | 219 ++++++++++++++++++++++ test/e2e/scaffold/apisix_deployer.go | 12 +- 9 files changed, 634 insertions(+), 8 deletions(-) create mode 100644 internal/adc/client/executor_test.go create mode 100644 test/e2e/apisix/conf_version.go diff --git a/Makefile b/Makefile index 599a47a9c..0530418f3 100644 --- a/Makefile +++ b/Makefile @@ -30,7 +30,7 @@ KIND_NAME ?= apisix-ingress-cluster KIND_NODE_IMAGE ?= kindest/node:v1.30.0@sha256:047357ac0cfea04663786a612ba1eaba9702bef25227a794b52890dd8bcd692e DASHBOARD_VERSION ?= dev -ADC_VERSION ?= 0.22.1 +ADC_VERSION ?= 0.27.1 DIR := $(shell pwd) diff --git a/api/adc/types.go b/api/adc/types.go index 1c6d46e42..4dde469a1 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/docs/en/latest/developer-guide.md b/docs/en/latest/developer-guide.md index 90fdacc78..6898fba9c 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 8b75236bc..5741181a4 100644 --- a/internal/adc/client/client.go +++ b/internal/adc/client/client.go @@ -50,6 +50,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 } @@ -67,6 +74,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), @@ -75,6 +83,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 @@ -264,6 +320,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) @@ -376,7 +482,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 5664b4f2b..c9add15b5 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 000000000..9e7ee71c0 --- /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 791e1780a..3b0889147 100644 --- a/internal/provider/apisix/provider.go +++ b/internal/provider/apisix/provider.go @@ -255,6 +255,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.readier.WaitReady(ctx, 5*time.Minute) initalSyncDelay := d.InitSyncDelay diff --git a/test/e2e/apisix/conf_version.go b/test/e2e/apisix/conf_version.go new file mode 100644 index 000000000..42735c925 --- /dev/null +++ b/test/e2e/apisix/conf_version.go @@ -0,0 +1,219 @@ +// 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), + }) + + // The Deployer interface is shared with the API7 control plane, which has no + // standalone Admin API, so the client lives on the APISIX deployer alone. + deployer, ok := s.Deployer.(*scaffold.APISIXDeployer) + Expect(ok).To(BeTrue(), "the standalone specs run on the APISIX deployer") + admin := 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 "_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 f68b99e2e..aa6d8fad3 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) } @@ -365,6 +365,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()