Merge pull request #1356 from openshift-cherrypick-robot/cherry-pick-1353-to-release-4.4

[release-4.4] Bug 1783284: feat(jitter): add jitter to controllers to smooth out our spiky resource

Author: openshift-merge-robot

cmd/olm/main.go

@@ -12,6 +12,7 @@ import (
 
 	configclientset "github.com/openshift/client-go/config/clientset/versioned"
 	configv1client "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	log "github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
@@ -195,7 +196,7 @@ func main() {
 		ctx,
 		olm.WithLogger(logger),
 		olm.WithWatchedNamespaces(namespaces...),
-		olm.WithResyncPeriod(*wakeupInterval),
+		olm.WithResyncPeriod(queueinformer.ResyncWithJitter(*wakeupInterval, 0.2)),
 		olm.WithExternalClient(crClient),
 		olm.WithOperatorClient(opClient),
 		olm.WithRestConfig(config),

pkg/controller/operators/catalog/operator.go

@@ -98,7 +98,8 @@ type Operator struct {
 type CatalogSourceSyncFunc func(logger *logrus.Entry, in *v1alpha1.CatalogSource) (out *v1alpha1.CatalogSource, continueSync bool, syncError error)
 
 // NewOperator creates a new Catalog Operator.
-func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clock, logger *logrus.Logger, resyncPeriod time.Duration, configmapRegistryImage, utilImage string, operatorNamespace string) (*Operator, error) {
+func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clock, logger *logrus.Logger, resync time.Duration, configmapRegistryImage, utilImage string, operatorNamespace string) (*Operator, error) {
+	resyncPeriod := queueinformer.ResyncWithJitter(resync, 0.2)
 	config, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
 	if err != nil {
 		return nil, err
@@ -149,7 +150,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now)
 
 	// Wire OLM CR sharedIndexInformers
-	crInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(op.client, resyncPeriod)
+	crInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(op.client, resyncPeriod())
 
 	// Wire CSVs
 	csvInformer := crInformerFactory.Operators().V1alpha1().ClusterServiceVersions()
@@ -249,7 +250,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	}
 
 	// Wire k8s sharedIndexInformers
-	k8sInformerFactory := informers.NewSharedInformerFactoryWithOptions(op.opClient.KubernetesInterface(), resyncPeriod)
+	k8sInformerFactory := informers.NewSharedInformerFactoryWithOptions(op.opClient.KubernetesInterface(), resyncPeriod())
 	sharedIndexInformers := []cache.SharedIndexInformer{}
 
 	// Wire Roles
@@ -321,7 +322,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	}
 
 	// Register CustomResourceDefinition QueueInformer
-	crdInformer := extinf.NewSharedInformerFactory(op.opClient.ApiextensionsV1beta1Interface(), resyncPeriod).Apiextensions().V1beta1().CustomResourceDefinitions()
+	crdInformer := extinf.NewSharedInformerFactory(op.opClient.ApiextensionsV1beta1Interface(), resyncPeriod()).Apiextensions().V1beta1().CustomResourceDefinitions()
 	op.lister.APIExtensionsV1beta1().RegisterCustomResourceDefinitionLister(crdInformer.Lister())
 	crdQueueInformer, err := queueinformer.NewQueueInformer(
 		ctx,
@@ -337,7 +338,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	}
 
 	// Namespace sync for resolving subscriptions
-	namespaceInformer := informers.NewSharedInformerFactory(op.opClient.KubernetesInterface(), resyncPeriod).Core().V1().Namespaces()
+	namespaceInformer := informers.NewSharedInformerFactory(op.opClient.KubernetesInterface(), resyncPeriod()).Core().V1().Namespaces()
 	op.lister.CoreV1().RegisterNamespaceLister(namespaceInformer.Lister())
 	op.nsResolveQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "resolver")
 	namespaceQueueInformer, err := queueinformer.NewQueueInformer(

pkg/controller/operators/olm/config.go

@@ -1,6 +1,7 @@
 package olm
 
 import (
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer"
 	"time"
 
 	"github.com/pkg/errors"
@@ -21,7 +22,7 @@ import (
 type OperatorOption func(*operatorConfig)
 
 type operatorConfig struct {
-	resyncPeriod      time.Duration
+	resyncPeriod      func() time.Duration
 	operatorNamespace string
 	watchedNamespaces []string
 	clock             utilclock.Clock
@@ -49,8 +50,8 @@ func newInvalidConfigError(name, msg string) error {
 func (o *operatorConfig) validate() (err error) {
 	// TODO: Add better config validation
 	switch {
-	case o.resyncPeriod < 0:
-		err = newInvalidConfigError("resync period", "must be >= 0")
+	case o.resyncPeriod == nil:
+		err = newInvalidConfigError("resync period", "must not be nil")
 	case o.operatorNamespace == metav1.NamespaceAll:
 		err = newInvalidConfigError("operator namespace", "must be a single namespace")
 	case len(o.watchedNamespaces) == 0:
@@ -80,7 +81,7 @@ func (o *operatorConfig) validate() (err error) {
 
 func defaultOperatorConfig() *operatorConfig {
 	return &operatorConfig{
-		resyncPeriod:      30 * time.Second,
+		resyncPeriod:      queueinformer.ResyncWithJitter(30 * time.Second, 0.2),
 		operatorNamespace: "default",
 		watchedNamespaces: []string{metav1.NamespaceAll},
 		clock:             utilclock.RealClock{},
@@ -91,9 +92,9 @@ func defaultOperatorConfig() *operatorConfig {
 	}
 }
 
-func WithResyncPeriod(period time.Duration) OperatorOption {
+func WithResyncPeriod(resyncPeriod func() time.Duration) OperatorOption {
 	return func(config *operatorConfig) {
-		config.resyncPeriod = period
+		config.resyncPeriod = resyncPeriod
 	}
 }
 

pkg/controller/operators/olm/operator.go

@@ -140,7 +140,7 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 	k8sSyncer := queueinformer.LegacySyncHandler(op.syncObject).ToSyncerWithDelete(op.handleDeletion)
 	for _, namespace := range config.watchedNamespaces {
 		// Wire CSVs
-		extInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(op.client, config.resyncPeriod, externalversions.WithNamespace(namespace))
+		extInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(op.client, config.resyncPeriod(), externalversions.WithNamespace(namespace))
 		csvInformer := extInformerFactory.Operators().V1alpha1().ClusterServiceVersions()
 		op.lister.OperatorsV1alpha1().RegisterClusterServiceVersionLister(namespace, csvInformer.Lister())
 		csvQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), fmt.Sprintf("%s/csv", namespace))
@@ -232,7 +232,7 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 		op.RegisterQueueInformer(subQueueInformer)
 
 		// Wire Deployments
-		k8sInformerFactory := informers.NewSharedInformerFactoryWithOptions(op.opClient.KubernetesInterface(), config.resyncPeriod, informers.WithNamespace(namespace))
+		k8sInformerFactory := informers.NewSharedInformerFactoryWithOptions(op.opClient.KubernetesInterface(), config.resyncPeriod(), informers.WithNamespace(namespace))
 		depInformer := k8sInformerFactory.Apps().V1().Deployments()
 		op.lister.AppsV1().RegisterDeploymentLister(namespace, depInformer.Lister())
 		depQueueInformer, err := queueinformer.NewQueueInformer(
@@ -360,7 +360,7 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 		return nil, err
 	}
 
-	k8sInformerFactory := informers.NewSharedInformerFactory(op.opClient.KubernetesInterface(), config.resyncPeriod)
+	k8sInformerFactory := informers.NewSharedInformerFactory(op.opClient.KubernetesInterface(), config.resyncPeriod())
 	clusterRoleInformer := k8sInformerFactory.Rbac().V1().ClusterRoles()
 	op.lister.RbacV1().RegisterClusterRoleLister(clusterRoleInformer.Lister())
 	clusterRoleQueueInformer, err := queueinformer.NewQueueInformer(
@@ -414,7 +414,7 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 	}
 
 	// Register APIService QueueInformer
-	apiServiceInformer := kagg.NewSharedInformerFactory(op.opClient.ApiregistrationV1Interface(), config.resyncPeriod).Apiregistration().V1().APIServices()
+	apiServiceInformer := kagg.NewSharedInformerFactory(op.opClient.ApiregistrationV1Interface(), config.resyncPeriod()).Apiregistration().V1().APIServices()
 	op.lister.APIRegistrationV1().RegisterAPIServiceLister(apiServiceInformer.Lister())
 	apiServiceQueueInformer, err := queueinformer.NewQueueInformer(
 		ctx,
@@ -431,7 +431,7 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 	}
 
 	// Register CustomResourceDefinition QueueInformer
-	crdInformer := extinf.NewSharedInformerFactory(op.opClient.ApiextensionsV1beta1Interface(), config.resyncPeriod).Apiextensions().V1beta1().CustomResourceDefinitions()
+	crdInformer := extinf.NewSharedInformerFactory(op.opClient.ApiextensionsV1beta1Interface(), config.resyncPeriod()).Apiextensions().V1beta1().CustomResourceDefinitions()
 	op.lister.APIExtensionsV1beta1().RegisterCustomResourceDefinitionLister(crdInformer.Lister())
 	crdQueueInformer, err := queueinformer.NewQueueInformer(
 		ctx,

pkg/controller/operators/olm/operator_test.go

@@ -8,6 +8,7 @@ import (
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"fmt"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer"
 	"math"
 	"math/big"
 	"reflect"
@@ -160,7 +161,7 @@ type fakeOperatorOption func(*fakeOperatorConfig)
 
 func withResyncPeriod(period time.Duration) fakeOperatorOption {
 	return func(config *fakeOperatorConfig) {
-		config.resyncPeriod = period
+		config.resyncPeriod = queueinformer.ResyncWithJitter(period, 0.1)
 	}
 }
 
@@ -257,7 +258,7 @@ func NewFakeOperator(ctx context.Context, options ...fakeOperatorOption) (*Opera
 	// Apply options to default config
 	config := &fakeOperatorConfig{
 		operatorConfig: &operatorConfig{
-			resyncPeriod:      5 * time.Minute,
+			resyncPeriod:      queueinformer.ResyncWithJitter(5 * time.Minute, 0.1),
 			operatorNamespace: "default",
 			watchedNamespaces: []string{metav1.NamespaceAll},
 			clock:             &utilclock.RealClock{},

pkg/lib/queueinformer/jitter.go

@@ -0,0 +1,32 @@
+package queueinformer
+
+import (
+	"math/rand"
+	"time"
+)
+
+const DefaultResyncPeriod = 15 * time.Minute
+
+// ResyncWithJitter takes a resync interval and adds jitter within a percent difference.
+// factor is a value between 0 and 1 indicating the amount of jitter
+// a factor of 0.2 and a period of 10m will have a range of 8 to 12 minutes (20%)
+func ResyncWithJitter(resyncPeriod time.Duration, factor float64) func() time.Duration {
+	return func() time.Duration {
+		if factor < 0.0 || factor > 1.0 {
+			return resyncPeriod
+		}
+		if resyncPeriod < 0.0 {
+			return DefaultResyncPeriod
+		}
+
+		// if we would wrap around, return resyncPeriod
+		if time.Duration((1+factor)*resyncPeriod.Minutes())*time.Minute < 0.0 {
+			return resyncPeriod
+		}
+
+		min := resyncPeriod.Minutes() * (1 - factor)
+		max := resyncPeriod.Minutes() * (1 + factor)
+
+		return time.Duration(min)*time.Minute + time.Duration(rand.Float64()*(max-min))*time.Minute
+	}
+}

pkg/lib/queueinformer/jitter_test.go

@@ -0,0 +1,92 @@
+package queueinformer
+
+import (
+	"math/rand"
+	"reflect"
+	"testing"
+	"testing/quick"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestResyncWithJitter(t *testing.T) {
+	type args struct {
+		resyncPeriod time.Duration
+		factor       float64
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantMin time.Duration
+		wantMax time.Duration
+	}{
+		{
+			name: "TypicalInput/Minutes",
+			args: args{
+				resyncPeriod: 15 * time.Minute,
+				factor:       0.2,
+			},
+			wantMin: 12 * time.Minute,
+			wantMax: 18 * time.Minute,
+		},
+		{
+			name: "TypicalInput/Hours",
+			args: args{
+				resyncPeriod: 10 * time.Hour,
+				factor:       0.1,
+			},
+			wantMin: 9 * time.Hour,
+			wantMax: 11 * time.Hour,
+		},
+		{
+			name: "BadInput/BadFactor",
+			args: args{
+				resyncPeriod: 10 * time.Hour,
+				factor:       -0.1,
+			},
+			wantMin: 10 * time.Hour,
+			wantMax: 10 * time.Hour,
+		},
+		{
+			name: "BadInput/BadResync",
+			args: args{
+				resyncPeriod: -10 * time.Hour,
+				factor:       0.1,
+			},
+			wantMin: DefaultResyncPeriod,
+			wantMax: DefaultResyncPeriod,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := ResyncWithJitter(tt.args.resyncPeriod, tt.args.factor)
+			require.True(t, got() >= tt.wantMin)
+			require.True(t, got() <= tt.wantMax)
+			require.True(t, got() != got() || tt.wantMax == tt.wantMin)
+		})
+	}
+}
+
+type float01 float64
+
+func (float01) Generate(rand *rand.Rand, size int) reflect.Value {
+	return reflect.ValueOf(float01(rand.Float64()))
+}
+
+type dur time.Duration
+
+func (dur) Generate(rand *rand.Rand, size int) reflect.Value {
+	return reflect.ValueOf(dur(rand.Uint64() / 2))
+}
+
+func TestGeneratesWithinRange(t *testing.T) {
+	f := func(resync dur, factor float01) bool {
+		resyncPeriod := time.Duration(resync)
+		min := float64(resyncPeriod.Nanoseconds()) * (1 - float64(factor))
+		max := float64(resyncPeriod.Nanoseconds()) * (1 + float64(factor))
+		d := ResyncWithJitter(resyncPeriod, float64(factor))()
+		return min < float64(d.Nanoseconds()) && float64(d.Nanoseconds()) < max
+	}
+	require.NoError(t, quick.Check(f, nil))
+}

scripts/build_local.sh

@@ -6,7 +6,7 @@
 set -e
 
 if [ -z "$NO_MINIKUBE" ]; then
-  pgrep -f "[m]inikube" >/dev/null || minikube start --kubernetes-version="v1.16.2" --extra-config=apiserver.v=4 || { echo 'Cannot start minikube.'; exit 1; }
+  pgrep -f "[m]inikube" >/dev/null || minikube start --kubernetes-version="v1.16.4" --extra-config=apiserver.v=4 || { echo 'Cannot start minikube.'; exit 1; }
   eval "$(minikube docker-env)" || { echo 'Cannot switch to minikube docker'; exit 1; }
   kubectl config use-context minikube
 fi