diff --git a/cmd/ingress_opts.go b/cmd/ingress_opts.go index 90f25d79..69b133f1 100644 --- a/cmd/ingress_opts.go +++ b/cmd/ingress_opts.go @@ -24,6 +24,7 @@ type ingressControllerOpts struct { SyncAPIURL string SyncAPINamespaceID string SyncAPIToken string + MaxConcurrentReconciles int `validate:"min=1"` } const ( @@ -39,6 +40,7 @@ const ( syncAPIURL = "sync-api-url" syncAPINamespaceID = "sync-api-namespace-id" syncAPIToken = "sync-api-token" //nolint:gosec + maxConcurrentReconciles = "max-concurrent-reconciles" ) func (s *ingressControllerOpts) setupFlags(flags *pflag.FlagSet) { @@ -53,6 +55,8 @@ func (s *ingressControllerOpts) setupFlags(flags *pflag.FlagSet) { flags.StringVar(&s.SyncAPIURL, syncAPIURL, "", "unified API sync URL") flags.StringVar(&s.SyncAPINamespaceID, syncAPINamespaceID, "", "unified API sync namespace ID") flags.StringVar(&s.SyncAPIToken, syncAPIToken, "", "unified API sync token") + flags.IntVar(&s.MaxConcurrentReconciles, maxConcurrentReconciles, 1, + "maximum number of concurrent ingress reconciles") } func (s *ingressControllerOpts) Validate() error { @@ -76,6 +80,7 @@ func (s *ingressControllerOpts) getIngressControllerOptions() ([]ingress.Option, ingress.WithNamespaces(s.Namespaces), ingress.WithAnnotationPrefix(s.AnnotationPrefix), ingress.WithControllerName(s.ClassName), + ingress.WithMaxConcurrentReconciles(s.MaxConcurrentReconciles), } if name, err := s.getGlobalSettings(); err != nil { return nil, err diff --git a/controllers/ingress/concurrency_test.go b/controllers/ingress/concurrency_test.go new file mode 100644 index 00000000..559a1e13 --- /dev/null +++ b/controllers/ingress/concurrency_test.go @@ -0,0 +1,21 @@ +package ingress + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestWithMaxConcurrentReconciles(t *testing.T) { + ic := &ingressController{} + + // default zero value leaves controller-runtime default in place + assert.Equal(t, 0, ic.maxConcurrentReconciles) + + WithMaxConcurrentReconciles(4)(ic) + assert.Equal(t, 4, ic.maxConcurrentReconciles) + + // a value < 1 is still recorded; SetupWithManager decides whether to apply it + WithMaxConcurrentReconciles(0)(ic) + assert.Equal(t, 0, ic.maxConcurrentReconciles) +} diff --git a/controllers/ingress/controller.go b/controllers/ingress/controller.go index cba30bb4..7ede231a 100644 --- a/controllers/ingress/controller.go +++ b/controllers/ingress/controller.go @@ -10,6 +10,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -55,6 +56,10 @@ type ingressController struct { // globalSettings defines which global settings object to watch globalSettings *types.NamespacedName + // maxConcurrentReconciles is the number of concurrent ingress reconciles. + // Defaults to 1 (controller-runtime default) when unset. + maxConcurrentReconciles int + // object Kinds are frequently used, do not change and are cached endpointsKind string ingressKind string @@ -119,6 +124,14 @@ func WithWatchSettings(name types.NamespacedName) Option { } } +// WithMaxConcurrentReconciles sets the maximum number of concurrent ingress +// reconciles. A value < 1 leaves the controller-runtime default (1) in place. +func WithMaxConcurrentReconciles(n int) Option { + return func(ic *ingressController) { + ic.maxConcurrentReconciles = n + } +} + // SetupWithManager sets up the controller with the Manager func (r *ingressController) SetupWithManager(mgr ctrl.Manager) error { r.Client = mgr.GetClient() @@ -132,6 +145,11 @@ func (r *ingressController) SetupWithManager(mgr ctrl.Manager) error { r.endpointsKind = generic.GVKForType[*corev1.Endpoints](r.Scheme).Kind r.ingressClassKind = generic.GVKForType[*networkingv1.IngressClass](r.Scheme).Kind + opts := controller.Options{} + if r.maxConcurrentReconciles > 0 { + opts.MaxConcurrentReconciles = r.maxConcurrentReconciles + } + err := ctrl.NewControllerManagedBy(mgr). Named(controllerName). For(&networkingv1.Ingress{}). @@ -146,6 +164,7 @@ func (r *ingressController) SetupWithManager(mgr ctrl.Manager) error { Watches(&corev1.Service{}, handler.EnqueueRequestsFromMapFunc(r.getDependantIngressFn(r.serviceKind))). Watches(&corev1.Endpoints{}, handler.EnqueueRequestsFromMapFunc(r.getDependantIngressFn(r.endpointsKind))). WithEventFilter(predicate.ResourceVersionChangedPredicate{}). + WithOptions(opts). Complete(r) if err != nil { return err diff --git a/pomerium/envoy/validate_cache.go b/pomerium/envoy/validate_cache.go new file mode 100644 index 00000000..521cf4e3 --- /dev/null +++ b/pomerium/envoy/validate_cache.go @@ -0,0 +1,34 @@ +// Package envoy contains functions for working with an embedded envoy binary. +package envoy + +import ( + "crypto/sha256" + "sync" +) + +// defaultValidationCache is the process-wide cache of the last successfully +// validated config hash. It is shared across concurrent reconciles. +var defaultValidationCache = &validationCache{} + +// validationCache caches the hash of the most recently successfully-validated +// config so that repeated validation of an unchanged config can skip the +// expensive embedded-envoy subprocess. It is safe for concurrent use. +type validationCache struct { + mu sync.RWMutex + lastHash [sha256.Size]byte + valid bool +} + +// hit reports whether hash matches the last successfully-validated config. +func (c *validationCache) hit(hash [sha256.Size]byte) bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.valid && c.lastHash == hash +} + +// store records hash as the last successfully-validated config. +func (c *validationCache) store(hash [sha256.Size]byte) { + c.mu.Lock() + defer c.mu.Unlock() + c.lastHash, c.valid = hash, true +} diff --git a/pomerium/envoy/validate_cache_test.go b/pomerium/envoy/validate_cache_test.go new file mode 100644 index 00000000..a6380a16 --- /dev/null +++ b/pomerium/envoy/validate_cache_test.go @@ -0,0 +1,44 @@ +package envoy + +import ( + "crypto/sha256" + "sync" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestValidationCache(t *testing.T) { + c := &validationCache{} + h1 := sha256.Sum256([]byte("config-1")) + h2 := sha256.Sum256([]byte("config-2")) + + // empty cache: never a hit + assert.False(t, c.hit(h1), "empty cache should not hit") + + // after storing h1, only h1 hits + c.store(h1) + assert.True(t, c.hit(h1), "stored hash should hit") + assert.False(t, c.hit(h2), "different hash should miss") + + // storing h2 replaces h1 (single-slot cache) + c.store(h2) + assert.True(t, c.hit(h2), "newly stored hash should hit") + assert.False(t, c.hit(h1), "previous hash should no longer hit") +} + +func TestValidationCacheConcurrent(t *testing.T) { + c := &validationCache{} + h := sha256.Sum256([]byte("config")) + c.store(h) + + // concurrent readers and writers must not race (run with -race). + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(2) + go func() { defer wg.Done(); _ = c.hit(h) }() + go func() { defer wg.Done(); c.store(h) }() + } + wg.Wait() + assert.True(t, c.hit(h)) +} diff --git a/pomerium/envoy/validate_envoy.go b/pomerium/envoy/validate_envoy.go index 7ceab8cd..f0e1e26f 100644 --- a/pomerium/envoy/validate_envoy.go +++ b/pomerium/envoy/validate_envoy.go @@ -6,7 +6,7 @@ package envoy import ( "context" - + "crypto/sha256" "os" "os/exec" "path/filepath" @@ -33,6 +33,14 @@ func Validate(ctx context.Context, bootstrap *envoy_config_bootstrap_v3.Bootstra return nil, err } + // Skip the (expensive) envoy subprocess if this exact config was already + // validated successfully. saveConfig revalidates the whole config on every + // change, so identical configs are validated repeatedly without this cache. + hash := sha256.Sum256(bs) + if defaultValidationCache.hit(hash) { + return &ValidateResult{Valid: true, Message: "OK (cached)"}, nil + } + cfgName := filepath.Join(os.TempDir(), id+".pb") err = os.WriteFile(cfgName, bs, ownerRW) if err != nil { @@ -62,6 +70,7 @@ func Validate(ctx context.Context, bootstrap *envoy_config_bootstrap_v3.Bootstra // all other errors are returned as errors return nil, err } + defaultValidationCache.store(hash) return &ValidateResult{ Valid: true, Message: "OK", diff --git a/pomerium/sync_databroker.go b/pomerium/sync_databroker.go index 27494219..7774a1ce 100644 --- a/pomerium/sync_databroker.go +++ b/pomerium/sync_databroker.go @@ -107,8 +107,53 @@ func (r *DataBrokerReconciler) Set(ctx context.Context, ics []*model.IngressConf if err != nil { return false, fmt.Errorf("get config: %w", err) } + + // Build the merged config using only cheap (non-envoy) per-ingress validation. + // The expensive full-config envoy validation runs a single time, in saveConfig. + // This avoids the O(n^2) behavior of validating a growing config once per ingress. + next := buildConfigCheap(ctx, ics) + + changed, err := r.saveConfig(ctx, prev, next, r.ConfigID) + if err == nil { + return changed, nil + } + + // The single full-config envoy validation failed. Rather than reject every + // ingress, fall back to validating ingresses incrementally to isolate and skip + // only the offending one(s). This path performs O(n) envoy validations, but is + // only reached when a batch fails to validate as a whole. + logger.Error(err, "batch config validation failed; falling back to incremental validation to isolate invalid ingress(es)") + next = r.buildConfigIncremental(ctx, ics) + return r.saveConfig(ctx, prev, next, r.ConfigID) +} + +// buildConfigCheap merges the routes for all ingresses into a single config, using +// only cheap (non-envoy) validation to skip individually-invalid ingresses. +func buildConfigCheap(ctx context.Context, ics []*model.IngressConfig) *pb.Config { + logger := log.FromContext(ctx) next := new(pb.Config) + for _, ic := range ics { + cfg := proto.Clone(next).(*pb.Config) + if err := multierror.Append( + upsertRoutes(ctx, cfg, ic), + validateCheap(ctx, cfg), + ).ErrorOrNil(); err != nil { + logger.Error(err, "skip ingress", "ingress", fmt.Sprintf("%s/%s", ic.Namespace, ic.Name)) + continue + } + addCerts(cfg, ic.Secrets) + next = cfg + } + return next +} +// buildConfigIncremental merges the routes for all ingresses into a single config, +// running the full envoy validation after each ingress so that an ingress which +// only fails full validation (not cheap validation) is isolated and skipped. This +// is the pre-caching behavior, retained as a fallback for the rare batch-failure case. +func (r *DataBrokerReconciler) buildConfigIncremental(ctx context.Context, ics []*model.IngressConfig) *pb.Config { + logger := log.FromContext(ctx) + next := new(pb.Config) for _, ic := range ics { cfg := proto.Clone(next).(*pb.Config) if err := multierror.Append( @@ -121,8 +166,7 @@ func (r *DataBrokerReconciler) Set(ctx context.Context, ics []*model.IngressConf addCerts(cfg, ic.Secrets) next = cfg } - - return r.saveConfig(ctx, prev, next, r.ConfigID) + return next } // SetConfig updates just the shared config settings diff --git a/pomerium/sync_test.go b/pomerium/sync_test.go new file mode 100644 index 00000000..d9fd51cd --- /dev/null +++ b/pomerium/sync_test.go @@ -0,0 +1,101 @@ +package pomerium + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/intstr" + + "github.com/pomerium/ingress-controller/model" + + _ "github.com/pomerium/ingress-controller/internal" +) + +// validIngressConfig returns a minimal, valid IngressConfig for the given name/host. +func validIngressConfig(name, host string) *model.IngressConfig { + pathTypePrefix := networkingv1.PathTypePrefix + return &model.IngressConfig{ + AnnotationPrefix: "ingress.pomerium.io", + Ingress: &networkingv1.Ingress{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: networkingv1.IngressSpec{ + Rules: []networkingv1.IngressRule{{ + Host: host, + IngressRuleValue: networkingv1.IngressRuleValue{ + HTTP: &networkingv1.HTTPIngressRuleValue{ + Paths: []networkingv1.HTTPIngressPath{{ + Path: "/", + PathType: &pathTypePrefix, + Backend: networkingv1.IngressBackend{ + Service: &networkingv1.IngressServiceBackend{ + Name: "service", + Port: networkingv1.ServiceBackendPort{Name: "http"}, + }, + }, + }}, + }, + }, + }}, + }, + }, + Services: map[types.NamespacedName]*corev1.Service{ + {Name: "service", Namespace: "default"}: { + ObjectMeta: metav1.ObjectMeta{Name: "service", Namespace: "default"}, + Spec: corev1.ServiceSpec{ + Ports: []corev1.ServicePort{{ + Name: "http", + Protocol: corev1.ProtocolTCP, + Port: 80, + TargetPort: intstr.FromInt(80), + }}, + }, + }, + }, + } +} + +// invalidIngressConfig returns an IngressConfig that fails route generation +// (empty host without the allow-empty-host annotation). +func invalidIngressConfig(name string) *model.IngressConfig { + ic := validIngressConfig(name, "") + return ic +} + +func TestBuildConfigCheap_AllValid(t *testing.T) { + ctx := context.Background() + ics := []*model.IngressConfig{ + validIngressConfig("a", "a.localhost.pomerium.io"), + validIngressConfig("b", "b.localhost.pomerium.io"), + } + + cfg := buildConfigCheap(ctx, ics) + require.NotNil(t, cfg) + assert.Len(t, cfg.GetRoutes(), 2, "both valid ingresses should produce a route") +} + +func TestBuildConfigCheap_SkipsInvalid(t *testing.T) { + ctx := context.Background() + ics := []*model.IngressConfig{ + validIngressConfig("a", "a.localhost.pomerium.io"), + invalidIngressConfig("bad"), // empty host -> route generation fails + validIngressConfig("c", "c.localhost.pomerium.io"), + } + + cfg := buildConfigCheap(ctx, ics) + require.NotNil(t, cfg) + // The invalid ingress is skipped; the two valid ones remain. + assert.Len(t, cfg.GetRoutes(), 2, "invalid ingress should be skipped, valid ones kept") + + hosts := map[string]bool{} + for _, r := range cfg.GetRoutes() { + hosts[r.GetFrom()] = true + } + assert.True(t, hosts["https://a.localhost.pomerium.io"], "route a should be present") + assert.True(t, hosts["https://c.localhost.pomerium.io"], "route c should be present") +} diff --git a/pomerium/validate.go b/pomerium/validate.go index 029db421..d32e8626 100644 --- a/pomerium/validate.go +++ b/pomerium/validate.go @@ -15,8 +15,10 @@ import ( "github.com/pomerium/ingress-controller/pomerium/envoy" ) -// validate validates pomerium config. -func validate(ctx context.Context, cfg *pb.Config, id string) error { +// validateOptions builds and validates pomerium options from the config without +// invoking the (expensive) embedded envoy subprocess. It catches route/policy and +// option-level errors cheaply. +func validateOptions(ctx context.Context, cfg *pb.Config) (*config.Options, error) { options := config.NewDefaultOptions() options.ApplySettings(ctx, cryptutil.NewCertificatesIndex(), cfg.GetSettings()) options.InsecureServer = true @@ -24,16 +26,34 @@ func validate(ctx context.Context, cfg *pb.Config, id string) error { for _, r := range cfg.GetRoutes() { p, err := config.NewPolicyFromProto(r) if err != nil { - return err + return nil, err } err = p.Validate() if err != nil { - return err + return nil, err } options.Policies = append(options.Policies, *p) } - err := options.Validate() + if err := options.Validate(); err != nil { + return nil, err + } + + return options, nil +} + +// validateCheap validates a pomerium config without invoking the embedded envoy +// subprocess. It is used to catch most invalid ingresses quickly, deferring the +// single expensive full-config envoy validation to saveConfig. +func validateCheap(ctx context.Context, cfg *pb.Config) error { + _, err := validateOptions(ctx, cfg) + return err +} + +// validate validates pomerium config, including a full bootstrap validation via +// the embedded envoy binary. +func validate(ctx context.Context, cfg *pb.Config, id string) error { + options, err := validateOptions(ctx, cfg) if err != nil { return err }