diff --git a/Makefile b/Makefile index 599a47a9..0530418f 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 1c6d46e4..4dde469a 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 90fdacc7..6898fba9 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 8b75236b..5741181a 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 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 791e1780..3b088914 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 00000000..42735c92 --- /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 f68b99e2..aa6d8fad 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()