Readiness probe

controllers/auth_config_controller.go

@@ -47,7 +47,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-const failedToCleanConfig = "failed to clean up all asynchronous workers"
+const (
+	failedToCleanConfig = "failed to clean up all asynchronous workers"
+
+	AuthConfigsReadyzSubpath = "authconfigs"
+)
 
 // AuthConfigReconciler reconciles an AuthConfig object
 type AuthConfigReconciler struct {
@@ -682,6 +686,22 @@ func (r *AuthConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Complete(r)
 }
 
+func (r *AuthConfigReconciler) Ready(includes, _ []string, _ bool) error {
+	if !utils.SliceContains(includes, AuthConfigsReadyzSubpath) {
+		return nil
+	}
+
+	for id, status := range r.StatusReport.ReadAll() {
+		switch status.Reason {
+		case api.StatusReasonReconciled:
+			continue
+		default:
+			return fmt.Errorf("authconfig is not ready: %s (reason: %s)", id, status.Reason)
+		}
+	}
+	return nil
+}
+
 func findIdentityConfigByName(identityConfigs []evaluators.IdentityConfig, name string) (*evaluators.IdentityConfig, error) {
 	for _, id := range identityConfigs {
 		if id.Name == name {

controllers/status_report.go

@@ -3,6 +3,8 @@ package controllers
 import (
 	"sync"
 	"time"
+
+	"github.com/kuadrant/authorino/pkg/utils"
 )
 
 func NewStatusReportMap() *StatusReportMap {
@@ -36,6 +38,13 @@ func (m *StatusReportMap) Set(id, reason, message string, hosts []string) {
 	}
 }
 
+func (m *StatusReportMap) ReadAll() map[string]StatusReport {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	return utils.CopyMap(m.statuses)
+}
+
 func (m *StatusReportMap) Clear(id string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

main.go

@@ -39,6 +39,7 @@ import (
 	api "github.com/kuadrant/authorino/api/v1beta1"
 	"github.com/kuadrant/authorino/controllers"
 	"github.com/kuadrant/authorino/pkg/evaluators"
+	"github.com/kuadrant/authorino/pkg/health"
 	"github.com/kuadrant/authorino/pkg/index"
 	"github.com/kuadrant/authorino/pkg/log"
 	"github.com/kuadrant/authorino/pkg/metrics"
@@ -49,6 +50,7 @@ import (
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	// +kubebuilder:scaffold:imports
 )
 
@@ -71,6 +73,7 @@ const (
 	envMaxHttpRequestBodySize         = "MAX_HTTP_REQUEST_BODY_SIZE" // in bytes
 
 	flagMetricsAddr          = "metrics-addr"
+	flagHealthProbeAddr      = "health-probe-addr"
 	flagEnableLeaderElection = "enable-leader-election"
 
 	defaultWatchNamespace                 = ""
@@ -89,6 +92,7 @@ const (
 	defaultEvaluatorCacheSize             = "1"
 	defaultDeepMetricsEnabled             = "false"
 	defaultMetricsAddr                    = ":8080"
+	defaultHealthProbeAddr                = ":8081"
 	defaultEnableLeaderElection           = false
 	defaultMaxHttpRequestBodySize         = "8192" // 8KB
 
@@ -135,9 +139,10 @@ func init() {
 }
 
 func main() {
-	var metricsAddr string
+	var metricsAddr, healthProbeAddr string
 	var enableLeaderElection bool
 	flag.StringVar(&metricsAddr, flagMetricsAddr, defaultMetricsAddr, "The address the metric endpoint binds to.")
+	flag.StringVar(&healthProbeAddr, flagHealthProbeAddr, defaultHealthProbeAddr, "The address the health probe endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, flagEnableLeaderElection, defaultEnableLeaderElection, "Enable leader election for status updater. Ensures only one instance of Authorino tries to update the status of reconciled resources.")
 	flag.Parse()
 
@@ -160,14 +165,16 @@ func main() {
 		envEvaluatorCacheSize, metadataCacheSize,
 		envDeepMetricsEnabled, deepMetricEnabled,
 		flagMetricsAddr, metricsAddr,
+		flagHealthProbeAddr, healthProbeAddr,
 		flagEnableLeaderElection, enableLeaderElection,
 	)
 
 	managerOptions := ctrl.Options{
-		Scheme:             scheme,
-		MetricsBindAddress: metricsAddr,
-		Port:               9443,
-		LeaderElection:     false,
+		Scheme:                 scheme,
+		MetricsBindAddress:     metricsAddr,
+		HealthProbeBindAddress: healthProbeAddr,
+		Port:                   9443,
+		LeaderElection:         false,
 	}
 
 	if watchNamespace != "" {
@@ -218,7 +225,21 @@ func main() {
 	startExtAuthServerHTTP(index)
 	startOIDCServer(index)
 
-	_ = mgr.AddMetricsExtraHandler("/server-metrics", promhttp.Handler())
+	if err := mgr.AddMetricsExtraHandler("/server-metrics", promhttp.Handler()); err != nil {
+		logger.Error(err, "unable to set up controller metrics server")
+		os.Exit(1)
+	}
+
+	if err := mgr.AddHealthzCheck("ping", healthz.Ping); err != nil {
+		logger.Error(err, "unable to set up controller health check")
+		os.Exit(1)
+	}
+
+	readinessCheck := health.NewHandler(controllers.AuthConfigsReadyzSubpath, health.Observe(authConfigReconciler))
+	if err := mgr.AddReadyzCheck(controllers.AuthConfigsReadyzSubpath, readinessCheck.HandleReadyzCheck); err != nil {
+		logger.Error(err, "unable to set up controller readiness check")
+		os.Exit(1)
+	}
 
 	signalHandler := ctrl.SetupSignalHandler()
 
@@ -236,6 +257,7 @@ func main() {
 	managerOptions.LeaderElection = enableLeaderElection
 	managerOptions.LeaderElectionID = fmt.Sprintf("%v.%v", hex.EncodeToString(leaderElectionId[:4]), leaderElectionIDSuffix)
 	managerOptions.MetricsBindAddress = "0"
+	managerOptions.HealthProbeBindAddress = "0"
 	statusUpdateManager, err := ctrl.NewManager(ctrl.GetConfigOrDie(), managerOptions)
 	if err != nil {
 		logger.Error(err, "unable to start status update manager")

pkg/health/health.go

@@ -0,0 +1,69 @@
+package health
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+)
+
+type Observable interface {
+	Ready(includes, excludes []string, verbose bool) error
+}
+
+type HealthzHandler interface {
+	Observe(...Observable)
+	HandleReadyzCheck(*http.Request) error
+}
+
+type HandlerOption func(HealthzHandler)
+
+func Observe(observables ...Observable) HandlerOption {
+	return func(h HealthzHandler) {
+		h.Observe(observables...)
+	}
+}
+
+func NewHandler(name string, options ...HandlerOption) HealthzHandler {
+	h := &handler{name: name}
+	for _, o := range options {
+		o(h)
+	}
+	return h
+}
+
+type handler struct {
+	name string
+
+	observables []Observable
+	mu          sync.RWMutex
+}
+
+func (h *handler) Observe(observables ...Observable) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.observables = append(h.observables, observables...)
+}
+
+func (h *handler) HandleReadyzCheck(req *http.Request) error {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	includes := req.URL.Query()["include"]
+	excludes := req.URL.Query()["exclude"]
+
+	// implicit include when requesting a specific check by name
+	if strings.HasSuffix(req.URL.Path, fmt.Sprintf("/%s", h.name)) || strings.HasSuffix(req.URL.Path, fmt.Sprintf("/%s/", h.name)) {
+		includes = append(includes, h.name)
+	}
+
+	// pass along the verbose parameter if present
+	_, verbose := req.URL.Query()["verbose"]
+
+	for _, reconciler := range h.observables {
+		if err := reconciler.Ready(includes, excludes, verbose); err != nil {
+			return err
+		}
+	}
+	return nil
+}

pkg/health/health_test.go

@@ -0,0 +1,109 @@
+package health
+
+import (
+	"fmt"
+	"net/http"
+	neturl "net/url"
+	"testing"
+
+	"github.com/kuadrant/authorino/pkg/utils"
+
+	"gotest.tools/assert"
+)
+
+type FakeObservableHealthy struct{}
+
+func (o *FakeObservableHealthy) Ready(_, _ []string, _ bool) error { return nil }
+
+type FakeObservableUnhealthy struct{}
+
+func (o *FakeObservableUnhealthy) Ready(_, _ []string, _ bool) error { return fmt.Errorf("unhealthy") }
+
+type FakeObservableFilterred struct {
+	checked []string
+}
+
+func (o *FakeObservableFilterred) Ready(includes, excludes []string, _ bool) error {
+	o.checked = includes
+
+	if utils.SliceContains(includes, "opt-in-unhealthy") {
+		return fmt.Errorf("opt-in-unhealthy not ready")
+	}
+
+	fmt.Println("includes: ", includes)
+	fmt.Println("excludes: ", excludes)
+
+	ready := utils.SliceContains(excludes, "opt-out-unhealthy")
+
+	if !ready {
+		o.checked = append(o.checked, "opt-out-unhealthy")
+		return fmt.Errorf("opt-out-unhealthy not ready")
+	}
+
+	if !utils.SliceContains(excludes, "opt-out-healthy") {
+		o.checked = append(o.checked, "opt-out-healthy")
+	}
+
+	return nil
+}
+
+func TestObserveHealthy(t *testing.T) {
+	h := NewHandler("foo", Observe(&FakeObservableHealthy{}))
+	err := h.HandleReadyzCheck(mockReq("http://localhost:8081/readyz"))
+	assert.NilError(t, err)
+}
+
+func TestObserveUnealthy(t *testing.T) {
+	h := NewHandler("foo", Observe(&FakeObservableUnhealthy{}))
+	err := h.HandleReadyzCheck(mockReq("http://localhost:8081/readyz"))
+	assert.ErrorContains(t, err, "unhealthy")
+}
+
+func TestObserveHeathyUnealthy(t *testing.T) {
+	h := NewHandler("foo", Observe(&FakeObservableHealthy{}, &FakeObservableUnhealthy{}))
+	err := h.HandleReadyzCheck(mockReq("http://localhost:8081/readyz"))
+	assert.ErrorContains(t, err, "unhealthy")
+}
+
+func TestObserveIncludeExclude(t *testing.T) {
+	o := &FakeObservableFilterred{checked: []string{}}
+	h := NewHandler("foo", Observe(o))
+	err := h.HandleReadyzCheck(mockReq("http://localhost:8081/readyz?include=opt-in-healthy&exclude=opt-out-unhealthy"))
+	assert.NilError(t, err)
+	assert.Equal(t, len(o.checked), 2)
+	assert.Equal(t, o.checked[0], "opt-in-healthy")
+	assert.Equal(t, o.checked[1], "opt-out-healthy")
+}
+
+func TestObserveIncludeUnhealthy(t *testing.T) {
+	o := &FakeObservableFilterred{checked: []string{}}
+	h := NewHandler("foo", Observe(o))
+	err := h.HandleReadyzCheck(mockReq("http://localhost:8081/readyz?include=opt-in-unhealthy"))
+	assert.ErrorContains(t, err, "opt-in-unhealthy not ready")
+	assert.Equal(t, len(o.checked), 1)
+	assert.Equal(t, o.checked[0], "opt-in-unhealthy")
+}
+
+func TestObserveExcludeUnhealthy(t *testing.T) {
+	o := &FakeObservableFilterred{checked: []string{}}
+	h := NewHandler("foo", Observe(o))
+	err := h.HandleReadyzCheck(mockReq("http://localhost:8081/readyz?exclude=opt-out-unhealthy"))
+	assert.NilError(t, err)
+	assert.Equal(t, len(o.checked), 1)
+	assert.Equal(t, o.checked[0], "opt-out-healthy")
+}
+
+func TestObserveIncludeImplicit(t *testing.T) {
+	o := &FakeObservableFilterred{checked: []string{}}
+	h := NewHandler("foo", Observe(o))
+	err := h.HandleReadyzCheck(mockReq("http://localhost:8081/readyz/foo"))
+	assert.ErrorContains(t, err, "opt-out-unhealthy not ready")
+	assert.Equal(t, len(o.checked), 2)
+	assert.Equal(t, o.checked[0], "foo")
+	assert.Equal(t, o.checked[1], "opt-out-unhealthy")
+}
+
+func mockReq(url string) *http.Request {
+	u, _ := neturl.Parse(url)
+	return &http.Request{URL: u}
+}

pkg/utils/utils.go

@@ -30,3 +30,20 @@ func SubtractSlice(sl1, sl2 []string) []string {
 	}
 	return diff
 }
+
+func SliceContains[T comparable](s []T, val T) bool {
+	for _, v := range s {
+		if v == val {
+			return true
+		}
+	}
+	return false
+}
+
+func CopyMap[T comparable, U any](m map[T]U) map[T]U {
+	m2 := make(map[T]U)
+	for k, v := range m {
+		m2[k] = v
+	}
+	return m2
+}

pkg/utils/utils_test.go

@@ -20,3 +20,26 @@ func TestSubtractSlice(t *testing.T) {
 	assert.Equal(t, strings.Join(SubtractSlice([]string{"a", "b", "c"}, []string{"c", "d"}), ""), "ab")
 	assert.Equal(t, strings.Join(SubtractSlice([]string{"a", "b", "c"}, []string{}), ""), "abc")
 }
+
+func TestSliceContains(t *testing.T) {
+	assert.Check(t, SliceContains([]string{"a", "b", "c"}, "a"))
+	assert.Check(t, SliceContains([]string{"a", "b", "c"}, "b"))
+	assert.Check(t, SliceContains([]string{"a", "b", "c"}, "c"))
+	assert.Check(t, !SliceContains([]string{"a", "b", "c"}, "d"))
+	assert.Check(t, SliceContains([]int{1, 2, 3}, 3))
+	assert.Check(t, !SliceContains([]int{1, 2, 3}, 4))
+}
+
+func TestCopyMap(t *testing.T) {
+	m1 := map[string]int{
+		"a": 1,
+		"b": 2,
+	}
+	m2 := CopyMap(m1)
+	assert.Check(t, &m1 != &m2)
+	assert.Equal(t, len(m1), len(m2))
+	assert.Equal(t, m1["a"], m2["a"])
+	assert.Equal(t, m1["b"], m2["b"])
+	m1["a"] = 3
+	assert.Check(t, m1["a"] != m2["a"])
+}