Add probes to IntegrationSource & -Sink deployments (#8867)

* Add probes to IntegrationSource deployments

* Fix auth-proxy SINK_URI missing value error

The auth-proxy container was crashing with "required key SINK_URI
missing value" due to a circular dependency in the reconciliation
order. The deployment (with auth-proxy) was being created before
the IntegrationSink status.Address was set, causing the auth-proxy
to fail during startup.

This commit fixes the issue by deriving the SINK_URI directly from
the sink name and namespace (using network.GetServiceHostname())
instead of reading it from status.Address. This matches how the
reconcileAddress() function constructs the URL, but without
requiring the status to be set first.

The same approach is now used for the SINK_AUDIENCE when OIDC
authentication is enabled.

This eliminates the circular dependency and ensures the auth-proxy
always has a valid SINK_URI, regardless of reconciliation timing.

* Add readiness check to auth-proxy

* Fix unit tests

* Use same timings for auth-proxy probes

Author: creydr

cmd/auth_proxy/main.go

@@ -22,9 +22,11 @@ package main
 
 import (
 	"context"
+	"errors"
 	"net"
 	"os"
 	"sync"
+	"time"
 
 	//nolint:gosec
 	"crypto/tls"
@@ -70,6 +72,7 @@ type envConfig struct {
 	TargetHTTPSPort int    `envconfig:"TARGET_HTTPS_PORT"  default:"8443"`
 	ProxyHTTPPort   int    `envconfig:"PROXY_HTTP_PORT" default:"3128"`
 	ProxyHTTPSPort  int    `envconfig:"PROXY_HTTPS_PORT" default:"3129"`
+	ProxyHealthPort int    `envconfig:"PROXY_HEALTH_PORT" default:"8090"`
 
 	SinkURI       string  `envconfig:"SINK_URI" required:"true"`
 	SinkNamespace string  `envconfig:"SINK_NAMESPACE" required:"true"`
@@ -143,6 +146,20 @@ func main() {
 		logger.Fatalw("Failed to start services", zap.Error(err))
 	}
 
+	// Start health server
+	healthServer := startHealthServer(config, handler)
+	go func() {
+		logger.Infof("Starting health server on port %d", config.ProxyHealthPort)
+		if err := healthServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			logger.Errorw("Health server error", zap.Error(err))
+		}
+	}()
+	defer func() {
+		if err := healthServer.Shutdown(context.Background()); err != nil {
+			logger.Errorw("Error shutting down health server", zap.Error(err))
+		}
+	}()
+
 	logger.Info("Starting auth proxy servers...")
 	if err = serverManager.StartServers(ctx); err != nil {
 		logger.Fatalw("StartServers() returned an error", zap.Error(err))
@@ -317,6 +334,34 @@ func (h *ProxyHandler) invalidateSubjectsCache(logger *zap.SugaredLogger) {
 	logger.Debug("Invalidated subjects cache due to EventPolicy change")
 }
 
+// startHealthServer creates and returns an HTTP server for health checks
+func startHealthServer(config envConfig, handler *ProxyHandler) *http.Server {
+	mux := http.NewServeMux()
+
+	// Readiness endpoint - checks if auth-proxy is ready to validate tokens
+	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
+		if handler.authVerifier.IsReady() {
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte("ready"))
+		} else {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			_, _ = w.Write([]byte("not ready"))
+		}
+	})
+
+	// Liveness endpoint - always returns OK if the server is running
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("alive"))
+	})
+
+	return &http.Server{
+		Addr:              fmt.Sprintf(":%d", config.ProxyHealthPort),
+		Handler:           mux,
+		ReadHeaderTimeout: 10 * time.Second,
+	}
+}
+
 func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ctx := h.withContext(r.Context())
 	logger := logging.FromContext(ctx)

pkg/auth/verifier.go

@@ -298,6 +298,15 @@ func (v *Verifier) initOIDCProvider(ctx context.Context, features feature.Flags)
 	return nil
 }
 
