Merge pull request #3053 from sbueringer/cert-watcher-plus-0.18

[release-0.18] 🌱 Add fsnotify watcher+polling

Author: k8s-ci-robot

examples/scratch-env/go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch/v5 v5.9.0 // indirect
+	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/zapr v1.3.0 // indirect
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect

examples/scratch-env/go.sum

@@ -12,6 +12,8 @@ github.com/evanphx/json-patch v4.12.0+incompatible h1:4onqiflcdA9EOZ4RxV643DvftH
 github.com/evanphx/json-patch v4.12.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/evanphx/json-patch/v5 v5.9.0 h1:kcBlZQbplgElYIlo/n1hJbls2z/1awpXxpRi0/FOJfg=
 github.com/evanphx/json-patch/v5 v5.9.0/go.mod h1:VNkHZ/282BpEyt/tObQO8s5CMPmYYq14uClGH4abBuQ=
+github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
+github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
 github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/zapr v1.3.0 h1:XGdV8XW8zdwFiwOA2Dryh1gj2KRQyOOoNmBy4EplIcQ=

go.mod

@@ -5,7 +5,7 @@ go 1.22.0
 require (
 	github.com/evanphx/json-patch v4.12.0+incompatible // Using v4 to match upstream
 	github.com/evanphx/json-patch/v5 v5.9.0
-	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/fsnotify/fsnotify v1.7.0
 	github.com/go-logr/logr v1.4.2
 	github.com/go-logr/zapr v1.3.0
 	github.com/google/go-cmp v0.6.0

pkg/certwatcher/certwatcher.go

@@ -20,10 +20,15 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"fmt"
 	"os"
 	"sync"
 	"time"
 
+	"github.com/fsnotify/fsnotify"
+	kerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/controller-runtime/pkg/certwatcher/metrics"
 	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
 )
