Skip to content

refactor: let adc client independent of external state #478

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bzp2010 wants to merge 2 commits into
base: master
Choose a base branch
from
Open

refactor: let adc client independent of external state #478

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
201d106
refactor: let adc client independent of external state
bzp2010 Sep 10, 2026
fc4f27d
fix e2e
bzp2010 Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
385 changes: 81 additions & 304 deletions internal/adc/client/client.go
View file
Open in desktop

Large diffs are not rendered by default.

126 changes: 17 additions & 109 deletions internal/adc/client/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import (
"io"
"net"
"net/http"
"os"
"strings"
"time"

Expand All @@ -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
Expand Down Expand Up @@ -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,
}
Expand All @@ -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) {
Expand All @@ -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,
}
Expand All @@ -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) {
Expand All @@ -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)
Expand All @@ -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)
}
Expand All @@ -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
Expand All @@ -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,
Expand All @@ -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 != "",
)
Expand Down
50 changes: 25 additions & 25 deletions internal/adc/client/executor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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.
Expand All @@ -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{},
}
}
Expand All @@ -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)
}

Expand All @@ -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)
}
Expand All @@ -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) {
Expand All @@ -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)
})
Expand All @@ -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)
}
Expand All @@ -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)
}
Expand All @@ -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")
Expand All @@ -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)
Expand Down
1 change: 0 additions & 1 deletion internal/adc/client/redaction_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}},
Expand Down
Loading
Loading