[Feature] Reconciliation Loop Interval option (#1395)

Author: ajanikow

CHANGELOG.md

@@ -11,6 +11,7 @@
 - (Bugfix) Fix CRD yaml (chart)
 - (Bugfix) (EE) Fix MemberMaintenance Context and ClusterMaintenance discovery
 - (Feature) Add proper Prometheus endpoint compression + 204 response code
+- (Feature) Reconciliation Loop Interval option
 
 ## [1.2.32](https://github.com/arangodb/kube-arangodb/tree/1.2.32) (2023-08-07)
 - (Feature) Backup lifetime - remove Backup once its lifetime has been reached

cmd/cmd.go

@@ -136,6 +136,8 @@ var (
 
 		alpineImage, metricsExporterImage, arangoImage string
 
+		reconciliationDelay time.Duration
+
 		singleMode bool
 		scope      string
 	}
@@ -224,6 +226,7 @@ func init() {
 	f.DurationVar(&shutdownOptions.delay, "shutdown.delay", defaultShutdownDelay, "The delay before running shutdown handlers")
 	f.DurationVar(&shutdownOptions.timeout, "shutdown.timeout", defaultShutdownTimeout, "Timeout for shutdown handlers")
 	f.BoolVar(&operatorOptions.scalingIntegrationEnabled, "internal.scaling-integration", false, "Enable Scaling Integration")
+	f.DurationVar(&operatorOptions.reconciliationDelay, "reconciliation.delay", 0, "Delay between reconciliation loops (<= 0 -> Disabled)")
 	f.Int64Var(&operatorKubernetesOptions.maxBatchSize, "kubernetes.max-batch-size", globals.DefaultKubernetesRequestBatchSize, "Size of batch during objects read")
 	f.Float32Var(&operatorKubernetesOptions.qps, "kubernetes.qps", kclient.DefaultQPS, "Number of queries per second for k8s API")
 	f.IntVar(&operatorKubernetesOptions.burst, "kubernetes.burst", kclient.DefaultBurst, "Burst for the k8s API")
@@ -530,6 +533,7 @@ func newOperatorConfigAndDeps(id, namespace, name string) (operator.Config, oper
 		ArangoImage:                 operatorOptions.arangoImage,
 		SingleMode:                  operatorOptions.singleMode,
 		Scope:                       scope,
+		ReconciliationDelay:         operatorOptions.reconciliationDelay,
 		ShutdownDelay:               shutdownOptions.delay,
 		ShutdownTimeout:             shutdownOptions.timeout,
 	}

pkg/deployment/deployment.go

@@ -60,6 +60,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/kerrors"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 	"github.com/arangodb/kube-arangodb/pkg/util/trigger"
 )
 
@@ -70,6 +71,7 @@ type Config struct {
 	ScalingIntegrationEnabled bool
 	OperatorImage             string
 	ArangoImage               string
+	ReconciliationDelay       time.Duration
 	Scope                     scope.Scope
 }
 
@@ -107,6 +109,8 @@ type Deployment struct {
 	uid       types.UID
 	namespace string
 
+	delayer timer.Delayer
+
 	currentObject       *api.ArangoDeployment
 	currentObjectStatus *api.DeploymentStatus
 	currentObjectLock   sync.RWMutex
@@ -256,6 +260,7 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoDeployment) (*De
 		stopCh:              make(chan struct{}),
 		agencyCache:         agency.NewCache(apiObject.GetNamespace(), apiObject.GetName(), apiObject.GetAcceptedSpec().Mode),
 		acs:                 acs.NewACS(apiObject.GetUID(), i),
+		delayer:             timer.NewDelayer(),
 	}
 
 	d.log = logger.WrapObj(d)

pkg/deployment/deployment_inspector.go

@@ -24,6 +24,8 @@ import (
 	"context"
 	"time"
 
+	"github.com/rs/zerolog/log"
+
 	"github.com/arangodb/kube-arangodb/pkg/apis/deployment"
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
 	"github.com/arangodb/kube-arangodb/pkg/deployment/features"
@@ -51,6 +53,11 @@ var (
 func (d *Deployment) inspectDeployment(lastInterval util.Interval) util.Interval {
 	start := time.Now()
 
+	if delay := d.delayer.Wait(); delay > 0 {
+		log.Info().Dur("delay", delay).Msgf("Reconciliation loop execution was delayed")
+	}
+	defer d.delayer.Delay(d.config.ReconciliationDelay)
+
 	ctxReconciliation, cancelReconciliation := globals.GetGlobalTimeouts().Reconciliation().WithTimeout(context.Background())
 	defer cancelReconciliation()
 	defer func() {

pkg/operator/operator.go

@@ -106,6 +106,7 @@ type Config struct {
 	ScalingIntegrationEnabled   bool
 	SingleMode                  bool
 	Scope                       scope.Scope
+	ReconciliationDelay         time.Duration
 	ShutdownDelay               time.Duration
 	ShutdownTimeout             time.Duration
 }

pkg/operator/operator_deployment.go

@@ -201,6 +201,7 @@ func (o *Operator) makeDeploymentConfigAndDeps() (deployment.Config, deployment.
 		ArangoImage:               o.ArangoImage,
 		AllowChaos:                o.Config.AllowChaos,
 		ScalingIntegrationEnabled: o.Config.ScalingIntegrationEnabled,
+		ReconciliationDelay:       o.Config.ReconciliationDelay,
 		Scope:                     o.Scope,
 	}
 	deps := deployment.Dependencies{

pkg/util/k8sutil/pods.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2023 ArangoDB GmbH, Cologne, Germany
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

pkg/util/timer/delayer.go

@@ -0,0 +1,77 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package timer
+
+import (
+	"sync"
+	"time"
+)
+
+func NewDelayer() Delayer {
+	return &delayer{}
+}
+
+type Delayer interface {
+	Delay(delay time.Duration)
+
+	Wait() time.Duration
+
+	Copy() Delayer
+}
+
+type delayer struct {
+	lock sync.Mutex
+
+	last, next time.Time
+}
+
+func (d *delayer) Copy() Delayer {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+
+	return &delayer{
+		last: d.last,
+		next: d.next,
+	}
+}
+
+func (d *delayer) Wait() time.Duration {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+
+	since := time.Until(d.next)
+
+	if since <= time.Millisecond {
+		return 0
+	}
+
+	time.Sleep(since)
+
+	return since
+}
+
+func (d *delayer) Delay(delay time.Duration) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+
+	d.last = time.Now()
+	d.next = d.last.Add(delay)
+}

pkg/util/timer/delayer_test.go

@@ -0,0 +1,78 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package timer
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func withTime(f func()) time.Duration {
+	now := time.Now()
+	f()
+	return time.Since(now)
+}
+
+func Test_Delayer(t *testing.T) {
+	d := NewDelayer()
+
+	t.Run("Ensure instant execution", func(t *testing.T) {
+		require.True(t, withTime(func() {
+			d.Wait()
+		}) < time.Millisecond)
+
+		require.True(t, withTime(func() {
+			d.Wait()
+		}) < time.Millisecond)
+	})
+
+	t.Run("Delay execution", func(t *testing.T) {
+		require.True(t, withTime(func() {
+			d.Delay(50 * time.Millisecond)
+			d.Wait()
+		}) >= 50*time.Millisecond)
+	})
+
+	t.Run("Delay execution, but allow multiple ones", func(t *testing.T) {
+		require.True(t, withTime(func() {
+			d.Delay(50 * time.Millisecond)
+			d.Wait()
+			d.Wait()
+			d.Wait()
+			d.Wait()
+		}) >= 50*time.Millisecond)
+	})
+
+	t.Run("Delay execution multiple times", func(t *testing.T) {
+		require.True(t, withTime(func() {
+			d.Delay(50 * time.Millisecond)
+			d.Wait()
+			d.Delay(50 * time.Millisecond)
+			d.Wait()
+			d.Delay(50 * time.Millisecond)
+			d.Wait()
+			d.Delay(50 * time.Millisecond)
+			d.Wait()
+		}) >= 200*time.Millisecond)
+	})
+}