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 80112024 refactor: let adc client independent of external state (#2865)
80112024 is described below
commit 80112024e0df674d900a1cb1da831f405179729a
Author: Zeping Bai <[email protected]>
AuthorDate: Thu Sep 10 14:37:39 2026 +0800
refactor: let adc client independent of external state (#2865)
---
internal/adc/client/client.go | 308 ++++++++--------------------
internal/adc/client/executor.go | 126 ++----------
internal/adc/client/executor_test.go | 50 ++---
internal/adc/client/redaction_test.go | 1 -
internal/provider/apisix/keyedmutex.go | 52 +++++
internal/provider/apisix/keyedmutex_test.go | 94 +++++++++
internal/provider/apisix/provider.go | 244 ++++++++++++++++++----
internal/provider/apisix/provider_test.go | 128 +++++++++++-
internal/provider/apisix/status.go | 14 +-
internal/webhook/v1/adc_validation.go | 1 -
pkg/metrics/metrics.go | 16 --
test/e2e/crds/v2/route.go | 1 -
12 files changed, 596 insertions(+), 439 deletions(-)
diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go
index e80d1b27..0e9e4249 100644
--- a/internal/adc/client/client.go
+++ b/internal/adc/client/client.go
@@ -15,11 +15,15 @@
// specific language governing permissions and limitations
// 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.
package client
import (
"context"
- "encoding/json"
"fmt"
"os"
"strings"
@@ -30,22 +34,13 @@ import (
"github.com/pkg/errors"
adctypes "github.com/apache/apisix-ingress-controller/api/adc"
- "github.com/apache/apisix-ingress-controller/internal/adc/cache"
- "github.com/apache/apisix-ingress-controller/internal/provider/common"
"github.com/apache/apisix-ingress-controller/internal/types"
pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
)
type Client struct {
- syncMu sync.RWMutex
- mu sync.Mutex
- *cache.Store
-
executor ADCExecutor
- ConfigManager *common.ConfigManager[types.NamespacedNameKind,
adctypes.Config]
- ADCDebugProvider *common.ADCDebugProvider
-
defaultMode string
// rebuiltMu guards rebuiltBaselines.
@@ -63,18 +58,13 @@ func New(log logr.Logger, defaultMode string, timeout
time.Duration) (*Client, e
if serverURL == "" {
serverURL = defaultHTTPADCExecutorAddr
}
- store := cache.NewStore(log)
- configManager := common.NewConfigManager[types.NamespacedNameKind,
adctypes.Config]()
logger := log.WithName("client")
logger.Info("ADC client initialized")
return &Client{
- Store: store,
rebuiltBaselines: make(map[string]struct{}),
executor: NewHTTPADCExecutor(log, serverURL, timeout),
- ConfigManager: configManager,
- ADCDebugProvider: common.NewADCDebugProvider(store,
configManager),
log: logger,
defaultMode: defaultMode,
}, nil
@@ -128,8 +118,9 @@ func isConfVersionRejection(err error) bool {
// (routes_conf_version, upstreams_conf_version, ...) and refuses a push that
moves back.
const confVersionField = "conf_version"
+// Task is a /validate request: one Kubernetes resource's translated result,
checked
+// against every GatewayProxy config it could target.
type Task struct {
- Key types.NamespacedNameKind
Name string
Labels map[string]string
Configs map[types.NamespacedNameKind]adctypes.Config
@@ -146,7 +137,6 @@ func (t Task) MarshalLog() any {
configNames = append(configNames, cfg.Name)
}
return map[string]any{
- "key": t.Key,
"name": t.Name,
"labels": t.Labels,
"resourceTypes": t.ResourceTypes,
@@ -155,119 +145,17 @@ func (t Task) MarshalLog() any {
}
}
-type StoreDelta struct {
- Deleted map[types.NamespacedNameKind]adctypes.Config
- Applied map[types.NamespacedNameKind]adctypes.Config
-}
-
-func (c *Client) applyStoreChanges(args Task, isDelete bool) (StoreDelta,
error) {
- c.mu.Lock()
- defer c.mu.Unlock()
-
- var delta StoreDelta
-
- if isDelete {
- delta.Deleted = c.ConfigManager.Get(args.Key)
- c.ConfigManager.Delete(args.Key)
- } else {
- deleted := c.ConfigManager.Update(args.Key, args.Configs)
- delta.Deleted = deleted
- delta.Applied = args.Configs
- }
-
- for _, cfg := range delta.Deleted {
- if err := c.Store.Delete(cfg.Name, args.ResourceTypes,
args.Labels); err != nil {
- c.log.Error(err, "store delete failed", "cfg", cfg,
"args", args)
- return StoreDelta{}, errors.Wrap(err,
fmt.Sprintf("store delete failed for config %s", cfg.Name))
- }
- }
-
- for _, cfg := range delta.Applied {
- if err := c.Insert(cfg.Name, args.ResourceTypes,
args.Resources, args.Labels); err != nil {
- c.log.Error(err, "store insert failed", "cfg", cfg,
"args", args)
- return StoreDelta{}, errors.Wrap(err,
fmt.Sprintf("store insert failed for config %s", cfg.Name))
- }
- }
-
- return delta, nil
-}
-
-func (c *Client) applySync(ctx context.Context, args Task, delta StoreDelta)
error {
- c.syncMu.RLock()
- defer c.syncMu.RUnlock()
-
- if len(delta.Deleted) > 0 {
- if err := c.sync(ctx, Task{
- Name: args.Name,
- Labels: args.Labels,
- ResourceTypes: args.ResourceTypes,
- Configs: delta.Deleted,
- }); err != nil {
- c.log.Error(err, "failed to sync deleted configs",
"args", args, "delta", delta)
- }
- }
-
- if len(delta.Applied) > 0 {
- return c.sync(ctx, Task{
- Name: args.Name,
- Labels: args.Labels,
- ResourceTypes: args.ResourceTypes,
- Configs: delta.Applied,
- Resources: args.Resources,
- })
- }
- return nil
-}
-
-func (c *Client) Update(ctx context.Context, args Task) error {
- delta, err := c.applyStoreChanges(args, false)
- if err != nil {
- return err
- }
- return c.applySync(ctx, args, delta)
-}
-
-func (c *Client) UpdateConfig(ctx context.Context, args Task) error {
- _, err := c.applyStoreChanges(args, false)
- return err
-}
-
-func (c *Client) Delete(ctx context.Context, args Task) error {
- delta, err := c.applyStoreChanges(args, true)
- if err != nil {
- return err
- }
- return c.applySync(ctx, args, delta)
-}
-
-// DeleteConfig removes the stored configuration for args.Key and reports what
-// it removed, so callers can skip a data plane sync when the key held nothing.
-func (c *Client) DeleteConfig(ctx context.Context, args Task) (StoreDelta,
error) {
- return c.applyStoreChanges(args, true)
-}
-
func (c *Client) Validate(ctx context.Context, task Task) error {
if len(task.Configs) == 0 || task.Resources == nil {
return nil
}
- fileIOStart := time.Now()
- syncFilePath, cleanup, err := prepareSyncFile(task.Resources)
- if err != nil {
- pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure",
time.Since(fileIOStart).Seconds())
- return err
- }
- pkgmetrics.RecordFileIODuration("prepare_sync_file",
adctypes.StatusSuccess, time.Since(fileIOStart).Seconds())
- defer cleanup()
-
- args2 := BuildADCExecuteArgs(syncFilePath, task.Labels,
task.ResourceTypes)
-
var errs types.ADCValidationErrors
for _, config := range task.Configs {
if config.BackendType == "" {
config.BackendType = c.defaultMode
}
- if err := c.executor.Validate(ctx, config, args2); err != nil {
+ if err := c.executor.Validate(ctx, config, task.Resources,
task.Labels, task.ResourceTypes); err != nil {
var validationErr types.ADCValidationError
if errors.As(err, &validationErr) {
errs.Errors = append(errs.Errors, validationErr)
@@ -283,56 +171,63 @@ func (c *Client) Validate(ctx context.Context, task Task)
error {
return nil
}
-func (c *Client) Sync(ctx context.Context)
(map[string]types.ADCExecutionErrors, error) {
- c.syncMu.Lock()
- defer c.syncMu.Unlock()
- c.log.Info("syncing all resources")
+// SyncInput is one GatewayProxy's complete sync unit. AIC builds it entirely
from its own
+// bookkeeping (which resources target this config, their merged translated
snapshot)
+// before handing it over -- this package never reaches back into AIC's state
to gather
+// anything itself, it only translates, sends, and interprets the response.
+type SyncInput struct {
+ // Name is the cacheKey: the GatewayProxy's own identity.
+ Name string
+ Config adctypes.Config
+ Resources *adctypes.Resources
+ ResourceTypes []string
+ Labels map[string]string
+}
- configs := c.ConfigManager.List()
+// MarshalLog implements logr.Marshaler so logging a SyncInput never dumps the
+// secret-bearing Resources body. Config redacts its own Token via
Config.MarshalJSON.
+func (in SyncInput) MarshalLog() any {
+ return map[string]any{
+ "name": in.Name,
+ "config": in.Config,
+ "labels": in.Labels,
+ "resourceTypes": in.ResourceTypes,
+ "resources": in.Resources.MarshalLog(),
+ }
+}
- if len(configs) == 0 {
- c.log.Info("no GatewayProxy configs provided")
+// 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 with multiple configs", "configs",
configs)
+ c.log.V(1).Info("syncing resources", "inputs", inputs)
failedMap := map[string]types.ADCExecutionErrors{}
- var failedConfigs []string
- for _, config := range configs {
- name := config.Name
- resources, err := c.GetResources(name)
- if err != nil {
- c.log.Error(err, "failed to get resources from store",
"name", name)
- failedConfigs = append(failedConfigs, name)
+ var failedNames []string
+ for _, in := range inputs {
+ if in.Resources == nil {
continue
}
- if resources == nil {
- continue
- }
- c.log.Info("syncing resources for config", "service_number",
len(resources.Services))
-
- if err := c.sync(ctx, Task{
- Name: name + "-sync",
- Configs: map[types.NamespacedNameKind]adctypes.Config{
- {}: config,
- },
- Resources: resources,
- }); err != nil {
- c.log.Error(err, "failed to sync resources", "name",
name)
- failedConfigs = append(failedConfigs, name)
+ 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[name] = execErrs
+ failedMap[in.Name] = execErrs
}
}
}
var err error
- if len(failedConfigs) > 0 {
+ if len(failedNames) > 0 {
err = fmt.Errorf("failed to sync %d configs: %s",
- len(failedConfigs),
- strings.Join(failedConfigs, ", "))
+ len(failedNames),
+ strings.Join(failedNames, ", "))
}
return failedMap, err
}
@@ -350,11 +245,11 @@ func (c *Client) Sync(ctx context.Context)
(map[string]types.ADCExecutionErrors,
// 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) {
+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, args)
+ err := c.executor.Execute(ctx, config, resources, labels, resourceTypes)
var alsoReport []types.ADCExecutionError
if standalone && !config.BypassCache && isConfVersionRejection(err) {
@@ -366,7 +261,7 @@ func (c *Client) push(ctx context.Context, config
adctypes.Config, args []string
pkgmetrics.RecordExecutionError(config.Name,
"conf_version_conflict")
config.BypassCache = true
- retryErr := c.executor.Execute(ctx, config, args)
+ 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
@@ -387,88 +282,49 @@ func (c *Client) push(ctx context.Context, config
adctypes.Config, args []string
return alsoReport, err
}
-func (c *Client) sync(ctx context.Context, task Task) error {
- c.log.V(1).Info("syncing resources", "task", task)
-
- if len(task.Configs) == 0 {
- c.log.Info("no adc configs provided")
- return nil
- }
+func (c *Client) syncOne(ctx context.Context, in SyncInput) error {
+ c.log.V(1).Info("syncing resources", "input", in)
var errs types.ADCExecutionErrors
- // Record file I/O duration
- fileIOStart := time.Now()
- // every task resources is the same, so we can use the first config to
prepare the sync file
- syncFilePath, cleanup, err := prepareSyncFile(task.Resources)
- if err != nil {
- pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure",
time.Since(fileIOStart).Seconds())
- return err
+ config := in.Config
+ if config.BackendType == "" {
+ config.BackendType = c.defaultMode
}
- pkgmetrics.RecordFileIODuration("prepare_sync_file",
adctypes.StatusSuccess, time.Since(fileIOStart).Seconds())
- defer cleanup()
- c.log.V(1).Info("prepared sync file", "path", syncFilePath)
-
- args := BuildADCExecuteArgs(syncFilePath, task.Labels,
task.ResourceTypes)
-
- for _, config := range task.Configs {
- // Record sync duration for each config
- startTime := time.Now()
- resourceType := strings.Join(task.ResourceTypes, ",")
- if resourceType == "" {
- resourceType = "all"
- }
- if config.BackendType == "" {
- config.BackendType = c.defaultMode
- }
- alsoReport, err := c.push(ctx, config, args)
- errs.Errors = append(errs.Errors, alsoReport...)
+ startTime := time.Now()
+ resourceType := strings.Join(in.ResourceTypes, ",")
+ if resourceType == "" {
+ resourceType = "all"
+ }
- duration := time.Since(startTime).Seconds()
+ alsoReport, err := c.push(ctx, config, in.Resources, in.Labels,
in.ResourceTypes)
+ errs.Errors = append(errs.Errors, alsoReport...)
- status := adctypes.StatusSuccess
- if err != nil {
- status = "failure"
- c.log.Error(err, "failed to execute adc command",
"config", config)
+ duration := time.Since(startTime).Seconds()
- var execErr types.ADCExecutionError
- if errors.As(err, &execErr) {
- errs.Errors = append(errs.Errors, execErr)
- pkgmetrics.RecordExecutionError(config.Name,
execErr.Name)
- } else {
- pkgmetrics.RecordExecutionError(config.Name,
"unknown")
- }
+ 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")
}
-
- // Record metrics
- pkgmetrics.RecordSyncDuration(config.Name, resourceType,
status, duration)
}
+ pkgmetrics.RecordSyncDuration(config.Name, resourceType, status,
duration)
+
if len(errs.Errors) > 0 {
return errs
}
return nil
}
-
-func prepareSyncFile(resources any) (string, func(), error) {
- data, err := json.Marshal(resources)
- if err != nil {
- return "", nil, err
- }
-
- tmpFile, err := os.CreateTemp("", "adc-task-*.json")
- if err != nil {
- return "", nil, err
- }
- cleanup := func() {
- _ = tmpFile.Close()
- _ = os.Remove(tmpFile.Name())
- }
- if _, err := tmpFile.Write(data); err != nil {
- cleanup()
- return "", nil, err
- }
-
- return tmpFile.Name(), cleanup, nil
-}
diff --git a/internal/adc/client/executor.go b/internal/adc/client/executor.go
index cfb46a36..6f02433b 100644
--- a/internal/adc/client/executor.go
+++ b/internal/adc/client/executor.go
@@ -26,7 +26,6 @@ import (
"io"
"net"
"net/http"
- "os"
"strings"
"time"
@@ -47,22 +46,8 @@ const (
)
type ADCExecutor interface {
- Execute(ctx context.Context, config adctypes.Config, args []string)
error
- Validate(ctx context.Context, config adctypes.Config, args []string)
error
-}
-
-func BuildADCExecuteArgs(filePath string, labels map[string]string, types
[]string) []string {
- args := []string{
- "sync",
- "-f", filePath,
- }
- for k, v := range labels {
- args = append(args, "--label-selector", k+"="+v)
- }
- for _, t := range types {
- args = append(args, "--include-resource-type", t)
- }
- return args
+ Execute(ctx context.Context, config adctypes.Config, resources
*adctypes.Resources, labels map[string]string, resourceTypes []string) error
+ Validate(ctx context.Context, config adctypes.Config, resources
*adctypes.Resources, labels map[string]string, resourceTypes []string) error
}
// ADCServerRequest represents the request body for ADC Server /sync endpoint
@@ -157,16 +142,16 @@ 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,
args []string) error {
- return e.runHTTPSync(ctx, config, args)
+func (e *HTTPADCExecutor) Execute(ctx context.Context, config adctypes.Config,
resources *adctypes.Resources, labels map[string]string, resourceTypes
[]string) error {
+ return e.runHTTPSync(ctx, config, resources, labels, resourceTypes)
}
-func (e *HTTPADCExecutor) Validate(ctx context.Context, config
adctypes.Config, args []string) error {
- return e.runHTTPValidate(ctx, config, args)
+func (e *HTTPADCExecutor) Validate(ctx context.Context, config
adctypes.Config, resources *adctypes.Resources, labels map[string]string,
resourceTypes []string) error {
+ 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, args []string) error {
+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,
}
@@ -180,7 +165,7 @@ func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context,
config adctypes.Confi
e.log.V(1).Info("running http sync", "serverAddrs", serverAddrs)
for _, addr := range serverAddrs {
- if err := e.runHTTPSyncForSingleServer(ctx, addr, config,
args); err != nil {
+ 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) {
@@ -199,7 +184,7 @@ func (e *HTTPADCExecutor) runHTTPSync(ctx context.Context,
config adctypes.Confi
return nil
}
-func (e *HTTPADCExecutor) runHTTPValidate(ctx context.Context, config
adctypes.Config, args []string) error {
+func (e *HTTPADCExecutor) runHTTPValidate(ctx context.Context, config
adctypes.Config, resources *adctypes.Resources, labels map[string]string,
resourceTypes []string) error {
var validationErr = types.ADCValidationError{
Name: config.Name,
}
@@ -211,7 +196,7 @@ func (e *HTTPADCExecutor) runHTTPValidate(ctx
context.Context, config adctypes.C
e.log.V(1).Info("running http validate", "serverAddrs", serverAddrs)
for _, addr := range serverAddrs {
- if err := e.runHTTPValidateForSingleServer(ctx, addr, config,
args); err != nil {
+ if err := e.runHTTPValidateForSingleServer(ctx, addr, config,
resources, labels, resourceTypes); err != nil {
e.log.Error(err, "failed to run http validate for
server", "server", addr)
var validationServerErr
types.ADCValidationServerAddrError
if errors.As(err, &validationServerErr) {
@@ -232,29 +217,15 @@ func (e *HTTPADCExecutor) runHTTPValidate(ctx
context.Context, config adctypes.C
}
// runHTTPSyncForSingleServer performs HTTP sync to a single ADC Server
-func (e *HTTPADCExecutor) runHTTPSyncForSingleServer(ctx context.Context,
serverAddr string, config adctypes.Config, args []string) error {
+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()
- // Parse args to extract labels, types, and file path
- labels, types, filePath, err := e.parseArgs(args)
- if err != nil {
- return fmt.Errorf("failed to parse args: %w", err)
- }
-
- // Load resources from file
- resources, err := e.loadResourcesFromFile(filePath)
- if err != nil {
- return fmt.Errorf("failed to load resources from file %s: %w",
filePath, err)
- }
-
- // Build HTTP request
- req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types,
resources, pathSync)
+ req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels,
resourceTypes, resources, pathSync)
if err != nil {
return fmt.Errorf("failed to build HTTP request: %w", err)
}
- // Send HTTP request
resp, err := e.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send HTTP request: %w", err)
@@ -265,25 +236,14 @@ func (e *HTTPADCExecutor) runHTTPSyncForSingleServer(ctx
context.Context, server
}
}()
- // Handle HTTP response
return e.handleHTTPResponse(resp, serverAddr)
}
-func (e *HTTPADCExecutor) runHTTPValidateForSingleServer(ctx context.Context,
serverAddr string, config adctypes.Config, args []string) error {
+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()
- labels, types, filePath, err := e.parseArgs(args)
- if err != nil {
- return fmt.Errorf("failed to parse args: %w", err)
- }
-
- resources, err := e.loadResourcesFromFile(filePath)
- if err != nil {
- return fmt.Errorf("failed to load resources from file %s: %w",
filePath, err)
- }
-
- req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels, types,
resources, pathValidate)
+ req, err := e.buildHTTPRequest(ctx, serverAddr, config, labels,
resourceTypes, resources, pathValidate)
if err != nil {
return fmt.Errorf("failed to build validate request: %w", err)
}
@@ -301,60 +261,8 @@ func (e *HTTPADCExecutor)
runHTTPValidateForSingleServer(ctx context.Context, se
return e.handleHTTPValidateResponse(resp, serverAddr)
}
-// parseArgs parses the command line arguments to extract labels, types, and
file path
-func (e *HTTPADCExecutor) parseArgs(args []string) (map[string]string,
[]string, string, error) {
- labels := make(map[string]string)
- var types []string
- var filePath string
-
- for i := 0; i < len(args); i++ {
- switch args[i] {
- case "-f":
- if i+1 < len(args) {
- filePath = args[i+1]
- i++
- }
- case "--label-selector":
- if i+1 < len(args) {
- labelPair := args[i+1]
- parts := strings.SplitN(labelPair, "=", 2)
- if len(parts) == 2 {
- labels[parts[0]] = parts[1]
- }
- i++
- }
- case "--include-resource-type":
- if i+1 < len(args) {
- types = append(types, args[i+1])
- i++
- }
- }
- }
-
- if filePath == "" {
- return nil, nil, "", errors.New("file path not found in args")
- }
-
- return labels, types, filePath, nil
-}
-
-// loadResourcesFromFile loads ADC resources from the specified file
-func (e *HTTPADCExecutor) loadResourcesFromFile(filePath string)
(*adctypes.Resources, error) {
- data, err := os.ReadFile(filePath)
- if err != nil {
- return nil, fmt.Errorf("failed to read file: %w", err)
- }
-
- var resources adctypes.Resources
- if err := json.Unmarshal(data, &resources); err != nil {
- return nil, fmt.Errorf("failed to unmarshal resources: %w", err)
- }
-
- return &resources, nil
-}
-
// buildHTTPRequest builds the HTTP request for ADC Server
-func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr
string, config adctypes.Config, labels map[string]string, types []string,
resources *adctypes.Resources, path string) (*http.Request, error) {
+func (e *HTTPADCExecutor) buildHTTPRequest(ctx context.Context, serverAddr
string, config adctypes.Config, labels map[string]string, resourceTypes
[]string, resources *adctypes.Resources, path string) (*http.Request, error) {
// Prepare request body
tlsVerify := config.TlsVerify
bypassCache := path == pathSync && config.BypassCache
@@ -365,7 +273,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx
context.Context, serverAddr strin
Server: strings.Split(serverAddr,
","),
Token: config.Token,
LabelSelector: labels,
- IncludeResourceType: types,
+ IncludeResourceType: resourceTypes,
TlsSkipVerify: ptr.To(!tlsVerify),
CaCert: config.CaCert,
CacheKey: config.Name,
@@ -389,7 +297,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx
context.Context, serverAddr strin
"cacheKey", config.Name,
"bypassCache", bypassCache,
"labelSelector", labels,
- "includeResourceType", types,
+ "includeResourceType", resourceTypes,
"tlsSkipVerify", !tlsVerify,
"hasCaCert", config.CaCert != "",
)
diff --git a/internal/adc/client/executor_test.go
b/internal/adc/client/executor_test.go
index 0d0eb3c8..609e0ee1 100644
--- a/internal/adc/client/executor_test.go
+++ b/internal/adc/client/executor_test.go
@@ -125,7 +125,7 @@ type fakeExecutor struct {
bypassSeq []bool
}
-func (f *fakeExecutor) Execute(_ context.Context, config adctypes.Config, _
[]string) error {
+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
@@ -135,7 +135,9 @@ func (f *fakeExecutor) Execute(_ context.Context, config
adctypes.Config, _ []st
return err
}
-func (f *fakeExecutor) Validate(context.Context, adctypes.Config, []string)
error { return nil }
+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.
@@ -158,12 +160,10 @@ func afterFirstSync(exec ADCExecutor) *Client {
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"},
- },
+func newSyncInput() SyncInput {
+ return SyncInput{
+ Name: "GatewayProxy/ns/name-sync",
+ Config: adctypes.Config{Name: "GatewayProxy/ns/name",
BackendType: "apisix-standalone"},
Resources: &adctypes.Resources{},
}
}
@@ -175,13 +175,13 @@ func
TestClientSyncRebuildsOnceAfterElectionThenReusesTheADCCache(t *testing.T)
// 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()))
+ 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.sync(context.Background(), newSyncTask()))
+ require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
assert.Equal(t, []bool{true, false, true}, exec.bypassSeq)
}
@@ -193,8 +193,8 @@ func
TestClientSyncRebuildsAgainWhenTheRebuildWasNotAccepted(t *testing.T) {
}}}
c := newTestClient(exec)
- require.Error(t, c.sync(context.Background(), newSyncTask()))
- require.NoError(t, c.sync(context.Background(), newSyncTask()))
+ require.Error(t, c.syncOne(context.Background(), newSyncInput()))
+ require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
assert.Equal(t, []bool{true, true}, exec.bypassSeq)
}
@@ -205,16 +205,16 @@ func
TestClientSyncRebuildsADCBaselineWhenTheDataPlaneRejectsThePush(t *testing.
// 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))
+ 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 task, it would reach the config the ConfigManager
holds and turn a
+ // 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, task.Configs[types.NamespacedNameKind{}].BypassCache,
- "the rebuild must not write BypassCache back into the task
config")
+ assert.False(t, in.Config.BypassCache,
+ "the rebuild must not write BypassCache back into the input's
config")
}
func TestClientSyncDoesNotRebuildOnUnrelatedFailures(t *testing.T) {
@@ -229,7 +229,7 @@ func TestClientSyncDoesNotRebuildOnUnrelatedFailures(t
*testing.T) {
exec := &fakeExecutor{errs: []error{err}}
c := afterFirstSync(exec)
- require.Error(t, c.sync(context.Background(),
newSyncTask()))
+ require.Error(t, c.syncOne(context.Background(),
newSyncInput()))
assert.Equal(t, []bool{false}, exec.bypassSeq)
})
@@ -242,7 +242,7 @@ func TestClientSyncRebuildsHoweverTheRejectionIsWorded(t
*testing.T) {
exec := &fakeExecutor{errs: []error{rejection("upstreams_conf_version
has moved backwards")}}
c := afterFirstSync(exec)
- require.NoError(t, c.sync(context.Background(), newSyncTask()))
+ require.NoError(t, c.syncOne(context.Background(), newSyncInput()))
assert.Equal(t, []bool{false, true}, exec.bypassSeq)
}
@@ -253,9 +253,9 @@ func TestClientSyncDoesNotRebuildOutsideStandalone(t
*testing.T) {
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))
+ 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)
}
@@ -266,7 +266,7 @@ func TestClientSyncSurfacesErrorWhenRebuildFails(t
*testing.T) {
exec := &fakeExecutor{errs: []error{confVersionError(),
rejection(`unrecognized key "bypassCache"`)}}
c := afterFirstSync(exec)
- err := c.sync(context.Background(), newSyncTask())
+ 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")
@@ -282,7 +282,7 @@ func TestClientSyncDoesNotReportTheSameRejectionTwice(t
*testing.T) {
exec := &fakeExecutor{errs: []error{confVersionError(),
confVersionError()}}
c := afterFirstSync(exec)
- err := c.sync(context.Background(), newSyncTask())
+ err := c.syncOne(context.Background(), newSyncInput())
var execErrs types.ADCExecutionErrors
require.ErrorAs(t, err, &execErrs)
diff --git a/internal/adc/client/redaction_test.go
b/internal/adc/client/redaction_test.go
index 0cbb1783..b76a6683 100644
--- a/internal/adc/client/redaction_test.go
+++ b/internal/adc/client/redaction_test.go
@@ -69,7 +69,6 @@ func TestTaskMarshalLogRedactsSecrets(t *testing.T) {
log := bufferLogger(&buf)
task := Task{
- Key: types.NamespacedNameKind{Namespace: "ns", Name:
"route-1", Kind: "ApisixRoute"},
Name: "ns/route-1",
Configs: map[types.NamespacedNameKind]adctypes.Config{
{}: {Name: "gw", Token: secretAdminKey, ServerAddrs:
[]string{"http://x"}},
diff --git a/internal/provider/apisix/keyedmutex.go
b/internal/provider/apisix/keyedmutex.go
new file mode 100644
index 00000000..b564dbe9
--- /dev/null
+++ b/internal/provider/apisix/keyedmutex.go
@@ -0,0 +1,52 @@
+// 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 "sync"
+
+// keyedMutex is a registry of per-key locks. It exists so that reading a
GatewayProxy's
+// current resource snapshot and pushing it can be one atomic step per
cacheKey: whichever
+// caller is granted a key's lock decides what to push only once it actually
holds the lock,
+// so nothing it sends can already be stale relative to whatever the other
caller committed
+// to the store before losing the race for the same key.
+//
+// The registry only grows -- entries are never evicted. Harmless in practice:
cacheKey
+// tracks a small, effectively fixed set of GatewayProxies for the life of the
process. This
+// mirrors ADC server's own per-cacheKey sync_lock.
+type keyedMutex struct {
+ mu sync.Mutex
+ locks map[string]*sync.Mutex
+}
+
+func newKeyedMutex() *keyedMutex {
+ return &keyedMutex{locks: make(map[string]*sync.Mutex)}
+}
+
+// Lock blocks until key's lock is held, and returns the func that releases it.
+func (k *keyedMutex) Lock(key string) func() {
+ k.mu.Lock()
+ l, ok := k.locks[key]
+ if !ok {
+ l = &sync.Mutex{}
+ k.locks[key] = l
+ }
+ k.mu.Unlock()
+
+ l.Lock()
+ return l.Unlock
+}
diff --git a/internal/provider/apisix/keyedmutex_test.go
b/internal/provider/apisix/keyedmutex_test.go
new file mode 100644
index 00000000..2c6f024c
--- /dev/null
+++ b/internal/provider/apisix/keyedmutex_test.go
@@ -0,0 +1,94 @@
+// 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 (
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// TestKeyedMutexSerializesTheSameKey covers what makes syncConfigNow correct:
two holders
+// of the same key must never be inside their critical section at the same
time.
+func TestKeyedMutexSerializesTheSameKey(t *testing.T) {
+ k := newKeyedMutex()
+ var busy atomic.Bool
+
+ var wg sync.WaitGroup
+ for range 8 {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ unlock := k.Lock("shared")
+ defer unlock()
+ if !busy.CompareAndSwap(false, true) {
+ t.Error("another holder was already in the
critical section")
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ busy.Store(false)
+ }()
+ }
+ wg.Wait()
+}
+
+// TestKeyedMutexDoesNotBlockDifferentKeys covers the other half: a busy key
must not stall
+// callers working on an unrelated one, or an immediate delete push for one
GatewayProxy
+// would wait behind a slow periodic sync of a completely different one.
+func TestKeyedMutexDoesNotBlockDifferentKeys(t *testing.T) {
+ k := newKeyedMutex()
+
+ unlockOne := k.Lock("one")
+ defer unlockOne()
+
+ done := make(chan struct{})
+ go func() {
+ unlockTwo := k.Lock("two")
+ defer unlockTwo()
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("locking a different key blocked behind an unrelated
key's holder")
+ }
+}
+
+// TestKeyedMutexUnlockReleasesTheKey covers that the returned func actually
frees the key
+// for the next caller, not just for the same goroutine that locked it.
+func TestKeyedMutexUnlockReleasesTheKey(t *testing.T) {
+ k := newKeyedMutex()
+
+ unlock := k.Lock("x")
+ unlock()
+
+ done := make(chan struct{})
+ go func() {
+ unlockAgain := k.Lock("x")
+ defer unlockAgain()
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("unlock did not release the key for the next caller")
+ }
+}
diff --git a/internal/provider/apisix/provider.go
b/internal/provider/apisix/provider.go
index 039e3d75..45a0179a 100644
--- a/internal/provider/apisix/provider.go
+++ b/internal/provider/apisix/provider.go
@@ -19,6 +19,8 @@ package apisix
import (
"context"
+ "errors"
+ "fmt"
"net/http"
"sync"
"time"
@@ -31,6 +33,7 @@ import (
adctypes "github.com/apache/apisix-ingress-controller/api/adc"
"github.com/apache/apisix-ingress-controller/api/v1alpha1"
apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
+ "github.com/apache/apisix-ingress-controller/internal/adc/cache"
adcclient
"github.com/apache/apisix-ingress-controller/internal/adc/client"
"github.com/apache/apisix-ingress-controller/internal/adc/translator"
"github.com/apache/apisix-ingress-controller/internal/controller/label"
@@ -51,12 +54,24 @@ const (
MinSyncPeriod = 1 * time.Second
)
+// apisixProvider owns AIC's own view of what should be live: which Kubernetes
resource
+// targets which GatewayProxy config (configManager) and the merged,
translated resource
+// snapshot per config (store). It builds the input the adc client package
needs and hands
+// it over on every call; the client package holds none of this state itself.
type apisixProvider struct {
provider.Options
sync.Mutex
translator *translator.Translator
+ store *cache.Store
+ configManager *common.ConfigManager[types.NamespacedNameKind,
adctypes.Config]
+ debugProvider *common.ADCDebugProvider
+
+ // syncLocks serializes, per cacheKey, reading that GatewayProxy's
current resource
+ // snapshot together with pushing it
+ syncLocks *keyedMutex
+
updater status.Updater
statusUpdateMap map[types.NamespacedNameKind][]string
@@ -82,19 +97,26 @@ func New(log logr.Logger, updater status.Updater, readier
readiness.ReadinessMan
return nil, err
}
+ store := cache.NewStore(logger)
+ configManager := common.NewConfigManager[types.NamespacedNameKind,
adctypes.Config]()
+
return &apisixProvider{
- client: cli,
- 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(),
+ Options: o,
+ translator: translator.NewTranslator(log,
o.ListenerPortMatchMode),
+ updater: updater,
+ readier: readier,
+ syncCh: make(chan struct{}, 1),
+ log: logger,
}, nil
}
func (d *apisixProvider) Register(pathPrefix string, mux *http.ServeMux) {
- d.client.ADCDebugProvider.SetupHandler(pathPrefix, mux)
+ d.debugProvider.SetupHandler(pathPrefix, mux)
}
func (d *apisixProvider) Update(ctx context.Context, tctx
*provider.TranslateContext, obj client.Object) error {
@@ -168,23 +190,17 @@ func (d *apisixProvider) Update(ctx context.Context, tctx
*provider.TranslateCon
defer d.syncNotify()
- task := adcclient.Task{
- Key: rk,
- Name: rk.String(),
- Labels: label.GenLabel(obj),
- Configs: configs,
- ResourceTypes: resourceTypes,
- Resources: &adctypes.Resources{
- GlobalRules: result.GlobalRules,
- PluginMetadata: result.PluginMetadata,
- Services: result.Services,
- SSLs: result.SSL,
- Consumers: result.Consumers,
- },
+ resources := &adctypes.Resources{
+ GlobalRules: result.GlobalRules,
+ PluginMetadata: result.PluginMetadata,
+ Services: result.Services,
+ SSLs: result.SSL,
+ Consumers: result.Consumers,
}
- d.log.V(1).Info("updating config", "task", task)
+ labels := label.GenLabel(obj)
+ d.log.V(1).Info("updating config", "resourceKey", rk, "configs",
configs, "resourceTypes", resourceTypes)
- return d.client.UpdateConfig(ctx, task)
+ return d.applyResourceState(rk, configs, resourceTypes, resources,
labels)
}
func (d *apisixProvider) Delete(ctx context.Context, obj client.Object) error {
@@ -222,25 +238,135 @@ func (d *apisixProvider) Delete(ctx context.Context, obj
client.Object) error {
// and it is not possible to perform scheduled synchronization
// on deleted gateway level resources
if len(resourceTypes) == 0 {
- return d.client.Delete(ctx, adcclient.Task{
- Key: nnk,
- Name: nnk.String(),
- Labels: labels,
- })
+ removed, err := d.removeResourceState(nnk, resourceTypes,
labels)
+ if err != nil {
+ return err
+ }
+ d.syncEvictedConfigsNow(ctx, removed, resourceTypes, labels)
+ return nil
+ }
+
+ removed, err := d.removeResourceState(nnk, resourceTypes, labels)
+ if err != nil {
+ return err
}
- delta, err := d.client.DeleteConfig(ctx, adcclient.Task{
- Key: nnk,
- Name: nnk.String(),
- Labels: labels,
- ResourceTypes: resourceTypes,
- })
- // Syncing pushes the whole store to every data plane. Objects this
controller
- // never configured delete nothing, and reconciles for them are
frequent, so
- // notify only when the store actually changed.
- if len(delta.Deleted) > 0 {
+ // Syncing pushes the whole store to every data plane. Objects this
controller never
+ // configured delete nothing, and reconciles for them are frequent, so
notify only
+ // when the store actually changed.
+ if len(removed) > 0 {
d.syncNotify()
}
- return err
+ return nil
+}
+
+// applyResourceState upserts a resource's config associations and its
contribution to each
+// target config's cached resource snapshot -- the AIC-side bookkeeping the
adc client
+// package no longer holds itself.
+func (d *apisixProvider) applyResourceState(
+ rk types.NamespacedNameKind,
+ configs map[types.NamespacedNameKind]adctypes.Config,
+ resourceTypes []string,
+ resources *adctypes.Resources,
+ labels map[string]string,
+) error {
+ d.Lock()
+ defer d.Unlock()
+
+ evicted := d.configManager.Update(rk, configs)
+ if err := d.evictFromStore(evicted, resourceTypes, labels); err != nil {
+ return err
+ }
+ for _, cfg := range configs {
+ if err := d.store.Insert(cfg.Name, resourceTypes, resources,
labels); err != nil {
+ return fmt.Errorf("store insert failed for config %s:
%w", cfg.Name, err)
+ }
+ }
+ return nil
+}
+
+// removeResourceState forgets a resource's config associations and evicts its
contribution
+// from each config it used to reference, returning those configs so an
immediate-push
+// caller (see syncEvictedConfigsNow) knows what to push right away.
+func (d *apisixProvider) removeResourceState(
+ rk types.NamespacedNameKind,
+ resourceTypes []string,
+ labels map[string]string,
+) (map[types.NamespacedNameKind]adctypes.Config, error) {
+ d.Lock()
+ defer d.Unlock()
+
+ evicted := d.configManager.Get(rk)
+ d.configManager.Delete(rk)
+ if err := d.evictFromStore(evicted, resourceTypes, labels); err != nil {
+ return nil, err
+ }
+ return evicted, nil
+}
+
+// evictFromStore deletes a resource's contribution from each of the given
configs' cached
+// snapshots. Callers must already hold d.Lock.
+func (d *apisixProvider) evictFromStore(
+ configs map[types.NamespacedNameKind]adctypes.Config,
+ resourceTypes []string,
+ labels map[string]string,
+) error {
+ for _, cfg := range configs {
+ if err := d.store.Delete(cfg.Name, resourceTypes, labels); err
!= nil {
+ return fmt.Errorf("store delete failed for config %s:
%w", cfg.Name, err)
+ }
+ }
+ return nil
+}
+
+// syncConfigNow reads name's current data (via build, called only once this
cacheKey's
+// lock is actually held) and pushes it -- one atomic read-then-push step per
cacheKey, so
+// whichever caller is granted the lock decides what to push only once it
holds it: nothing
+// it sends can already be stale relative to whatever the other caller
committed to the
+// store before losing the race for the same key. See keyedMutex.
+func (d *apisixProvider) syncConfigNow(
+ ctx context.Context,
+ name string,
+ build func() (adcclient.SyncInput, error),
+) (types.ADCExecutionErrors, error) {
+ unlock := d.syncLocks.Lock(name)
+ defer unlock()
+
+ input, err := build()
+ if err != nil {
+ return types.ADCExecutionErrors{}, err
+ }
+ failedMap, err := d.client.Sync(ctx, []adcclient.SyncInput{input})
+ return failedMap[name], err
+}
+
+// syncEvictedConfigsNow pushes an empty resource set for each of the given
configs
+// immediately, instead of waiting for the next scheduled sync round --
through the same
+// per-cacheKey lock the periodic sync uses, so it can never race a periodic
round for the
+// same GatewayProxy. Used only when the deleted resource is a Gateway or
IngressClass --
+// resourceTypes is empty for those, so the preceding removeResourceState call
already
+// reset each config's whole cached snapshot via Store.Delete, and that reset
should reach
+// the data plane promptly. Failures are logged, not surfaced as a status
update -- this
+// mirrors the deferred path, which only reports through the next scheduled
sync round.
+func (d *apisixProvider) syncEvictedConfigsNow(
+ ctx context.Context,
+ configs map[types.NamespacedNameKind]adctypes.Config,
+ resourceTypes []string,
+ labels map[string]string,
+) {
+ for _, cfg := range configs {
+ _, err := d.syncConfigNow(ctx, cfg.Name, func()
(adcclient.SyncInput, error) {
+ return adcclient.SyncInput{
+ Name: cfg.Name,
+ Config: cfg,
+ Resources: &adctypes.Resources{},
+ ResourceTypes: resourceTypes,
+ Labels: labels,
+ }, nil
+ })
+ if err != nil {
+ d.log.Error(err, "failed to sync deleted config",
"config", cfg)
+ }
+ }
}
func (d *apisixProvider) buildConfig(tctx *provider.TranslateContext, nnk
types.NamespacedNameKind) (map[types.NamespacedNameKind]adctypes.Config, error)
{
@@ -298,10 +424,35 @@ func (d *apisixProvider) Start(ctx context.Context) error
{
}
}
+// sync pushes every GatewayProxy AIC currently knows about, config by config
-- each
+// one's current resource snapshot is only read once syncConfigNow actually
holds that
+// cacheKey's lock, so a slow round can never push a snapshot that was already
stale by the
+// time its turn came up. All of this round's results are still collected into
one
+// statusesMap and handed to handleADCExecutionErrors together, exactly as a
single batched
+// sync would: that logic diffs against last round's full picture, not
per-config.
func (d *apisixProvider) sync(ctx context.Context) error {
- statusesMap, err := d.client.Sync(ctx)
+ configs := d.configManager.List()
+
+ statusesMap := map[string]types.ADCExecutionErrors{}
+ var errs []error
+ for _, config := range configs {
+ execErrs, err := d.syncConfigNow(ctx, config.Name, func()
(adcclient.SyncInput, error) {
+ resources, err := d.store.GetResources(config.Name)
+ if err != nil {
+ return adcclient.SyncInput{},
fmt.Errorf("failed to get resources from store: %w", err)
+ }
+ return adcclient.SyncInput{Name: config.Name, Config:
config, Resources: resources}, nil
+ })
+ if err != nil {
+ errs = append(errs, fmt.Errorf("config %s: %w",
config.Name, err))
+ }
+ if len(execErrs.Errors) > 0 {
+ statusesMap[config.Name] = execErrs
+ }
+ }
+
d.handleADCExecutionErrors(statusesMap)
- return err
+ return errors.Join(errs...)
}
func (d *apisixProvider) syncNotify() {
@@ -330,12 +481,17 @@ func (d *apisixProvider) updateConfigForGatewayProxy(tctx
*provider.TranslateCon
nnk := utils.NamespacedNameKind(gp)
if config == nil {
- d.client.ConfigManager.DeleteConfig(nnk)
+ d.Lock()
+ d.configManager.DeleteConfig(nnk)
+ d.Unlock()
return nil
}
+
referrers := tctx.GatewayProxyReferrers[utils.NamespacedName(gp)]
- d.client.ConfigManager.SetConfigRefs(nnk, referrers)
- d.client.ConfigManager.UpdateConfig(nnk, *config)
+ d.Lock()
+ d.configManager.SetConfigRefs(nnk, referrers)
+ d.configManager.UpdateConfig(nnk, *config)
+ d.Unlock()
d.syncNotify()
return nil
}
diff --git a/internal/provider/apisix/provider_test.go
b/internal/provider/apisix/provider_test.go
index e3be9b13..edd182a8 100644
--- a/internal/provider/apisix/provider_test.go
+++ b/internal/provider/apisix/provider_test.go
@@ -19,33 +19,58 @@ package apisix
import (
"context"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "sync"
"testing"
"time"
"github.com/go-logr/logr"
+ "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
adctypes "github.com/apache/apisix-ingress-controller/api/adc"
+ "github.com/apache/apisix-ingress-controller/internal/adc/cache"
adcclient
"github.com/apache/apisix-ingress-controller/internal/adc/client"
+ "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"
)
+// withMockADCServer starts an ADC server stub and points ADC_SERVER_URL at it
for the
+// duration of the test. The handler itself is how a test inspects what it
received.
+func withMockADCServer(t *testing.T, handler http.HandlerFunc) {
+ t.Helper()
+ server := httptest.NewServer(handler)
+ t.Setenv("ADC_SERVER_URL", server.URL)
+ t.Cleanup(server.Close)
+}
+
+// newTestProvider builds a minimally-wired apisixProvider against the given
mock ADC
+// server -- every field Client/Delete/sync touch, none of the
manager/controller ones.
+func newTestProvider(t *testing.T) *apisixProvider {
+ t.Helper()
+ 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(),
+ }
+}
+
// TestDeleteNotifiesSyncOnlyWhenConfigWasRemoved covers the cost side of route
// ownership: a sync pushes the whole store to every data plane, and reconciles
// for routes this controller never configured are frequent (any EndpointSlice
// event on a shared backend enqueues them), so those must not notify.
func TestDeleteNotifiesSyncOnlyWhenConfigWasRemoved(t *testing.T) {
- cli, err := adcclient.New(logr.Discard(), ProviderTypeAPISIX,
time.Second)
- require.NoError(t, err)
-
- d := &apisixProvider{
- client: cli,
- syncCh: make(chan struct{}, 1),
- log: logr.Discard(),
- }
+ d := newTestProvider(t)
route := &gatewayv1.HTTPRoute{
TypeMeta: metav1.TypeMeta{
@@ -58,10 +83,95 @@ func TestDeleteNotifiesSyncOnlyWhenConfigWasRemoved(t
*testing.T) {
require.NoError(t, d.Delete(context.Background(), route))
require.Empty(t, d.syncCh, "a route this controller never configured
must not trigger a sync")
- cli.ConfigManager.Update(utils.NamespacedNameKind(route),
map[types.NamespacedNameKind]adctypes.Config{
+ d.configManager.Update(utils.NamespacedNameKind(route),
map[types.NamespacedNameKind]adctypes.Config{
{Namespace: "default", Name: "proxy", Kind: "GatewayProxy"}:
{Name: "proxy"},
})
require.NoError(t, d.Delete(context.Background(), route))
require.Len(t, d.syncCh, 1, "removing configuration this controller
pushed must trigger a sync")
}
+
+// TestDeleteTriggersImmediateSyncForEvictedConfigs covers the immediate-push
branch of
+// Delete: a Gateway going away must reach the data plane right away -- an
empty resource
+// set for the config it referenced -- not wait for the next scheduled sync
round.
+func TestDeleteTriggersImmediateSyncForEvictedConfigs(t *testing.T) {
+ var mu sync.Mutex
+ var received []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()
+ received = append(received, req)
+ mu.Unlock()
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(adctypes.SyncResult{Status:
adctypes.StatusSuccess})
+ })
+
+ d := newTestProvider(t)
+
+ gw := &gatewayv1.Gateway{
+ TypeMeta: metav1.TypeMeta{
+ Kind: "Gateway",
+ APIVersion: gatewayv1.GroupVersion.String(),
+ },
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "gw"},
+ }
+ d.configManager.Update(utils.NamespacedNameKind(gw),
map[types.NamespacedNameKind]adctypes.Config{
+ {Namespace: "default", Name: "proxy", Kind: "GatewayProxy"}: {
+ Name: "proxy",
+ BackendType: "apisix",
+ ServerAddrs: []string{"http://apisix:9080"},
+ },
+ })
+
+ require.NoError(t, d.Delete(context.Background(), gw))
+
+ mu.Lock()
+ defer mu.Unlock()
+ require.Len(t, received, 1, "deleting a Gateway must push immediately,
not wait for the next scheduled round")
+ assert.Equal(t, "proxy", received[0].Task.Opts.CacheKey)
+ assert.Empty(t, received[0].Task.Config.Services, "the evicted config's
push must carry an empty resource set")
+}
+
+// TestSyncStillPushesHealthyConfigsWhenAnotherFails covers sync's error
aggregation: one
+// GatewayProxy's push failing must not stop the others in the same round from
being
+// attempted, and the failure must still be reported.
+func TestSyncStillPushesHealthyConfigsWhenAnotherFails(t *testing.T) {
+ var mu sync.Mutex
+ seen := map[string]bool{}
+
+ withMockADCServer(t, func(w http.ResponseWriter, r *http.Request) {
+ var req adcclient.ADCServerRequest
+ require.NoError(t, json.NewDecoder(r.Body).Decode(&req))
+ mu.Lock()
+ seen[req.Task.Opts.CacheKey] = true
+ mu.Unlock()
+ if req.Task.Opts.CacheKey == "bad" {
+ w.WriteHeader(http.StatusInternalServerError)
+ _, _ = w.Write([]byte(`{"message": "boom"}`))
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+ _ = json.NewEncoder(w).Encode(adctypes.SyncResult{Status:
adctypes.StatusSuccess})
+ })
+
+ d := newTestProvider(t)
+ for _, name := range []string{"bad", "good"} {
+ key := types.NamespacedNameKind{Namespace: "default", Name:
name, Kind: "GatewayProxy"}
+ d.configManager.UpdateConfig(key, adctypes.Config{
+ Name: name,
+ BackendType: "apisix",
+ ServerAddrs: []string{"http://apisix:9080"},
+ })
+ }
+
+ err := d.sync(context.Background())
+ require.Error(t, err, "one config failing must still be reported")
+ assert.Contains(t, err.Error(), "bad")
+
+ mu.Lock()
+ defer mu.Unlock()
+ assert.True(t, seen["bad"], "the failing config must still have been
attempted")
+ assert.True(t, seen["good"], "a config failing must not stop the others
from being pushed")
+}
diff --git a/internal/provider/apisix/status.go
b/internal/provider/apisix/status.go
index a1d857eb..e2f82bd0 100644
--- a/internal/provider/apisix/status.go
+++ b/internal/provider/apisix/status.go
@@ -109,7 +109,7 @@ func (d *apisixProvider) updateStatus(nnk
types.NamespacedNameKind, condition me
}),
})
case types.KindHTTPRoute:
- parentRefs :=
d.client.ConfigManager.GetConfigRefsByResourceKey(nnk)
+ parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk)
d.log.V(1).Info("updating HTTPRoute status", "parentRefs",
parentRefs)
gatewayRefs := map[types.NamespacedNameKind]struct{}{}
for _, parentRef := range parentRefs {
@@ -145,7 +145,7 @@ func (d *apisixProvider) updateStatus(nnk
types.NamespacedNameKind, condition me
}),
})
case types.KindUDPRoute:
- parentRefs :=
d.client.ConfigManager.GetConfigRefsByResourceKey(nnk)
+ parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk)
d.log.V(1).Info("updating UDPRoute status", "parentRefs",
parentRefs)
gatewayRefs := map[types.NamespacedNameKind]struct{}{}
for _, parentRef := range parentRefs {
@@ -181,7 +181,7 @@ func (d *apisixProvider) updateStatus(nnk
types.NamespacedNameKind, condition me
}),
})
case types.KindTCPRoute:
- parentRefs :=
d.client.ConfigManager.GetConfigRefsByResourceKey(nnk)
+ parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk)
d.log.V(1).Info("updating TCPRoute status", "parentRefs",
parentRefs)
gatewayRefs := map[types.NamespacedNameKind]struct{}{}
for _, parentRef := range parentRefs {
@@ -217,7 +217,7 @@ func (d *apisixProvider) updateStatus(nnk
types.NamespacedNameKind, condition me
}),
})
case types.KindGRPCRoute:
- parentRefs :=
d.client.ConfigManager.GetConfigRefsByResourceKey(nnk)
+ parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk)
d.log.V(1).Info("updating GRPCRoute status", "parentRefs",
parentRefs)
gatewayRefs := map[types.NamespacedNameKind]struct{}{}
for _, parentRef := range parentRefs {
@@ -279,7 +279,7 @@ func (d *apisixProvider) handleEmptyFailedStatuses(
failedStatus types.ADCExecutionServerAddrError,
statusUpdateMap map[types.NamespacedNameKind][]string,
) {
- resource, err := d.client.GetResources(configName)
+ resource, err := d.store.GetResources(configName)
if err != nil {
d.log.Error(err, "failed to get resources from store",
"configName", configName)
return
@@ -297,7 +297,7 @@ func (d *apisixProvider) handleEmptyFailedStatuses(
d.addResourceToStatusUpdateMap(obj.GetLabels(),
failedStatus.Error(), statusUpdateMap)
}
- globalRules, err := d.client.ListGlobalRules(configName)
+ globalRules, err := d.store.ListGlobalRules(configName)
if err != nil {
d.log.Error(err, "failed to list global rules", "configName",
configName)
return
@@ -319,7 +319,7 @@ func (d *apisixProvider) handleDetailedFailedStatuses(
return
}
id := status.Event.ResourceID
- labels, err := d.client.GetResourceLabel(configName,
status.Event.ResourceType, id)
+ labels, err := d.store.GetResourceLabel(configName,
status.Event.ResourceType, id)
if err != nil {
d.log.Error(err, "failed to get resource label",
"configName", configName,
diff --git a/internal/webhook/v1/adc_validation.go
b/internal/webhook/v1/adc_validation.go
index 2c980c0f..d4c45f9a 100644
--- a/internal/webhook/v1/adc_validation.go
+++ b/internal/webhook/v1/adc_validation.go
@@ -217,7 +217,6 @@ func (v *adcAdmissionValidator)
buildIngressClassConfigs(ctx context.Context, ob
func (v *adcAdmissionValidator) newTask(obj client.Object, configs
map[internaltypes.NamespacedNameKind]adctypes.Config, resourceTypes []string,
result *adctranslator.TranslateResult) *adcclient.Task {
return &adcclient.Task{
- Key: utils.NamespacedNameKind(obj),
Name: utils.NamespacedNameKind(obj).String(),
Labels: label.GenLabel(obj),
Configs: configs,
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index c9537fe1..4f1b6ff0 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -58,16 +58,6 @@ var (
Help: "Current length of the status update queue",
},
)
-
- // File I/O operation duration histogram
- FileIODuration = prometheus.NewHistogramVec(
- prometheus.HistogramOpts{
- Name: "apisix_ingress_file_io_duration_seconds",
- Help: "Time spent on file I/O operations",
- Buckets: prometheus.DefBuckets,
- },
- []string{"operation", "status"},
- )
)
// init registers all metrics with the global prometheus registry
@@ -78,7 +68,6 @@ func init() {
ADCSyncTotal,
ADCExecutionErrors,
StatusUpdateQueueLength,
- FileIODuration,
)
}
@@ -107,8 +96,3 @@ func IncStatusQueueLength() {
func DecStatusQueueLength() {
StatusUpdateQueueLength.Dec()
}
-
-// RecordFileIODuration records the duration of a file I/O operation
-func RecordFileIODuration(operation, status string, duration float64) {
- FileIODuration.WithLabelValues(operation, status).Observe(duration)
-}
diff --git a/test/e2e/crds/v2/route.go b/test/e2e/crds/v2/route.go
index 4159a42b..67b4a0b7 100644
--- a/test/e2e/crds/v2/route.go
+++ b/test/e2e/crds/v2/route.go
@@ -170,7 +170,6 @@ spec:
Expect(bodyStr).Should(ContainSubstring("apisix_ingress_adc_sync_duration_seconds"))
Expect(bodyStr).Should(ContainSubstring("apisix_ingress_adc_sync_total"))
Expect(bodyStr).Should(ContainSubstring("apisix_ingress_status_update_queue_length"))
-
Expect(bodyStr).Should(ContainSubstring("apisix_ingress_file_io_duration_seconds"))
}
It("Basic", func() {
test(apisixRouteSpec)
