diff --git a/README.md b/README.md index f0ae9fe..f713077 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,8 @@ task docker # Build Docker image | `--metrics-port` | Port for Prometheus metrics and health checks | `8080` | No | | `--log-level` | Log level (debug, info, warn, error) | `info` | No | | `--dry-run` | Dry-run mode: do not make actual DNS changes | `false` | No | +| `--apply-timeout` | Maximum time to spend on one batch of record changes. Changes not reached in time are skipped, the request fails with a 503 that names the limit, and ExternalDNS retries them on its next sync | `5m` | No | +| `--apply-concurrency` | Number of record changes applied in parallel | `4` | No | ### Single-Cluster (Flags Only) diff --git a/go.mod b/go.mod index c54888c..07a2cbd 100644 --- a/go.mod +++ b/go.mod @@ -18,6 +18,7 @@ require ( require ( github.com/aws/aws-sdk-go v1.55.5 // indirect github.com/beorn7/perks v1.0.1 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.12.2 // indirect diff --git a/go.sum b/go.sum index 2d31709..99d1774 100644 --- a/go.sum +++ b/go.sum @@ -2,6 +2,8 @@ github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/internal/provider/apply_test.go b/internal/provider/apply_test.go new file mode 100644 index 0000000..321bae6 --- /dev/null +++ b/internal/provider/apply_test.go @@ -0,0 +1,276 @@ +package provider + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + "sigs.k8s.io/external-dns/endpoint" + "sigs.k8s.io/external-dns/plan" + extdnsprovider "sigs.k8s.io/external-dns/provider" + "sigs.k8s.io/external-dns/provider/webhook" +) + +type slowAPI struct { + delay time.Duration + panicOn string + inFlight atomic.Int32 + maxSeen atomic.Int32 + mu sync.Mutex + calls []string +} + +func (a *slowAPI) wait(ctx context.Context, call string) error { + n := a.inFlight.Add(1) + defer a.inFlight.Add(-1) + for { + seen := a.maxSeen.Load() + if n <= seen || a.maxSeen.CompareAndSwap(seen, n) { + break + } + } + a.mu.Lock() + a.calls = append(a.calls, call) + a.mu.Unlock() + if a.panicOn != "" && strings.HasSuffix(call, a.panicOn) { + panic("injected") + } + + select { + case <-time.After(a.delay): + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (a *slowAPI) funcs() interceptor.Funcs { + return interceptor.Funcs{ + Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error { + if err := a.wait(ctx, "create "+obj.GetName()); err != nil { + return err + } + return c.Create(ctx, obj, opts...) + }, + Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error { + if err := a.wait(ctx, "delete "+obj.GetName()); err != nil { + return err + } + return c.Delete(ctx, obj, opts...) + }, + } +} + +func setupSlowServer(t *testing.T, api *slowAPI, cfg *Config, objects ...client.Object) (*Server, client.Client) { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, dnsv1alpha1.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + + objects = append(objects, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}}, + &dnsv1alpha1.DNSZone{ + ObjectMeta: metav1.ObjectMeta{Name: "example-com", Namespace: "default"}, + Spec: dnsv1alpha1.DNSZoneSpec{DomainName: "example.com"}, + }, + ) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objects...). + WithInterceptorFuncs(api.funcs()). + Build() + + logger := logrus.New() + logger.SetLevel(logrus.PanicLevel) + + src := NewZoneSource("test", fakeClient, ZoneSourceConfig{}, cfg, logger) + p, err := NewProvider(cfg, "test-owner", []*ZoneSource{src}) + require.NoError(t, err) + require.NoError(t, p.registry.Refresh(context.Background())) + + return NewServer(p, cfg), fakeClient +} + +func serveWebhook(t *testing.T, s *Server) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + srv := s.newWebhookServer(listener.Addr().String()) + go func() { _ = srv.Serve(listener) }() + t.Cleanup(func() { _ = srv.Close() }) + + return "http://" + listener.Addr().String() +} + +func createChanges(n int) *plan.Changes { + changes := &plan.Changes{} + for i := range n { + changes.Create = append(changes.Create, &endpoint.Endpoint{ + DNSName: fmt.Sprintf("app%d.example.com", i), + Targets: endpoint.Targets{"192.0.2.1"}, + RecordType: endpoint.RecordTypeA, + }) + } + return changes +} + +func jsonReader(v any) (io.Reader, error) { + b, err := json.Marshal(v) + if err != nil { + return nil, err + } + return bytes.NewReader(b), nil +} + +func TestConfig_WebhookWriteTimeoutExceedsApplyTimeout(t *testing.T) { + for _, cfg := range []*Config{{}, DefaultConfig(), {ApplyTimeout: 20 * time.Minute}} { + assert.Greater(t, cfg.WebhookWriteTimeout(), cfg.EffectiveApplyTimeout()) + } +} + +func TestApplyChanges_SlowBatchCompletesOverHTTP(t *testing.T) { + api := &slowAPI{delay: 50 * time.Millisecond} + cfg := &Config{ApplyTimeout: 5 * time.Second, ApplyConcurrency: 4} + s, c := setupSlowServer(t, api, cfg) + url := serveWebhook(t, s) + + client, err := webhook.NewWebhookProvider(url) + require.NoError(t, err) + + start := time.Now() + require.NoError(t, client.ApplyChanges(context.Background(), createChanges(20))) + elapsed := time.Since(start) + + var list dnsv1alpha1.DNSRecordSetList + require.NoError(t, c.List(context.Background(), &list)) + assert.Len(t, list.Items, 20) + assert.LessOrEqual(t, int(api.maxSeen.Load()), 4) + assert.Greater(t, int(api.maxSeen.Load()), 1) + assert.Less(t, elapsed, 20*api.delay, "changes should be applied in parallel") +} + +func TestApplyChanges_OverrunReturnsSoftErrorNotEOF(t *testing.T) { + api := &slowAPI{delay: 100 * time.Millisecond} + cfg := &Config{ApplyTimeout: 250 * time.Millisecond, ApplyConcurrency: 1} + s, c := setupSlowServer(t, api, cfg) + url := serveWebhook(t, s) + + client, err := webhook.NewWebhookProvider(url) + require.NoError(t, err) + + err = client.ApplyChanges(context.Background(), createChanges(10)) + require.Error(t, err) + assert.True(t, errors.Is(err, extdnsprovider.SoftError), + "external-dns must see a soft error so it retries instead of exiting, got: %v", err) + assert.NotContains(t, err.Error(), "EOF") + + var list dnsv1alpha1.DNSRecordSetList + require.NoError(t, c.List(context.Background(), &list)) + assert.NotEmpty(t, list.Items) + assert.Less(t, len(list.Items), 10) +} + +func TestApplyChanges_OverrunResponseNamesTheLimit(t *testing.T) { + api := &slowAPI{delay: 100 * time.Millisecond} + cfg := &Config{ApplyTimeout: 250 * time.Millisecond, ApplyConcurrency: 1} + s, _ := setupSlowServer(t, api, cfg) + url := serveWebhook(t, s) + + body, err := jsonReader(createChanges(10)) + require.NoError(t, err) + req, err := http.NewRequest(http.MethodPost, url+"/records", body) + require.NoError(t, err) + req.Header.Set(ContentTypeHeader, MediaTypeFormatAndVersion) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err, "the server must answer rather than drop the connection") + defer func() { _ = resp.Body.Close() }() + + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + msg, err := io.ReadAll(resp.Body) + require.NoError(t, err) + assert.Contains(t, string(msg), "apply timeout of 250ms reached") + assert.Contains(t, string(msg), "of 10 changes") + assert.Contains(t, string(msg), "--apply-timeout") +} + +func TestApplyChanges_DeleteRunsBeforeCreateOfSameRecordSet(t *testing.T) { + name := GenerateRecordSetName("app.example.com", endpoint.RecordTypeA, "") + existing := &dnsv1alpha1.DNSRecordSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Labels: map[string]string{LabelOwner: "test-owner", LabelManagedBy: ManagedByValue}, + }, + Spec: dnsv1alpha1.DNSRecordSetSpec{ + DNSZoneRef: corev1.LocalObjectReference{Name: "example-com"}, + RecordType: dnsv1alpha1.RRTypeA, + }, + } + + api := &slowAPI{delay: time.Millisecond} + cfg := &Config{ApplyTimeout: 5 * time.Second, ApplyConcurrency: 4} + s, c := setupSlowServer(t, api, cfg, existing) + + changes := &plan.Changes{ + Create: []*endpoint.Endpoint{ + {DNSName: "app.example.com", Targets: endpoint.Targets{"192.0.2.2"}, RecordType: endpoint.RecordTypeA}, + {DNSName: "new.example.com", Targets: endpoint.Targets{"192.0.2.3"}, RecordType: endpoint.RecordTypeA}, + }, + Delete: []*endpoint.Endpoint{ + {DNSName: "app.example.com", Targets: endpoint.Targets{"192.0.2.1"}, RecordType: endpoint.RecordTypeA}, + {DNSName: "old.example.com", Targets: endpoint.Targets{"192.0.2.4"}, RecordType: endpoint.RecordTypeA}, + }, + } + require.NoError(t, s.provider.ApplyChanges(context.Background(), changes)) + + var got dnsv1alpha1.DNSRecordSet + require.NoError(t, c.Get(context.Background(), types.NamespacedName{Name: name, Namespace: "default"}, &got)) + require.Len(t, got.Spec.Records, 1) + assert.Equal(t, "192.0.2.2", got.Spec.Records[0].A.Content) + + api.mu.Lock() + defer api.mu.Unlock() + require.Len(t, api.calls, 4) + assert.Equal(t, "delete "+name, api.calls[0]) + assert.True(t, strings.HasPrefix(api.calls[3], "delete old-example-com-a-"), + "unrelated deletes run after creates, got %v", api.calls) +} + +func TestApplyChanges_PanicInOneChangeDoesNotAbortBatch(t *testing.T) { + api := &slowAPI{ + delay: time.Millisecond, + panicOn: GenerateRecordSetName("app0.example.com", endpoint.RecordTypeA, ""), + } + cfg := &Config{ApplyTimeout: 5 * time.Second, ApplyConcurrency: 2} + s, c := setupSlowServer(t, api, cfg) + + require.NoError(t, s.provider.ApplyChanges(context.Background(), createChanges(4))) + + var list dnsv1alpha1.DNSRecordSetList + require.NoError(t, c.List(context.Background(), &list)) + assert.Len(t, list.Items, 3) +} diff --git a/internal/provider/config.go b/internal/provider/config.go index e66212d..ef766a5 100644 --- a/internal/provider/config.go +++ b/internal/provider/config.go @@ -17,6 +17,41 @@ type Config struct { MetricsPort int `yaml:"metricsPort"` LogLevel string `yaml:"logLevel"` DryRun bool `yaml:"dryRun"` + + // ApplyTimeout bounds how long a single POST /records may spend applying + // changes. The webhook server's write timeout is derived from it so the + // handler always answers before the server closes the connection. + ApplyTimeout time.Duration `yaml:"applyTimeout"` + // ApplyConcurrency is the number of record changes applied in parallel. + ApplyConcurrency int `yaml:"applyConcurrency"` +} + +const ( + defaultApplyTimeout = 5 * time.Minute + defaultApplyConcurrency = 4 + writeTimeoutGrace = 30 * time.Second +) + +// EffectiveApplyTimeout returns ApplyTimeout, or the default when unset. +func (c *Config) EffectiveApplyTimeout() time.Duration { + if c.ApplyTimeout <= 0 { + return defaultApplyTimeout + } + return c.ApplyTimeout +} + +// EffectiveApplyConcurrency returns ApplyConcurrency, or the default when unset. +func (c *Config) EffectiveApplyConcurrency() int { + if c.ApplyConcurrency <= 0 { + return defaultApplyConcurrency + } + return c.ApplyConcurrency +} + +// WebhookWriteTimeout is the webhook server's write timeout. It exceeds the +// apply timeout so a request that runs out of time still gets a response. +func (c *Config) WebhookWriteTimeout() time.Duration { + return c.EffectiveApplyTimeout() + writeTimeoutGrace } // ZoneSourceConfig holds connection and filtering settings for a single @@ -65,5 +100,8 @@ func DefaultConfig() *Config { BindAddress: "0.0.0.0", MetricsPort: 8080, LogLevel: "info", + + ApplyTimeout: defaultApplyTimeout, + ApplyConcurrency: defaultApplyConcurrency, } } diff --git a/internal/provider/provider.go b/internal/provider/provider.go index abc896e..161278b 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -3,6 +3,7 @@ package provider import ( "context" "fmt" + "sync" log "github.com/sirupsen/logrus" dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" @@ -108,31 +109,108 @@ func (p *Provider) resolveZoneForRecordSet(rs *dnsv1alpha1.DNSRecordSet, src *Zo return zone } +// IncompleteApplyError reports that the context ended before every change in +// a batch was attempted. Changes that were not attempted are left for the next +// sync. +type IncompleteApplyError struct { + Attempted int + Total int + Err error +} + +func (e *IncompleteApplyError) Error() string { + return fmt.Sprintf("attempted %d of %d changes before stopping: %v", e.Attempted, e.Total, e.Err) +} + +func (e *IncompleteApplyError) Unwrap() error { return e.Err } + +type changeFunc func(context.Context, *endpoint.Endpoint) error + // ApplyChanges applies the given changes to DNS records, routing each operation -// to the zone source that owns the target zone. +// to the zone source that owns the target zone. Changes within a phase run in +// parallel, bounded by the configured concurrency. A delete whose record set is +// recreated in the same batch runs before the creates; every other delete runs +// last. Per-change failures are logged and do not fail the batch. If ctx ends +// first, the remaining changes are skipped and an *IncompleteApplyError is +// returned. func (p *Provider) ApplyChanges(ctx context.Context, changes *plan.Changes) error { + total := len(changes.Create) + len(changes.UpdateNew) + len(changes.Delete) p.logger.Debugf("Applying changes: %d creates, %d updates, %d deletes", len(changes.Create), len(changes.UpdateNew), len(changes.Delete)) + created := make(map[string]bool, len(changes.Create)) for _, ep := range changes.Create { - if err := p.createRecord(ctx, ep); err != nil { - p.logger.Errorf("Failed to create record for %s: %v", ep.DNSName, err) + created[recordSetNameFor(ep)] = true + } + var replacedDeletes, deletes []*endpoint.Endpoint + for _, ep := range changes.Delete { + if created[recordSetNameFor(ep)] { + replacedDeletes = append(replacedDeletes, ep) + } else { + deletes = append(deletes, ep) } } - for _, ep := range changes.UpdateNew { - if err := p.updateRecord(ctx, ep); err != nil { - p.logger.Errorf("Failed to update record for %s: %v", ep.DNSName, err) + phases := []struct { + op string + eps []*endpoint.Endpoint + apply changeFunc + }{ + {"delete", replacedDeletes, p.deleteRecord}, + {"create", changes.Create, p.createRecord}, + {"update", changes.UpdateNew, p.updateRecord}, + {"delete", deletes, p.deleteRecord}, + } + + attempted := 0 + for _, phase := range phases { + attempted += p.applyPhase(ctx, phase.op, phase.eps, phase.apply) + if err := ctx.Err(); err != nil { + return &IncompleteApplyError{Attempted: attempted, Total: total, Err: err} } } + return nil +} - for _, ep := range changes.Delete { - if err := p.deleteRecord(ctx, ep); err != nil { - p.logger.Errorf("Failed to delete record for %s: %v", ep.DNSName, err) +func (p *Provider) applyPhase(ctx context.Context, op string, eps []*endpoint.Endpoint, apply changeFunc) int { + sem := make(chan struct{}, p.config.EffectiveApplyConcurrency()) + var wg sync.WaitGroup + attempted := 0 + + for _, ep := range eps { + select { + case <-ctx.Done(): + case sem <- struct{}{}: } + if ctx.Err() != nil { + break + } + attempted++ + wg.Add(1) + go func(ep *endpoint.Endpoint) { + defer wg.Done() + defer func() { <-sem }() + if err := p.applyOne(ctx, apply, ep); err != nil { + p.logger.Errorf("Failed to %s record for %s: %v", op, ep.DNSName, err) + } + }(ep) } - return nil + wg.Wait() + return attempted +} + +func (p *Provider) applyOne(ctx context.Context, apply changeFunc, ep *endpoint.Endpoint) (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("panic: %v", r) + } + }() + return apply(ctx, ep) +} + +func recordSetNameFor(ep *endpoint.Endpoint) string { + return GenerateRecordSetName(ep.DNSName, ep.RecordType, ep.SetIdentifier) } func (p *Provider) createRecord(ctx context.Context, ep *endpoint.Endpoint) error { @@ -198,8 +276,7 @@ func (p *Provider) deleteRecord(ctx context.Context, ep *endpoint.Endpoint) erro return fmt.Errorf("no zone found for domain %s: %w", ep.DNSName, err) } - name := GenerateRecordSetName(ep.DNSName, ep.RecordType, ep.SetIdentifier) - return match.Source.RecordSets().Delete(ctx, name, match.Zone.Namespace) + return match.Source.RecordSets().Delete(ctx, recordSetNameFor(ep), match.Zone.Namespace) } // AdjustEndpoints modifies endpoints before they are processed by the plan. diff --git a/internal/provider/server.go b/internal/provider/server.go index e3f8be6..6782da0 100644 --- a/internal/provider/server.go +++ b/internal/provider/server.go @@ -3,6 +3,7 @@ package provider import ( "context" "encoding/json" + "errors" "fmt" "net" "net/http" @@ -72,17 +73,7 @@ func (s *Server) Start(ctx context.Context) error { // Start webhook server webhookAddr := fmt.Sprintf("%s:%d", s.config.BindAddress, s.config.Port) - webhookMux := http.NewServeMux() - webhookMux.HandleFunc("/", s.instrumentHandler("/", s.negotiateHandler)) - webhookMux.HandleFunc("/records", s.instrumentHandler("/records", s.recordsHandler)) - webhookMux.HandleFunc("/adjustendpoints", s.instrumentHandler("/adjustendpoints", s.adjustEndpointsHandler)) - - s.webhookServer = &http.Server{ - Addr: webhookAddr, - Handler: webhookMux, - ReadTimeout: 30 * time.Second, - WriteTimeout: 30 * time.Second, - } + s.webhookServer = s.newWebhookServer(webhookAddr) webhookListener, err := net.Listen("tcp", webhookAddr) if err != nil { @@ -97,6 +88,20 @@ func (s *Server) Start(ctx context.Context) error { return nil } +func (s *Server) newWebhookServer(addr string) *http.Server { + mux := http.NewServeMux() + mux.HandleFunc("/", s.instrumentHandler("/", s.negotiateHandler)) + mux.HandleFunc("/records", s.instrumentHandler("/records", s.recordsHandler)) + mux.HandleFunc("/adjustendpoints", s.instrumentHandler("/adjustendpoints", s.adjustEndpointsHandler)) + + return &http.Server{ + Addr: addr, + Handler: mux, + ReadTimeout: 30 * time.Second, + WriteTimeout: s.config.WebhookWriteTimeout(), + } +} + // Shutdown gracefully shuts down both servers. func (s *Server) Shutdown(ctx context.Context) error { log.Info("Shutting down servers...") @@ -200,7 +205,21 @@ func (s *Server) handleApplyChanges(w http.ResponseWriter, r *http.Request) { return } - if err := s.provider.ApplyChanges(r.Context(), &changes); err != nil { + timeout := s.config.EffectiveApplyTimeout() + ctx, cancel := context.WithTimeout(r.Context(), timeout) + defer cancel() + + err := s.provider.ApplyChanges(ctx, &changes) + var incomplete *IncompleteApplyError + if errors.As(err, &incomplete) { + msg := fmt.Sprintf("apply timeout of %s reached: attempted %d of %d changes, the rest will be retried on the next sync; "+ + "raise --apply-timeout or --apply-concurrency to fit larger batches in one request", + timeout, incomplete.Attempted, incomplete.Total) + log.Error(msg) + http.Error(w, msg, http.StatusServiceUnavailable) + return + } + if err != nil { log.Errorf("Failed to apply changes: %v", err) http.Error(w, "internal server error", http.StatusInternalServerError) return diff --git a/main.go b/main.go index 508914f..09da469 100644 --- a/main.go +++ b/main.go @@ -32,6 +32,8 @@ func main() { logLevel string dryRun bool configPath string + applyTimeout time.Duration + applyConcurrency int ) flag.StringVar(&ownerID, "owner-id", "", "Owner ID to identify this ExternalDNS instance (required)") @@ -44,6 +46,8 @@ func main() { flag.StringVar(&logLevel, "log-level", "info", "Log level (debug, info, warn, error)") flag.BoolVar(&dryRun, "dry-run", false, "Dry-run mode: do not make actual DNS changes") flag.StringVar(&configPath, "config", "", "Path to YAML config file for additional zone sources") + flag.DurationVar(&applyTimeout, "apply-timeout", 5*time.Minute, "Maximum time to spend applying one batch of record changes; changes not reached in time are retried on the next sync") + flag.IntVar(&applyConcurrency, "apply-concurrency", 4, "Number of record changes applied in parallel") flag.Parse() cfg := &provider.Config{ @@ -53,6 +57,9 @@ func main() { MetricsPort: metricsPort, LogLevel: logLevel, DryRun: dryRun, + + ApplyTimeout: applyTimeout, + ApplyConcurrency: applyConcurrency, } if configPath != "" { @@ -73,6 +80,11 @@ func main() { }} } + if applyTimeout <= 0 || applyConcurrency <= 0 { + fmt.Fprintln(os.Stderr, "error: --apply-timeout and --apply-concurrency must be positive") + os.Exit(1) + } + if cfg.OwnerID == "" { fmt.Fprintln(os.Stderr, "error: --owner-id is required") flag.Usage()