feat: add timeout to monitoring tasks (active and passive)

plaffitt · plaffitt · commit 70e8756ae45f · 2025-10-07T12:58:59.000+02:00
diff --git a/api/kuik/v1alpha1/imagemonitor_types.go b/api/kuik/v1alpha1/imagemonitor_types.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"strings"
+	"time"
 
 	"github.com/cespare/xxhash"
 	"github.com/enix/kube-image-keeper/internal/registry"
@@ -132,7 +133,7 @@ func (i *ImageMonitor) GetPullSecrets(ctx context.Context, c client.Client) (sec
 	return image.GetPullSecrets(ctx, c)
 }
 
-func (i *ImageMonitor) Monitor(ctx context.Context, k8sClient client.Client, httpMethod string) error {
+func (i *ImageMonitor) Monitor(ctx context.Context, k8sClient client.Client, httpMethod string, timeout time.Duration) error {
 	patch := client.MergeFrom(i.DeepCopy())
 	i.Status.Upstream.LastMonitor = metav1.Now()
 	if err := k8sClient.Status().Patch(ctx, i, patch); err != nil {
@@ -141,9 +142,10 @@ func (i *ImageMonitor) Monitor(ctx context.Context, k8sClient client.Client, htt
 
 	patch = client.MergeFrom(i.DeepCopy())
 	pullSecrets, pullSecretsErr := i.GetPullSecrets(ctx, k8sClient)
+	client := registry.NewClient(nil, nil).WithPullSecrets(pullSecrets)
 
 	var lastErr error
-	if desc, err := registry.ReadDescriptor(httpMethod, i.Reference(), pullSecrets, nil, nil); err != nil {
+	if desc, err := client.ReadDescriptor(httpMethod, i.Reference(), timeout); err != nil {
 		i.Status.Upstream.LastError = err.Error()
 		lastErr = err
 		var te *transport.Error
diff --git a/api/kuik/v1alpha1/registrymonitor_types.go b/api/kuik/v1alpha1/registrymonitor_types.go
@@ -24,6 +24,9 @@ type RegistryMonitorSpec struct {
 	// +kubebuilder:validation:Enum=HEAD;GET
 	// +default:value="HEAD"
 	Method string `json:"method"`
+	// Timeout is the maximum duration of a monitoring task
+	// +default:value="30s"
+	Timeout metav1.Duration `json:"timeout"`
 }
 
 // RegistryMonitorStatus defines the observed state of RegistryMonitor.
diff --git a/api/kuik/v1alpha1/zz_generated.deepcopy.go b/api/kuik/v1alpha1/zz_generated.deepcopy.go
diff --git a/config/crd/bases/kuik.enix.io_registrymonitors.yaml b/config/crd/bases/kuik.enix.io_registrymonitors.yaml
@@ -86,12 +86,17 @@ spec:
                 description: Registry is the registry to monitor for image updates,
                   it filters local image to check upstream
                 type: string
+              timeout:
+                default: 30s
+                description: Timeout is the maximum duration of a monitoring task
+                type: string
             required:
             - interval
             - maxPerInterval
             - method
             - parallel
             - registry
+            - timeout
             type: object
           status:
             description: RegistryMonitorStatus defines the observed state of RegistryMonitor.
diff --git a/helm/kube-image-keeper/crds/kuik.enix.io_registrymonitors.yaml b/helm/kube-image-keeper/crds/kuik.enix.io_registrymonitors.yaml
@@ -85,12 +85,17 @@ spec:
                 description: Registry is the registry to monitor for image updates,
                   it filters local image to check upstream
                 type: string
+              timeout:
+                default: 30s
+                description: Timeout is the maximum duration of a monitoring task
+                type: string
             required:
             - interval
             - maxPerInterval
             - method
             - parallel
             - registry
+            - timeout
             type: object
           status:
             description: RegistryMonitorStatus defines the observed state of RegistryMonitor.
diff --git a/helm/kube-image-keeper/templates/mutatingwebhookconfiguration.yaml b/helm/kube-image-keeper/templates/mutatingwebhookconfiguration.yaml
@@ -1,3 +1,4 @@
+{{- if gt (len .Values.configuration.routing.strategies) 0 }}
 apiVersion: admissionregistration.k8s.io/v1
 kind: MutatingWebhookConfiguration
 metadata:
@@ -34,4 +35,4 @@ webhooks:
     resources:
     - pods
   sideEffects: None
-
+{{- end -}}
diff --git a/helm/kube-image-keeper/values.yaml b/helm/kube-image-keeper/values.yaml
@@ -103,6 +103,8 @@ registryMonitors:
     parallel: 1
     # -- HTTP method to use to monitor an image of this registry
     method: HEAD
+    # -- Maximum duration of a monitoring task
+    timeout: 30s
   # -- RegistryMonitors to create on install and upgrade, if name is not provided, defaults to registry value.
   items:
     docker.io:
@@ -139,7 +141,9 @@ configuration:
   monitoring:
     enabled: true
   routing:
-    activeCheck: false
+    activeCheck:
+      enabled: true
+      timeout: 1s
     strategies: []
       # - paths:
       #     - enix/x509-certificate-exporter
diff --git a/internal/config/config.go b/internal/config/config.go
@@ -36,7 +36,10 @@ func load(provider koanf.Provider, parser koanf.Parser) (*Config, error) {
 
 	k.UnmarshalWithConf("", config, koanf.UnmarshalConf{
 		DecoderConfig: &mapstructure.DecoderConfig{
-			DecodeHook: mapstructure.ComposeDecodeHookFunc(stringToRegexp),
+			DecodeHook: mapstructure.ComposeDecodeHookFunc(
+				mapstructure.StringToTimeDurationHookFunc(),
+				stringToRegexp,
+			),
 		},
 	})
 
diff --git a/internal/controller/core/pod_controller.go b/internal/controller/core/pod_controller.go
@@ -155,6 +155,7 @@ func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		MaxPerInterval: 1,
 		Parallel:       1,
 		Method:         http.MethodHead,
+		Timeout:        metav1.Duration{Duration: 5 * time.Second},
 	}
 
 	// TODO: read this from config instead
diff --git a/internal/controller/kuik/registrymonitor_controller.go b/internal/controller/kuik/registrymonitor_controller.go
@@ -174,7 +174,7 @@ func (r *RegistryMonitorReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			logImageMonitor.Info("monitoring image")
 			// TODO: add to SetupWithManager Watches(&source.Channel{Source: eventChannel}, &handler.EnqueueRequestForObject{})
 			// push an event in the channel when this function is done
-			if err := imageMonitor.Monitor(logf.IntoContext(ctx, logImageMonitor), r.Client, registryMonitor.Spec.Method); err != nil {
+			if err := imageMonitor.Monitor(logf.IntoContext(ctx, logImageMonitor), r.Client, registryMonitor.Spec.Method, registryMonitor.Spec.Timeout.Duration); err != nil {
 				logImageMonitor.Info("failed to monitor image", "error", err.Error())
 			}
 			logImageMonitor.Info("image monitored with success")
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
@@ -1,16 +1,17 @@
 package registry
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
 	"net/http"
 	"slices"
 	"strings"
+	"time"
 
 	"github.com/cespare/xxhash/v2"
 	"github.com/distribution/reference"
-	"github.com/google/go-containerregistry/pkg/authn"
 	"github.com/google/go-containerregistry/pkg/name"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
@@ -19,8 +20,40 @@ import (
 
 type descriptorReader func(ref name.Reference, options ...remote.Option) (*v1.Descriptor, error)
 
-func ReadDescriptor(httpMethod string, imageName string, pullSecrets []corev1.Secret, insecureRegistries []string, rootCAs *x509.CertPool) (*v1.Descriptor, error) {
-	keychains, err := GetKeychains(imageName, pullSecrets)
+type Client struct {
+	insecureRegistries []string
+	rootCAs            *x509.CertPool
+}
+
+type AuthenticatedClient struct {
+	*Client
+	pullSecrets []corev1.Secret
+}
+
+func NewClient(insecureRegistries []string, rootCAs *x509.CertPool) *Client {
+	return &Client{
+		insecureRegistries: insecureRegistries,
+		rootCAs:            rootCAs,
+	}
+}
+
+func (c *Client) options(ctx context.Context, ref name.Reference) []remote.Option {
+	transport := unauthenticatedTransport(ref.Context().RegistryStr(), c.insecureRegistries, c.rootCAs)
+	return []remote.Option{
+		remote.WithTransport(transport),
+		remote.WithContext(ctx),
+	}
+}
+
+func (c *Client) WithPullSecrets(pullSecrets []corev1.Secret) *AuthenticatedClient {
+	return &AuthenticatedClient{
+		Client:      c,
+		pullSecrets: pullSecrets,
+	}
+}
+
+func (a *AuthenticatedClient) ReadDescriptor(httpMethod string, imageName string, timeout time.Duration) (*v1.Descriptor, error) {
+	keychains, err := GetKeychains(imageName, a.pullSecrets)
 	if err != nil {
 		return nil, err
 	}
@@ -30,9 +63,14 @@ func ReadDescriptor(httpMethod string, imageName string, pullSecrets []corev1.Se
 		return nil, err
 	}
 
+	// global timeout for all keychains
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	options := a.options(ctx, sourceRef)
+	defer cancel()
+
 	var returnedErr error
 	for _, keychain := range keychains {
-		opts := options(sourceRef, keychain, insecureRegistries, rootCAs)
+		opts := append([]remote.Option{remote.WithAuthFromKeychain(keychain)}, options...)
 		desc, err := getReader(httpMethod)(sourceRef, opts...)
 
 		if err == nil { // stops at the first success
@@ -100,11 +138,3 @@ func unauthenticatedTransport(registry string, insecureRegistries []string, root
 
 	return transport
 }
-
-func options(ref name.Reference, keychain authn.Keychain, insecureRegistries []string, rootCAs *x509.CertPool) []remote.Option {
-	transport := unauthenticatedTransport(ref.Context().RegistryStr(), insecureRegistries, rootCAs)
-	return []remote.Option{
-		remote.WithAuthFromKeychain(keychain),
-		remote.WithTransport(transport),
-	}
-}
diff --git a/internal/registry/routing/routing.go b/internal/registry/routing/routing.go
@@ -2,13 +2,20 @@ package routing
 
 import (
 	"regexp"
+	"time"
 
 	kuikv1alpha1 "github.com/enix/kube-image-keeper/api/kuik/v1alpha1"
 )
 
+type ActiveCheck struct {
+	Enabled bool          `koanf:"enabled"`
+	Timeout time.Duration `koanf:"timeout"`
+	// Timeoutstr string        `koanf:"timeout"`
+}
+
 type Routing struct {
-	Strategies  []Strategy `koanf:"strategies"`
-	ActiveCheck bool       `koanf:"activeCheck"`
+	Strategies  []Strategy  `koanf:"strategies"`
+	ActiveCheck ActiveCheck `koanf:"activeCheck"`
 }
 
 type Strategy struct {
diff --git a/internal/webhook/core/v1/pod_webhook.go b/internal/webhook/core/v1/pod_webhook.go
@@ -108,6 +108,9 @@ func (d *PodCustomDefaulter) handleContainer(ctx context.Context, container *cor
 	}
 
 	for _, reg := range matchingStrategy.Registries {
+		if reg == registry {
+			continue // don't check the same registry twice
+		}
 		alternativeRef := reg + "/" + imageReference.Path
 		alternativeStatus, err := d.getImageStatus(ctx, alternativeRef)
 		if err != nil {
@@ -134,13 +137,13 @@ func (d *PodCustomDefaulter) getImageStatus(ctx context.Context, reference strin
 		return kuikv1alpha1.ImageMonitorStatusUpstream(""), err
 	}
 
-	if d.Routing.ActiveCheck {
+	if d.Routing.ActiveCheck.Enabled {
 		registryMonitor, err := imageMonitor.GetRegistryMonitor(ctx, d.Client)
 		if err != nil {
 			return kuikv1alpha1.ImageMonitorStatusUpstream(""), err
 		}
 
-		imageMonitor.Monitor(ctx, d.Client, registryMonitor.Spec.Method)
+		err = imageMonitor.Monitor(ctx, d.Client, registryMonitor.Spec.Method, d.Routing.ActiveCheck.Timeout)
 	}
 
 	return imageMonitor.Status.Upstream.Status, nil

Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,9 @@ type RegistryMonitorSpec struct {`
`24`	`24`	`// +kubebuilder:validation:Enum=HEAD;GET`
`25`	`25`	`// +default:value="HEAD"`
`26`	`26`	Method string `json:"method"`
	`27`	`+ // Timeout is the maximum duration of a monitoring task`
	`28`	`+ // +default:value="30s"`
	`29`	+ Timeout metav1.Duration `json:"timeout"`
`27`	`30`	`}`
`28`	`31`
`29`	`32`	`// RegistryMonitorStatus defines the observed state of RegistryMonitor.`
Original file line number	Diff line number	Diff line change
`@@ -155,6 +155,7 @@ func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {`
`155`	`155`	`MaxPerInterval: 1,`
`156`	`156`	`Parallel: 1,`
`157`	`157`	`Method: http.MethodHead,`
	`158`	`+ Timeout: metav1.Duration{Duration: 5 * time.Second},`
`158`	`159`	`}`
`159`	`160`
`160`	`161`	`// TODO: read this from config instead`
Original file line number	Diff line number	Diff line change
`@@ -174,7 +174,7 @@ func (r *RegistryMonitorReconciler) Reconcile(ctx context.Context, req ctrl.Requ`
`174`	`174`	`logImageMonitor.Info("monitoring image")`
`175`	`175`	`// TODO: add to SetupWithManager Watches(&source.Channel{Source: eventChannel}, &handler.EnqueueRequestForObject{})`
`176`	`176`	`// push an event in the channel when this function is done`
`177`		`- if err := imageMonitor.Monitor(logf.IntoContext(ctx, logImageMonitor), r.Client, registryMonitor.Spec.Method); err != nil {`
	`177`	`+ if err := imageMonitor.Monitor(logf.IntoContext(ctx, logImageMonitor), r.Client, registryMonitor.Spec.Method, registryMonitor.Spec.Timeout.Duration); err != nil {`
`178`	`178`	`logImageMonitor.Info("failed to monitor image", "error", err.Error())`
`179`	`179`	`}`
`180`	`180`	`logImageMonitor.Info("image monitored with success")`