refactor: fightThreshold global variable (#1815)

* Adds DefaultFightThreshold to register.go
* Adds fightThreshold member to the Detector struct

Author: Camila-B

cmd/reconciler/main.go

@@ -71,7 +71,7 @@ var (
 	hydratedLinkDir = flag.String("hydrated-link", "rev",
 		"The name of (a symlink to) the source directory under --hydrated-root, which contains the hydrated configs")
 	fightDetectionThreshold = flag.Float64(
-		"fight-detection-threshold", 5.0,
+		"fight-detection-threshold", configsync.DefaultFightThreshold,
 		"The rate of updates per minute to an API Resource at which the Syncer logs warnings about too many updates to the resource.")
 	fullSyncPeriod = flag.Duration("full-sync-period", configsync.DefaultReconcilerFullSyncPeriod,
 		"Period of time between forced re-syncs from source (even without a new commit).")

e2e/testcases/remediator_test.go

@@ -56,7 +56,7 @@ func TestSurfaceFightError(t *testing.T) {
 	nt.Must(nomostest.ValidateMetrics(nt,
 		nomostest.ReconcilerErrorMetrics(nt, rootSyncLabels, commitHash, metrics.ErrorSummary{})))
 
-	// Make the # of updates exceed the fightThreshold defined in pkg/syncer/reconcile/fight_detector.go
+	// Make the # of updates exceed the DefaultFightThreshold defined in pkg/api/configsync/register.go
 	ctx, cancel := context.WithCancel(t.Context())
 	nt.T.Cleanup(func() {
 		cancel() // cancel in case an assertion failed early

pkg/api/configsync/register.go

@@ -90,6 +90,17 @@ const (
 
 	// DefaultHelmReleaseNamespace is the default namespace for a Helm Release which does not have a namespace specified
 	DefaultHelmReleaseNamespace = "default"
+
+	// DefaultFightThreshold is the default threshold of updates per minute to a
+	// resource before Config Sync reports a fight. A fight occurs when the
+	// Remediator and another process are both trying to manage the same
+	// resource.
+	// When a fight is detected, an error is logged, surfaced in the
+	// RSync status, and emitted as a metric.
+	// This value was chosen because a resource being updated more than 5 times
+	// per minute is a clear sign of a problem, while less frequent updates are
+	// unlikely to be fights.
+	DefaultFightThreshold = 5.0
 )
 
 // SourceFormat specifies how the Importer should parse the repository.

pkg/reconciler/reconciler.go

@@ -149,7 +149,6 @@ type RootOptions struct {
 func Run(opts Options) {
 	// Start listening to signals
 	signalCtx := signals.SetupSignalHandler()
-	fight.SetFightThreshold(opts.FightDetectionThreshold)
 
 	// Get a config to talk to the apiserver.
 	apiServerTimeout, err := time.ParseDuration(opts.APIServerTimeout)
@@ -190,7 +189,7 @@ func Run(opts Options) {
 	// Configure the Applier.
 	applySetID := applyset.IDFromSync(opts.SyncName, opts.ReconcilerScope)
 	genericClient := syncerclient.New(cl, metrics.APICallDuration)
-	baseApplier, err := reconcile.NewApplierForMultiRepo(cfg, genericClient, applySetID)
+	baseApplier, err := reconcile.NewApplierForMultiRepo(cfg, genericClient, applySetID, opts.FightDetectionThreshold)
 	if err != nil {
 		klog.Fatalf("Instantiating Applier: %v", err)
 	}

pkg/syncer/reconcile/apply.go

@@ -75,11 +75,11 @@ type clientApplier struct {
 var _ Applier = &clientApplier{}
 
 // NewApplierForMultiRepo returns a new clientApplier for callers with multi repo feature enabled.
-func NewApplierForMultiRepo(cfg *rest.Config, client *syncerclient.Client, applySetID string) (Applier, error) {
-	return newApplier(cfg, client, applySetID)
+func NewApplierForMultiRepo(cfg *rest.Config, client *syncerclient.Client, applySetID string, fightThreshold float64) (Applier, error) {
+	return newApplier(cfg, client, applySetID, fightThreshold)
 }
 
-func newApplier(cfg *rest.Config, client *syncerclient.Client, applySetID string) (Applier, error) {
+func newApplier(cfg *rest.Config, client *syncerclient.Client, applySetID string, fightThreshold float64) (Applier, error) {
 	c, err := dynamic.NewForConfig(cfg)
 	if err != nil {
 		return nil, err
@@ -100,7 +100,7 @@ func newApplier(cfg *rest.Config, client *syncerclient.Client, applySetID string
 		discoveryClient:  dc,
 		openAPIResources: oa,
 		client:           client,
-		fights:           fight.NewDetector(),
+		fights:           fight.NewDetector(fightThreshold),
 		applySetID:       applySetID,
 	}, nil
 }

pkg/syncer/reconcile/fight/detector.go

@@ -24,20 +24,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-// fightThreshold is the threshold of updates per minute at which we log to Info
-// that the Syncer is fighting over a resource on the API Server with some
-// other process.
-//
-// This value was chosen arbitrarily as updates occurring more frequently are
-// obviously problems, and we don't really care about less frequent updates.
-var fightThreshold = 5.0
-
-// SetFightThreshold updates the maximum allowed rate of updates to a resource
-// per minute before we begin logging errors to the user.
-func SetFightThreshold(updatesPerMinute float64) {
-	fightThreshold = updatesPerMinute
-}
-
 // Detector uses a linear differential equation to estimate the frequency
 // of updates to resources, then logs to klog.Warning when it detects resources
 // needing updates too frequently.
@@ -59,13 +45,18 @@ type Detector struct {
 	fights map[core.ID]*fight
 
 	fLogger *logger
+
+	// fightThreshold is the maximum allowed rate of updates to a resource
+	// per minute before errors are logged to the user, surfaced on the RSync status and emitted as a metric
+	fightThreshold float64
 }
 
 // NewDetector instantiates a fight detector.
-func NewDetector() Detector {
+func NewDetector(threshold float64) Detector {
 	return Detector{
-		fights:  make(map[core.ID]*fight),
-		fLogger: newLogger(),
+		fights:         make(map[core.ID]*fight),
+		fLogger:        newLogger(),
+		fightThreshold: threshold,
 	}
 }
 
@@ -79,14 +70,14 @@ func (d *Detector) DetectFight(now time.Time, obj client.Object) (bool, status.R
 	if d.fights[id] == nil {
 		d.fights[id] = &fight{}
 	}
-	if frequency := d.fights[id].refreshUpdateFrequency(now); frequency >= fightThreshold {
+	if frequency := d.fights[id].refreshUpdateFrequency(now); frequency >= d.fightThreshold {
 		fightErr := status.FightError(frequency, obj)
 		return d.fLogger.logFight(now, fightErr), fightErr
 	}
 	return false, nil
 }
 
-// fight estimates how often a specific API resource is updated by the Syncer.
+// fight estimates how often a specific API resource is updated by the Remediator.
 type fight struct {
 	// heat is an estimate of the number of times a resource is updated per minute.
 	// It decays exponentially with time when there are no updates to a resource.

pkg/syncer/reconcile/fight/detector_test.go

@@ -110,6 +110,7 @@ func TestFight(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			fightThreshold := 5.0
 			now := time.Now()
 			f := fight{
 				heat: tc.startHeat,
@@ -185,7 +186,7 @@ func TestFightDetector(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			fd := NewDetector()
+			fd := NewDetector(5.0)
 
 			now := time.Now()
 			for id, updates := range tc.updates {
@@ -195,7 +196,7 @@ func TestFightDetector(t *testing.T) {
 				logged := false
 				for i, update := range updates {
 					logErr, fightErr := fd.DetectFight(now.Add(update), u)
-					if i+1 >= int(fightThreshold) {
+					if i+1 >= int(fd.fightThreshold) {
 						require.Error(t, fightErr)
 						aboveThreshold = true
 						if logged {