add coalescing reconciler to enable debouncing of reconciles

Author: devigned

pkg/coalescing/mocks/coalescing_mock.go

@@ -0,0 +1,22 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Run go generate to regenerate this mock.
+//go:generate ../../../hack/tools/bin/mockgen -destination coalescing_mock.go -package mock_coalescing -source ../reconciler.go ReconcileCacher
+//go:generate ../../../hack/tools/bin/mockgen -destination reconciler_mock.go -package mock_coalescing sigs.k8s.io/controller-runtime/pkg/reconcile Reconciler
+//go:generate /usr/bin/env bash -c "cat ../../../hack/boilerplate/boilerplate.generatego.txt coalescing_mock.go > _coalescing_mock.go && mv _coalescing_mock.go coalescing_mock.go"
+//go:generate /usr/bin/env bash -c "cat ../../../hack/boilerplate/boilerplate.generatego.txt reconciler_mock.go > _reconciler_mock.go && mv _reconciler_mock.go reconciler_mock.go"
+package mock_coalescing //nolint

pkg/coalescing/mocks/doc.go

@@ -0,0 +1,121 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package coalescing
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/pkg/errors"
+	"go.opentelemetry.io/otel/api/trace"
+	"go.opentelemetry.io/otel/label"
+	"sigs.k8s.io/cluster-api-provider-azure/util/cache/ttllru"
+	"sigs.k8s.io/cluster-api-provider-azure/util/tele"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+type (
+	// ReconcileCache uses and underlying time to live last recently used cache to track high frequency requests.
+	// A reconciler should call ShouldProcess to determine if the key has expired. If the key has expired, a zero value
+	// time.Time and true is returned. If the key has not expired, the expiration and false is returned. Upon successful
+	// reconciliation a reconciler should call Reconciled to update the cache expiry.
+	ReconcileCache struct {
+		lastSuccessfulReconciliationCache ttllru.PeekingCacher
+	}
+
+	// ReconcileCacher describes an interface for determining if a request should be reconciled through a call to
+	// ShouldProcess and if ok, reset the cool down through a call to Reconciled
+	ReconcileCacher interface {
+		ShouldProcess(key string) (expiration time.Time, ok bool)
+		Reconciled(key string)
+	}
+
+	// reconciler is the caching reconciler middleware that uses the cache or
+	reconciler struct {
+		upstream reconcile.Reconciler
+		cache    ReconcileCacher
+		log      logr.Logger
+	}
+)
+
+// NewRequestCache creates a new instance of a ReconcileCache given a specified window of expiration
+func NewRequestCache(window time.Duration) (*ReconcileCache, error) {
+	cache, err := ttllru.New(1024, window)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to build ttllru cache")
+	}
+
+	return &ReconcileCache{
+		lastSuccessfulReconciliationCache: cache,
+	}, nil
+}
+
+// ShouldProcess determines if the key has expired. If the key has expired, a zero value
+// time.Time and true is returned. If the key has not expired, the expiration and false is returned.
+func (cache *ReconcileCache) ShouldProcess(key string) (time.Time, bool) {
+	_, expiration, ok := cache.lastSuccessfulReconciliationCache.Peek(key)
+	return expiration, !ok
+}
+
+// Reconciled updates the cache expiry for a given key
+func (cache *ReconcileCache) Reconciled(key string) {
+	cache.lastSuccessfulReconciliationCache.Add(key, nil)
+}
+
+// NewReconciler returns a reconcile wrapper that will delay new reconcile.Requests
+// after the cache expiry of the request string key.
+// A successful reconciliation is defined as as one where no error is returned
+func NewReconciler(upstream reconcile.Reconciler, cache ReconcileCacher, log logr.Logger) reconcile.Reconciler {
+	return &reconciler{
+		upstream: upstream,
+		cache:    cache,
+		log:      log.WithName("CoalescingReconciler"),
+	}
+}
+
+// Reconcile sends a request to the upstream reconciler if the request is outside of the debounce window
+func (rc *reconciler) Reconcile(ctx context.Context, r reconcile.Request) (reconcile.Result, error) {
+	ctx, span := tele.Tracer().Start(ctx, "controllers.reconciler.Reconcile",
+		trace.WithAttributes(
+			label.String("namespace", r.Namespace),
+			label.String("name", r.Name),
+		))
+	defer span.End()
+
+	log := rc.log.WithValues("request", r.String())
+
+	if expiration, ok := rc.cache.ShouldProcess(r.String()); !ok {
+		log.V(4).Info("not processing", "expiration", expiration, "timeUntil", time.Until(expiration))
+		var requeueAfter = time.Until(expiration)
+		if requeueAfter < 1*time.Second {
+			requeueAfter = 1 * time.Second
+		}
+		return reconcile.Result{RequeueAfter: requeueAfter}, nil
+	}
+
+	log.V(4).Info("processing")
+	result, err := rc.upstream.Reconcile(ctx, r)
+	if err != nil {
+		log.V(4).Info("not successful")
+		return result, err
+	}
+
+	log.V(4).Info("successful")
+	rc.cache.Reconciled(r.String())
+	return result, nil
+}

