fix: SA authentication and change how errors in requests are handled (#79)

* fix: add token URL parameter to SetConfigOptions for SA authentication

* fix: enforce strict execution order for DNS changes to prevent orphaned records and mitigate quota issues

* add test, fix lint, add new option to documentation

Author: ondbeh

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).

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")

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,52 +94,70 @@ 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 {
 		workerChannel <- task
 	}
 	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:

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) {

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
 }

pkg/stackit/options_test.go

@@ -9,39 +9,47 @@ import (
 func TestMissingBaseURL(t *testing.T) {
 	t.Parallel()
 
-	options, err := SetConfigOptions("", "", "")
+	options, err := SetConfigOptions("", "", "", "")
 	assert.ErrorContains(t, err, "base-url")
 	assert.Nil(t, options)
 }
 
 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)
 }
 
 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)
 }
 
 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)
 }
 
 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)
 }