This is an automated email from the ASF dual-hosted git repository.
bzp2010 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 cff8979d refactor: clearly boundaries between provider and ADC client
(#2869)
cff8979d is described below
commit cff8979d2b602f95af3e623c0d40bdd08dd1402f
Author: Zeping Bai <[email protected]>
AuthorDate: Fri Sep 11 13:42:42 2026 +0800
refactor: clearly boundaries between provider and ADC client (#2869)
---
internal/adc/client/client.go | 204 ++++------------------
internal/adc/client/executor.go | 114 +++++++------
internal/adc/client/executor_test.go | 207 ++++-------------------
internal/adc/client/standalone_syncer.go | 120 +++++++++++++
internal/provider/apisix/provider.go | 97 +++++++++--
internal/provider/apisix/provider_test.go | 13 +-
internal/provider/apisix/sync_baseline_test.go | 223 +++++++++++++++++++++++++
pkg/metrics/metrics.go | 42 +++++
8 files changed, 603 insertions(+), 417 deletions(-)
diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go
index 0e9e4249..ae86aae2 100644
--- a/internal/adc/client/client.go
+++ b/internal/adc/client/client.go
@@ -16,18 +16,17 @@
// under the License.
// Package client talks to the ADC server: given a fully-prepared sync or
validate
-// request, it translates it to ADC's wire format, sends it, and interprets
the response.
-// It holds no bookkeeping of its own about which Kubernetes resource maps to
which
-// GatewayProxy, or what a GatewayProxy's current resource snapshot is -- that
is AIC's own
-// state, owned by the caller and handed in as input on every call.
+// request, it translates it to ADC's wire format, sends it once, and
interprets the
+// response into a typed error. It holds no bookkeeping of its own: not which
Kubernetes
+// resource maps to which GatewayProxy, not a GatewayProxy's current resource
snapshot,
+// and not whether a data plane's diff baseline can be trusted. All of that is
AIC's own
+// state, owned by the caller, which also owns every decision to retry.
package client
import (
"context"
- "fmt"
"os"
"strings"
- "sync"
"time"
"github.com/go-logr/logr"
@@ -43,13 +42,6 @@ type Client struct {
defaultMode string
- // rebuiltMu guards rebuiltBaselines.
- rebuiltMu sync.Mutex
- // rebuiltBaselines holds the cacheKeys whose ADC baseline this
leadership term has
- // already re-derived from the data plane. A key missing from it is
synced with
- // bypassCache first. See InvalidateADCCache.
- rebuiltBaselines map[string]struct{}
-
log logr.Logger
}
@@ -63,54 +55,23 @@ func New(log logr.Logger, defaultMode string, timeout
time.Duration) (*Client, e
logger.Info("ADC client initialized")
return &Client{
- rebuiltBaselines: make(map[string]struct{}),
- executor: NewHTTPADCExecutor(log, serverURL, timeout),
- log: logger,
- defaultMode: defaultMode,
+ executor: NewHTTPADCExecutor(log, serverURL, timeout),
+ log: logger,
+ defaultMode: defaultMode,
}, nil
}
-// InvalidateADCCache forgets which ADC baselines are known to be current, so
that the
-// next sync of each cacheKey re-derives its baseline from the data plane.
-//
-// It is called on leader acquisition, which is the one moment a stale
baseline can enter
-// the picture. The ADC server is a sidecar that outlives the controller
process: losing
-// the lease terminates the manager container but not the sidecar, so what ADC
holds for a
-// cacheKey -- the last synced content plus the conf_version it generated --
can still be
-// the snapshot this pod left behind in an earlier term, while the leader in
between kept
-// pushing and moved the data plane's conf_version past it. APISIX standalone
requires
-// those versions to be monotonic and refuses the whole configuration
otherwise.
-func (c *Client) InvalidateADCCache() {
- c.rebuiltMu.Lock()
- defer c.rebuiltMu.Unlock()
- clear(c.rebuiltBaselines)
-}
-
-func (c *Client) baselineIsCurrent(cacheKey string) bool {
- c.rebuiltMu.Lock()
- defer c.rebuiltMu.Unlock()
- _, ok := c.rebuiltBaselines[cacheKey]
- return ok
-}
-
-func (c *Client) markBaselineCurrent(cacheKey string) {
- c.rebuiltMu.Lock()
- defer c.rebuiltMu.Unlock()
- c.rebuiltBaselines[cacheKey] = struct{}{}
-}
-
-// isConfVersionRejection reports whether the data plane refused the push
because of a
-// conf_version, which is the one rejection re-deriving the baseline can
answer.
+// IsConfVersionRejection reports whether err is the data plane refusing a
push because
+// its conf_version is behind. That is the one rejection a caller can answer,
by asking
+// ADC to rebuild its diff baseline from the data plane
(SyncInput.Config.BypassCache) and
+// syncing again. This package never makes that decision; it only lets a
caller recognize
+// the case.
//
// It matches the field name, not the sentence. conf_version is part of the
standalone
-// admin API -- we send those keys ourselves -- so any rejection that concerns
it names it,
-// whatever prose APISIX wraps it in. Matching the sentence would tie us to
prose APISIX is
-// free to reword; matching the field only breaks if it renames the API.
-//
-// This backs the safety net, not the fix. A baseline is rebuilt on leader
acquisition,
-// which is where staleness comes from, so if this ever stopped firing the
reported bug
-// would not come back with it.
-func isConfVersionRejection(err error) bool {
+// admin API, callers send those keys themselves, so any rejection that
concerns it names
+// the field whatever prose APISIX wraps it in. Matching the sentence would
tie this to
+// prose APISIX is free to reword; matching the field only breaks if it
renames the API.
+func IsConfVersionRejection(err error) bool {
return err != nil && strings.Contains(err.Error(), confVersionField)
}
@@ -196,135 +157,38 @@ func (in SyncInput) MarshalLog() any {
}
}
-// Sync pushes every given SyncInput to its data plane in one sweep, and
reports the
-// parsed, typed error for each one that failed, keyed by its Name -- an input
whose name
-// is absent from the returned map genuinely succeeded. It never returns a raw
HTTP status
-// or body; every response ADC can send back is already interpreted by the
time it gets
-// here.
-func (c *Client) Sync(ctx context.Context, inputs []SyncInput)
(map[string]types.ADCExecutionErrors, error) {
- if len(inputs) == 0 {
- return nil, nil
- }
- c.log.V(1).Info("syncing resources", "inputs", inputs)
-
- failedMap := map[string]types.ADCExecutionErrors{}
- var failedNames []string
- for _, in := range inputs {
- if in.Resources == nil {
- continue
- }
- if err := c.syncOne(ctx, in); err != nil {
- c.log.Error(err, "failed to sync resources", "name",
in.Name)
- failedNames = append(failedNames, in.Name)
- var execErrs types.ADCExecutionErrors
- if errors.As(err, &execErrs) {
- failedMap[in.Name] = execErrs
- }
- }
- }
-
- var err error
- if len(failedNames) > 0 {
- err = fmt.Errorf("failed to sync %d configs: %s",
- len(failedNames),
- strings.Join(failedNames, ", "))
- }
- return failedMap, err
-}
-
-// push syncs one config through the ADC server, re-deriving the baseline ADC
diffs against
-// whenever that baseline cannot be trusted. Beside the error to report it
returns the ones
-// to report next to it, which a rebuild that failed leaves behind.
+// Sync sends in to its data plane once and returns the parsed, typed error if
the push
+// failed, or nil if it succeeded. It never returns a raw HTTP status or body:
every
+// response ADC can send back is already interpreted by the time it gets here,
into a
+// types.ADCExecutionServerAddrError. The raw status this call's own metrics
are labeled
+// with never leaves this function.
//
-// The ADC sidecar outlives the controller process, so the baseline it holds
for a cacheKey
-// may be one an earlier leadership term left behind. It is re-derived from
the data plane
-// the first time this term syncs the key, before anything can be pushed from
it, and only
-// a sync ADC accepts settles the question.
-//
-// Rebuilding on leader acquisition covers where staleness comes from. The
safety net covers
-// what it cannot foresee -- another writer on this data plane, a desync no
leadership change
-// explains -- and a conf_version the data plane refuses is the only way any
of that shows
-// itself. Re-read the data plane and push again.
-func (c *Client) push(ctx context.Context, config adctypes.Config, resources
*adctypes.Resources, labels map[string]string, resourceTypes []string)
([]types.ADCExecutionError, error) {
- standalone := config.BackendType == backendAPISIXStandalone
- config.BypassCache = standalone && !c.baselineIsCurrent(config.Name)
-
- err := c.executor.Execute(ctx, config, resources, labels, resourceTypes)
-
- var alsoReport []types.ADCExecutionError
- if standalone && !config.BypassCache && isConfVersionRejection(err) {
- c.log.Info("data plane rejected a stale conf_version,
rebuilding the ADC baseline",
- "config", config.Name, "error", err.Error())
- // Keep the rejection visible even when the sync recovers. The
rebuild is not rate
- // limited, so a rejection on every sync -- someone else
writing to this data plane --
- // turns every sync into a full fetch and diff, and this
counter is what says so.
- pkgmetrics.RecordExecutionError(config.Name,
"conf_version_conflict")
-
- config.BypassCache = true
- retryErr := c.executor.Execute(ctx, config, resources, labels,
resourceTypes)
-
- // Report the rejection as well. On its own a failed rebuild
says nothing about what it
- // was rebuilding for, and it is the rejection that names the
cause -- an ADC server too
- // old to know bypassCache, say, answers with a schema error
that points nowhere near
- // it. Unless the rebuild was rejected the same way, in which
case saying it twice only
- // pads the status message.
- var rejected types.ADCExecutionError
- if retryErr != nil && retryErr.Error() != err.Error() &&
errors.As(err, &rejected) {
- alsoReport = append(alsoReport, rejected)
- }
- err = retryErr
- }
-
- // Only a sync ADC accepted proves its baseline is now derived from the
data plane.
- if err == nil && config.BypassCache {
- c.markBaselineCurrent(config.Name)
+// It never retries. A caller that retries (see IsConfVersionRejection) may
call this more
+// than once for what is, from the outside, one logical sync; this call's own
duration and
+// (on failure) error are recorded here regardless, so each underlying HTTP
round trip
+// stays individually visible, but only the caller knows when that logical
sync is
+// actually over, and owns whatever metric reflects that.
+func (c *Client) Sync(ctx context.Context, in SyncInput) error {
+ if in.Resources == nil {
+ return nil
}
- return alsoReport, err
-}
-
-func (c *Client) syncOne(ctx context.Context, in SyncInput) error {
c.log.V(1).Info("syncing resources", "input", in)
- var errs types.ADCExecutionErrors
-
config := in.Config
if config.BackendType == "" {
config.BackendType = c.defaultMode
}
startTime := time.Now()
- resourceType := strings.Join(in.ResourceTypes, ",")
- if resourceType == "" {
- resourceType = "all"
- }
-
- alsoReport, err := c.push(ctx, config, in.Resources, in.Labels,
in.ResourceTypes)
- errs.Errors = append(errs.Errors, alsoReport...)
-
- duration := time.Since(startTime).Seconds()
+ statusCode, err := c.executor.Execute(ctx, config, in.Resources,
in.Labels, in.ResourceTypes)
status := adctypes.StatusSuccess
if err != nil {
status = "failure"
c.log.Error(err, "failed to sync with ADC", "config", config)
-
- var execErr types.ADCExecutionError
- if errors.As(err, &execErr) {
- errs.Errors = append(errs.Errors, execErr)
- pkgmetrics.RecordExecutionError(config.Name,
execErr.Name)
- } else {
- errs.Errors = append(errs.Errors,
types.ADCExecutionError{
- Name: config.Name,
- FailedErrors:
[]types.ADCExecutionServerAddrError{{Err: err.Error()}},
- })
- pkgmetrics.RecordExecutionError(config.Name, "unknown")
- }
+ pkgmetrics.RecordClientSyncError(config.Name, statusCode)
}
+ pkgmetrics.RecordClientSyncDuration(config.Name, status,
time.Since(startTime).Seconds())
- pkgmetrics.RecordSyncDuration(config.Name, resourceType, status,
duration)
-
- if len(errs.Errors) > 0 {
- return errs
- }
- return nil
+ return err
}
diff --git a/internal/adc/client/executor.go b/internal/adc/client/executor.go
index 6f02433b..a6c6f057 100644
--- a/internal/adc/client/executor.go
+++ b/internal/adc/client/executor.go
@@ -42,11 +42,19 @@ const (
pathSync = "/sync"
pathValidate = "/validate"
- backendAPISIXStandalone = "apisix-standalone"
+ // BackendAPISIXStandalone is the one backend type this package
resolves a
+ // multi-address ServerAddrs into a single joined sync target for. It
is exported so
+ // apisixProvider, which owns the conf_version rebuild decision, can
recognize the
+ // same backend type without repeating the string.
+ BackendAPISIXStandalone = "apisix-standalone"
)
type ADCExecutor interface {
- Execute(ctx context.Context, config adctypes.Config, resources
*adctypes.Resources, labels map[string]string, resourceTypes []string) error
+ // Execute performs one sync and returns the raw HTTP status ADC
answered with (0 if
+ // the call never got a response at all, e.g. a transport failure)
alongside the
+ // parsed error, if any. The status code is for Client.Sync's own
metrics; nothing
+ // outside this package ever sees it.
+ Execute(ctx context.Context, config adctypes.Config, resources
*adctypes.Resources, labels map[string]string, resourceTypes []string)
(statusCode int, err error)
Validate(ctx context.Context, config adctypes.Config, resources
*adctypes.Resources, labels map[string]string, resourceTypes []string) error
}
@@ -142,7 +150,7 @@ func NewHTTPADCExecutor(log logr.Logger, serverURL string,
timeout time.Duration
}
// Execute implements the ADCExecutor interface using HTTP calls
-func (e *HTTPADCExecutor) Execute(ctx context.Context, config adctypes.Config,
resources *adctypes.Resources, labels map[string]string, resourceTypes
[]string) error {
+func (e *HTTPADCExecutor) Execute(ctx context.Context, config adctypes.Config,
resources *adctypes.Resources, labels map[string]string, resourceTypes
[]string) (int, error) {
return e.runHTTPSync(ctx, config, resources, labels, resourceTypes)
}
@@ -150,38 +158,65 @@ func (e *HTTPADCExecutor) Validate(ctx context.Context,
config adctypes.Config,
return e.runHTTPValidate(ctx, config, resources, labels, resourceTypes)
}
-// runHTTPSync performs HTTP sync to ADC Server for each server address
-func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context, config
adctypes.Config, resources *adctypes.Resources, labels map[string]string,
resourceTypes []string) error {
- var execErrs = types.ADCExecutionError{
- Name: config.Name,
+// runHTTPSync sends config in one /sync request and returns the parsed
failure, if any.
+// A sync is one request whatever config.ServerAddrs holds: apisix-standalone
joins every
+// entry with commas because ADC addresses them together as one logical
destination
+// (buildHTTPRequest splits them back apart into the request body), every
other backend
+// type takes the first entry only, since a GatewayProxy is expected to
resolve to one
+// address there even though nothing enforces it yet. Deciding how many
addresses a
+// GatewayProxy has belongs to the caller that built config.ServerAddrs.
+//
+// A GatewayProxy with no resolved address is a sync failure for
apisix-standalone (the
+// data plane it configures is unreachable, e.g. scaled to zero), and a no-op
for every
+// other backend type, which pushes per address and so has nothing to push.
+//
+// This package never decides whether to retry the failure; callers interpret
it and ask
+// again if they choose to. The returned status code is 0 whenever no HTTP
response came
+// back at all (no address to sync to, or the request never reached ADC or got
answered).
+func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context, config
adctypes.Config, resources *adctypes.Resources, labels map[string]string,
resourceTypes []string) (int, error) {
+ standalone := config.BackendType == BackendAPISIXStandalone
+ if len(config.ServerAddrs) == 0 {
+ if standalone {
+ return 0, types.ADCExecutionServerAddrError{Err: "no
data plane address to sync apisix-standalone config to"}
+ }
+ return 0, nil
}
- serverAddrs := func() []string {
- if config.BackendType == backendAPISIXStandalone {
- return []string{strings.Join(config.ServerAddrs, ",")}
+ target := syncTargetAddr(config)
+ e.log.V(1).Info("running http sync", "server", target)
+
+ ctx, cancel := context.WithTimeout(ctx, e.httpClient.Timeout)
+ defer cancel()
+
+ req, err := e.buildHTTPRequest(ctx, target, config, labels,
resourceTypes, resources, pathSync)
+ if err != nil {
+ return 0, types.ADCExecutionServerAddrError{ServerAddr: target,
Err: fmt.Sprintf("failed to build HTTP request: %s", err)}
+ }
+
+ resp, err := e.httpClient.Do(req)
+ if err != nil {
+ return 0, types.ADCExecutionServerAddrError{ServerAddr: target,
Err: fmt.Sprintf("failed to send HTTP request: %s", err)}
+ }
+ defer func() {
+ if closeErr := resp.Body.Close(); closeErr != nil {
+ e.log.Error(closeErr, "failed to close response body")
}
- return config.ServerAddrs
}()
- e.log.V(1).Info("running http sync", "serverAddrs", serverAddrs)
- for _, addr := range serverAddrs {
- if err := e.runHTTPSyncForSingleServer(ctx, addr, config,
resources, labels, resourceTypes); err != nil {
- e.log.Error(err, "failed to run http sync for server",
"server", addr)
- var execErr types.ADCExecutionServerAddrError
- if errors.As(err, &execErr) {
- execErrs.FailedErrors =
append(execErrs.FailedErrors, execErr)
- } else {
- execErrs.FailedErrors =
append(execErrs.FailedErrors, types.ADCExecutionServerAddrError{
- ServerAddr: addr,
- Err: err.Error(),
- })
- }
- }
+ if err := e.handleHTTPResponse(resp, target); err != nil {
+ e.log.Error(err, "failed to run http sync", "server", target)
+ return resp.StatusCode, err
}
- if len(execErrs.FailedErrors) > 0 {
- return execErrs
+ return resp.StatusCode, nil
+}
+
+// syncTargetAddr resolves config.ServerAddrs into what one /sync request
targets. Callers
+// must have already handled an empty ServerAddrs (see runHTTPSync).
+func syncTargetAddr(config adctypes.Config) string {
+ if config.BackendType == BackendAPISIXStandalone {
+ return strings.Join(config.ServerAddrs, ",")
}
- return nil
+ return config.ServerAddrs[0]
}
func (e *HTTPADCExecutor) runHTTPValidate(ctx context.Context, config
adctypes.Config, resources *adctypes.Resources, labels map[string]string,
resourceTypes []string) error {
@@ -216,29 +251,6 @@ func (e *HTTPADCExecutor) runHTTPValidate(ctx
context.Context, config adctypes.C
return nil
}
-// runHTTPSyncForSingleServer performs HTTP sync to a single ADC Server
-func (e *HTTPADCExecutor) runHTTPSyncForSingleServer(ctx context.Context,
serverAddr string, config adctypes.Config, resources *adctypes.Resources,
labels map[string]string, resourceTypes []string) error {
- ctx, cancel := context.WithTimeout(ctx, e.httpClient.Timeout)
- defer cancel()
-
- req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels,
resourceTypes, resources, pathSync)
- if err != nil {
- return fmt.Errorf("failed to build HTTP request: %w", err)
- }
-
- resp, err := e.httpClient.Do(req)
- if err != nil {
- return fmt.Errorf("failed to send HTTP request: %w", err)
- }
- defer func() {
- if closeErr := resp.Body.Close(); closeErr != nil {
- e.log.Error(closeErr, "failed to close response body")
- }
- }()
-
- return e.handleHTTPResponse(resp, serverAddr)
-}
-
func (e *HTTPADCExecutor) runHTTPValidateForSingleServer(ctx context.Context,
serverAddr string, config adctypes.Config, resources *adctypes.Resources,
labels map[string]string, resourceTypes []string) error {
ctx, cancel := context.WithTimeout(ctx, e.httpClient.Timeout)
defer cancel()
diff --git a/internal/adc/client/executor_test.go
b/internal/adc/client/executor_test.go
index 609e0ee1..12b0cfb4 100644
--- a/internal/adc/client/executor_test.go
+++ b/internal/adc/client/executor_test.go
@@ -118,177 +118,6 @@ func rejection(reason string) error {
}
}
-// fakeExecutor answers each Execute call with the next error in errs, and
records the
-// BypassCache flag it was called with.
-type fakeExecutor struct {
- errs []error
- bypassSeq []bool
-}
-
-func (f *fakeExecutor) Execute(_ context.Context, config adctypes.Config, _
*adctypes.Resources, _ map[string]string, _ []string) error {
- f.bypassSeq = append(f.bypassSeq, config.BypassCache)
- if len(f.errs) == 0 {
- return nil
- }
- err := f.errs[0]
- f.errs = f.errs[1:]
- return err
-}
-
-func (f *fakeExecutor) Validate(context.Context, adctypes.Config,
*adctypes.Resources, map[string]string, []string) error {
- return nil
-}
-
-// newTestClient starts out as a controller that has just been elected: no ADC
baseline is
-// known to be current, so the first sync of a cacheKey rebuilds it.
-func newTestClient(exec ADCExecutor) *Client {
- return &Client{
- executor: exec,
- rebuiltBaselines: make(map[string]struct{}),
- log: logr.Discard(),
- }
-}
-
-// afterFirstSync is the state a controller settles into once the first sync
of its term
-// has landed: the ADC baseline for this cacheKey is known to be derived from
the data
-// plane, so nothing rebuilds it again unless the data plane says otherwise.
-func afterFirstSync(exec ADCExecutor) *Client {
- c := newTestClient(exec)
- c.markBaselineCurrent(syncTaskCacheKey)
- return c
-}
-
-const syncTaskCacheKey = "GatewayProxy/ns/name"
-
-func newSyncInput() SyncInput {
- return SyncInput{
- Name: "GatewayProxy/ns/name-sync",
- Config: adctypes.Config{Name: "GatewayProxy/ns/name",
BackendType: "apisix-standalone"},
- Resources: &adctypes.Resources{},
- }
-}
-
-func TestClientSyncRebuildsOnceAfterElectionThenReusesTheADCCache(t
*testing.T) {
- exec := &fakeExecutor{}
- c := newTestClient(exec)
-
- // The sidecar may still hold a baseline from an earlier term, so the
first sync of a
- // cacheKey re-derives it from the data plane. Once ADC has accepted
that sync, its
- // baseline is current and later syncs diff against it.
- require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
- require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
- assert.Equal(t, []bool{true, false}, exec.bypassSeq)
-
- // Winning the election again puts every baseline back in doubt.
- c.InvalidateADCCache()
- require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
- assert.Equal(t, []bool{true, false, true}, exec.bypassSeq)
-}
-
-func TestClientSyncRebuildsAgainWhenTheRebuildWasNotAccepted(t *testing.T) {
- // Nothing proves the baseline is current except ADC accepting the sync
that rebuilt it.
- exec := &fakeExecutor{errs: []error{types.ADCExecutionError{
- Name: "GatewayProxy/ns/name",
- FailedErrors: []types.ADCExecutionServerAddrError{{Err:
"connection refused"}},
- }}}
- c := newTestClient(exec)
-
- require.Error(t, c.syncOne(context.Background(), newSyncInput()))
- require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
-
- assert.Equal(t, []bool{true, true}, exec.bypassSeq)
-}
-
-func TestClientSyncRebuildsADCBaselineWhenTheDataPlaneRejectsThePush(t
*testing.T) {
- exec := &fakeExecutor{errs: []error{confVersionError()}}
- c := afterFirstSync(exec)
-
- // The data plane holds a conf_version newer than the one the ADC
baseline carries, so
- // the push is rejected. The retry rebuilds that baseline from the data
plane.
- in := newSyncInput()
- require.NoError(t, c.syncOne(context.Background(), in))
-
- assert.Equal(t, []bool{false, true}, exec.bypassSeq)
-
- // BypassCache is scoped to the request that recovers from the
rejection. Were it to
- // survive in the input, it would reach the config ConfigManager holds
and turn a
- // one-off rebuild into a data plane fetch on every later sync.
- assert.False(t, in.Config.BypassCache,
- "the rebuild must not write BypassCache back into the input's
config")
-}
-
-func TestClientSyncDoesNotRebuildOnUnrelatedFailures(t *testing.T) {
- // Re-deriving the baseline answers a stale conf_version and nothing
else. A data plane
- // that cannot be reached, or one that refuses the configuration on its
merits, is not a
- // question the baseline can answer, and a rebuild would only cost a
fetch.
- for name, err := range map[string]error{
- "unreachable": rejection("connection refused"),
- "invalid plugins": rejection(`failed to check the configuration
of plugin limit-count: value should match only one schema`),
- } {
- t.Run(name, func(t *testing.T) {
- exec := &fakeExecutor{errs: []error{err}}
- c := afterFirstSync(exec)
-
- require.Error(t, c.syncOne(context.Background(),
newSyncInput()))
-
- assert.Equal(t, []bool{false}, exec.bypassSeq)
- })
- }
-}
-
-func TestClientSyncRebuildsHoweverTheRejectionIsWorded(t *testing.T) {
- // The rejection is recognised by the field it names, not by the
sentence around it:
- // conf_version is part of the standalone admin API, the wording is
APISIX's to change.
- exec := &fakeExecutor{errs: []error{rejection("upstreams_conf_version
has moved backwards")}}
- c := afterFirstSync(exec)
-
- require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
-
- assert.Equal(t, []bool{false, true}, exec.bypassSeq)
-}
-
-func TestClientSyncDoesNotRebuildOutsideStandalone(t *testing.T) {
- // conf_version, and the whole notion of a version the data plane can
refuse, only
- // exists in standalone mode.
- exec := &fakeExecutor{errs: []error{confVersionError()}}
- c := afterFirstSync(exec)
-
- in := newSyncInput()
- in.Config = adctypes.Config{Name: "GatewayProxy/ns/name", BackendType:
"apisix"}
- require.Error(t, c.syncOne(context.Background(), in))
-
- assert.Equal(t, []bool{false}, exec.bypassSeq)
-}
-
-func TestClientSyncSurfacesErrorWhenRebuildFails(t *testing.T) {
- // An ADC server older than 0.27.0 answers the rebuild with a schema
error, which on
- // its own points nowhere near the cause.
- exec := &fakeExecutor{errs: []error{confVersionError(),
rejection(`unrecognized key "bypassCache"`)}}
- c := afterFirstSync(exec)
-
- err := c.syncOne(context.Background(), newSyncInput())
-
- require.Error(t, err, "a rebuild that still fails must not be
swallowed")
- assert.Equal(t, []bool{false, true}, exec.bypassSeq, "the rebuild is
attempted once, not in a loop")
- assert.Contains(t, err.Error(), "conf_version must be greater than or
equal to",
- "the rejection that triggered the rebuild must stay in the
reported error")
- assert.Contains(t, err.Error(), `unrecognized key "bypassCache"`,
- "so must the reason the rebuild itself failed")
-}
-
-func TestClientSyncDoesNotReportTheSameRejectionTwice(t *testing.T) {
- // Someone else keeps writing to this data plane, so the rebuilt
baseline is stale again
- // by the time it is pushed. Reporting that one rejection twice only
pads the status.
- exec := &fakeExecutor{errs: []error{confVersionError(),
confVersionError()}}
- c := afterFirstSync(exec)
-
- err := c.syncOne(context.Background(), newSyncInput())
-
- var execErrs types.ADCExecutionErrors
- require.ErrorAs(t, err, &execErrs)
- assert.Len(t, execErrs.Errors, 1)
-}
-
func httpResponse(status int, body string) *http.Response {
return &http.Response{StatusCode: status, Body:
io.NopCloser(strings.NewReader(body))}
}
@@ -466,6 +295,32 @@ func
TestHandleHTTPResponse422CarriesBothFailedAndEndpointStatusesButErrPicksFai
require.Len(t, addrErr.EndpointStatuses, 2)
}
+func TestRunHTTPSyncFailsForAStandaloneConfigWithNoAddress(t *testing.T) {
+ // A GatewayProxy resolving to no data plane address (scaled to zero,
say) must sync
+ // fail for apisix-standalone, which pushes the whole config to one
destination, not
+ // pass as a no-op.
+ e := &HTTPADCExecutor{log: logr.Discard()}
+
+ statusCode, err := e.runHTTPSync(context.Background(),
+ adctypes.Config{Name: "gw", BackendType:
BackendAPISIXStandalone}, &adctypes.Resources{}, nil, nil)
+
+ var addrErr types.ADCExecutionServerAddrError
+ require.ErrorAs(t, err, &addrErr)
+ assert.Contains(t, addrErr.Err, "no data plane address")
+ assert.Zero(t, statusCode, "no HTTP response was ever involved")
+}
+
+func TestRunHTTPSyncNoOpsForANonStandaloneConfigWithNoAddress(t *testing.T) {
+ // Every other backend type pushes per address, so no address is
nothing to push.
+ e := &HTTPADCExecutor{log: logr.Discard()}
+
+ statusCode, err := e.runHTTPSync(context.Background(),
+ adctypes.Config{Name: "gw", BackendType: "apisix"},
&adctypes.Resources{}, nil, nil)
+
+ assert.NoError(t, err)
+ assert.Zero(t, statusCode)
+}
+
func TestDistinctReasonsJoinsWithoutDuplicates(t *testing.T) {
reasons := distinctReasons([]adctypes.SyncStatus{
{Reason: "unknown plugin foo"},
@@ -510,10 +365,10 @@ func TestHandleHTTPResponseAppliedSucceeds(t *testing.T) {
}
func TestIsConfVersionRejection(t *testing.T) {
- assert.False(t, isConfVersionRejection(nil))
- assert.False(t, isConfVersionRejection(errors.New("context deadline
exceeded")))
- assert.False(t, isConfVersionRejection(rejection("connection refused")))
- assert.True(t, isConfVersionRejection(confVersionError()))
- assert.True(t, isConfVersionRejection(rejection("routes_conf_version
has moved backwards")),
+ assert.False(t, IsConfVersionRejection(nil))
+ assert.False(t, IsConfVersionRejection(errors.New("context deadline
exceeded")))
+ assert.False(t, IsConfVersionRejection(rejection("connection refused")))
+ assert.True(t, IsConfVersionRejection(confVersionError()))
+ assert.True(t, IsConfVersionRejection(rejection("routes_conf_version
has moved backwards")),
"the field is what names the rejection, not the sentence")
}
diff --git a/internal/adc/client/standalone_syncer.go
b/internal/adc/client/standalone_syncer.go
new file mode 100644
index 00000000..87cd0291
--- /dev/null
+++ b/internal/adc/client/standalone_syncer.go
@@ -0,0 +1,120 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package client
+
+import (
+ "context"
+ "sync"
+
+ "github.com/go-logr/logr"
+
+ pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
+)
+
+// StandaloneSyncer drives apisix-standalone's diff-baseline recovery on top
of the
+// one-shot Client. It is only for apisix-standalone: no other backend type
keeps a
+// conf_version, so no other backend type needs any of this.
+//
+// APISIX standalone keeps a monotonic conf_version per resource type and
refuses a whole
+// push whose version is behind the data plane's. ADC diffs against a cached
baseline to
+// build that push, and the baseline can go stale two ways:
+//
+// - Across a leadership change. The ADC server is a sidecar that outlives
the manager
+// container, so what it holds for a cacheKey can be the snapshot this pod
left behind
+// in an earlier term, while the leader in between moved the data plane's
conf_version
+// past it. InvalidateBaselines on leader acquisition forces the first
push of every
+// cacheKey this term to re-derive its baseline from the data plane.
+// - Within a term, from a desync no leadership change explains, e.g.
another writer on
+// the same data plane. A conf_version the data plane refuses is the only
way that
+// shows itself; Sync answers it with one rebuild-and-retry.
+type StandaloneSyncer struct {
+ client *Client
+ log logr.Logger
+
+ mu sync.Mutex
+ rebuilt map[string]struct{}
+}
+
+func NewStandaloneSyncer(c *Client, log logr.Logger) *StandaloneSyncer {
+ return &StandaloneSyncer{
+ client: c,
+ log: log.WithName("standalone-syncer"),
+ rebuilt: make(map[string]struct{}),
+ }
+}
+
+// InvalidateBaselines forgets every rebuilt-this-term record, so the next
push of each
+// cacheKey re-derives its baseline. Call on leader acquisition.
+func (s *StandaloneSyncer) InvalidateBaselines() {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ clear(s.rebuilt)
+}
+
+func (s *StandaloneSyncer) isCurrent(cacheKey string) bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ _, ok := s.rebuilt[cacheKey]
+ return ok
+}
+
+func (s *StandaloneSyncer) markCurrent(cacheKey string) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.rebuilt[cacheKey] = struct{}{}
+}
+
+// Sync pushes in, rebuilding ADC's baseline first if this term has not yet
pushed this
+// cacheKey, and once more if the data plane rejects the push over a stale
conf_version.
+//
+// It returns every error worth reporting, so 0, 1, or 2 of them: the final
failure, plus
+// the conf_version rejection that triggered a rebuild which then failed for a
different
+// reason (on its own that failure points nowhere near its cause, e.g. an ADC
server too
+// old to know bypassCache answers with a schema error). BypassCache is scoped
to the call
+// that recovers from a rejection and never written back into in.
+func (s *StandaloneSyncer) Sync(ctx context.Context, in SyncInput) []error {
+ in.Config.BypassCache = !s.isCurrent(in.Name)
+ err := s.client.Sync(ctx, in)
+
+ var report []error
+ if !in.Config.BypassCache && IsConfVersionRejection(err) {
+ s.log.Info("data plane rejected a stale conf_version,
rebuilding the ADC baseline",
+ "config", in.Name, "error", err.Error())
+ // The rebuild is not rate limited, so a rejection on every
push (someone else
+ // writing to this data plane) turns every push into a full
fetch and diff, and
+ // this counter is what says so.
+ pkgmetrics.RecordExecutionError(in.Name,
"conf_version_conflict")
+
+ rejection := err
+ in.Config.BypassCache = true
+ err = s.client.Sync(ctx, in)
+
+ if err != nil && err.Error() != rejection.Error() {
+ report = append(report, rejection)
+ }
+ }
+
+ // Only a push ADC accepted proves its baseline is now derived from the
data plane.
+ if err == nil && in.Config.BypassCache {
+ s.markCurrent(in.Name)
+ }
+ if err != nil {
+ report = append(report, err)
+ }
+ return report
+}
diff --git a/internal/provider/apisix/provider.go
b/internal/provider/apisix/provider.go
index 45a0179a..3b8139b9 100644
--- a/internal/provider/apisix/provider.go
+++ b/internal/provider/apisix/provider.go
@@ -22,6 +22,7 @@ import (
"errors"
"fmt"
"net/http"
+ "strings"
"sync"
"time"
@@ -43,6 +44,7 @@ import (
"github.com/apache/apisix-ingress-controller/internal/provider/common"
"github.com/apache/apisix-ingress-controller/internal/types"
"github.com/apache/apisix-ingress-controller/internal/utils"
+ pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
)
const (
@@ -72,6 +74,11 @@ type apisixProvider struct {
// snapshot together with pushing it
syncLocks *keyedMutex
+ // standaloneSyncer owns apisix-standalone's ADC diff-baseline
recovery: the
+ // BypassCache decision, the one retry for a stale conf_version, and
the per-term
+ // record of which cacheKeys it has rebuilt. Unused for every other
backend type.
+ standaloneSyncer *adcclient.StandaloneSyncer
+
updater status.Updater
statusUpdateMap map[types.NamespacedNameKind][]string
@@ -101,17 +108,18 @@ func New(log logr.Logger, updater status.Updater, readier
readiness.ReadinessMan
configManager := common.NewConfigManager[types.NamespacedNameKind,
adctypes.Config]()
return &apisixProvider{
- client: cli,
- store: store,
- configManager: configManager,
- debugProvider: common.NewADCDebugProvider(store, configManager),
- syncLocks: newKeyedMutex(),
- Options: o,
- translator: translator.NewTranslator(log,
o.ListenerPortMatchMode),
- updater: updater,
- readier: readier,
- syncCh: make(chan struct{}, 1),
- log: logger,
+ client: cli,
+ store: store,
+ configManager: configManager,
+ debugProvider: common.NewADCDebugProvider(store,
configManager),
+ syncLocks: newKeyedMutex(),
+ standaloneSyncer: adcclient.NewStandaloneSyncer(cli, logger),
+ Options: o,
+ translator: translator.NewTranslator(log,
o.ListenerPortMatchMode),
+ updater: updater,
+ readier: readier,
+ syncCh: make(chan struct{}, 1),
+ log: logger,
}, nil
}
@@ -335,8 +343,69 @@ func (d *apisixProvider) syncConfigNow(
if err != nil {
return types.ADCExecutionErrors{}, err
}
- failedMap, err := d.client.Sync(ctx, []adcclient.SyncInput{input})
- return failedMap[name], err
+ execErrs := d.pushConfig(ctx, input)
+ if len(execErrs.Errors) > 0 {
+ return execErrs, execErrs
+ }
+ return execErrs, nil
+}
+
+// pushConfig sends input to its data plane and shapes whatever failed into
the form
+// status reporting consumes. apisix-standalone goes through standaloneSyncer,
which may
+// rebuild ADC's diff baseline and retry once; every other backend type is a
single
+// one-shot push through the adc client, which never retries.
+//
+// Metrics are recorded here, once per call, around whichever of those two
logical syncs
+// ran: the adc client itself records nothing, since a caller that retries may
drive it
+// more than once for what is, from the outside, one sync attempt, and only
this layer
+// knows when that attempt is actually over.
+func (d *apisixProvider) pushConfig(ctx context.Context, input
adcclient.SyncInput) types.ADCExecutionErrors {
+ backend := input.Config.BackendType
+ if backend == "" {
+ backend = d.DefaultBackendMode
+ }
+
+ startTime := time.Now()
+ resourceType := strings.Join(input.ResourceTypes, ",")
+ if resourceType == "" {
+ resourceType = "all"
+ }
+
+ var errs []error
+ if backend == adcclient.BackendAPISIXStandalone {
+ errs = d.standaloneSyncer.Sync(ctx, input)
+ } else if err := d.client.Sync(ctx, input); err != nil {
+ errs = []error{err}
+ }
+
+ status := adctypes.StatusSuccess
+ if len(errs) > 0 {
+ status = "failure"
+ errorType := "unknown"
+ var addrErr types.ADCExecutionServerAddrError
+ if errors.As(errs[len(errs)-1], &addrErr) {
+ errorType = "sync_failed"
+ }
+ pkgmetrics.RecordExecutionError(input.Name, errorType)
+ }
+ pkgmetrics.RecordSyncDuration(input.Name, resourceType, status,
time.Since(startTime).Seconds())
+
+ var execErrs types.ADCExecutionErrors
+ for _, err := range errs {
+ execErrs.Errors = append(execErrs.Errors,
toADCExecutionError(input.Name, err))
+ }
+ return execErrs
+}
+
+// toADCExecutionError shapes one sync error into the per-config form status
reporting
+// consumes. A parsed per-server error travels through with its structured
detail intact;
+// anything else becomes a bare message.
+func toADCExecutionError(name string, err error) types.ADCExecutionError {
+ var addrErr types.ADCExecutionServerAddrError
+ if errors.As(err, &addrErr) {
+ return types.ADCExecutionError{Name: name, FailedErrors:
[]types.ADCExecutionServerAddrError{addrErr}}
+ }
+ return types.ADCExecutionError{Name: name, FailedErrors:
[]types.ADCExecutionServerAddrError{{Err: err.Error()}}}
}
// syncEvictedConfigsNow pushes an empty resource set for each of the given
configs
@@ -386,7 +455,7 @@ func (d *apisixProvider) Start(ctx context.Context) error {
// one thing that leaves the ADC sidecar holding a baseline from an
earlier term: it
// survives the manager container, the configuration it was derived
from does not.
// Rebuild every baseline from the data plane before syncing from it.
- d.client.InvalidateADCCache()
+ d.standaloneSyncer.InvalidateBaselines()
d.log.Info("starting provider, waiting for readiness")
d.readier.WaitReady(ctx, 5*time.Minute)
diff --git a/internal/provider/apisix/provider_test.go
b/internal/provider/apisix/provider_test.go
index edd182a8..66a91b7f 100644
--- a/internal/provider/apisix/provider_test.go
+++ b/internal/provider/apisix/provider_test.go
@@ -56,12 +56,13 @@ func newTestProvider(t *testing.T) *apisixProvider {
cli, err := adcclient.New(logr.Discard(), ProviderTypeAPISIX,
time.Second)
require.NoError(t, err)
return &apisixProvider{
- client: cli,
- store: cache.NewStore(logr.Discard()),
- configManager:
common.NewConfigManager[types.NamespacedNameKind, adctypes.Config](),
- syncLocks: newKeyedMutex(),
- syncCh: make(chan struct{}, 1),
- log: logr.Discard(),
+ client: cli,
+ store: cache.NewStore(logr.Discard()),
+ configManager:
common.NewConfigManager[types.NamespacedNameKind, adctypes.Config](),
+ syncLocks: newKeyedMutex(),
+ standaloneSyncer: adcclient.NewStandaloneSyncer(cli,
logr.Discard()),
+ syncCh: make(chan struct{}, 1),
+ log: logr.Discard(),
}
}
diff --git a/internal/provider/apisix/sync_baseline_test.go
b/internal/provider/apisix/sync_baseline_test.go
new file mode 100644
index 00000000..1ba98f89
--- /dev/null
+++ b/internal/provider/apisix/sync_baseline_test.go
@@ -0,0 +1,223 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package apisix
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "sync"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ adctypes "github.com/apache/apisix-ingress-controller/api/adc"
+ adcclient
"github.com/apache/apisix-ingress-controller/internal/adc/client"
+)
+
+// The ADC diff baseline lives on apisixProvider now: it decides when ADC's
cached view of
+// a data plane cannot be trusted (BypassCache), retries a stale-conf_version
rejection
+// once against a rebuilt baseline, and records which cacheKeys this
leadership term has
+// already rebuilt. The adc client only sends one request and parses the
reply. These
+// exercise that decision through the real HTTP path.
+
+type adcResp struct {
+ status int
+ body any
+}
+
+func respOK() adcResp {
+ return adcResp{status: http.StatusOK, body: adctypes.SyncResult{Status:
adctypes.StatusSuccess}}
+}
+
+func respRejected(reason string) adcResp {
+ return adcResp{
+ status: http.StatusUnprocessableEntity,
+ body: adctypes.SyncResult{
+ Status: "all_failed",
+ Failed: []adctypes.SyncStatus{{Reason: reason}},
+ },
+ }
+}
+
+func respConfVersionRejected() adcResp {
+ return respRejected("upstreams_conf_version must be greater than or
equal to (1779434128737)")
+}
+
+// scriptedADC stands up a mock ADC server that answers each request with the
next
+// response in the script (repeating the last once the script runs out), and
returns a
+// snapshot func for the requests it received.
+func scriptedADC(t *testing.T, script ...adcResp) func()
[]adcclient.ADCServerRequest {
+ t.Helper()
+ var mu sync.Mutex
+ var got []adcclient.ADCServerRequest
+ withMockADCServer(t, func(w http.ResponseWriter, r *http.Request) {
+ var req adcclient.ADCServerRequest
+ require.NoError(t, json.NewDecoder(r.Body).Decode(&req))
+ mu.Lock()
+ i := len(got)
+ got = append(got, req)
+ mu.Unlock()
+ resp := script[min(i, len(script)-1)]
+ w.WriteHeader(resp.status)
+ if resp.body != nil {
+ _ = json.NewEncoder(w).Encode(resp.body)
+ }
+ })
+ return func() []adcclient.ADCServerRequest {
+ mu.Lock()
+ defer mu.Unlock()
+ return append([]adcclient.ADCServerRequest(nil), got...)
+ }
+}
+
+func standaloneInput() adcclient.SyncInput {
+ return adcclient.SyncInput{
+ Name: "proxy",
+ Config: adctypes.Config{
+ Name: "proxy",
+ BackendType: adcclient.BackendAPISIXStandalone,
+ ServerAddrs: []string{"http://apisix:9180"},
+ },
+ Resources: &adctypes.Resources{},
+ }
+}
+
+func bypassSeq(reqs []adcclient.ADCServerRequest) []bool {
+ seq := make([]bool, len(reqs))
+ for i, req := range reqs {
+ seq[i] = req.Task.Opts.BypassCache
+ }
+ return seq
+}
+
+func TestPushRebuildsBaselineOncePerTermThenReusesIt(t *testing.T) {
+ reqs := scriptedADC(t, respOK())
+ d := newTestProvider(t)
+ in := standaloneInput()
+
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+ d.standaloneSyncer.InvalidateBaselines()
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+
+ assert.Equal(t, []bool{true, false, true}, bypassSeq(reqs()),
+ "the first push of a term rebuilds the baseline, later ones
reuse it, a new term rebuilds again")
+}
+
+func TestPushRebuildsAgainWhenTheRebuildWasNotAccepted(t *testing.T) {
+ // Nothing proves the baseline current except ADC accepting the push
that rebuilt it.
+ reqs := scriptedADC(t, respRejected("connection refused"), respOK())
+ d := newTestProvider(t)
+ in := standaloneInput()
+
+ require.NotEmpty(t, d.pushConfig(context.Background(), in).Errors)
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+
+ assert.Equal(t, []bool{true, true}, bypassSeq(reqs()))
+}
+
+func TestPushRebuildsBaselineWhenTheDataPlaneRejectsAStaleConfVersion(t
*testing.T) {
+ reqs := scriptedADC(t, respOK(), respConfVersionRejected(), respOK())
+ d := newTestProvider(t)
+ in := standaloneInput()
+
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors) //
settles the baseline
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors) //
rejected, then retried with a rebuild
+
+ assert.Equal(t, []bool{true, false, true}, bypassSeq(reqs()))
+ assert.False(t, in.Config.BypassCache,
+ "the rebuild must not write BypassCache back into the caller's
input")
+}
+
+func TestPushDoesNotRebuildOnUnrelatedFailures(t *testing.T) {
+ // Re-deriving the baseline answers a stale conf_version and nothing
else. An
+ // unreachable data plane, or one refusing the configuration on its
merits, is not a
+ // question a rebuild can answer.
+ for name, reason := range map[string]string{
+ "unreachable": "connection refused",
+ "invalid plugins": `failed to check the configuration of plugin
limit-count: value should match only one schema`,
+ } {
+ t.Run(name, func(t *testing.T) {
+ reqs := scriptedADC(t, respOK(), respRejected(reason))
+ d := newTestProvider(t)
+ in := standaloneInput()
+
+ require.Empty(t, d.pushConfig(context.Background(),
in).Errors)
+ require.NotEmpty(t, d.pushConfig(context.Background(),
in).Errors)
+
+ assert.Equal(t, []bool{true, false}, bypassSeq(reqs()))
+ })
+ }
+}
+
+func TestPushDoesNotRebuildOutsideStandalone(t *testing.T) {
+ // conf_version, and the whole notion of a version the data plane can
refuse, only
+ // exists in standalone mode.
+ reqs := scriptedADC(t, respConfVersionRejected())
+ d := newTestProvider(t)
+ in := standaloneInput()
+ in.Config.BackendType = "apisix"
+
+ require.NotEmpty(t, d.pushConfig(context.Background(), in).Errors)
+
+ assert.Equal(t, []bool{false}, bypassSeq(reqs()))
+}
+
+func TestPushSurfacesBothReasonsWhenTheRebuildAlsoFails(t *testing.T) {
+ // An ADC server older than 0.27.0 answers the rebuild with a schema
error, which on
+ // its own points nowhere near the cause. The rejection that triggered
it must stay.
+ reqs := scriptedADC(t, respOK(), respConfVersionRejected(),
respRejected(`unrecognized key "bypassCache"`))
+ d := newTestProvider(t)
+ in := standaloneInput()
+
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+ execErrs := d.pushConfig(context.Background(), in)
+
+ require.NotEmpty(t, execErrs.Errors)
+ msg := execErrs.Error()
+ assert.Contains(t, msg, "conf_version must be greater than or equal to")
+ assert.Contains(t, msg, `unrecognized key "bypassCache"`)
+ assert.Len(t, reqs(), 3, "the rebuild is attempted once, not in a loop")
+}
+
+func TestPushDoesNotReportTheSameRejectionTwice(t *testing.T) {
+ // Someone else keeps writing to this data plane, so the rebuilt
baseline is stale
+ // again by the time it is pushed. Reporting that one rejection twice
only pads status.
+ scriptedADC(t, respOK(), respConfVersionRejected(),
respConfVersionRejected())
+ d := newTestProvider(t)
+ in := standaloneInput()
+
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+ execErrs := d.pushConfig(context.Background(), in)
+
+ assert.Len(t, execErrs.Errors, 1)
+}
+
+func TestPushRebuildsHoweverTheRejectionIsWorded(t *testing.T) {
+ // The rejection is recognised by the field it names, not the sentence
around it.
+ reqs := scriptedADC(t, respOK(), respRejected("upstreams_conf_version
has moved backwards"), respOK())
+ d := newTestProvider(t)
+ in := standaloneInput()
+
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+ require.Empty(t, d.pushConfig(context.Background(), in).Errors)
+
+ assert.Equal(t, []bool{true, false, true}, bypassSeq(reqs()))
+}
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index 4f1b6ff0..7242f46b 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -18,6 +18,8 @@
package metrics
import (
+ "strconv"
+
"github.com/prometheus/client_golang/prometheus"
"sigs.k8s.io/controller-runtime/pkg/metrics"
)
@@ -51,6 +53,33 @@ var (
[]string{"config_name", "error_type"},
)
+ // ADC client Sync call duration histogram. Distinct in scope from
ADCSyncDuration:
+ // that one covers a whole logical sync, which for apisix-standalone
recovering from a
+ // stale conf_version can drive more than one underlying call; this one
is exactly one
+ // Client.Sync call, one HTTP round trip to ADC. A retry made above the
client is a
+ // second, independent sample here.
+ ADCClientSyncDuration = prometheus.NewHistogramVec(
+ prometheus.HistogramOpts{
+ Name:
"apisix_ingress_adc_client_sync_duration_seconds",
+ Help: "Time spent on a single adc client Sync call
(one HTTP round trip to ADC)",
+ Buckets: prometheus.DefBuckets,
+ },
+ []string{"config_name", "status"},
+ )
+
+ // ADC client Sync call errors counter. Same config_name/error_type
label shape as
+ // ADCExecutionErrors, but error_type here is the raw HTTP status ADC
answered the
+ // call with ("0" when the call never got a response at all, e.g. a
transport
+ // failure) rather than a semantic category: this is the client's own
per-call view,
+ // entirely internal to how Client.Sync went, and never leaves the adc
client package.
+ ADCClientSyncErrors = prometheus.NewCounterVec(
+ prometheus.CounterOpts{
+ Name: "apisix_ingress_adc_client_sync_errors",
+ Help: "Total number of adc client Sync call failures,
by the raw ADC HTTP status code",
+ },
+ []string{"config_name", "error_type"},
+ )
+
// Status update channel queue length gauge
StatusUpdateQueueLength = prometheus.NewGauge(
prometheus.GaugeOpts{
@@ -67,6 +96,8 @@ func init() {
ADCSyncDuration,
ADCSyncTotal,
ADCExecutionErrors,
+ ADCClientSyncDuration,
+ ADCClientSyncErrors,
StatusUpdateQueueLength,
)
}
@@ -82,6 +113,17 @@ func RecordExecutionError(configName, errorType string) {
ADCExecutionErrors.WithLabelValues(configName, errorType).Inc()
}
+// RecordClientSyncDuration records the duration of a single adc client Sync
call.
+func RecordClientSyncDuration(configName, status string, duration float64) {
+ ADCClientSyncDuration.WithLabelValues(configName,
status).Observe(duration)
+}
+
+// RecordClientSyncError records a single adc client Sync call failure, by the
raw HTTP
+// status ADC answered with (0 when no response was received at all).
+func RecordClientSyncError(configName string, statusCode int) {
+ ADCClientSyncErrors.WithLabelValues(configName,
strconv.Itoa(statusCode)).Inc()
+}
+
// UpdateStatusQueueLength updates the status update queue length gauge
func UpdateStatusQueueLength(length float64) {
StatusUpdateQueueLength.Set(length)
