From 201d106e5ae6ce85f8211af9ba722e63e795094b Mon Sep 17 00:00:00 2001 From: bzp2010 Date: Thu, 10 Sep 2026 16:29:40 +0800 Subject: [PATCH 1/2] refactor: let adc client independent of external state --- internal/adc/client/client.go | 385 ++++---------------- internal/adc/client/executor.go | 126 +------ internal/adc/client/executor_test.go | 50 +-- internal/adc/client/redaction_test.go | 1 - internal/provider/api7ee/provider.go | 282 ++++++++++++-- internal/provider/api7ee/provider_test.go | 201 ++++++++++ internal/provider/api7ee/status.go | 16 +- internal/provider/apisix/provider.go | 246 ++++++++++--- internal/provider/apisix/provider_test.go | 128 ++++++- internal/provider/apisix/status.go | 16 +- internal/provider/common/configmanager.go | 12 - internal/provider/common/immediatesync.go | 71 ++++ internal/provider/common/keyedmutex.go | 55 +++ internal/provider/common/keyedmutex_test.go | 94 +++++ internal/webhook/v1/adc_validation.go | 1 - pkg/metrics/metrics.go | 16 - test/e2e/crds/v2/route.go | 1 - 17 files changed, 1128 insertions(+), 573 deletions(-) create mode 100644 internal/provider/api7ee/provider_test.go create mode 100644 internal/provider/common/immediatesync.go create mode 100644 internal/provider/common/keyedmutex.go create mode 100644 internal/provider/common/keyedmutex_test.go diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go index 83d617a1f..0e9e42496 100644 --- a/internal/adc/client/client.go +++ b/internal/adc/client/client.go @@ -15,14 +15,17 @@ // 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" - "slices" "strings" "sync" "time" @@ -31,23 +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/controller/label" - "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. @@ -66,18 +59,12 @@ func New(log logr.Logger, defaultMode string, timeout time.Duration) (*Client, e 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 @@ -131,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 @@ -149,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, @@ -158,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() - - args := 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, args); 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) @@ -286,55 +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 - } - - 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 } @@ -352,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) { @@ -368,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 @@ -389,165 +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.Labels) > 0 { - // only keep the resource key label for filtering resources - task.Labels = map[string]string{label.LabelResourceKey: task.Labels[label.LabelResourceKey]} - } - - 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 - // for global rules, we need to list all global rules and set it to the task resources - if slices.Contains(task.ResourceTypes, "global_rule") { - for _, config := range task.Configs { - globalRules, err := c.ListGlobalRules(config.Name) - if err != nil { - return err - } - var globalrule adctypes.GlobalRule - if len(globalRules) > 0 { - merged := make(adctypes.Plugins) - for _, item := range globalRules { - for k, v := range item.Plugins { - merged[k] = v - } - } - globalrule = adctypes.GlobalRule(merged) - } - if task.Resources == nil { - task.Resources = &adctypes.Resources{} - } - - task.Resources.GlobalRules = globalrule - c.log.V(1).Info("syncing resources global rules", "globalRules", task.Resources.GlobalRules) - - 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", "success", time.Since(fileIOStart).Seconds()) - defer cleanup() - - args := BuildADCExecuteArgs(syncFilePath, task.Labels, task.ResourceTypes) - - // 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 - } - - err = c.executor.Execute(ctx, config, args) - duration := time.Since(startTime).Seconds() - - status := "success" - if err != nil { - status = "failure" - c.log.Error(err, "failed to execute adc command", "config", config) - - 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") - } - } - - // Record metrics - pkgmetrics.RecordSyncDuration(config.Name, resourceType, status, duration) - } - - if len(errs.Errors) > 0 { - return errs - } - return nil + config := in.Config + if config.BackendType == "" { + config.BackendType = c.defaultMode } - // 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 + startTime := time.Now() + resourceType := strings.Join(in.ResourceTypes, ",") + if resourceType == "" { + resourceType = "all" } - 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...) + alsoReport, err := c.push(ctx, config, in.Resources, in.Labels, in.ResourceTypes) + errs.Errors = append(errs.Errors, alsoReport...) - duration := time.Since(startTime).Seconds() + duration := time.Since(startTime).Seconds() - status := adctypes.StatusSuccess - if err != nil { - status = "failure" - c.log.Error(err, "failed to execute adc command", "config", config) - - 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 92feb7d90..0576cc562 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 a7f998e68..4f9aef390 100644 --- a/internal/adc/client/executor_test.go +++ b/internal/adc/client/executor_test.go @@ -123,7 +123,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 @@ -133,7 +133,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. @@ -156,12 +158,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{}, } } @@ -173,13 +173,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) } @@ -191,8 +191,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) } @@ -203,16 +203,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) { @@ -227,7 +227,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) }) @@ -240,7 +240,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) } @@ -251,9 +251,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) } @@ -264,7 +264,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") @@ -280,7 +280,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 0cbb17830..b76a6683f 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/api7ee/provider.go b/internal/provider/api7ee/provider.go index 38a0567cb..22198a36a 100644 --- a/internal/provider/api7ee/provider.go +++ b/internal/provider/api7ee/provider.go @@ -19,7 +19,10 @@ package api7ee import ( "context" + "errors" + "fmt" "net/http" + "sync" "sync/atomic" "time" @@ -32,6 +35,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,9 +55,23 @@ const ( RetryMaxDelay = 1000 * time.Second ) +// api7eeProvider 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 api7eeProvider struct { + 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 *common.KeyedMutex + updater status.Updater statusUpdateMap map[types.NamespacedNameKind][]string @@ -81,19 +99,28 @@ func New(log logr.Logger, updater status.Updater, readier readiness.ReadinessMan return nil, err } + logger := log.WithName("provider") + + store := cache.NewStore(logger) + configManager := common.NewConfigManager[types.NamespacedNameKind, adctypes.Config]() + return &api7eeProvider{ - client: cli, - Options: o, - translator: translator.NewTranslator(log, o.ListenerPortMatchMode), - updater: updater, - readier: readier, - syncCh: make(chan struct{}, 1), - log: log.WithName("provider"), + client: cli, + store: store, + configManager: configManager, + debugProvider: common.NewADCDebugProvider(store, configManager), + syncLocks: common.NewKeyedMutex(), + Options: o, + translator: translator.NewTranslator(log, o.ListenerPortMatchMode), + updater: updater, + readier: readier, + syncCh: make(chan struct{}, 1), + log: logger, }, nil } func (d *api7eeProvider) Register(pathPrefix string, mux *http.ServeMux) { - d.client.ADCDebugProvider.SetupHandler(pathPrefix, mux) + d.debugProvider.SetupHandler(pathPrefix, mux) } func (d *api7eeProvider) Update(ctx context.Context, tctx *provider.TranslateContext, obj client.Object) error { @@ -169,29 +196,31 @@ func (d *api7eeProvider) Update(ctx context.Context, tctx *provider.TranslateCon return nil } - nnk := utils.NamespacedNameKind(obj) + resources := &adctypes.Resources{ + GlobalRules: result.GlobalRules, + PluginMetadata: result.PluginMetadata, + Services: result.Services, + SSLs: result.SSL, + Consumers: result.Consumers, + } + labels := label.GenLabel(obj) - task := adcclient.Task{ - Key: nnk, - Name: nnk.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, - }, + evicted, err := d.applyResourceState(rk, configs, resourceTypes, resources, labels) + if err != nil { + return err } if !d.startUpSync.Load() { d.log.V(1).Info("startup synchronization not completed, skip sync", "object", obj) - return d.client.UpdateConfig(ctx, task) + return nil } - return d.client.Update(ctx, task) + // A GatewayProxy this resource no longer targets must still lose this resource's + // contribution on the data plane. That push is best-effort, the same as the deferred + // path's: only the periodic sync ever surfaces its failures as a status update. + d.syncEvictedConfigsNow(ctx, evicted, resourceTypes, labels) + + return d.pushConfigsNow(ctx, configs, resourceTypes, resources, labels) } func (d *api7eeProvider) Delete(ctx context.Context, obj client.Object) error { @@ -225,12 +254,165 @@ func (d *api7eeProvider) Delete(ctx context.Context, obj client.Object) error { } nnk := utils.NamespacedNameKind(obj) - return d.client.Delete(ctx, adcclient.Task{ - Key: nnk, - Name: nnk.String(), - Labels: labels, - ResourceTypes: resourceTypes, - }) + + removed, err := d.removeResourceState(nnk, resourceTypes, labels) + if err != nil { + return err + } + + // A deleted resource always pushes right away rather than waiting for the next + // scheduled round, and -- like the deferred path -- only logs a push failure instead + // of surfacing it, since there is no per-object status to report it against once the + // object itself is gone. + d.syncEvictedConfigsNow(ctx, removed, resourceTypes, labels) + 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. It returns the configs this resource no longer +// references (if any), so the caller can push their eviction right away instead of +// waiting for the next scheduled sync round. +func (d *api7eeProvider) applyResourceState( + rk types.NamespacedNameKind, + configs map[types.NamespacedNameKind]adctypes.Config, + resourceTypes []string, + resources *adctypes.Resources, + labels map[string]string, +) (map[types.NamespacedNameKind]adctypes.Config, error) { + d.Lock() + defer d.Unlock() + + evicted := d.configManager.Update(rk, configs) + if err := d.evictFromStore(evicted, resourceTypes, labels); err != nil { + return nil, err + } + for _, cfg := range configs { + if err := d.store.Insert(cfg.Name, resourceTypes, resources, labels); err != nil { + return nil, fmt.Errorf("store insert failed for config %s: %w", cfg.Name, err) + } + } + return evicted, 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 *api7eeProvider) 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 *api7eeProvider) 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 common.KeyedMutex. +func (d *api7eeProvider) 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 +} + +// pushConfigsNow pushes resources immediately to every one of the given configs, through +// the same per-cacheKey lock the periodic sync uses, and reports the combined push error. +// This is the immediate half of Update: it only runs once startup synchronization has +// completed, mirroring the old Client.Update. Before that, applyResourceState alone is +// enough -- the startup sync and the periodic ticker eventually push it. +func (d *api7eeProvider) pushConfigsNow( + ctx context.Context, + configs map[types.NamespacedNameKind]adctypes.Config, + resourceTypes []string, + resources *adctypes.Resources, + labels map[string]string, +) error { + var errs []error + for _, cfg := range configs { + _, err := d.syncConfigNow(ctx, cfg.Name, func() (adcclient.SyncInput, error) { + mergedResources, err := common.WithMergedGlobalRules(d.store, cfg.Name, resourceTypes, resources) + if err != nil { + return adcclient.SyncInput{}, err + } + return adcclient.SyncInput{ + Name: cfg.Name, + Config: cfg, + Resources: mergedResources, + ResourceTypes: resourceTypes, + Labels: common.ResourceKeyLabels(labels), + }, nil + }) + if err != nil { + errs = append(errs, fmt.Errorf("config %s: %w", cfg.Name, err)) + } + } + return errors.Join(errs...) +} + +// 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. 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 *api7eeProvider) 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) { + resources, err := common.WithMergedGlobalRules(d.store, cfg.Name, resourceTypes, &adctypes.Resources{}) + if err != nil { + return adcclient.SyncInput{}, err + } + return adcclient.SyncInput{ + Name: cfg.Name, + Config: cfg, + Resources: resources, + ResourceTypes: resourceTypes, + Labels: common.ResourceKeyLabels(labels), + }, nil + }) + if err != nil { + d.log.Error(err, "failed to sync deleted config", "config", cfg) + } + } } func (d *api7eeProvider) Start(ctx context.Context) error { @@ -285,10 +467,35 @@ func (d *api7eeProvider) syncNotify() { } } +// 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 *api7eeProvider) 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 *api7eeProvider) handleADCExecutionErrors(statusesMap map[string]types.ADCExecutionErrors) { @@ -310,12 +517,17 @@ func (d *api7eeProvider) 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/api7ee/provider_test.go b/internal/provider/api7ee/provider_test.go new file mode 100644 index 000000000..dc6bd8915 --- /dev/null +++ b/internal/provider/api7ee/provider_test.go @@ -0,0 +1,201 @@ +// 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 api7ee + +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 api7eeProvider against the given mock ADC +// server -- every field Delete/sync touch, none of the manager/controller ones. +func newTestProvider(t *testing.T) *api7eeProvider { + t.Helper() + cli, err := adcclient.New(logr.Discard(), ProviderTypeAPI7EE, time.Second) + require.NoError(t, err) + return &api7eeProvider{ + client: cli, + store: cache.NewStore(logr.Discard()), + configManager: common.NewConfigManager[types.NamespacedNameKind, adctypes.Config](), + syncLocks: common.NewKeyedMutex(), + syncCh: make(chan struct{}, 1), + log: logr.Discard(), + } +} + +// TestDeletePushesImmediatelyRegardlessOfStartup covers what sets api7ee apart from +// apisix: every Delete pushes right away, whether or not startup synchronization has +// completed -- unlike Update, which defers to the periodic sync until it has. +func TestDeletePushesImmediatelyRegardlessOfStartup(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) + // startUpSync is deliberately left false: Delete must not wait for it. + + route := &gatewayv1.HTTPRoute{ + TypeMeta: metav1.TypeMeta{ + Kind: "HTTPRoute", + APIVersion: gatewayv1.GroupVersion.String(), + }, + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "route"}, + } + d.configManager.Update(utils.NamespacedNameKind(route), 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(), route)) + + mu.Lock() + defer mu.Unlock() + require.Len(t, received, 1, "deleting a route must push immediately, not wait for the next scheduled round") + assert.Equal(t, "proxy", received[0].Task.Opts.CacheKey) +} + +// 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") +} + +// TestPushConfigsNowMergesGlobalRulesFromStore covers the fork-specific piece the client +// package no longer holds: global_rule is a singleton per config, not partitioned by +// label, so an immediate push scoped to "global_rule" must carry every contribution +// currently in store for that config, not just the one this call is pushing. +func TestPushConfigsNowMergesGlobalRulesFromStore(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) + cfg := adctypes.Config{Name: "proxy", BackendType: "apisix", ServerAddrs: []string{"http://apisix:9080"}} + + // Another ApisixGlobalRule already contributed a rule to this same config's store + // entry before this call. + require.NoError(t, d.store.Insert(cfg.Name, []string{"global_rule"}, &adctypes.Resources{ + GlobalRules: adctypes.GlobalRule{"limit-count": map[string]any{"count": float64(1)}}, + }, map[string]string{"k8s/resource-key": "ApisixGlobalRule/default/one"})) + + configs := map[types.NamespacedNameKind]adctypes.Config{ + {Namespace: "default", Name: "proxy", Kind: "GatewayProxy"}: cfg, + } + resources := &adctypes.Resources{ + GlobalRules: adctypes.GlobalRule{"key-auth": map[string]any{"key": "k"}}, + } + labels := map[string]string{"k8s/resource-key": "ApisixGlobalRule/default/two"} + + // pushConfigsNow is the immediate half of Update, called only after applyResourceState + // has already put this call's own contribution in the store -- mirror that here. + require.NoError(t, d.store.Insert(cfg.Name, []string{"global_rule"}, resources, labels)) + + require.NoError(t, d.pushConfigsNow(context.Background(), configs, []string{"global_rule"}, resources, labels)) + + mu.Lock() + defer mu.Unlock() + require.Len(t, received, 1) + assert.Contains(t, received[0].Task.Config.GlobalRules, "limit-count", + "the other ApisixGlobalRule's contribution must not be dropped by this push") + assert.Contains(t, received[0].Task.Config.GlobalRules, "key-auth", + "this push's own contribution must still be included") +} diff --git a/internal/provider/api7ee/status.go b/internal/provider/api7ee/status.go index 0c2427c12..583d98fb0 100644 --- a/internal/provider/api7ee/status.go +++ b/internal/provider/api7ee/status.go @@ -109,7 +109,7 @@ func (d *api7eeProvider) updateStatus(nnk types.NamespacedNameKind, condition me }), }) case types.KindHTTPRoute: - parentRefs := d.client.ConfigManager.GetConfigRefsByResourceKey(nnk) + parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk) gatewayRefs := map[types.NamespacedNameKind]struct{}{} for _, parentRef := range parentRefs { if parentRef.Kind == types.KindGateway { @@ -144,7 +144,7 @@ func (d *api7eeProvider) 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 { @@ -180,7 +180,7 @@ func (d *api7eeProvider) 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 { @@ -216,7 +216,7 @@ func (d *api7eeProvider) 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 { @@ -252,7 +252,7 @@ func (d *api7eeProvider) updateStatus(nnk types.NamespacedNameKind, condition me }), }) case types.KindTLSRoute: - parentRefs := d.client.ConfigManager.GetConfigRefsByResourceKey(nnk) + parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk) d.log.V(1).Info("updating TLSRoute status", "parentRefs", parentRefs) gatewayRefs := map[types.NamespacedNameKind]struct{}{} for _, parentRef := range parentRefs { @@ -314,7 +314,7 @@ func (d *api7eeProvider) 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 @@ -332,7 +332,7 @@ func (d *api7eeProvider) 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 @@ -349,7 +349,7 @@ func (d *api7eeProvider) handleDetailedFailedStatuses( ) { for _, status := range failedStatus.FailedStatuses { 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/provider/apisix/provider.go b/internal/provider/apisix/provider.go index aff19913c..f3247f1e4 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" @@ -32,6 +34,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" @@ -53,12 +56,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 *common.KeyedMutex + updater status.Updater statusUpdateMap map[types.NamespacedNameKind][]string @@ -82,19 +97,28 @@ func New(log logr.Logger, updater status.Updater, readier readiness.ReadinessMan return nil, err } + logger := log.WithName("provider") + + 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: log.WithName("provider"), + client: cli, + store: store, + configManager: configManager, + debugProvider: common.NewADCDebugProvider(store, configManager), + syncLocks: common.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 { @@ -172,23 +196,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 { @@ -226,25 +244,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 common.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) { @@ -300,10 +428,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() { @@ -332,12 +485,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 e3be9b137..24d793e15 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: common.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 0c9c997d9..0cb2555c6 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 { @@ -253,7 +253,7 @@ func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition me }), }) case types.KindTLSRoute: - parentRefs := d.client.ConfigManager.GetConfigRefsByResourceKey(nnk) + parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk) d.log.V(1).Info("updating TLSRoute status", "parentRefs", parentRefs) gatewayRefs := map[types.NamespacedNameKind]struct{}{} for _, parentRef := range parentRefs { @@ -315,7 +315,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 @@ -333,7 +333,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 @@ -356,7 +356,7 @@ func (d *apisixProvider) handleDetailedFailedStatuses( } 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/provider/common/configmanager.go b/internal/provider/common/configmanager.go index 4e1d529b3..b913bb920 100644 --- a/internal/provider/common/configmanager.go +++ b/internal/provider/common/configmanager.go @@ -36,12 +36,6 @@ func NewConfigManager[K comparable, T any]() *ConfigManager[K, T] { } } -func (s *ConfigManager[K, T]) GetConfigRefs(key K) []K { - s.mu.Lock() - defer s.mu.Unlock() - return s.configRefs[key] -} - func (s *ConfigManager[K, T]) GetConfigRefsByResourceKey(key K) []K { s.mu.Lock() defer s.mu.Unlock() @@ -124,12 +118,6 @@ func (s *ConfigManager[K, T]) Update( return discard } -func (s *ConfigManager[K, T]) Set(key K, cfg T) { - s.mu.Lock() - defer s.mu.Unlock() - s.configs[key] = cfg -} - func (s *ConfigManager[K, T]) Delete(key K) { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/provider/common/immediatesync.go b/internal/provider/common/immediatesync.go new file mode 100644 index 000000000..6a7f82ef0 --- /dev/null +++ b/internal/provider/common/immediatesync.go @@ -0,0 +1,71 @@ +// 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 common + +import ( + "fmt" + "slices" + + 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/controller/label" +) + +// ResourceKeyLabels narrows labels down to just the resource-key label, which is what an +// immediate, single-resource push to ADC uses as its label selector: it must touch only +// the resources this one Kubernetes object contributed, not everything else that happens +// to share its kind, name, or namespace. The richer label set (kind, name, namespace) +// stays in the store's own bookkeeping, which keys entries by all of them. +func ResourceKeyLabels(labels map[string]string) map[string]string { + if len(labels) == 0 { + return nil + } + return map[string]string{label.LabelResourceKey: labels[label.LabelResourceKey]} +} + +// WithMergedGlobalRules returns resources with GlobalRules replaced by every global_rule +// contribution currently in store for cfgName, merged into one -- but only when +// resourceTypes actually names "global_rule" as one of the types this push is scoped to. +// +// global_rule is a singleton object per config, not partitioned by label the way a route +// or a consumer is: an immediate push that carries only the just-translated object's own +// contribution would silently drop every other source's rules. The deferred, store-wide +// sync never needs this, since it already reads the whole store (see Store.GetResources). +func WithMergedGlobalRules(store *cache.Store, cfgName string, resourceTypes []string, resources *adctypes.Resources) (*adctypes.Resources, error) { + if !slices.Contains(resourceTypes, adctypes.TypeGlobalRule) { + return resources, nil + } + + items, err := store.ListGlobalRules(cfgName) + if err != nil { + return nil, fmt.Errorf("failed to list global rules for config %s: %w", cfgName, err) + } + merged := make(adctypes.Plugins) + for _, item := range items { + for k, v := range item.Plugins { + merged[k] = v + } + } + + out := &adctypes.Resources{} + if resources != nil { + *out = *resources + } + out.GlobalRules = adctypes.GlobalRule(merged) + return out, nil +} diff --git a/internal/provider/common/keyedmutex.go b/internal/provider/common/keyedmutex.go new file mode 100644 index 000000000..9535028af --- /dev/null +++ b/internal/provider/common/keyedmutex.go @@ -0,0 +1,55 @@ +// 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 common + +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. +// +// It is shared by every provider (apisix, api7ee, ...) so each gets the same per-cacheKey +// serialization instead of reimplementing it. +// +// 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/common/keyedmutex_test.go b/internal/provider/common/keyedmutex_test.go new file mode 100644 index 000000000..cd0072d09 --- /dev/null +++ b/internal/provider/common/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 common + +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/webhook/v1/adc_validation.go b/internal/webhook/v1/adc_validation.go index 434115fba..ab2432092 100644 --- a/internal/webhook/v1/adc_validation.go +++ b/internal/webhook/v1/adc_validation.go @@ -218,7 +218,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 c9537fe12..4f1b6ff0e 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 2d98d36c4..52a7fcaf1 100644 --- a/test/e2e/crds/v2/route.go +++ b/test/e2e/crds/v2/route.go @@ -172,7 +172,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) From fc4f27da0d374af513f76d149a900fd0e5f9d916 Mon Sep 17 00:00:00 2001 From: bzp2010 Date: Thu, 10 Sep 2026 17:15:40 +0800 Subject: [PATCH 2/2] fix e2e --- internal/provider/api7ee/provider.go | 8 +++--- internal/provider/api7ee/provider_test.go | 33 +++++++++++++++++++++++ internal/provider/apisix/provider.go | 4 +-- internal/provider/common/immediatesync.go | 13 +++++++++ 4 files changed, 52 insertions(+), 6 deletions(-) diff --git a/internal/provider/api7ee/provider.go b/internal/provider/api7ee/provider.go index 22198a36a..aa55cfb4e 100644 --- a/internal/provider/api7ee/provider.go +++ b/internal/provider/api7ee/provider.go @@ -364,7 +364,7 @@ func (d *api7eeProvider) pushConfigsNow( ) error { var errs []error for _, cfg := range configs { - _, err := d.syncConfigNow(ctx, cfg.Name, func() (adcclient.SyncInput, error) { + execErrs, err := d.syncConfigNow(ctx, cfg.Name, func() (adcclient.SyncInput, error) { mergedResources, err := common.WithMergedGlobalRules(d.store, cfg.Name, resourceTypes, resources) if err != nil { return adcclient.SyncInput{}, err @@ -378,7 +378,7 @@ func (d *api7eeProvider) pushConfigsNow( }, nil }) if err != nil { - errs = append(errs, fmt.Errorf("config %s: %w", cfg.Name, err)) + errs = append(errs, common.PushError(cfg.Name, execErrs, err)) } } return errors.Join(errs...) @@ -396,7 +396,7 @@ func (d *api7eeProvider) syncEvictedConfigsNow( labels map[string]string, ) { for _, cfg := range configs { - _, err := d.syncConfigNow(ctx, cfg.Name, func() (adcclient.SyncInput, error) { + execErrs, err := d.syncConfigNow(ctx, cfg.Name, func() (adcclient.SyncInput, error) { resources, err := common.WithMergedGlobalRules(d.store, cfg.Name, resourceTypes, &adctypes.Resources{}) if err != nil { return adcclient.SyncInput{}, err @@ -410,7 +410,7 @@ func (d *api7eeProvider) syncEvictedConfigsNow( }, nil }) if err != nil { - d.log.Error(err, "failed to sync deleted config", "config", cfg) + d.log.Error(common.PushError(cfg.Name, execErrs, err), "failed to sync deleted config", "config", cfg) } } } diff --git a/internal/provider/api7ee/provider_test.go b/internal/provider/api7ee/provider_test.go index dc6bd8915..2614da166 100644 --- a/internal/provider/api7ee/provider_test.go +++ b/internal/provider/api7ee/provider_test.go @@ -150,6 +150,39 @@ func TestSyncStillPushesHealthyConfigsWhenAnotherFails(t *testing.T) { assert.True(t, seen["good"], "a config failing must not stop the others from being pushed") } +// TestPushConfigsNowSurfacesTheDataPlaneRejectionReason covers a regression: Update's +// immediate push must return the actual reason the data plane rejected a resource for +// (e.g. "custom plugin (non-existent-plugin) not found"), not just a generic "failed to +// sync N configs" wrapper -- that reason is what a resource controller puts into the +// resource's own status condition, and status.go tests for it verbatim. +func TestPushConfigsNowSurfacesTheDataPlaneRejectionReason(t *testing.T) { + const rejectReason = "custom plugin (non-existent-plugin) not found" + + withMockADCServer(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(adctypes.SyncResult{ + Status: adctypes.StatusFailed, + FailedCount: 1, + Failed: []adctypes.SyncStatus{{ + Event: adctypes.StatusEvent{ResourceType: "route", ResourceID: "r1"}, + Reason: rejectReason, + }}, + }) + }) + + d := newTestProvider(t) + cfg := adctypes.Config{Name: "proxy", BackendType: "apisix", ServerAddrs: []string{"http://apisix:9080"}} + configs := map[types.NamespacedNameKind]adctypes.Config{ + {Namespace: "default", Name: "proxy", Kind: "GatewayProxy"}: cfg, + } + resources := &adctypes.Resources{} + + err := d.pushConfigsNow(context.Background(), configs, nil, resources, nil) + require.Error(t, err) + assert.Contains(t, err.Error(), rejectReason, + "the data plane's rejection reason must reach the caller, not just a generic wrapper") +} + // TestPushConfigsNowMergesGlobalRulesFromStore covers the fork-specific piece the client // package no longer holds: global_rule is a singleton per config, not partitioned by // label, so an immediate push scoped to "global_rule" must carry every contribution diff --git a/internal/provider/apisix/provider.go b/internal/provider/apisix/provider.go index f3247f1e4..a5eb97a34 100644 --- a/internal/provider/apisix/provider.go +++ b/internal/provider/apisix/provider.go @@ -360,7 +360,7 @@ func (d *apisixProvider) syncEvictedConfigsNow( labels map[string]string, ) { for _, cfg := range configs { - _, err := d.syncConfigNow(ctx, cfg.Name, func() (adcclient.SyncInput, error) { + execErrs, err := d.syncConfigNow(ctx, cfg.Name, func() (adcclient.SyncInput, error) { return adcclient.SyncInput{ Name: cfg.Name, Config: cfg, @@ -370,7 +370,7 @@ func (d *apisixProvider) syncEvictedConfigsNow( }, nil }) if err != nil { - d.log.Error(err, "failed to sync deleted config", "config", cfg) + d.log.Error(common.PushError(cfg.Name, execErrs, err), "failed to sync deleted config", "config", cfg) } } } diff --git a/internal/provider/common/immediatesync.go b/internal/provider/common/immediatesync.go index 6a7f82ef0..367a7f578 100644 --- a/internal/provider/common/immediatesync.go +++ b/internal/provider/common/immediatesync.go @@ -24,6 +24,7 @@ import ( 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/controller/label" + "github.com/apache/apisix-ingress-controller/internal/types" ) // ResourceKeyLabels narrows labels down to just the resource-key label, which is what an @@ -69,3 +70,15 @@ func WithMergedGlobalRules(store *cache.Store, cfgName string, resourceTypes []s out.GlobalRules = adctypes.GlobalRule(merged) return out, nil } + +// PushError picks what to report for a failed immediate push: execErrs, when the data +// plane is what rejected it, carries the actual reason (e.g. "custom plugin +// (non-existent-plugin) not found") that a caller surfaces as a resource's status message. +// The generic err from syncConfigNow itself -- the build callback failing, say -- carries +// none of that, so it is only a fallback. +func PushError(cacheKey string, execErrs types.ADCExecutionErrors, err error) error { + if len(execErrs.Errors) > 0 { + return execErrs + } + return fmt.Errorf("config %s: %w", cacheKey, err) +}