From ff8caa66c708872e1f697371b666a1ea8d565784 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Christoph=20St=C3=A4bler?= Date: Tue, 3 Feb 2026 12:01:10 +0100 Subject: [PATCH 1/5] Make auth-proxy query EventPolicies dynamically IntegrationSink was baking EventPolicies into the AUTH_POLICIES env var, requiring deployment rollouts whenever policies changed. This caused test failures because old pods with stale policies continued serving traffic during RollingUpdate. Change auth-proxy to query EventPolicies dynamically using a namespace-scoped informer, similar to how Broker and Channel work. This eliminates deployment rollouts when EventPolicies change. - Add knative-eventing-eventpolicy-reader ClusterRole - Create namespace-scoped EventPolicy informer in auth-proxy - Add parent resource env vars to identify which resource to query policies for - Create RoleBinding in sink's namespace for EventPolicy access - Remove AUTH_POLICIES env var from deployment spec - Add test coverage for OIDC-enabled deployments with RoleBindings --- cmd/auth_proxy/main.go | 121 +++++++++--- config/core/roles/clusterrole-namespaced.yaml | 13 ++ .../integration/sink/integrationsink.go | 94 +++++++-- .../integration/sink/integrationsink_test.go | 187 +++++++++++++++--- .../sink/resources/container_image.go | 77 +++++--- .../sink/resources/container_image_test.go | 2 +- .../testing/v1alpha1/integrationsink.go | 12 ++ 7 files changed, 423 insertions(+), 83 deletions(-) diff --git a/cmd/auth_proxy/main.go b/cmd/auth_proxy/main.go index 7ca98bdafd8..b53b34427f1 100644 --- a/cmd/auth_proxy/main.go +++ b/cmd/auth_proxy/main.go @@ -22,7 +22,6 @@ package main import ( "context" - "encoding/json" "net" "os" @@ -36,11 +35,16 @@ import ( "github.com/kelseyhightower/envconfig" "go.uber.org/zap" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/kubernetes" "k8s.io/utils/ptr" cmdbroker "knative.dev/eventing/cmd/broker" "knative.dev/eventing/pkg/apis/feature" "knative.dev/eventing/pkg/auth" + eventingclient "knative.dev/eventing/pkg/client/clientset/versioned" + eventinginformers "knative.dev/eventing/pkg/client/informers/externalversions" + eventingv1alpha1listers "knative.dev/eventing/pkg/client/listers/eventing/v1alpha1" "knative.dev/eventing/pkg/eventingtls" "knative.dev/eventing/pkg/kncloudevents" "knative.dev/pkg/apis" @@ -69,7 +73,11 @@ type envConfig struct { SinkNamespace string `envconfig:"SINK_NAMESPACE" required:"true"` SinkAudience *string `envconfig:"SINK_AUDIENCE"` - AuthPolicies string `envconfig:"AUTH_POLICIES"` + // Parent resource information for dynamic EventPolicy lookup + ParentAPIVersion string `envconfig:"PARENT_API_VERSION" required:"true"` + ParentKind string `envconfig:"PARENT_KIND" required:"true"` + ParentName string `envconfig:"PARENT_NAME" required:"true"` + ParentNamespace string `envconfig:"PARENT_NAMESPACE" required:"true"` SinkTLSCertPath *string `envconfig:"SINK_TLS_CERT_FILE"` SinkTLSKeyPath *string `envconfig:"SINK_TLS_KEY_FILE"` @@ -79,13 +87,14 @@ type envConfig struct { // ProxyHandler handles HTTP requests and performs authentication/authorization // before forwarding to the target service type ProxyHandler struct { - kubeClient kubernetes.Interface - withContext func(ctx context.Context) context.Context - authVerifier *auth.Verifier - httpProxy *httputil.ReverseProxy - httpsProxy *httputil.ReverseProxy - config envConfig - authSubjects []auth.SubjectsWithFilters + kubeClient kubernetes.Interface + withContext func(ctx context.Context) context.Context + authVerifier *auth.Verifier + httpProxy *httputil.ReverseProxy + httpsProxy *httputil.ReverseProxy + config envConfig + eventPolicyLister eventingv1alpha1listers.EventPolicyLister + parentResourceGVK schema.GroupVersionKind } func main() { @@ -104,7 +113,14 @@ func main() { featureStore := setupFeatureStore(ctx, logger, configMapWatcher) - handler, err := createProxyHandler(ctx, config, logger, featureStore, configMapWatcher) + // Create namespace-scoped EventPolicy informer + eventPolicyLister, eventPolicyInformer, err := setupEventPolicyInformer(ctx, config, logger) + if err != nil { + logger.Fatalw("Failed to setup EventPolicy informer", zap.Error(err)) + } + informers = append(informers, eventPolicyInformer) + + handler, err := createProxyHandler(ctx, config, logger, featureStore, configMapWatcher, eventPolicyLister) if err != nil { logger.Fatalw("Failed to create proxy handler", zap.Error(err)) } @@ -166,21 +182,45 @@ func setupFeatureStore(_ context.Context, logger *zap.SugaredLogger, configMapWa return featureStore } -// createProxyHandler creates and configures the proxy handler -func createProxyHandler(ctx context.Context, config envConfig, logger *zap.SugaredLogger, featureStore *feature.Store, configMapWatcher *configmap.InformedWatcher) (*ProxyHandler, error) { - var authSubjects []auth.SubjectsWithFilters +// setupEventPolicyInformer creates a namespace-scoped EventPolicy informer +func setupEventPolicyInformer(ctx context.Context, config envConfig, logger *zap.SugaredLogger) (eventingv1alpha1listers.EventPolicyLister, controller.Informer, error) { + cfg := injection.GetConfig(ctx) - if len(config.AuthPolicies) > 0 { - if err := json.Unmarshal([]byte(config.AuthPolicies), &authSubjects); err != nil { - return nil, fmt.Errorf("failed to parse policies: %w", err) - } + // Create eventing client + eventingClient, err := eventingclient.NewForConfig(cfg) + if err != nil { + return nil, nil, fmt.Errorf("failed to create eventing client: %w", err) } + // Create namespace-scoped informer factory + eventingInformerFactory := eventinginformers.NewSharedInformerFactoryWithOptions( + eventingClient, + 0, // resyncPeriod - 0 means no resync + eventinginformers.WithNamespace(config.ParentNamespace), + ) + + eventPolicyInformer := eventingInformerFactory.Eventing().V1alpha1().EventPolicies() + + logger.Infof("Created namespace-scoped EventPolicy informer for namespace %s", config.ParentNamespace) + + return eventPolicyInformer.Lister(), eventPolicyInformer.Informer(), nil +} + +// createProxyHandler creates and configures the proxy handler +func createProxyHandler(ctx context.Context, config envConfig, logger *zap.SugaredLogger, featureStore *feature.Store, configMapWatcher *configmap.InformedWatcher, eventPolicyLister eventingv1alpha1listers.EventPolicyLister) (*ProxyHandler, error) { + // Parse parent resource GVK + gv, err := schema.ParseGroupVersion(config.ParentAPIVersion) + if err != nil { + return nil, fmt.Errorf("failed to parse parent API version %q: %w", config.ParentAPIVersion, err) + } + parentGVK := gv.WithKind(config.ParentKind) + handler := &ProxyHandler{ - kubeClient: kubeclient.Get(ctx), - authVerifier: auth.NewVerifier(ctx, nil, nil, configMapWatcher), - config: config, - authSubjects: authSubjects, + kubeClient: kubeclient.Get(ctx), + authVerifier: auth.NewVerifier(ctx, nil, nil, configMapWatcher), + config: config, + eventPolicyLister: eventPolicyLister, + parentResourceGVK: parentGVK, } handler.withContext = func(ctx context.Context) context.Context { @@ -251,7 +291,15 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { logger.Debugf("Handling request to %s", r.RequestURI) - err := h.authVerifier.VerifyRequestFromSubjectsWithFilters(ctx, features, h.config.SinkAudience, h.authSubjects, h.config.SinkNamespace, r, w) + // Get EventPolicies dynamically for the parent resource + authSubjects, err := h.getAuthSubjects(logger) + if err != nil { + logger.Errorw("Failed to get EventPolicies", zap.Error(err)) + http.Error(w, "Internal server error", http.StatusInternalServerError) + return + } + + err = h.authVerifier.VerifyRequestFromSubjectsWithFilters(ctx, features, h.config.SinkAudience, authSubjects, h.config.SinkNamespace, r, w) if err != nil { logger.Debugw("Failed to verify AuthN and AuthZ", zap.Error(err)) return @@ -266,6 +314,35 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } +// getAuthSubjects retrieves EventPolicies for the parent resource and converts them to SubjectsWithFilters +func (h *ProxyHandler) getAuthSubjects(logger *zap.SugaredLogger) ([]auth.SubjectsWithFilters, error) { + // Get EventPolicies applying to this resource + policies, err := auth.GetEventPoliciesForResource( + h.eventPolicyLister, + h.parentResourceGVK, + metav1.ObjectMeta{ + Name: h.config.ParentName, + Namespace: h.config.ParentNamespace, + }, + ) + if err != nil { + return nil, fmt.Errorf("failed to get EventPolicies: %w", err) + } + + logger.Debugf("Found %d EventPolicies for %s/%s", len(policies), h.config.ParentNamespace, h.config.ParentName) + + // Convert EventPolicies to SubjectsWithFilters + subjectsWithFilters := make([]auth.SubjectsWithFilters, 0, len(policies)) + for _, policy := range policies { + subjectsWithFilters = append(subjectsWithFilters, auth.SubjectsWithFilters{ + Subjects: policy.Status.From, + Filters: policy.Spec.Filters, + }) + } + + return subjectsWithFilters, nil +} + // httpReverseProxy creates a reverse proxy for HTTP traffic to the target service func httpReverseProxy(config envConfig) (*httputil.ReverseProxy, error) { httpTarget := fmt.Sprintf("http://%s:%d", config.TargetHost, config.TargetHTTPPort) diff --git a/config/core/roles/clusterrole-namespaced.yaml b/config/core/roles/clusterrole-namespaced.yaml index d1f6bae5121..22f66163a55 100644 --- a/config/core/roles/clusterrole-namespaced.yaml +++ b/config/core/roles/clusterrole-namespaced.yaml @@ -115,3 +115,16 @@ rules: - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev", "sinks.knative.dev"] resources: ["*"] verbs: ["get", "list", "watch"] +--- +kind: ClusterRole +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: knative-eventing-eventpolicy-reader + labels: + rbac.authorization.k8s.io/aggregate-to-view: "true" + app.kubernetes.io/version: devel + app.kubernetes.io/name: knative-eventing +rules: + - apiGroups: ["eventing.knative.dev"] + resources: ["eventpolicies"] + verbs: ["get", "list", "watch"] diff --git a/pkg/reconciler/integration/sink/integrationsink.go b/pkg/reconciler/integration/sink/integrationsink.go index b664781063c..5d452ca30be 100644 --- a/pkg/reconciler/integration/sink/integrationsink.go +++ b/pkg/reconciler/integration/sink/integrationsink.go @@ -114,7 +114,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, sink *sinks.IntegrationS } logger.Debugw("Reconciling IntegrationSink auth-proxy RBAC") - _, err = r.reconcileAuthProxyRBAC(ctx, sink) + err = r.reconcileAuthProxyRBAC(ctx, sink) if err != nil { logging.FromContext(ctx).Errorw("Error reconciling auth-proxy RBAC", zap.Error(err)) return err @@ -149,7 +149,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, sink *sinks.IntegrationS } func (r *Reconciler) reconcileDeployment(ctx context.Context, sink *sinks.IntegrationSink, authProxyImage string, featureFlags feature.Flags) (*v1.Deployment, error) { - expected, err := resources.MakeDeploymentSpec(sink, authProxyImage, featureFlags, r.trustBundleConfigMapLister, r.eventPolicyLister) + expected, err := resources.MakeDeploymentSpec(sink, authProxyImage, featureFlags, r.trustBundleConfigMapLister) if err != nil { return nil, fmt.Errorf("failed to create deployment template: %w", err) } @@ -434,41 +434,103 @@ func (r *Reconciler) serviceChanged(existingSvc corev1.ServiceSpec, wantSvc core return !equality.Semantic.DeepDerivative(wantSvc, existingSvc) } -func (r *Reconciler) reconcileAuthProxyRBAC(ctx context.Context, sink *sinks.IntegrationSink) (*rbacv1.RoleBinding, error) { +func (r *Reconciler) reconcileAuthProxyRBAC(ctx context.Context, sink *sinks.IntegrationSink) error { features := feature.FromContext(ctx) + // 1. Reconcile EventPolicy access RBAC (RoleBinding in sink's namespace) + // This allows auth-proxy to read EventPolicies to enforce authorization + if err := r.reconcileEventPolicyRBAC(ctx, sink, features); err != nil { + return fmt.Errorf("failed to reconcile EventPolicy RBAC: %w", err) + } + + // 2. Reconcile ConfigMap access RBAC (RoleBinding in knative-eventing namespace) + // This allows auth-proxy to read logging configuration from the knative-eventing namespace + err := r.reconcileConfigMapAccessRBAC(ctx, sink, features) + if err != nil { + return fmt.Errorf("failed to reconcile ConfigMap access RBAC: %w", err) + } + + return nil +} + +// reconcileEventPolicyRBAC reconciles the namespace-scoped RoleBinding +// that allows auth-proxy to read EventPolicies in the sink's namespace. +// The RoleBinding references the knative-eventing-eventpolicy-reader ClusterRole. +func (r *Reconciler) reconcileEventPolicyRBAC(ctx context.Context, sink *sinks.IntegrationSink, features feature.Flags) error { + if !features.IsOIDCAuthentication() { + // EventPolicy RoleBinding has OwnerReferences, so it'll be garbage collected + // when the sink is deleted. No explicit cleanup needed when OIDC is disabled. + return nil + } + + if err := r.reconcileEventPolicyRoleBinding(ctx, sink); err != nil { + return err + } + + return nil +} + +// reconcileConfigMapAccessRBAC reconciles the RoleBinding in knative-eventing namespace +// that allows auth-proxy to read ConfigMaps (for logging configuration). +func (r *Reconciler) reconcileConfigMapAccessRBAC(ctx context.Context, sink *sinks.IntegrationSink, features feature.Flags) error { expected, err := resources.MakeAuthProxyRoleBindings(sink, r.integrationSinkLister, features) if err != nil { - return nil, fmt.Errorf("creating auth proxy rolebinding: %w", err) + return fmt.Errorf("creating auth proxy rolebinding: %w", err) } if !features.IsOIDCAuthentication() { - return nil, r.deleteIntegrationSinkRBAC(ctx, expected) + return r.deleteIntegrationSinkRBAC(ctx, expected) } rolebinding, err := r.rolebindingLister.RoleBindings(expected.Namespace).Get(expected.Name) if apierrors.IsNotFound(err) { - created, err := r.kubeClientSet.RbacV1().RoleBindings(system.Namespace()).Create(ctx, expected, metav1.CreateOptions{}) + _, err = r.kubeClientSet.RbacV1().RoleBindings(system.Namespace()).Create(ctx, expected, metav1.CreateOptions{}) if err != nil { - return nil, err + return err } - - return created, nil + return nil } if err != nil { - return nil, fmt.Errorf("failed to get rolebinding %s/%s: %w", expected.GetNamespace(), expected.GetName(), err) + return fmt.Errorf("failed to get rolebinding %s/%s: %w", expected.GetNamespace(), expected.GetName(), err) } if !r.rolebindingChanged(rolebinding, expected) { - return rolebinding, nil + return nil } expected.ResourceVersion = rolebinding.ResourceVersion - updated, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, expected, metav1.UpdateOptions{}) + _, err = r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, expected, metav1.UpdateOptions{}) if err != nil { - return nil, err + return err } - return updated, nil + return nil +} + +func (r *Reconciler) reconcileEventPolicyRoleBinding(ctx context.Context, sink *sinks.IntegrationSink) error { + expected := resources.MakeEventPolicyRoleBinding(sink) + + rb, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Get(ctx, expected.Name, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + _, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Create(ctx, expected, metav1.CreateOptions{}) + if err != nil { + return fmt.Errorf("creating EventPolicy RoleBinding: %w", err) + } + return nil + } + if err != nil { + return fmt.Errorf("getting EventPolicy RoleBinding: %w", err) + } + + // Update if subjects changed + if !equality.Semantic.DeepEqual(rb.Subjects, expected.Subjects) { + rb.Subjects = expected.Subjects + _, err = r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, rb, metav1.UpdateOptions{}) + if err != nil { + return fmt.Errorf("updating EventPolicy RoleBinding: %w", err) + } + } + + return nil } func (r *Reconciler) deleteIntegrationSinkRBAC(ctx context.Context, rolebinding *rbacv1.RoleBinding) error { @@ -488,6 +550,10 @@ func (r *Reconciler) deleteIntegrationSinkRBAC(ctx context.Context, rolebinding return fmt.Errorf("failed to delete auth-proxy rolebinding %s/%s: %w", rolebinding.Namespace, rolebinding.Name, err) } + // Note: EventPolicy Role and RoleBinding have OwnerReferences set to the IntegrationSink, + // so they will be automatically garbage collected when the sink is deleted. + // We don't need to explicitly delete them here. + return nil } diff --git a/pkg/reconciler/integration/sink/integrationsink_test.go b/pkg/reconciler/integration/sink/integrationsink_test.go index 0ba1b91f9fb..dc490928369 100644 --- a/pkg/reconciler/integration/sink/integrationsink_test.go +++ b/pkg/reconciler/integration/sink/integrationsink_test.go @@ -17,45 +17,41 @@ limitations under the License. package sink import ( + "context" "fmt" "sync/atomic" + "testing" cmlisters "github.com/cert-manager/cert-manager/pkg/client/listers/certmanager/v1" - - "knative.dev/eventing/pkg/certificates" - - "knative.dev/eventing/pkg/reconciler/integration" - - "k8s.io/apimachinery/pkg/util/intstr" - "k8s.io/utils/ptr" - "knative.dev/pkg/apis" - duckv1 "knative.dev/pkg/apis/duck/v1" - "knative.dev/pkg/network" - appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/intstr" clientgotesting "k8s.io/client-go/testing" - sinksv1alpha1 "knative.dev/eventing/pkg/apis/sinks/v1alpha1" - fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake" - "knative.dev/eventing/pkg/client/injection/reconciler/sinks/v1alpha1/integrationsink" - fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake" - "knative.dev/pkg/kmeta" - "knative.dev/pkg/logging" - - "context" - - . "knative.dev/eventing/pkg/reconciler/testing/v1" - . "knative.dev/eventing/pkg/reconciler/testing/v1alpha1" - + "k8s.io/utils/ptr" + "knative.dev/pkg/apis" + duckv1 "knative.dev/pkg/apis/duck/v1" "knative.dev/pkg/client/injection/ducks/duck/v1/addressable" "knative.dev/pkg/configmap" "knative.dev/pkg/controller" + "knative.dev/pkg/kmeta" + "knative.dev/pkg/logging" logtesting "knative.dev/pkg/logging/testing" + "knative.dev/pkg/network" . "knative.dev/pkg/reconciler/testing" + "knative.dev/pkg/system" - "testing" + "knative.dev/eventing/pkg/apis/feature" + sinksv1alpha1 "knative.dev/eventing/pkg/apis/sinks/v1alpha1" + "knative.dev/eventing/pkg/certificates" + fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake" + "knative.dev/eventing/pkg/client/injection/reconciler/sinks/v1alpha1/integrationsink" + "knative.dev/eventing/pkg/reconciler/integration" + . "knative.dev/eventing/pkg/reconciler/testing/v1" + . "knative.dev/eventing/pkg/reconciler/testing/v1alpha1" + fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake" ) const ( @@ -151,6 +147,54 @@ func TestReconcile(t *testing.T) { WithIntegrationSinkPropagateDeploymenteStatus(makeDeploymentStatus(&conditionTrue)), ), }}, + }, { + Name: "creates deployment with auth-proxy and RoleBindings when OIDC is enabled", + Objects: []runtime.Object{ + NewIntegrationSink(sinkName, testNS, + WithIntegrationSinkUID(sinkUID), + WithIntegrationSinkSpec(makeIntegrationSinkSpec()), + ), + }, + Key: testNS + "/" + sinkName, + Ctx: feature.ToContext(context.Background(), feature.Flags{ + feature.OIDCAuthentication: feature.Enabled, + }), + WantCreates: []runtime.Object{ + makeEventPolicyRoleBinding(), + makeConfigMapAccessRoleBinding(), + makeDeploymentWithAuthProxy(NewIntegrationSink(sinkName, testNS, + WithIntegrationSinkUID(sinkUID), + WithIntegrationSinkSpec(makeIntegrationSinkSpec())), + nil), + makeService(deploymentName, testNS), + }, + WantEvents: []string{ + Eventf(corev1.EventTypeNormal, deploymentCreated, "Deployment created %q", deploymentName), + Eventf(corev1.EventTypeNormal, serviceCreated, "Service created %q", deploymentName), + Eventf(corev1.EventTypeNormal, sinkReconciled, `IntegrationSink reconciled: "%s/%s"`, testNS, sinkName), + }, + WantStatusUpdates: []clientgotesting.UpdateActionImpl{{ + Object: NewIntegrationSink(sinkName, testNS, + WithIntegrationSinkUID(sinkUID), + WithIntegrationSinkAddressableReady(), + WithIntegrationSinkAddress(duckv1.Addressable{ + Name: ptr.To("http"), + URL: &apis.URL{ + Scheme: "http", + Host: network.GetServiceHostname(deploymentName, testNS), + }, + Audience: ptr.To("sinks.knative.dev/integrationsink/" + testNS + "/" + sinkName), + }), + WithIntegrationSinkSpec(makeIntegrationSinkSpec()), + WithIntegrationSinkEventPoliciesReady("DefaultAuthorizationMode", `Default authz mode is ""`), + WithIntegrationSinkTrustBundlePropagatedReady(), + WithInitIntegrationSinkConditions, + WithIntegrationSinkPropagateDeploymenteStatus(&appsv1.Deployment{ + Status: appsv1.DeploymentStatus{}, + }), + ), + }}, + SkipNamespaceValidation: true, }} logger := logtesting.TestLogger(t) @@ -328,6 +372,48 @@ func makeDeployment(sink *sinksv1alpha1.IntegrationSink, ready *corev1.Condition return d } +func makeDeploymentWithAuthProxy(sink *sinksv1alpha1.IntegrationSink, ready *corev1.ConditionStatus) runtime.Object { + d := makeDeployment(sink, ready).(*appsv1.Deployment) + + // Remove ports from sink container (auth-proxy becomes the entry point) + d.Spec.Template.Spec.Containers[0].Ports = nil + + // Add auth-proxy container + authProxyContainer := corev1.Container{ + Name: "auth-proxy", + ImagePullPolicy: corev1.PullIfNotPresent, + Ports: []corev1.ContainerPort{ + {ContainerPort: 3128, Protocol: corev1.ProtocolTCP, Name: "http"}, + {ContainerPort: 3129, Protocol: corev1.ProtocolTCP, Name: "https"}, + }, + Env: []corev1.EnvVar{ + {Name: "TARGET_HTTP_PORT", Value: "8080"}, + {Name: "TARGET_HTTPS_PORT", Value: "8443"}, + {Name: "PROXY_HTTP_PORT", Value: "3128"}, + {Name: "PROXY_HTTPS_PORT", Value: "3129"}, + {Name: "SYSTEM_NAMESPACE", Value: system.Namespace()}, + {Name: "PARENT_API_VERSION", Value: "sinks.knative.dev/v1alpha1"}, + {Name: "PARENT_KIND", Value: "IntegrationSink"}, + {Name: "PARENT_NAME", Value: sink.Name}, + {Name: "PARENT_NAMESPACE", Value: sink.Namespace}, + {Name: "SINK_NAMESPACE", Value: sink.Namespace}, + {Name: "SINK_TLS_CERT_FILE", Value: "/etc/" + certificates.CertificateName(sink.Name) + "/tls.crt"}, + {Name: "SINK_TLS_KEY_FILE", Value: "/etc/" + certificates.CertificateName(sink.Name) + "/tls.key"}, + {Name: "SINK_TLS_CA_FILE", Value: "/etc/" + certificates.CertificateName(sink.Name) + "/ca.crt"}, + }, + VolumeMounts: []corev1.VolumeMount{ + { + Name: certificates.CertificateName(sink.Name), + MountPath: "/etc/" + certificates.CertificateName(sink.Name), + ReadOnly: true, + }, + }, + } + + d.Spec.Template.Spec.Containers = append(d.Spec.Template.Spec.Containers, authProxyContainer) + return d +} + func makeService(name, namespace string) *corev1.Service { return &corev1.Service{ TypeMeta: metav1.TypeMeta{ @@ -407,3 +493,56 @@ func makeDeploymentStatus(ready *corev1.ConditionStatus) *appsv1.Deployment { }, } } + +func makeEventPolicyRoleBinding() *rbacv1.RoleBinding { + return &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: sinkName + "-eventpolicy-reader", + Namespace: testNS, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "sinks.knative.dev/v1alpha1", + Kind: "IntegrationSink", + Name: sinkName, + UID: sinkUID, + Controller: ptr.To(true), + BlockOwnerDeletion: ptr.To(true), + }, + }, + Labels: integration.Labels(sinkName), + }, + RoleRef: rbacv1.RoleRef{ + APIGroup: "rbac.authorization.k8s.io", + Kind: "ClusterRole", + Name: "knative-eventing-eventpolicy-reader", + }, + Subjects: []rbacv1.Subject{ + { + Kind: "ServiceAccount", + Namespace: testNS, + Name: "default", + }, + }, + } +} + +func makeConfigMapAccessRoleBinding() *rbacv1.RoleBinding { + return &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: "eventing-auth-proxy", + Namespace: system.Namespace(), + }, + RoleRef: rbacv1.RoleRef{ + APIGroup: "rbac.authorization.k8s.io", + Kind: "Role", + Name: "knative-eventing-auth-proxy", + }, + Subjects: []rbacv1.Subject{ + { + Kind: "ServiceAccount", + Namespace: testNS, + Name: "default", + }, + }, + } +} diff --git a/pkg/reconciler/integration/sink/resources/container_image.go b/pkg/reconciler/integration/sink/resources/container_image.go index 08adf52dc9a..a150e5d6e17 100644 --- a/pkg/reconciler/integration/sink/resources/container_image.go +++ b/pkg/reconciler/integration/sink/resources/container_image.go @@ -17,7 +17,6 @@ limitations under the License. package resources import ( - "encoding/json" "fmt" "os" @@ -33,9 +32,7 @@ import ( commonv1a1 "knative.dev/eventing/pkg/apis/common/integration/v1alpha1" "knative.dev/eventing/pkg/apis/feature" "knative.dev/eventing/pkg/apis/sinks/v1alpha1" - "knative.dev/eventing/pkg/auth" "knative.dev/eventing/pkg/certificates" - alpha1 "knative.dev/eventing/pkg/client/listers/eventing/v1alpha1" v1alpha1listers "knative.dev/eventing/pkg/client/listers/sinks/v1alpha1" "knative.dev/eventing/pkg/eventingtls" "knative.dev/eventing/pkg/reconciler/integration" @@ -47,7 +44,7 @@ const ( AuthProxyRolebindingName = "eventing-auth-proxy" ) -func MakeDeploymentSpec(sink *v1alpha1.IntegrationSink, authProxyImage string, featureFlags feature.Flags, trustBundleConfigmapLister corev1listers.ConfigMapLister, eventPolicyLister alpha1.EventPolicyLister) (*appsv1.Deployment, error) { +func MakeDeploymentSpec(sink *v1alpha1.IntegrationSink, authProxyImage string, featureFlags feature.Flags, trustBundleConfigmapLister corev1listers.ConfigMapLister) (*appsv1.Deployment, error) { deploy := &appsv1.Deployment{ TypeMeta: metav1.TypeMeta{ APIVersion: "apps/v1", @@ -116,11 +113,7 @@ func MakeDeploymentSpec(sink *v1alpha1.IntegrationSink, authProxyImage string, f if featureFlags.IsOIDCAuthentication() { // add auth-proxy - - proxyVars, err := makeAuthProxyEnv(sink, featureFlags, eventPolicyLister) - if err != nil { - return nil, fmt.Errorf("failed to make auth proxy env vars: %w", err) - } + proxyVars := makeAuthProxyEnv(sink, featureFlags) deploy.Spec.Template.Spec.Containers = append(deploy.Spec.Template.Spec.Containers, corev1.Container{ Name: "auth-proxy", @@ -246,20 +239,48 @@ func MakeAuthProxyRoleBindings(sink *v1alpha1.IntegrationSink, sinkLister v1alph return &rb, nil } -func makeAuthProxyEnv(sink *v1alpha1.IntegrationSink, featureFlags feature.Flags, eventPolicyLister alpha1.EventPolicyLister) ([]corev1.EnvVar, error) { - sinkAddress := eventingtls.GetHttpsAddress(sink.Status.Addresses) - if sinkAddress == nil { - sinkAddress = sink.Status.Address +// MakeEventPolicyRole creates a namespace-scoped Role for auth-proxy to access EventPolicies +// MakeEventPolicyRoleBinding creates a RoleBinding for auth-proxy to access EventPolicies. +// It references the knative-eventing-eventpolicy-reader ClusterRole which grants +// read access to EventPolicies. +func MakeEventPolicyRoleBinding(sink *v1alpha1.IntegrationSink) *rbacv1.RoleBinding { + saName := makeServiceAccountName(sink) + if saName == "" { + saName = "default" } - policies, err := auth.SubjectWithFiltersFromPolicyRef(eventPolicyLister, sink.Namespace, sink.Status.Policies) - if err != nil { - return nil, fmt.Errorf("failed to build auth proxy policies env vars: %w", err) + return &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: makeEventPolicyRoleBindingName(sink), + Namespace: sink.Namespace, + OwnerReferences: []metav1.OwnerReference{ + *kmeta.NewControllerRef(sink), + }, + Labels: integration.Labels(sink.Name), + }, + RoleRef: rbacv1.RoleRef{ + APIGroup: rbacv1.GroupName, + Kind: "ClusterRole", + Name: "knative-eventing-eventpolicy-reader", + }, + Subjects: []rbacv1.Subject{ + { + Kind: "ServiceAccount", + Namespace: sink.Namespace, + Name: saName, + }, + }, } +} - policiesJson, err := json.Marshal(policies) - if err != nil { - return nil, fmt.Errorf("failed to parse policies for auth proxy env vars: %w", err) +func makeEventPolicyRoleBindingName(sink *v1alpha1.IntegrationSink) string { + return fmt.Sprintf("%s-eventpolicy-reader", sink.Name) +} + +func makeAuthProxyEnv(sink *v1alpha1.IntegrationSink, featureFlags feature.Flags) []corev1.EnvVar { + sinkAddress := eventingtls.GetHttpsAddress(sink.Status.Addresses) + if sinkAddress == nil { + sinkAddress = sink.Status.Address } envVars := []corev1.EnvVar{ @@ -284,8 +305,20 @@ func makeAuthProxyEnv(sink *v1alpha1.IntegrationSink, featureFlags feature.Flags Value: system.Namespace(), }, { - Name: "AUTH_POLICIES", - Value: string(policiesJson), + Name: "PARENT_API_VERSION", + Value: v1alpha1.SchemeGroupVersion.String(), + }, + { + Name: "PARENT_KIND", + Value: "IntegrationSink", + }, + { + Name: "PARENT_NAME", + Value: sink.Name, + }, + { + Name: "PARENT_NAMESPACE", + Value: sink.Namespace, }, { Name: "SINK_NAMESPACE", @@ -323,7 +356,7 @@ func makeAuthProxyEnv(sink *v1alpha1.IntegrationSink, featureFlags feature.Flags }) } - return envVars, nil + return envVars } func makeEnv(sink *v1alpha1.IntegrationSink, featureFlags feature.Flags) []corev1.EnvVar { diff --git a/pkg/reconciler/integration/sink/resources/container_image_test.go b/pkg/reconciler/integration/sink/resources/container_image_test.go index f65dc09c45b..b66f95475d1 100644 --- a/pkg/reconciler/integration/sink/resources/container_image_test.go +++ b/pkg/reconciler/integration/sink/resources/container_image_test.go @@ -157,7 +157,7 @@ func TestNewSQSContainerSink(t *testing.T) { }, } - got, _ := MakeDeploymentSpec(sink, "unused", feature.Flags{}, nil, nil) + got, _ := MakeDeploymentSpec(sink, "unused", feature.Flags{}, nil) sortOpts := []cmp.Option{ cmp.Transformer("SortEnvVars", func(in corev1.Container) corev1.Container { out := in diff --git a/pkg/reconciler/testing/v1alpha1/integrationsink.go b/pkg/reconciler/testing/v1alpha1/integrationsink.go index 8583498c5fe..a7088033da3 100644 --- a/pkg/reconciler/testing/v1alpha1/integrationsink.go +++ b/pkg/reconciler/testing/v1alpha1/integrationsink.go @@ -87,8 +87,20 @@ func WithIntegrationSinkEventPoliciesReadyBecauseOIDCDisabled() IntegrationSinkO } } +func WithIntegrationSinkEventPoliciesReadyBecauseNoPolicyAndOIDCEnabled() IntegrationSinkOption { + return func(s *v1alpha1.IntegrationSink) { + s.Status.MarkEventPoliciesTrueWithReason("DefaultAuthorizationMode", "Default authz mode is %q", feature.AuthorizationAllowSameNamespace) + } +} + func WithIntegrationSinkTrustBundlePropagatedReady() IntegrationSinkOption { return func(s *v1alpha1.IntegrationSink) { s.Status.MarkTrustBundlePropagated() } } + +func WithIntegrationSinkEventPoliciesReady(reason, message string) IntegrationSinkOption { + return func(s *v1alpha1.IntegrationSink) { + s.Status.MarkEventPoliciesTrueWithReason(reason, message) + } +} From cc43b46ca440ec17f336f918422555e09634b826 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Christoph=20St=C3=A4bler?= Date: Tue, 3 Feb 2026 17:07:13 +0100 Subject: [PATCH 2/5] Get resync period from context --- cmd/auth_proxy/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/auth_proxy/main.go b/cmd/auth_proxy/main.go index b53b34427f1..5d2b8e7fe3b 100644 --- a/cmd/auth_proxy/main.go +++ b/cmd/auth_proxy/main.go @@ -195,7 +195,7 @@ func setupEventPolicyInformer(ctx context.Context, config envConfig, logger *zap // Create namespace-scoped informer factory eventingInformerFactory := eventinginformers.NewSharedInformerFactoryWithOptions( eventingClient, - 0, // resyncPeriod - 0 means no resync + controller.GetResyncPeriod(ctx), eventinginformers.WithNamespace(config.ParentNamespace), ) From 3825d8b810fd2e53642e511af85d23ffecd12225 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Christoph=20St=C3=A4bler?= Date: Wed, 4 Feb 2026 08:12:40 +0100 Subject: [PATCH 3/5] Use rolebindingLister instead of kubeclient directly --- pkg/reconciler/integration/sink/integrationsink.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/reconciler/integration/sink/integrationsink.go b/pkg/reconciler/integration/sink/integrationsink.go index 5d452ca30be..f0c6131778e 100644 --- a/pkg/reconciler/integration/sink/integrationsink.go +++ b/pkg/reconciler/integration/sink/integrationsink.go @@ -509,7 +509,7 @@ func (r *Reconciler) reconcileConfigMapAccessRBAC(ctx context.Context, sink *sin func (r *Reconciler) reconcileEventPolicyRoleBinding(ctx context.Context, sink *sinks.IntegrationSink) error { expected := resources.MakeEventPolicyRoleBinding(sink) - rb, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Get(ctx, expected.Name, metav1.GetOptions{}) + rb, err := r.rolebindingLister.RoleBindings(expected.Namespace).Get(expected.Name) if apierrors.IsNotFound(err) { _, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Create(ctx, expected, metav1.CreateOptions{}) if err != nil { From 660865a263290de4c7e3ae27d363e65f459952e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Christoph=20St=C3=A4bler?= Date: Wed, 4 Feb 2026 08:19:42 +0100 Subject: [PATCH 4/5] Delete EventPolicy RBAC when OIDC gets disabled --- .../integration/sink/integrationsink.go | 60 ++++++++++--------- 1 file changed, 33 insertions(+), 27 deletions(-) diff --git a/pkg/reconciler/integration/sink/integrationsink.go b/pkg/reconciler/integration/sink/integrationsink.go index f0c6131778e..f7ce5335c8d 100644 --- a/pkg/reconciler/integration/sink/integrationsink.go +++ b/pkg/reconciler/integration/sink/integrationsink.go @@ -457,14 +457,31 @@ func (r *Reconciler) reconcileAuthProxyRBAC(ctx context.Context, sink *sinks.Int // that allows auth-proxy to read EventPolicies in the sink's namespace. // The RoleBinding references the knative-eventing-eventpolicy-reader ClusterRole. func (r *Reconciler) reconcileEventPolicyRBAC(ctx context.Context, sink *sinks.IntegrationSink, features feature.Flags) error { + expected := resources.MakeEventPolicyRoleBinding(sink) + if !features.IsOIDCAuthentication() { - // EventPolicy RoleBinding has OwnerReferences, so it'll be garbage collected - // when the sink is deleted. No explicit cleanup needed when OIDC is disabled. + return r.deleteEventPolicyRBAC(ctx, expected) + } + + rb, err := r.rolebindingLister.RoleBindings(expected.Namespace).Get(expected.Name) + if apierrors.IsNotFound(err) { + _, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Create(ctx, expected, metav1.CreateOptions{}) + if err != nil { + return fmt.Errorf("creating EventPolicy RoleBinding: %w", err) + } return nil } + if err != nil { + return fmt.Errorf("getting EventPolicy RoleBinding: %w", err) + } - if err := r.reconcileEventPolicyRoleBinding(ctx, sink); err != nil { - return err + // Update if subjects changed + if !equality.Semantic.DeepEqual(rb.Subjects, expected.Subjects) { + rb.Subjects = expected.Subjects + _, err = r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, rb, metav1.UpdateOptions{}) + if err != nil { + return fmt.Errorf("updating EventPolicy RoleBinding: %w", err) + } } return nil @@ -479,7 +496,7 @@ func (r *Reconciler) reconcileConfigMapAccessRBAC(ctx context.Context, sink *sin } if !features.IsOIDCAuthentication() { - return r.deleteIntegrationSinkRBAC(ctx, expected) + return r.deleteConfigMapAccessRBAC(ctx, expected) } rolebinding, err := r.rolebindingLister.RoleBindings(expected.Namespace).Get(expected.Name) @@ -506,34 +523,27 @@ func (r *Reconciler) reconcileConfigMapAccessRBAC(ctx context.Context, sink *sin return nil } -func (r *Reconciler) reconcileEventPolicyRoleBinding(ctx context.Context, sink *sinks.IntegrationSink) error { - expected := resources.MakeEventPolicyRoleBinding(sink) - - rb, err := r.rolebindingLister.RoleBindings(expected.Namespace).Get(expected.Name) +func (r *Reconciler) deleteConfigMapAccessRBAC(ctx context.Context, rolebinding *rbacv1.RoleBinding) error { + _, err := r.rolebindingLister.RoleBindings(rolebinding.Namespace).Get(rolebinding.Name) if apierrors.IsNotFound(err) { - _, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Create(ctx, expected, metav1.CreateOptions{}) - if err != nil { - return fmt.Errorf("creating EventPolicy RoleBinding: %w", err) - } return nil } if err != nil { - return fmt.Errorf("getting EventPolicy RoleBinding: %w", err) + return fmt.Errorf("failed to get rolebinding %s/%s: %w", rolebinding.Namespace, rolebinding.Name, err) } - // Update if subjects changed - if !equality.Semantic.DeepEqual(rb.Subjects, expected.Subjects) { - rb.Subjects = expected.Subjects - _, err = r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, rb, metav1.UpdateOptions{}) - if err != nil { - return fmt.Errorf("updating EventPolicy RoleBinding: %w", err) - } + err = r.kubeClientSet.RbacV1().RoleBindings(rolebinding.Namespace).Delete(ctx, rolebinding.Name, metav1.DeleteOptions{}) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return fmt.Errorf("failed to delete auth-proxy rolebinding %s/%s: %w", rolebinding.Namespace, rolebinding.Name, err) } return nil } -func (r *Reconciler) deleteIntegrationSinkRBAC(ctx context.Context, rolebinding *rbacv1.RoleBinding) error { +func (r *Reconciler) deleteEventPolicyRBAC(ctx context.Context, rolebinding *rbacv1.RoleBinding) error { _, err := r.rolebindingLister.RoleBindings(rolebinding.Namespace).Get(rolebinding.Name) if apierrors.IsNotFound(err) { return nil @@ -547,13 +557,9 @@ func (r *Reconciler) deleteIntegrationSinkRBAC(ctx context.Context, rolebinding return nil } if err != nil { - return fmt.Errorf("failed to delete auth-proxy rolebinding %s/%s: %w", rolebinding.Namespace, rolebinding.Name, err) + return fmt.Errorf("failed to delete auth-proxy eventpolicy rolebinding %s/%s: %w", rolebinding.Namespace, rolebinding.Name, err) } - // Note: EventPolicy Role and RoleBinding have OwnerReferences set to the IntegrationSink, - // so they will be automatically garbage collected when the sink is deleted. - // We don't need to explicitly delete them here. - return nil } From 5c9fe25a262b9c1b2c4653e08d26da856193e4ed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Christoph=20St=C3=A4bler?= Date: Wed, 4 Feb 2026 09:40:58 +0100 Subject: [PATCH 5/5] Recreate subjectsWithFilters only on eventPolicy changes --- cmd/auth_proxy/main.go | 59 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) diff --git a/cmd/auth_proxy/main.go b/cmd/auth_proxy/main.go index 5d2b8e7fe3b..3a757172328 100644 --- a/cmd/auth_proxy/main.go +++ b/cmd/auth_proxy/main.go @@ -24,6 +24,7 @@ import ( "context" "net" "os" + "sync" //nolint:gosec "crypto/tls" @@ -38,6 +39,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" "k8s.io/utils/ptr" cmdbroker "knative.dev/eventing/cmd/broker" "knative.dev/eventing/pkg/apis/feature" @@ -95,6 +97,10 @@ type ProxyHandler struct { config envConfig eventPolicyLister eventingv1alpha1listers.EventPolicyLister parentResourceGVK schema.GroupVersionKind + + // Cache for subjects with filters + subjectsWithFiltersMu sync.RWMutex + cachedSubjectsWithFilters []auth.SubjectsWithFilters } func main() { @@ -125,6 +131,9 @@ func main() { logger.Fatalw("Failed to create proxy handler", zap.Error(err)) } + // Register event handler to invalidate cache when EventPolicies change + registerEventPolicyHandler(eventPolicyInformer, handler, logger) + serverManager, err := createServerManager(ctx, config, handler, logger, configMapWatcher) if err != nil { logger.Fatalw("Failed to create server manager", zap.Error(err)) @@ -183,7 +192,7 @@ func setupFeatureStore(_ context.Context, logger *zap.SugaredLogger, configMapWa } // setupEventPolicyInformer creates a namespace-scoped EventPolicy informer -func setupEventPolicyInformer(ctx context.Context, config envConfig, logger *zap.SugaredLogger) (eventingv1alpha1listers.EventPolicyLister, controller.Informer, error) { +func setupEventPolicyInformer(ctx context.Context, config envConfig, logger *zap.SugaredLogger) (eventingv1alpha1listers.EventPolicyLister, cache.SharedIndexInformer, error) { cfg := injection.GetConfig(ctx) // Create eventing client @@ -284,6 +293,30 @@ func startServices(ctx context.Context, informers []controller.Informer, configM return nil } +// registerEventPolicyHandler registers an event handler on the EventPolicy informer +// to invalidate the cache when policies change +func registerEventPolicyHandler(eventPolicyInformer cache.SharedIndexInformer, handler *ProxyHandler, logger *zap.SugaredLogger) { + _, _ = eventPolicyInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + handler.invalidateSubjectsCache(logger) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + handler.invalidateSubjectsCache(logger) + }, + DeleteFunc: func(obj interface{}) { + handler.invalidateSubjectsCache(logger) + }, + }) +} + +// invalidateSubjectsCache clears the cached subjects with filters +func (h *ProxyHandler) invalidateSubjectsCache(logger *zap.SugaredLogger) { + h.subjectsWithFiltersMu.Lock() + defer h.subjectsWithFiltersMu.Unlock() + h.cachedSubjectsWithFilters = nil + logger.Debug("Invalidated subjects cache due to EventPolicy change") +} + func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { ctx := h.withContext(r.Context()) logger := logging.FromContext(ctx) @@ -316,6 +349,26 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // getAuthSubjects retrieves EventPolicies for the parent resource and converts them to SubjectsWithFilters func (h *ProxyHandler) getAuthSubjects(logger *zap.SugaredLogger) ([]auth.SubjectsWithFilters, error) { + // Try to use cached value (fast path with read lock) + h.subjectsWithFiltersMu.RLock() + if h.cachedSubjectsWithFilters != nil { + cached := h.cachedSubjectsWithFilters + h.subjectsWithFiltersMu.RUnlock() + logger.Debug("Using cached subjects with filters") + return cached, nil + } + h.subjectsWithFiltersMu.RUnlock() + + // Cache miss - acquire write lock and compute + h.subjectsWithFiltersMu.Lock() + defer h.subjectsWithFiltersMu.Unlock() + + // Double-check after acquiring write lock (another goroutine might have populated the cache) + if h.cachedSubjectsWithFilters != nil { + logger.Debug("Using cached subjects with filters (after lock)") + return h.cachedSubjectsWithFilters, nil + } + // Get EventPolicies applying to this resource policies, err := auth.GetEventPoliciesForResource( h.eventPolicyLister, @@ -340,6 +393,10 @@ func (h *ProxyHandler) getAuthSubjects(logger *zap.SugaredLogger) ([]auth.Subjec }) } + // Store in cache + h.cachedSubjectsWithFilters = subjectsWithFilters + logger.Debugf("Cached %d subjects with filters for %s/%s", len(subjectsWithFilters), h.config.ParentNamespace, h.config.ParentName) + return subjectsWithFilters, nil }