This is an automated email from the ASF dual-hosted git repository. bzp2010 pushed a commit to branch bzp/feat-refactor-provider-layers in repository https://gitbox.apache.org/repos/asf/apisix-ingress-controller.git
commit feecbb3f5c79494c2b1592a405f75ac55aa24f3a Author: bzp2010 <[email protected]> AuthorDate: Wed Sep 9 15:07:14 2026 +0800 refactor: let adc client independent of external state --- internal/adc/client/client.go | 303 +++++++++------------------------- internal/adc/client/executor.go | 120 ++------------ internal/adc/client/executor_test.go | 50 +++--- internal/adc/client/redaction_test.go | 1 - internal/provider/apisix/provider.go | 200 +++++++++++++++++----- internal/provider/apisix/status.go | 14 +- internal/webhook/v1/adc_validation.go | 1 - pkg/metrics/metrics.go | 16 -- 8 files changed, 285 insertions(+), 420 deletions(-) diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go index b3db60ca..8611310f 100644 --- a/internal/adc/client/client.go +++ b/internal/adc/client/client.go @@ -15,11 +15,15 @@ // specific language governing permissions and limitations // under the License. +// Package client talks to the ADC server: given a fully-prepared sync or validate +// request, it translates it to ADC's wire format, sends it, and interprets the response. +// It holds no bookkeeping of its own about which Kubernetes resource maps to which +// GatewayProxy, or what a GatewayProxy's current resource snapshot is -- that is AIC's own +// state, owned by the caller and handed in as input on every call. package client import ( "context" - "encoding/json" "fmt" "os" "strings" @@ -30,22 +34,13 @@ import ( "github.com/pkg/errors" adctypes "github.com/apache/apisix-ingress-controller/api/adc" - "github.com/apache/apisix-ingress-controller/internal/adc/cache" - "github.com/apache/apisix-ingress-controller/internal/provider/common" "github.com/apache/apisix-ingress-controller/internal/types" pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics" ) type Client struct { - syncMu sync.RWMutex - mu sync.Mutex - *cache.Store - executor ADCExecutor - ConfigManager *common.ConfigManager[types.NamespacedNameKind, adctypes.Config] - ADCDebugProvider *common.ADCDebugProvider - defaultMode string // rebuiltMu guards rebuiltBaselines. @@ -63,18 +58,13 @@ func New(log logr.Logger, defaultMode string, timeout time.Duration) (*Client, e if serverURL == "" { serverURL = defaultHTTPADCExecutorAddr } - store := cache.NewStore(log) - configManager := common.NewConfigManager[types.NamespacedNameKind, adctypes.Config]() logger := log.WithName("client") logger.Info("ADC client initialized") return &Client{ - Store: store, rebuiltBaselines: make(map[string]struct{}), executor: NewHTTPADCExecutor(log, serverURL, timeout), - ConfigManager: configManager, - ADCDebugProvider: common.NewADCDebugProvider(store, configManager), log: logger, defaultMode: defaultMode, }, nil @@ -128,8 +118,9 @@ func isConfVersionRejection(err error) bool { // (routes_conf_version, upstreams_conf_version, ...) and refuses a push that moves back. const confVersionField = "conf_version" +// Task is a /validate request: one Kubernetes resource's translated result, checked +// against every GatewayProxy config it could target. type Task struct { - Key types.NamespacedNameKind Name string Labels map[string]string Configs map[types.NamespacedNameKind]adctypes.Config @@ -146,7 +137,6 @@ func (t Task) MarshalLog() any { configNames = append(configNames, cfg.Name) } return map[string]any{ - "key": t.Key, "name": t.Name, "labels": t.Labels, "resourceTypes": t.ResourceTypes, @@ -155,118 +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) -} - -func (c *Client) DeleteConfig(ctx context.Context, args Task) error { - _, err := c.applyStoreChanges(args, true) - return err -} - func (c *Client) Validate(ctx context.Context, task Task) error { if len(task.Configs) == 0 || task.Resources == nil { return nil } - fileIOStart := time.Now() - syncFilePath, cleanup, err := prepareSyncFile(task.Resources) - if err != nil { - pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure", time.Since(fileIOStart).Seconds()) - return err - } - pkgmetrics.RecordFileIODuration("prepare_sync_file", adctypes.StatusSuccess, time.Since(fileIOStart).Seconds()) - defer cleanup() - - args2 := BuildADCExecuteArgs(syncFilePath, task.Labels, task.ResourceTypes) - var errs types.ADCValidationErrors for _, config := range task.Configs { if config.BackendType == "" { config.BackendType = c.defaultMode } - if err := c.executor.Validate(ctx, config, args2); err != nil { + if err := c.executor.Validate(ctx, config, task.Resources, task.Labels, task.ResourceTypes); err != nil { var validationErr types.ADCValidationError if errors.As(err, &validationErr) { errs.Errors = append(errs.Errors, validationErr) @@ -282,56 +171,63 @@ func (c *Client) Validate(ctx context.Context, task Task) error { return nil } -func (c *Client) Sync(ctx context.Context) (map[string]types.ADCExecutionErrors, error) { - c.syncMu.Lock() - defer c.syncMu.Unlock() - c.log.Info("syncing all resources") +// SyncInput is one GatewayProxy's complete sync unit. AIC builds it entirely from its own +// bookkeeping (which resources target this config, their merged translated snapshot) +// before handing it over -- this package never reaches back into AIC's state to gather +// anything itself, it only translates, sends, and interprets the response. +type SyncInput struct { + // Name is the cacheKey: the GatewayProxy's own identity. + Name string + Config adctypes.Config + Resources *adctypes.Resources + ResourceTypes []string + Labels map[string]string +} - configs := c.ConfigManager.List() +// MarshalLog implements logr.Marshaler so logging a SyncInput never dumps the +// secret-bearing Resources body. Config redacts its own Token via Config.MarshalJSON. +func (in SyncInput) MarshalLog() any { + return map[string]any{ + "name": in.Name, + "config": in.Config, + "labels": in.Labels, + "resourceTypes": in.ResourceTypes, + "resources": in.Resources.MarshalLog(), + } +} - if len(configs) == 0 { - c.log.Info("no GatewayProxy configs provided") +// Sync pushes every given SyncInput to its data plane in one sweep, and reports the +// parsed, typed error for each one that failed, keyed by its Name -- an input whose name +// is absent from the returned map genuinely succeeded. It never returns a raw HTTP status +// or body; every response ADC can send back is already interpreted by the time it gets +// here. +func (c *Client) Sync(ctx context.Context, inputs []SyncInput) (map[string]types.ADCExecutionErrors, error) { + if len(inputs) == 0 { return nil, nil } - - c.log.V(1).Info("syncing resources with multiple configs", "configs", configs) + c.log.V(1).Info("syncing resources", "inputs", inputs) failedMap := map[string]types.ADCExecutionErrors{} - var failedConfigs []string - for _, config := range configs { - name := config.Name - resources, err := c.GetResources(name) - if err != nil { - c.log.Error(err, "failed to get resources from store", "name", name) - failedConfigs = append(failedConfigs, name) + var failedNames []string + for _, in := range inputs { + if in.Resources == nil { continue } - if resources == nil { - continue - } - c.log.Info("syncing resources for config", "service_number", len(resources.Services)) - - if err := c.sync(ctx, Task{ - Name: name + "-sync", - Configs: map[types.NamespacedNameKind]adctypes.Config{ - {}: config, - }, - Resources: resources, - }); err != nil { - c.log.Error(err, "failed to sync resources", "name", name) - failedConfigs = append(failedConfigs, name) + if err := c.syncOne(ctx, in); err != nil { + c.log.Error(err, "failed to sync resources", "name", in.Name) + failedNames = append(failedNames, in.Name) var execErrs types.ADCExecutionErrors if errors.As(err, &execErrs) { - failedMap[name] = execErrs + failedMap[in.Name] = execErrs } } } var err error - if len(failedConfigs) > 0 { + if len(failedNames) > 0 { err = fmt.Errorf("failed to sync %d configs: %s", - len(failedConfigs), - strings.Join(failedConfigs, ", ")) + len(failedNames), + strings.Join(failedNames, ", ")) } return failedMap, err } @@ -349,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) { @@ -365,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 @@ -386,88 +282,45 @@ func (c *Client) push(ctx context.Context, config adctypes.Config, args []string return alsoReport, err } -func (c *Client) sync(ctx context.Context, task Task) error { - c.log.V(1).Info("syncing resources", "task", task) - - if len(task.Configs) == 0 { - c.log.Info("no adc configs provided") - return nil - } +func (c *Client) syncOne(ctx context.Context, in SyncInput) error { + c.log.V(1).Info("syncing resources", "input", in) var errs types.ADCExecutionErrors - // Record file I/O duration - fileIOStart := time.Now() - // every task resources is the same, so we can use the first config to prepare the sync file - syncFilePath, cleanup, err := prepareSyncFile(task.Resources) - if err != nil { - pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure", time.Since(fileIOStart).Seconds()) - return err + config := in.Config + if config.BackendType == "" { + config.BackendType = c.defaultMode } - pkgmetrics.RecordFileIODuration("prepare_sync_file", adctypes.StatusSuccess, time.Since(fileIOStart).Seconds()) - defer cleanup() - c.log.V(1).Info("prepared sync file", "path", syncFilePath) - - args := BuildADCExecuteArgs(syncFilePath, task.Labels, task.ResourceTypes) - - for _, config := range task.Configs { - // Record sync duration for each config - startTime := time.Now() - resourceType := strings.Join(task.ResourceTypes, ",") - if resourceType == "" { - resourceType = "all" - } - if config.BackendType == "" { - config.BackendType = c.defaultMode - } - alsoReport, err := c.push(ctx, config, args) - errs.Errors = append(errs.Errors, alsoReport...) + startTime := time.Now() + resourceType := strings.Join(in.ResourceTypes, ",") + if resourceType == "" { + resourceType = "all" + } - duration := time.Since(startTime).Seconds() + alsoReport, err := c.push(ctx, config, in.Resources, in.Labels, in.ResourceTypes) + errs.Errors = append(errs.Errors, alsoReport...) - status := adctypes.StatusSuccess - if err != nil { - status = "failure" - c.log.Error(err, "failed to execute adc command", "config", config) + duration := time.Since(startTime).Seconds() - var execErr types.ADCExecutionError - if errors.As(err, &execErr) { - errs.Errors = append(errs.Errors, execErr) - pkgmetrics.RecordExecutionError(config.Name, execErr.Name) - } else { - pkgmetrics.RecordExecutionError(config.Name, "unknown") - } + status := adctypes.StatusSuccess + if err != nil { + status = "failure" + c.log.Error(err, "failed to 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) } + pkgmetrics.RecordSyncDuration(config.Name, resourceType, status, duration) + if len(errs.Errors) > 0 { return errs } return nil } - -func prepareSyncFile(resources any) (string, func(), error) { - data, err := json.Marshal(resources) - if err != nil { - return "", nil, err - } - - tmpFile, err := os.CreateTemp("", "adc-task-*.json") - if err != nil { - return "", nil, err - } - cleanup := func() { - _ = tmpFile.Close() - _ = os.Remove(tmpFile.Name()) - } - if _, err := tmpFile.Write(data); err != nil { - cleanup() - return "", nil, err - } - - return tmpFile.Name(), cleanup, nil -} diff --git a/internal/adc/client/executor.go b/internal/adc/client/executor.go index cfb46a36..b3d612d9 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,58 +261,6 @@ 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) { // Prepare request body diff --git a/internal/adc/client/executor_test.go b/internal/adc/client/executor_test.go index 0d0eb3c8..609e0ee1 100644 --- a/internal/adc/client/executor_test.go +++ b/internal/adc/client/executor_test.go @@ -125,7 +125,7 @@ type fakeExecutor struct { bypassSeq []bool } -func (f *fakeExecutor) Execute(_ context.Context, config adctypes.Config, _ []string) error { +func (f *fakeExecutor) Execute(_ context.Context, config adctypes.Config, _ *adctypes.Resources, _ map[string]string, _ []string) error { f.bypassSeq = append(f.bypassSeq, config.BypassCache) if len(f.errs) == 0 { return nil @@ -135,7 +135,9 @@ func (f *fakeExecutor) Execute(_ context.Context, config adctypes.Config, _ []st return err } -func (f *fakeExecutor) Validate(context.Context, adctypes.Config, []string) error { return nil } +func (f *fakeExecutor) Validate(context.Context, adctypes.Config, *adctypes.Resources, map[string]string, []string) error { + return nil +} // newTestClient starts out as a controller that has just been elected: no ADC baseline is // known to be current, so the first sync of a cacheKey rebuilds it. @@ -158,12 +160,10 @@ func afterFirstSync(exec ADCExecutor) *Client { const syncTaskCacheKey = "GatewayProxy/ns/name" -func newSyncTask() Task { - return Task{ - Name: "GatewayProxy/ns/name-sync", - Configs: map[types.NamespacedNameKind]adctypes.Config{ - {}: {Name: "GatewayProxy/ns/name", BackendType: "apisix-standalone"}, - }, +func newSyncInput() SyncInput { + return SyncInput{ + Name: "GatewayProxy/ns/name-sync", + Config: adctypes.Config{Name: "GatewayProxy/ns/name", BackendType: "apisix-standalone"}, Resources: &adctypes.Resources{}, } } @@ -175,13 +175,13 @@ func TestClientSyncRebuildsOnceAfterElectionThenReusesTheADCCache(t *testing.T) // The sidecar may still hold a baseline from an earlier term, so the first sync of a // cacheKey re-derives it from the data plane. Once ADC has accepted that sync, its // baseline is current and later syncs diff against it. - require.NoError(t, c.sync(context.Background(), newSyncTask())) - require.NoError(t, c.sync(context.Background(), newSyncTask())) + require.NoError(t, c.syncOne(context.Background(), newSyncInput())) + require.NoError(t, c.syncOne(context.Background(), newSyncInput())) assert.Equal(t, []bool{true, false}, exec.bypassSeq) // Winning the election again puts every baseline back in doubt. c.InvalidateADCCache() - require.NoError(t, c.sync(context.Background(), newSyncTask())) + require.NoError(t, c.syncOne(context.Background(), newSyncInput())) assert.Equal(t, []bool{true, false, true}, exec.bypassSeq) } @@ -193,8 +193,8 @@ func TestClientSyncRebuildsAgainWhenTheRebuildWasNotAccepted(t *testing.T) { }}} c := newTestClient(exec) - require.Error(t, c.sync(context.Background(), newSyncTask())) - require.NoError(t, c.sync(context.Background(), newSyncTask())) + require.Error(t, c.syncOne(context.Background(), newSyncInput())) + require.NoError(t, c.syncOne(context.Background(), newSyncInput())) assert.Equal(t, []bool{true, true}, exec.bypassSeq) } @@ -205,16 +205,16 @@ func TestClientSyncRebuildsADCBaselineWhenTheDataPlaneRejectsThePush(t *testing. // The data plane holds a conf_version newer than the one the ADC baseline carries, so // the push is rejected. The retry rebuilds that baseline from the data plane. - task := newSyncTask() - require.NoError(t, c.sync(context.Background(), task)) + in := newSyncInput() + require.NoError(t, c.syncOne(context.Background(), in)) assert.Equal(t, []bool{false, true}, exec.bypassSeq) // BypassCache is scoped to the request that recovers from the rejection. Were it to - // survive in the task, it would reach the config the ConfigManager holds and turn a + // survive in the input, it would reach the config ConfigManager holds and turn a // one-off rebuild into a data plane fetch on every later sync. - assert.False(t, task.Configs[types.NamespacedNameKind{}].BypassCache, - "the rebuild must not write BypassCache back into the task config") + assert.False(t, in.Config.BypassCache, + "the rebuild must not write BypassCache back into the input's config") } func TestClientSyncDoesNotRebuildOnUnrelatedFailures(t *testing.T) { @@ -229,7 +229,7 @@ func TestClientSyncDoesNotRebuildOnUnrelatedFailures(t *testing.T) { exec := &fakeExecutor{errs: []error{err}} c := afterFirstSync(exec) - require.Error(t, c.sync(context.Background(), newSyncTask())) + require.Error(t, c.syncOne(context.Background(), newSyncInput())) assert.Equal(t, []bool{false}, exec.bypassSeq) }) @@ -242,7 +242,7 @@ func TestClientSyncRebuildsHoweverTheRejectionIsWorded(t *testing.T) { exec := &fakeExecutor{errs: []error{rejection("upstreams_conf_version has moved backwards")}} c := afterFirstSync(exec) - require.NoError(t, c.sync(context.Background(), newSyncTask())) + require.NoError(t, c.syncOne(context.Background(), newSyncInput())) assert.Equal(t, []bool{false, true}, exec.bypassSeq) } @@ -253,9 +253,9 @@ func TestClientSyncDoesNotRebuildOutsideStandalone(t *testing.T) { exec := &fakeExecutor{errs: []error{confVersionError()}} c := afterFirstSync(exec) - task := newSyncTask() - task.Configs[types.NamespacedNameKind{}] = adctypes.Config{Name: "GatewayProxy/ns/name", BackendType: "apisix"} - require.Error(t, c.sync(context.Background(), task)) + in := newSyncInput() + in.Config = adctypes.Config{Name: "GatewayProxy/ns/name", BackendType: "apisix"} + require.Error(t, c.syncOne(context.Background(), in)) assert.Equal(t, []bool{false}, exec.bypassSeq) } @@ -266,7 +266,7 @@ func TestClientSyncSurfacesErrorWhenRebuildFails(t *testing.T) { exec := &fakeExecutor{errs: []error{confVersionError(), rejection(`unrecognized key "bypassCache"`)}} c := afterFirstSync(exec) - err := c.sync(context.Background(), newSyncTask()) + err := c.syncOne(context.Background(), newSyncInput()) require.Error(t, err, "a rebuild that still fails must not be swallowed") assert.Equal(t, []bool{false, true}, exec.bypassSeq, "the rebuild is attempted once, not in a loop") @@ -282,7 +282,7 @@ func TestClientSyncDoesNotReportTheSameRejectionTwice(t *testing.T) { exec := &fakeExecutor{errs: []error{confVersionError(), confVersionError()}} c := afterFirstSync(exec) - err := c.sync(context.Background(), newSyncTask()) + err := c.syncOne(context.Background(), newSyncInput()) var execErrs types.ADCExecutionErrors require.ErrorAs(t, err, &execErrs) diff --git a/internal/adc/client/redaction_test.go b/internal/adc/client/redaction_test.go index 0cbb1783..b76a6683 100644 --- a/internal/adc/client/redaction_test.go +++ b/internal/adc/client/redaction_test.go @@ -69,7 +69,6 @@ func TestTaskMarshalLogRedactsSecrets(t *testing.T) { log := bufferLogger(&buf) task := Task{ - Key: types.NamespacedNameKind{Namespace: "ns", Name: "route-1", Kind: "ApisixRoute"}, Name: "ns/route-1", Configs: map[types.NamespacedNameKind]adctypes.Config{ {}: {Name: "gw", Token: secretAdminKey, ServerAddrs: []string{"http://x"}}, diff --git a/internal/provider/apisix/provider.go b/internal/provider/apisix/provider.go index 86d7f532..6bb01e04 100644 --- a/internal/provider/apisix/provider.go +++ b/internal/provider/apisix/provider.go @@ -19,6 +19,8 @@ package apisix import ( "context" + "errors" + "fmt" "net/http" "sync" "time" @@ -31,6 +33,7 @@ import ( adctypes "github.com/apache/apisix-ingress-controller/api/adc" "github.com/apache/apisix-ingress-controller/api/v1alpha1" apiv2 "github.com/apache/apisix-ingress-controller/api/v2" + "github.com/apache/apisix-ingress-controller/internal/adc/cache" adcclient "github.com/apache/apisix-ingress-controller/internal/adc/client" "github.com/apache/apisix-ingress-controller/internal/adc/translator" "github.com/apache/apisix-ingress-controller/internal/controller/label" @@ -51,12 +54,20 @@ 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 + updater status.Updater statusUpdateMap map[types.NamespacedNameKind][]string @@ -82,19 +93,25 @@ func New(log logr.Logger, updater status.Updater, readier readiness.ReadinessMan return nil, err } + store := cache.NewStore(logger) + configManager := common.NewConfigManager[types.NamespacedNameKind, adctypes.Config]() + return &apisixProvider{ - client: cli, - Options: o, - translator: translator.NewTranslator(log, o.ListenerPortMatchMode), - updater: updater, - readier: readier, - syncCh: make(chan struct{}, 1), - log: logger, + client: cli, + store: store, + configManager: configManager, + debugProvider: common.NewADCDebugProvider(store, configManager), + Options: o, + translator: translator.NewTranslator(log, o.ListenerPortMatchMode), + updater: updater, + readier: readier, + syncCh: make(chan struct{}, 1), + log: logger, }, nil } func (d *apisixProvider) Register(pathPrefix string, mux *http.ServeMux) { - d.client.ADCDebugProvider.SetupHandler(pathPrefix, mux) + d.debugProvider.SetupHandler(pathPrefix, mux) } func (d *apisixProvider) Update(ctx context.Context, tctx *provider.TranslateContext, obj client.Object) error { @@ -168,23 +185,17 @@ func (d *apisixProvider) Update(ctx context.Context, tctx *provider.TranslateCon defer d.syncNotify() - task := adcclient.Task{ - Key: rk, - Name: rk.String(), - Labels: label.GenLabel(obj), - Configs: configs, - ResourceTypes: resourceTypes, - Resources: &adctypes.Resources{ - GlobalRules: result.GlobalRules, - PluginMetadata: result.PluginMetadata, - Services: result.Services, - SSLs: result.SSL, - Consumers: result.Consumers, - }, + resources := &adctypes.Resources{ + GlobalRules: result.GlobalRules, + PluginMetadata: result.PluginMetadata, + Services: result.Services, + SSLs: result.SSL, + Consumers: result.Consumers, } - d.log.V(1).Info("updating config", "task", task) + labels := label.GenLabel(obj) + d.log.V(1).Info("updating config", "resourceKey", rk, "configs", configs, "resourceTypes", resourceTypes) - return d.client.UpdateConfig(ctx, task) + return d.applyResourceState(rk, configs, resourceTypes, resources, labels) } func (d *apisixProvider) Delete(ctx context.Context, obj client.Object) error { @@ -222,19 +233,97 @@ 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 } + defer d.syncNotify() - return d.client.DeleteConfig(ctx, adcclient.Task{ - Key: nnk, - Name: nnk.String(), - Labels: labels, - ResourceTypes: resourceTypes, - }) + _, err := d.removeResourceState(nnk, resourceTypes, labels) + return err +} + +// 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() + + discarded := d.configManager.Update(rk, configs) + + for _, cfg := range discarded { + if err := d.store.Delete(cfg.Name, resourceTypes, labels); err != nil { + return fmt.Errorf("store delete failed for config %s: %w", cfg.Name, 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() + + removed := d.configManager.Get(rk) + d.configManager.Delete(rk) + for _, cfg := range removed { + if err := d.store.Delete(cfg.Name, resourceTypes, labels); err != nil { + return nil, fmt.Errorf("store delete failed for config %s: %w", cfg.Name, err) + } + } + return removed, nil +} + +// syncEvictedConfigsNow pushes an empty resource set for each of the given configs +// immediately, instead of waiting for the next scheduled sync round. 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, +) { + if len(configs) == 0 { + return + } + inputs := make([]adcclient.SyncInput, 0, len(configs)) + for _, cfg := range configs { + inputs = append(inputs, adcclient.SyncInput{ + Name: cfg.Name, + Config: cfg, + Resources: &adctypes.Resources{}, + ResourceTypes: resourceTypes, + Labels: labels, + }) + } + if _, err := d.client.Sync(ctx, inputs); err != nil { + d.log.Error(err, "failed to sync deleted configs", "configs", configs) + } } func (d *apisixProvider) buildConfig(tctx *provider.TranslateContext, nnk types.NamespacedNameKind) (map[types.NamespacedNameKind]adctypes.Config, error) { @@ -293,9 +382,37 @@ func (d *apisixProvider) Start(ctx context.Context) error { } func (d *apisixProvider) sync(ctx context.Context) error { - statusesMap, err := d.client.Sync(ctx) + inputs, resourceErr := d.buildSyncInputs() + statusesMap, syncErr := d.client.Sync(ctx, inputs) d.handleADCExecutionErrors(statusesMap) - return err + return errors.Join(resourceErr, syncErr) +} + +// buildSyncInputs organizes this round's full config set -- every GatewayProxy AIC +// currently knows about, each with its merged translated resource snapshot -- into the +// input the adc client package needs. The client package never gathers this itself. +func (d *apisixProvider) buildSyncInputs() ([]adcclient.SyncInput, error) { + configs := d.configManager.List() + if len(configs) == 0 { + return nil, nil + } + + inputs := make([]adcclient.SyncInput, 0, len(configs)) + var errs []error + for _, config := range configs { + resources, err := d.store.GetResources(config.Name) + if err != nil { + d.log.Error(err, "failed to get resources from store", "name", config.Name) + errs = append(errs, fmt.Errorf("config %s: %w", config.Name, err)) + continue + } + inputs = append(inputs, adcclient.SyncInput{ + Name: config.Name, + Config: config, + Resources: resources, + }) + } + return inputs, errors.Join(errs...) } func (d *apisixProvider) syncNotify() { @@ -324,12 +441,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/status.go b/internal/provider/apisix/status.go index a1d857eb..e2f82bd0 100644 --- a/internal/provider/apisix/status.go +++ b/internal/provider/apisix/status.go @@ -109,7 +109,7 @@ func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition me }), }) case types.KindHTTPRoute: - parentRefs := d.client.ConfigManager.GetConfigRefsByResourceKey(nnk) + parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk) d.log.V(1).Info("updating HTTPRoute status", "parentRefs", parentRefs) gatewayRefs := map[types.NamespacedNameKind]struct{}{} for _, parentRef := range parentRefs { @@ -145,7 +145,7 @@ func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition me }), }) case types.KindUDPRoute: - parentRefs := d.client.ConfigManager.GetConfigRefsByResourceKey(nnk) + parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk) d.log.V(1).Info("updating UDPRoute status", "parentRefs", parentRefs) gatewayRefs := map[types.NamespacedNameKind]struct{}{} for _, parentRef := range parentRefs { @@ -181,7 +181,7 @@ func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition me }), }) case types.KindTCPRoute: - parentRefs := d.client.ConfigManager.GetConfigRefsByResourceKey(nnk) + parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk) d.log.V(1).Info("updating TCPRoute status", "parentRefs", parentRefs) gatewayRefs := map[types.NamespacedNameKind]struct{}{} for _, parentRef := range parentRefs { @@ -217,7 +217,7 @@ func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition me }), }) case types.KindGRPCRoute: - parentRefs := d.client.ConfigManager.GetConfigRefsByResourceKey(nnk) + parentRefs := d.configManager.GetConfigRefsByResourceKey(nnk) d.log.V(1).Info("updating GRPCRoute status", "parentRefs", parentRefs) gatewayRefs := map[types.NamespacedNameKind]struct{}{} for _, parentRef := range parentRefs { @@ -279,7 +279,7 @@ func (d *apisixProvider) handleEmptyFailedStatuses( failedStatus types.ADCExecutionServerAddrError, statusUpdateMap map[types.NamespacedNameKind][]string, ) { - resource, err := d.client.GetResources(configName) + resource, err := d.store.GetResources(configName) if err != nil { d.log.Error(err, "failed to get resources from store", "configName", configName) return @@ -297,7 +297,7 @@ func (d *apisixProvider) handleEmptyFailedStatuses( d.addResourceToStatusUpdateMap(obj.GetLabels(), failedStatus.Error(), statusUpdateMap) } - globalRules, err := d.client.ListGlobalRules(configName) + globalRules, err := d.store.ListGlobalRules(configName) if err != nil { d.log.Error(err, "failed to list global rules", "configName", configName) return @@ -319,7 +319,7 @@ func (d *apisixProvider) handleDetailedFailedStatuses( return } id := status.Event.ResourceID - labels, err := d.client.GetResourceLabel(configName, status.Event.ResourceType, id) + labels, err := d.store.GetResourceLabel(configName, status.Event.ResourceType, id) if err != nil { d.log.Error(err, "failed to get resource label", "configName", configName, diff --git a/internal/webhook/v1/adc_validation.go b/internal/webhook/v1/adc_validation.go index 2c980c0f..d4c45f9a 100644 --- a/internal/webhook/v1/adc_validation.go +++ b/internal/webhook/v1/adc_validation.go @@ -217,7 +217,6 @@ func (v *adcAdmissionValidator) buildIngressClassConfigs(ctx context.Context, ob func (v *adcAdmissionValidator) newTask(obj client.Object, configs map[internaltypes.NamespacedNameKind]adctypes.Config, resourceTypes []string, result *adctranslator.TranslateResult) *adcclient.Task { return &adcclient.Task{ - Key: utils.NamespacedNameKind(obj), Name: utils.NamespacedNameKind(obj).String(), Labels: label.GenLabel(obj), Configs: configs, diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index c9537fe1..4f1b6ff0 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -58,16 +58,6 @@ var ( Help: "Current length of the status update queue", }, ) - - // File I/O operation duration histogram - FileIODuration = prometheus.NewHistogramVec( - prometheus.HistogramOpts{ - Name: "apisix_ingress_file_io_duration_seconds", - Help: "Time spent on file I/O operations", - Buckets: prometheus.DefBuckets, - }, - []string{"operation", "status"}, - ) ) // init registers all metrics with the global prometheus registry @@ -78,7 +68,6 @@ func init() { ADCSyncTotal, ADCExecutionErrors, StatusUpdateQueueLength, - FileIODuration, ) } @@ -107,8 +96,3 @@ func IncStatusQueueLength() { func DecStatusQueueLength() { StatusUpdateQueueLength.Dec() } - -// RecordFileIODuration records the duration of a file I/O operation -func RecordFileIODuration(operation, status string, duration float64) { - FileIODuration.WithLabelValues(operation, status).Observe(duration) -}