+// IsReady returns true if the OIDC provider has been initialized and the verifier
+// is ready to validate tokens. This is used by health checks to ensure the auth-proxy
+// doesn't receive traffic before it can properly validate authentication.
+func (v *Verifier) IsReady() bool {
+	v.m.RLock()
+	defer v.m.RUnlock()
+	return v.provider != nil
+}
+
 func (v *Verifier) getHTTPClientForKubeAPIServer() (*http.Client, error) {
 	client, err := rest.HTTPClientFor(v.restConfig)
 	if err != nil {

pkg/reconciler/integration/helper.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 func GenerateEnvVarsFromStruct(prefix string, s interface{}) []corev1.EnvVar {
@@ -132,3 +133,54 @@ func Labels(name string) map[string]string {
 		"app.kubernetes.io/name": name,
 	}
 }
+
+func LivenessProbe(port int32, scheme corev1.URIScheme) *corev1.Probe {
+	return &corev1.Probe{
+		FailureThreshold: 3,
+		ProbeHandler: corev1.ProbeHandler{
+			HTTPGet: &corev1.HTTPGetAction{
+				Path:   "/q/health/live",
+				Port:   intstr.FromInt32(port),
+				Scheme: scheme,
+			},
+		},
+		InitialDelaySeconds: 5,
+		PeriodSeconds:       10,
+		SuccessThreshold:    1,
+		TimeoutSeconds:      10,
+	}
+}
+
+func ReadinessProbe(port int32, scheme corev1.URIScheme) *corev1.Probe {
+	return &corev1.Probe{
+		FailureThreshold: 3,
+		ProbeHandler: corev1.ProbeHandler{
+			HTTPGet: &corev1.HTTPGetAction{
+				Path:   "/q/health/ready",
+				Port:   intstr.FromInt32(port),
+				Scheme: scheme,
+			},
+		},
+		InitialDelaySeconds: 5,
+		PeriodSeconds:       10,
+		SuccessThreshold:    1,
+		TimeoutSeconds:      10,
+	}
+}
+
+func StartupProbe(port int32, scheme corev1.URIScheme) *corev1.Probe {
+	return &corev1.Probe{
+		FailureThreshold: 3,
+		ProbeHandler: corev1.ProbeHandler{
+			HTTPGet: &corev1.HTTPGetAction{
+				Path:   "/q/health/started",
+				Port:   intstr.FromInt32(port),
+				Scheme: scheme,
+			},
+		},
+		InitialDelaySeconds: 5,
+		PeriodSeconds:       10,
+		SuccessThreshold:    1,
+		TimeoutSeconds:      10,
+	}
+}

pkg/reconciler/integration/sink/integrationsink_test.go

@@ -363,6 +363,48 @@ func makeDeployment(sink *sinksv1alpha1.IntegrationSink, ready *corev1.Condition
 									ReadOnly:  true,
 								},
 							},
+							LivenessProbe: &corev1.Probe{
+								FailureThreshold: 3,
+								ProbeHandler: corev1.ProbeHandler{
+									HTTPGet: &corev1.HTTPGetAction{
+										Path:   "/q/health/live",
+										Port:   intstr.FromInt32(8080),
+										Scheme: corev1.URISchemeHTTP,
+									},
+								},
+								InitialDelaySeconds: 5,
+								PeriodSeconds:       10,
+								SuccessThreshold:    1,
+								TimeoutSeconds:      10,
+							},
+							ReadinessProbe: &corev1.Probe{
+								FailureThreshold: 3,
+								ProbeHandler: corev1.ProbeHandler{
+									HTTPGet: &corev1.HTTPGetAction{
+										Path:   "/q/health/ready",
+										Port:   intstr.FromInt32(8080),
+										Scheme: corev1.URISchemeHTTP,
+									},
+								},
+								InitialDelaySeconds: 5,
+								PeriodSeconds:       10,
+								SuccessThreshold:    1,
+								TimeoutSeconds:      10,
+							},
+							StartupProbe: &corev1.Probe{
+								FailureThreshold: 3,
+								ProbeHandler: corev1.ProbeHandler{
+									HTTPGet: &corev1.HTTPGetAction{
+										Path:   "/q/health/started",
+										Port:   intstr.FromInt32(8080),
+										Scheme: corev1.URISchemeHTTP,
+									},
+								},
+								InitialDelaySeconds: 5,
+								PeriodSeconds:       10,
+								SuccessThreshold:    1,
+								TimeoutSeconds:      10,
+							},
 						},
 					},
 				},
@@ -385,21 +427,25 @@ func makeDeploymentWithAuthProxy(sink *sinksv1alpha1.IntegrationSink, ready *cor
 		Ports: []corev1.ContainerPort{
 			{ContainerPort: 3128, Protocol: corev1.ProtocolTCP, Name: "http"},
 			{ContainerPort: 3129, Protocol: corev1.ProtocolTCP, Name: "https"},
+			{ContainerPort: 8090, Protocol: corev1.ProtocolTCP, Name: "health"},
 		},
 		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: "PROXY_HEALTH_PORT", Value: "8090"},
 			{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_URI", Value: "http://" + deploymentName + "." + sink.Namespace + ".svc.cluster.local"},
 			{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"},
+			{Name: "SINK_AUDIENCE", Value: "sinks.knative.dev/integrationsink/" + sink.Namespace + "/" + sink.Name},
 		},
 		VolumeMounts: []corev1.VolumeMount{
 			{
@@ -408,6 +454,48 @@ func makeDeploymentWithAuthProxy(sink *sinksv1alpha1.IntegrationSink, ready *cor
 				ReadOnly:  true,
 			},
 		},
+		ReadinessProbe: &corev1.Probe{
+			ProbeHandler: corev1.ProbeHandler{
+				HTTPGet: &corev1.HTTPGetAction{
+					Path:   "/readyz",
+					Port:   intstr.FromInt32(8090),
+					Scheme: corev1.URISchemeHTTP,
+				},
+			},
+			InitialDelaySeconds: 5,
+			PeriodSeconds:       10,
+			SuccessThreshold:    1,
+			FailureThreshold:    3,
+			TimeoutSeconds:      10,
+		},
+		LivenessProbe: &corev1.Probe{
+			ProbeHandler: corev1.ProbeHandler{
+				HTTPGet: &corev1.HTTPGetAction{
+					Path:   "/healthz",
+					Port:   intstr.FromInt32(8090),
+					Scheme: corev1.URISchemeHTTP,
+				},
+			},
+			InitialDelaySeconds: 5,
+			PeriodSeconds:       10,
+			SuccessThreshold:    1,
+			FailureThreshold:    3,
+			TimeoutSeconds:      10,
+		},
+		StartupProbe: &corev1.Probe{
+			ProbeHandler: corev1.ProbeHandler{
+				HTTPGet: &corev1.HTTPGetAction{
+					Path:   "/readyz",
+					Port:   intstr.FromInt32(8090),
+					Scheme: corev1.URISchemeHTTP,
+				},
+			},
+			InitialDelaySeconds: 5,
+			PeriodSeconds:       10,
+			SuccessThreshold:    1,
+			FailureThreshold:    3,
+			TimeoutSeconds:      10,
+		},
 	}
 
 	d.Spec.Template.Spec.Containers = append(d.Spec.Template.Spec.Containers, authProxyContainer)