diff --git a/README.md b/README.md index 3f1031c..2ca0157 100644 --- a/README.md +++ b/README.md @@ -231,6 +231,7 @@ Below are the options that are available. - `--auth-key-path`/`AUTH_KEY_PATH` (required): Defines the file path of the service account key for the STACKIT API. Prefer using a Kubernetes Secret mounted as a file and set `AUTH_KEY_PATH` to the in-container path (e.g. `/var/run/secrets/stackit/sa.json`). +- `--token-url`/`TOKEN_URL` (optional): Specifies alternative URL for authentication with service account key (default "https://service-account.api.stackit.cloud/token"). - `--worker`/`WORKER` (optional): Specifies the number of workers to employ for querying the API. Given that we need to iterate over all zones and records, it can be parallelized. However, it is important to avoid setting this number excessively high to prevent receiving 429 rate limiting from the API (default 10). diff --git a/cmd/webhook/cmd/root.go b/cmd/webhook/cmd/root.go index f72d2c7..1f05107 100644 --- a/cmd/webhook/cmd/root.go +++ b/cmd/webhook/cmd/root.go @@ -22,6 +22,7 @@ var ( apiPort string authBearerToken string authKeyPath string + tokenUrl string baseUrl string projectID string worker int @@ -45,7 +46,7 @@ var rootCmd = &cobra.Command{ endpointDomainFilter := endpoint.DomainFilter{Filters: domainFilter} - stackitConfigOptions, err := stackit.SetConfigOptions(baseUrl, authBearerToken, authKeyPath) + stackitConfigOptions, err := stackit.SetConfigOptions(baseUrl, authBearerToken, authKeyPath, tokenUrl) if err != nil { panic(err) } @@ -76,9 +77,8 @@ var rootCmd = &cobra.Command{ func getLogger() *zap.Logger { cfg := zap.Config{ - Level: zap.NewAtomicLevelAt(getZapLogLevel()), - Encoding: "json", // or "console" - // ... other zap configuration as needed + Level: zap.NewAtomicLevelAt(getZapLogLevel()), + Encoding: "json", OutputPaths: []string{"stdout"}, ErrorOutputPaths: []string{"stderr"}, } @@ -116,12 +116,10 @@ func init() { rootCmd.PersistentFlags().StringVar(&apiPort, "api-port", "8888", "Specifies the port to listen on.") rootCmd.PersistentFlags().StringVar(&authBearerToken, "auth-token", "", "Defines the authentication token for the STACKIT API. Mutually exclusive with 'auth-key-path'.") rootCmd.PersistentFlags().StringVar(&authKeyPath, "auth-key-path", "", "Defines the file path of the service account key for the STACKIT API. Mutually exclusive with 'auth-token'.") + rootCmd.PersistentFlags().StringVar(&tokenUrl, "token-url", "", "Defines the authentication token endpoint for the STACKIT API.") rootCmd.PersistentFlags().StringVar(&baseUrl, "base-url", "https://dns.api.stackit.cloud", " Identifies the Base URL for utilizing the API.") rootCmd.PersistentFlags().StringVar(&projectID, "project-id", "", "Specifies the project id of the STACKIT project.") - rootCmd.PersistentFlags().IntVar(&worker, "worker", 10, "Specifies the number "+ - "of workers to employ for querying the API. Given that we need to iterate over all zones and "+ - "records, it can be parallelized. However, it is important to avoid setting this number "+ - "excessively high to prevent receiving 429 rate limiting from the API.") + rootCmd.PersistentFlags().IntVar(&worker, "worker", 10, "Specifies the number of workers to employ for querying the API. Given that we need to iterate over all zones and records, it can be parallelized. However, it is important to avoid setting this number excessively high to prevent receiving 429 rate limiting from the API.") rootCmd.PersistentFlags().StringArrayVar(&domainFilter, "domain-filter", []string{}, "Establishes a filter for DNS zone names") rootCmd.PersistentFlags().BoolVar(&dryRun, "dry-run", false, "Specifies whether to perform a dry run.") rootCmd.PersistentFlags().StringVar(&logLevel, "log-level", "info", "Specifies the log level. Possible values are: debug, info, warn, error") diff --git a/internal/stackitprovider/apply_changes.go b/internal/stackitprovider/apply_changes.go index 5518d8b..bc8076b 100644 --- a/internal/stackitprovider/apply_changes.go +++ b/internal/stackitprovider/apply_changes.go @@ -2,6 +2,7 @@ package stackitprovider import ( "context" + "errors" "fmt" "sync" @@ -11,31 +12,72 @@ import ( "sigs.k8s.io/external-dns/plan" ) -// ApplyChanges applies a given set of changes in a given zone. +// ApplyChanges applies a given set of DNS changes to the STACKIT DNS API. +// It enforces a strict phase-based execution order to prevent orphaned records +// and mitigate quota limit issues (e.g., max 10k records per zone). +// Deletions are processed before creations to free up zone quota. func (d *StackitDNSProvider) ApplyChanges(ctx context.Context, changes *plan.Changes) error { - // Preallocate to avoid repeated growth (prealloc) - totalTasks := len(changes.Create) + len(changes.UpdateNew) + len(changes.Delete) - tasks := make([]changeTask, 0, totalTasks) - - // create rr set. POST /v1/projects/{projectId}/zones/{zoneId}/rrsets - tasks = append(tasks, d.buildRRSetTasks(changes.Create, CREATE)...) - // update rr set. PATCH /v1/projects/{projectId}/zones/{zoneId}/rrsets/{rrSetId} - tasks = append(tasks, d.buildRRSetTasks(changes.UpdateNew, UPDATE)...) + if len(changes.Create)+len(changes.UpdateNew)+len(changes.Delete) == 0 { + return nil + } d.logger.Info("records to delete", zap.String("records", fmt.Sprintf("%v", changes.Delete))) - // delete rr set. DELETE /v1/projects/{projectId}/zones/{zoneId}/rrsets/{rrSetId} - tasks = append(tasks, d.buildRRSetTasks(changes.Delete, DELETE)...) - zones, err := d.zoneFetcherClient.zones(ctx) if err != nil { return err } - return d.handleRRSetWithWorkers(ctx, tasks, zones) + // Separate ownership records (TXT) from target records (A, CNAME, etc.) + // to enforce strict dependency ordering and prevent orphaned records. + deleteTXT, deleteOther := splitTXTAndOther(changes.Delete) + updateTXT, updateOther := splitTXTAndOther(changes.UpdateNew) + createTXT, createOther := splitTXTAndOther(changes.Create) + + // Execution order is critical. + // 1. Delete targets first, then their TXT ownership records. + // 2. Update TXT ownerships, then targets. + // 3. Create TXT ownerships first, then create targets. + batches := [][]changeTask{ + d.buildRRSetTasks(deleteOther, DELETE), + d.buildRRSetTasks(deleteTXT, DELETE), + d.buildRRSetTasks(updateTXT, UPDATE), + d.buildRRSetTasks(updateOther, UPDATE), + d.buildRRSetTasks(createTXT, CREATE), + d.buildRRSetTasks(createOther, CREATE), + } + + for _, batch := range batches { + if len(batch) == 0 { + continue + } + + // If any batch fails (e.g., hitting a quota limit), the entire sync loop aborts. + // This leaves the DNS state consistent for the next retry attempt. + if err := d.handleRRSetWithWorkers(ctx, batch, zones); err != nil { + return err + } + } + + return nil +} + +// splitTXTAndOther separates TXT records from all other record types. +// External-DNS relies on TXT records to track ownership. +func splitTXTAndOther(endpoints []*endpoint.Endpoint) ([]*endpoint.Endpoint, []*endpoint.Endpoint) { + var txt, other []*endpoint.Endpoint + for _, ep := range endpoints { + if ep.RecordType == "TXT" { + txt = append(txt, ep) + } else { + other = append(other, ep) + } + } + + return txt, other } -// handleRRSetWithWorkers handles the given endpoints with workers to optimize speed. +// buildRRSetTasks wraps endpoint changes into executable tasks for the worker pool. func (d *StackitDNSProvider) buildRRSetTasks( endpoints []*endpoint.Endpoint, action string, @@ -52,19 +94,25 @@ func (d *StackitDNSProvider) buildRRSetTasks( return tasks } -// handleRRSetWithWorkers handles the given endpoints with workers to optimize speed. +// handleRRSetWithWorkers processes a batch of DNS changes concurrently. +// It implements a fail-fast mechanism: if any worker encounters an error +// (like a 4xx quota limit reached), it cancels the context to stop remaining queued tasks, +// preventing an API DoS. func (d *StackitDNSProvider) handleRRSetWithWorkers( ctx context.Context, tasks []changeTask, zones []stackitdnsclient.Zone, ) error { + cancelCtx, cancel := context.WithCancel(ctx) + defer cancel() + workerChannel := make(chan changeTask, len(tasks)) errorChannel := make(chan error, len(tasks)) var wg sync.WaitGroup for i := 0; i < d.workers; i++ { wg.Add(1) - go d.changeWorker(ctx, workerChannel, errorChannel, zones, &wg) + go d.changeWorker(cancelCtx, workerChannel, errorChannel, zones, &wg) } for _, task := range tasks { @@ -72,32 +120,44 @@ func (d *StackitDNSProvider) handleRRSetWithWorkers( } close(workerChannel) - // capture first error - var err error + var firstErr error for i := 0; i < len(tasks); i++ { - err = <-errorChannel - if err != nil { - break + err := <-errorChannel + if err != nil && firstErr == nil { + if !errors.Is(err, context.Canceled) { + firstErr = err + d.logger.Error("error encountered during batch processing, canceling remaining tasks", zap.Error(err)) + // Fail fast: signal all active and pending workers to abort. + cancel() + } } } // wait until all workers have finished wg.Wait() - return err + return firstErr } -// changeWorker is a worker that handles changes passed by a channel. +// changeWorker listens for tasks on the workerChannel and executes the appropriate API call. +// It respects context cancellation to safely abort pending operations. func (d *StackitDNSProvider) changeWorker( ctx context.Context, - changes chan changeTask, - errorChannel chan error, + changes <-chan changeTask, + errorChannel chan<- error, zones []stackitdnsclient.Zone, wg *sync.WaitGroup, ) { defer wg.Done() for change := range changes { + // Check for context cancellation before processing the next task. + if err := ctx.Err(); err != nil { + errorChannel <- err + + continue + } + var err error switch change.action { case CREATE: diff --git a/internal/stackitprovider/apply_changes_test.go b/internal/stackitprovider/apply_changes_test.go index b831b5a..d1a5714 100644 --- a/internal/stackitprovider/apply_changes_test.go +++ b/internal/stackitprovider/apply_changes_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "sync/atomic" "testing" stackitdnsclient "github.com/stackitcloud/stackit-sdk-go/services/dns/v1api" @@ -198,6 +199,52 @@ func TestPartialUpdate(t *testing.T) { assert.True(t, rrSetUpdated, "rrset was not updated") } +func TestFailFastCancellation(t *testing.T) { + t.Parallel() + ctx := context.Background() + + validZoneResponse := getValidResponseZoneAllBytes(t) + + var requestCount atomic.Int32 + mux := http.NewServeMux() + server := httptest.NewServer(mux) + defer server.Close() + + setUpCommonEndpoints(mux, validZoneResponse, http.StatusOK) + + mux.HandleFunc("/v1/projects/1234/zones/1234/rrsets", func(w http.ResponseWriter, r *http.Request) { + requestCount.Add(1) + w.Header().Set("Content-Type", "application/json") + // Force the first requests to fail with a quota-like error + w.WriteHeader(http.StatusUnprocessableEntity) + w.Write([]byte(`{"message": "quota exceeded"}`)) + }) + + stackitDnsProvider, err := getDefaultTestProvider(server) + assert.NoError(t, err) + + // Create a large batch of changes to ensure the queue fills up and tests the cancellation + endpoints := make([]*endpoint.Endpoint, 0, 50) + for i := 0; i < 50; i++ { + endpoints = append(endpoints, &endpoint.Endpoint{ + DNSName: fmt.Sprintf("test%d.com", i), + Targets: endpoint.Targets{"1.2.3.4"}, + RecordType: "A", + }) + } + + changes := &plan.Changes{ + Create: endpoints, + } + + err = stackitDnsProvider.ApplyChanges(ctx, changes) + assert.Error(t, err) + + // If fail-fast is working, the request count should be significantly less than 50 + // because the context cancellation stops the remaining workers from executing HTTP requests. + assert.Less(t, int(requestCount.Load()), 50, "expected fail-fast to cancel remaining requests") +} + // setUpCommonEndpoints for all change types. func setUpCommonEndpoints(mux *http.ServeMux, responseZone []byte, responseZoneCode int) { mux.HandleFunc("/v1/projects/1234/zones", func(w http.ResponseWriter, r *http.Request) { diff --git a/pkg/stackit/options.go b/pkg/stackit/options.go index 69fc2b0..d8e508b 100644 --- a/pkg/stackit/options.go +++ b/pkg/stackit/options.go @@ -1,6 +1,7 @@ package stackit import ( + "context" "fmt" "net/http" "time" @@ -13,7 +14,7 @@ import ( // passed bearerToken and keyPath parameters. If no baseURL or an invalid // combination of auth options is given (neither or both), the function returns // an error. -func SetConfigOptions(baseURL, bearerToken, keyPath string) ([]stackitconfig.ConfigurationOption, error) { +func SetConfigOptions(baseURL, bearerToken, keyPath, tokenURL string) ([]stackitconfig.ConfigurationOption, error) { if len(baseURL) == 0 { return nil, fmt.Errorf("base-url is required") } @@ -35,6 +36,10 @@ func SetConfigOptions(baseURL, bearerToken, keyPath string) ([]stackitconfig.Con if bearerTokenSet { return append(options, stackitconfig.WithToken(bearerToken)), nil } + if len(tokenURL) > 0 { + options = append(options, stackitconfig.WithTokenEndpoint(tokenURL)) + } + options = append(options, stackitconfig.WithBackgroundTokenRefresh(context.Background())) return append(options, stackitconfig.WithServiceAccountKeyPath(keyPath)), nil } diff --git a/pkg/stackit/options_test.go b/pkg/stackit/options_test.go index 9a024c0..1526f72 100644 --- a/pkg/stackit/options_test.go +++ b/pkg/stackit/options_test.go @@ -9,7 +9,7 @@ import ( func TestMissingBaseURL(t *testing.T) { t.Parallel() - options, err := SetConfigOptions("", "", "") + options, err := SetConfigOptions("", "", "", "") assert.ErrorContains(t, err, "base-url") assert.Nil(t, options) } @@ -17,7 +17,7 @@ func TestMissingBaseURL(t *testing.T) { func TestBothAuthOptionsMissing(t *testing.T) { t.Parallel() - options, err := SetConfigOptions("https://example.com", "", "") + options, err := SetConfigOptions("https://example.com", "", "", "") assert.ErrorContains(t, err, "auth-token or auth-key-path") assert.Nil(t, options) } @@ -25,7 +25,7 @@ func TestBothAuthOptionsMissing(t *testing.T) { func TestBothAuthOptionsSet(t *testing.T) { t.Parallel() - options, err := SetConfigOptions("https://example.com", "token", "key/path") + options, err := SetConfigOptions("https://example.com", "token", "key/path", "") assert.ErrorContains(t, err, "auth-token or auth-key-path") assert.Nil(t, options) } @@ -33,7 +33,7 @@ func TestBothAuthOptionsSet(t *testing.T) { func TestBearerTokenSet(t *testing.T) { t.Parallel() - options, err := SetConfigOptions("https://example.com", "token", "") + options, err := SetConfigOptions("https://example.com", "token", "", "") assert.NoError(t, err) assert.Len(t, options, 3) } @@ -41,7 +41,15 @@ func TestBearerTokenSet(t *testing.T) { func TestKeyPathSet(t *testing.T) { t.Parallel() - options, err := SetConfigOptions("https://example.com", "", "key/path") + options, err := SetConfigOptions("https://example.com", "", "key/path", "") assert.NoError(t, err) - assert.Len(t, options, 3) + assert.Len(t, options, 4) +} + +func TestKeyPathAndURLSet(t *testing.T) { + t.Parallel() + + options, err := SetConfigOptions("https://example.com", "", "key/path", "https://alternative.url.stackit.cloud/token") + assert.NoError(t, err) + assert.Len(t, options, 5) }