[Bugfix] Memory leaks due to incorrect time.After function usage (#1045)

CHANGELOG.md

@@ -4,6 +4,7 @@
 - (Feature) Add ArangoDeployment ServerGroupStatus
 - (Feature) (EE) Ordered Member IDs
 - (Refactor) Deprecate ForeachServerGroup, ForeachServerInGroups and ForServerGroup functions and refactor code accordingly
+- (Bugfix) Memory leaks due to incorrect time.After function usage
 
 ## [1.2.15](https://github.com/arangodb/kube-arangodb/tree/1.2.15) (2022-07-20)
 - (Bugfix) Ensure pod names not too long

pkg/deployment/chaos/monkey.go

@@ -23,13 +23,13 @@ package chaos
 import (
 	"context"
 	"math/rand"
-	"time"
 
 	"github.com/rs/zerolog"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/arangodb/kube-arangodb/pkg/logging"
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 )
 
 var (
@@ -78,7 +78,7 @@ func (m Monkey) Run(stopCh <-chan struct{}) {
 		}
 
 		select {
-		case <-time.After(spec.Chaos.GetInterval()):
+		case <-timer.After(spec.Chaos.GetInterval()):
 			// Continue
 		case <-stopCh:
 			// We're done

pkg/deployment/cluster_scaling_integration.go

@@ -35,6 +35,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 	"github.com/arangodb/kube-arangodb/pkg/util/globals"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 )
 
 var ciLogger = logging.Global().RegisterAndGetLogger("deployment-ci", logging.Info)
@@ -143,7 +144,7 @@ func (ci *clusterScalingIntegration) ListenForClusterEvents(stopCh <-chan struct
 		}
 
 		select {
-		case <-time.After(time.Second * 2):
+		case <-timer.After(time.Second * 2):
 			// Continue
 		case <-stopCh:
 			// We're done

pkg/deployment/resources/pdbs.go

@@ -34,6 +34,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 	"github.com/arangodb/kube-arangodb/pkg/util/globals"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 )
 
 func min(a int, b int) int {
@@ -220,7 +221,7 @@ func (r *Resources) ensurePDBForGroup(ctx context.Context, group api.ServerGroup
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case <-time.After(time.Second):
+		case <-timer.After(time.Second):
 		}
 	}
 }

pkg/exporter/monitor.go

@@ -37,6 +37,7 @@ import (
 
 	"github.com/arangodb/kube-arangodb/pkg/apis/shared"
 	"github.com/arangodb/kube-arangodb/pkg/logging"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 )
 
 const (
@@ -92,7 +93,7 @@ func (m monitor) UpdateMonitorStatus(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case <-time.After(sleep):
+		case <-timer.After(sleep):
 			continue
 		}
 	}

pkg/operator/operator.go

@@ -57,6 +57,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/constants"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
 	"github.com/arangodb/kube-arangodb/pkg/util/probe"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 )
 
 const (
@@ -200,7 +201,7 @@ func (o *Operator) handleShutdown() {
 	case <-shutdownCh:
 		o.log.Info("All deployments terminated, exiting.")
 		return
-	case <-time.After(o.Config.ShutdownTimeout):
+	case <-timer.After(o.Config.ShutdownTimeout):
 		o.log.Info("Timeout reached before all deployments terminated, exiting.")
 		return
 	}

pkg/replication/deployment_replication.go

@@ -38,6 +38,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
 	"github.com/arangodb/kube-arangodb/pkg/util/retry"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 	"github.com/arangodb/kube-arangodb/pkg/util/trigger"
 )
 
@@ -178,7 +179,7 @@ func (dr *DeploymentReplication) run() {
 		case <-dr.inspectTrigger.Done():
 			inspectionInterval = dr.inspectDeploymentReplication(inspectionInterval)
 
-		case <-time.After(inspectionInterval):
+		case <-timer.After(inspectionInterval):
 			// Trigger inspection
 			dr.inspectTrigger.Trigger()
 			// Backoff with next interval

pkg/storage/local_storage.go

@@ -37,6 +37,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
 	"github.com/arangodb/kube-arangodb/pkg/util/retry"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 	"github.com/arangodb/kube-arangodb/pkg/util/trigger"
 )
 
@@ -279,7 +280,7 @@ func (ls *LocalStorage) run() {
 				recentInspectionErrors = 0
 			}
 
-		case <-time.After(inspectionInterval):
+		case <-timer.After(inspectionInterval):
 			// Trigger inspection
 			ls.inspectTrigger.Trigger()
 			// Backoff with next interval

pkg/storage/pv_cleanup.go

@@ -34,6 +34,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/storage/provisioner"
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 	"github.com/arangodb/kube-arangodb/pkg/util/trigger"
 )
 
@@ -78,7 +79,7 @@ func (c *pvCleaner) Run(stopCh <-chan struct{}) {
 			return
 		case <-c.trigger.Done():
 			// Continue
-		case <-time.After(delay):
+		case <-timer.After(delay):
 			// Continue
 		}
 	}

pkg/util/interval.go

@@ -22,6 +22,8 @@ package util
 
 import (
 	"time"
+
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 )
 
 // Interval is a specialization of Duration so we can add some
@@ -60,5 +62,5 @@ func (i Interval) Backoff(factor float64, maxInt Interval) Interval {
 // After waits for the interval to elapse and then sends the current time
 // on the returned channel.
 func (i Interval) After() <-chan time.Time {
-	return time.After(time.Duration(i))
+	return timer.After(time.Duration(i))
 }

pkg/util/probe/liveness.go

@@ -24,6 +24,8 @@ import (
 	"net/http"
 	"sync"
 	"time"
+
+	"github.com/arangodb/kube-arangodb/pkg/util/timer"
 )
 
 const (
@@ -83,7 +85,7 @@ func (p *LivenessProbe) waitUntilNotLocked(timeout time.Duration) bool {
 		select {
 		case <-w:
 			// continue
-		case <-time.After(time.Until(deadline)):
+		case <-timer.After(time.Until(deadline)):
 			// Timeout
 			return false
 		}

pkg/util/timer/after.go

@@ -0,0 +1,44 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2022 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 "time"
+
+func After(duration time.Duration) <-chan time.Time {
+	r := make(chan time.Time)
+
+	go func() {
+		defer close(r)
+
+		tc := time.NewTimer(duration)
+		defer func() {
+			// Cleanup channel
+			for range tc.C {
+
+			}
+		}()
+		defer tc.Stop()
+
+		r <- <-tc.C
+	}()
+
+	return r
+}