pkg/coalescing/mocks/reconciler_mock.go

@@ -0,0 +1,104 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package coalescing
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	. "github.com/onsi/gomega"
+	gtypes "github.com/onsi/gomega/types"
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/types"
+	mock_coalescing "sigs.k8s.io/cluster-api-provider-azure/pkg/coalescing/mocks"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+func TestCoalescingReconciler_Reconcile(t *testing.T) {
+	var (
+		defaultRequest = reconcile.Request{
+			NamespacedName: types.NamespacedName{
+				Name:      "aName",
+				Namespace: "aNamespace",
+			},
+		}
+
+		defaultRequestKey = "aNamespace/aName"
+	)
+
+	cases := []struct {
+		Name       string
+		Reconciler func(g *WithT, cacherMock *mock_coalescing.MockReconcileCacher, mockReconciler *mock_coalescing.MockReconciler) reconcile.Reconciler
+		Request    reconcile.Request
+		MatchThis  gtypes.GomegaMatcher
+		Error      string
+	}{
+		{
+			Name: "should call upstream reconciler if key does not exist in cache",
+			Reconciler: func(g *WithT, cacherMock *mock_coalescing.MockReconcileCacher, mockReconciler *mock_coalescing.MockReconciler) reconcile.Reconciler {
+				cacherMock.EXPECT().ShouldProcess(defaultRequestKey).Return(time.Now(), true)
+				cacherMock.EXPECT().Reconciled(defaultRequestKey)
+				mockReconciler.EXPECT().Reconcile(gomock.Any(), defaultRequest)
+				return NewReconciler(mockReconciler, cacherMock, log.NullLogger{})
+			},
+			Request:   defaultRequest,
+			MatchThis: Equal(0 * time.Second),
+		},
+		{
+			Name: "should not call upstream reconciler if key does exists in cache and is not expired",
+			Reconciler: func(g *WithT, cacherMock *mock_coalescing.MockReconcileCacher, mockReconciler *mock_coalescing.MockReconciler) reconcile.Reconciler {
+				cacherMock.EXPECT().ShouldProcess(defaultRequestKey).Return(time.Now().Add(30*time.Second), false)
+				return NewReconciler(mockReconciler, cacherMock, log.NullLogger{})
+			},
+			Request:   defaultRequest,
+			MatchThis: And(BeNumerically("<=", 30*time.Second), BeNumerically(">", 29*time.Second)),
+		},
+		{
+			Name: "should call upstream reconciler if key does not exist in cache and return error",
+			Reconciler: func(g *WithT, cacherMock *mock_coalescing.MockReconcileCacher, mockReconciler *mock_coalescing.MockReconciler) reconcile.Reconciler {
+				cacherMock.EXPECT().ShouldProcess(defaultRequestKey).Return(time.Now(), true)
+				mockReconciler.EXPECT().Reconcile(gomock.Any(), defaultRequest).Return(reconcile.Result{}, errors.New("boom"))
+				return NewReconciler(mockReconciler, cacherMock, log.NullLogger{})
+			},
+			Request:   defaultRequest,
+			MatchThis: Equal(0 * time.Second),
+			Error:     "boom",
+		},
+	}
+
+	for _, c := range cases {
+		c := c
+		t.Run(c.Name, func(t *testing.T) {
+			g := NewWithT(t)
+			mockCtrl := gomock.NewController(t)
+			defer mockCtrl.Finish()
+			cacherMock := mock_coalescing.NewMockReconcileCacher(mockCtrl)
+			reconcilerMock := mock_coalescing.NewMockReconciler(mockCtrl)
+			subject := c.Reconciler(g, cacherMock, reconcilerMock)
+			result, err := subject.Reconcile(context.Background(), c.Request)
+			if c.Error != "" || err != nil {
+				g.Expect(err).To(And(HaveOccurred(), MatchError(c.Error)))
+				return
+			}
+
+			g.Expect(result.RequeueAfter).To(c.MatchThis)
+		})
+	}
+}