@@ -40,6 +45,7 @@ type CertWatcher struct {
 	sync.RWMutex
 
 	currentCert *tls.Certificate
+	watcher     *fsnotify.Watcher
 	interval    time.Duration
 
 	certPath string
@@ -53,13 +59,25 @@ type CertWatcher struct {
 
 // New returns a new CertWatcher watching the given certificate and key.
 func New(certPath, keyPath string) (*CertWatcher, error) {
+	var err error
+
 	cw := &CertWatcher{
 		certPath: certPath,
 		keyPath:  keyPath,
 		interval: defaultWatchInterval,
 	}
 
-	return cw, cw.ReadCertificate()
+	// Initial read of certificate and key.
+	if err := cw.ReadCertificate(); err != nil {
+		return nil, err
+	}
+
+	cw.watcher, err = fsnotify.NewWatcher()
+	if err != nil {
+		return nil, err
+	}
+
+	return cw, nil
 }
 
 // WithWatchInterval sets the watch interval and returns the CertWatcher pointer
@@ -88,14 +106,35 @@ func (cw *CertWatcher) GetCertificate(_ *tls.ClientHelloInfo) (*tls.Certificate,
 
 // Start starts the watch on the certificate and key files.
 func (cw *CertWatcher) Start(ctx context.Context) error {
+	files := sets.New(cw.certPath, cw.keyPath)
+
+	{
+		var watchErr error
+		if err := wait.PollUntilContextTimeout(ctx, 1*time.Second, 10*time.Second, true, func(ctx context.Context) (done bool, err error) {
+			for _, f := range files.UnsortedList() {
+				if err := cw.watcher.Add(f); err != nil {
+					watchErr = err
+					return false, nil //nolint:nilerr // We want to keep trying.
+				}
+				// We've added the watch, remove it from the set.
+				files.Delete(f)
+			}
+			return true, nil
+		}); err != nil {
+			return fmt.Errorf("failed to add watches: %w", kerrors.NewAggregate([]error{err, watchErr}))
+		}
+	}
+
+	go cw.Watch()
+
 	ticker := time.NewTicker(cw.interval)
 	defer ticker.Stop()
 
-	log.Info("Starting certificate watcher")
+	log.Info("Starting certificate poll+watcher", "interval", cw.interval)
 	for {
 		select {
 		case <-ctx.Done():
-			return nil
+			return cw.watcher.Close()
 		case <-ticker.C:
 			if err := cw.ReadCertificate(); err != nil {
 				log.Error(err, "failed read certificate")
@@ -104,11 +143,26 @@ func (cw *CertWatcher) Start(ctx context.Context) error {
 	}
 }
 
-// Watch used to read events from the watcher's channel and reacts to changes,
-// it has currently no function and it's left here for backward compatibility until a future release.
-//
-// Deprecated: fsnotify has been removed and Start() is now polling instead.
+// Watch reads events from the watcher's channel and reacts to changes.
 func (cw *CertWatcher) Watch() {
+	for {
+		select {
+		case event, ok := <-cw.watcher.Events:
+			// Channel is closed.
+			if !ok {
+				return
+			}
+
+			cw.handleEvent(event)
+		case err, ok := <-cw.watcher.Errors:
+			// Channel is closed.
+			if !ok {
+				return
+			}
+
+			log.Error(err, "certificate watch error")
+		}
+	}
 }
 
 // updateCachedCertificate checks if the new certificate differs from the cache,
@@ -166,3 +220,23 @@ func (cw *CertWatcher) ReadCertificate() error {
 	}
 	return nil
 }
+
+func (cw *CertWatcher) handleEvent(event fsnotify.Event) {
+	// Only care about events which may modify the contents of the file.
+	switch {
+	case event.Op.Has(fsnotify.Write):
+	case event.Op.Has(fsnotify.Create):
+	case event.Op.Has(fsnotify.Chmod), event.Op.Has(fsnotify.Remove):
+		// If the file was removed or renamed, re-add the watch to the previous name
+		if err := cw.watcher.Add(event.Name); err != nil {
+			log.Error(err, "error re-watching file")
+		}
+	default:
+		return
+	}
+
+	log.V(1).Info("certificate event", "event", event)
+	if err := cw.ReadCertificate(); err != nil {
+		log.Error(err, "error re-reading certificate")
+	}
+}

pkg/certwatcher/certwatcher_test.go

@@ -77,12 +77,12 @@ var _ = Describe("CertWatcher", func() {
 			Expect(err).ToNot(HaveOccurred())
 		})
 
-		startWatcher := func() (done <-chan struct{}) {
+		startWatcher := func(interval time.Duration) (done <-chan struct{}) {
 			doneCh := make(chan struct{})
 			go func() {
 				defer GinkgoRecover()
 				defer close(doneCh)
-				Expect(watcher.WithWatchInterval(time.Second).Start(ctx)).To(Succeed())
+				Expect(watcher.WithWatchInterval(interval).Start(ctx)).To(Succeed())
 			}()
 			// wait till we read first cert
 			Eventually(func() error {
@@ -93,14 +93,16 @@ var _ = Describe("CertWatcher", func() {
 		}
 
 		It("should read the initial cert/key", func() {
-			doneCh := startWatcher()
+			// This test verifies the initial read succeeded. So interval doesn't matter.
+			doneCh := startWatcher(10 * time.Second)
 
 			ctxCancel()
 			Eventually(doneCh, "4s").Should(BeClosed())
 		})
 
 		It("should reload currentCert when changed", func() {
-			doneCh := startWatcher()
+			// This test verifies fsnotify detects the cert change. So interval doesn't matter.
+			doneCh := startWatcher(10 * time.Second)
 			called := atomic.Int64{}
 			watcher.RegisterCallback(func(crt tls.Certificate) {
 				called.Add(1)
@@ -124,7 +126,8 @@ var _ = Describe("CertWatcher", func() {
 		})
 
 		It("should reload currentCert when changed with rename", func() {
-			doneCh := startWatcher()
+			// This test verifies fsnotify detects the cert change. So interval doesn't matter.
+			doneCh := startWatcher(10 * time.Second)
 			called := atomic.Int64{}
 			watcher.RegisterCallback(func(crt tls.Certificate) {
 				called.Add(1)
@@ -154,7 +157,8 @@ var _ = Describe("CertWatcher", func() {
 		})
 
 		It("should reload currentCert after move out", func() {
-			doneCh := startWatcher()
+			// This test verifies poll works, so we'll use 1s as interval (fsnotify doesn't detect this change).
+			doneCh := startWatcher(1 * time.Second)
 			called := atomic.Int64{}
 			watcher.RegisterCallback(func(crt tls.Certificate) {
 				called.Add(1)
@@ -190,7 +194,8 @@ var _ = Describe("CertWatcher", func() {
 			})
 
 			It("should get updated on successful certificate read", func() {
-				doneCh := startWatcher()
+				// This test verifies fsnotify, so interval doesn't matter.
+				doneCh := startWatcher(10 * time.Second)
 
 				Eventually(func() error {
 					readCertificateTotalAfter := testutil.ToFloat64(metrics.ReadCertificateTotal)
@@ -205,7 +210,8 @@ var _ = Describe("CertWatcher", func() {
 			})
 
 			It("should get updated on read certificate errors", func() {
-				doneCh := startWatcher()
+				// This test works with fsnotify, so interval doesn't matter.
+				doneCh := startWatcher(10 * time.Second)
 
 				Eventually(func() error {
 					readCertificateTotalAfter := testutil.ToFloat64(metrics.ReadCertificateTotal)