fix: avoid O(n^2) envoy validation on initial sync; cache validation; add reconcile concurrency flag

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

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)
+}

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

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
+}

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))
+}

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

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

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