[Feature] Extract Agency timeout (#1344)

CHANGELOG.md

@@ -4,6 +4,8 @@
 - (Improvement) Block traffic on the services if there is more than 1 active leader in ActiveFailover mode
 - (Improvement) Improve master endpoint validation.
 - (Feature) Agency Improvements
+- (Bugfix) Fix agency timeout
+- (Improvement) Extract Agency Timeout
 
 ## [1.2.30](https://github.com/arangodb/kube-arangodb/tree/1.2.30) (2023-06-16)
 - (Feature) AgencyCache Interface

pkg/deployment/agency/loader.go

@@ -32,6 +32,8 @@ import (
 func getLoader[T interface{}]() agencyCache.StateLoader[T] {
 	loader := getLoaderBase[T]()
 
+	loader = TimeoutLoader[T](loader, globals.GetGlobalTimeouts().Agency().Get())
+
 	loader = InvalidateOnErrorLoader[T](loader)
 
 	loader = DelayLoader[T](loader, agencyCache.GlobalConfig().RefreshDelay)
@@ -103,18 +105,12 @@ func (s *simpleStateLoader[T]) Refresh(ctx context.Context, discovery agencyCach
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	lctx, cancel := globals.GetGlobalTimeouts().Agency().WithTimeout(ctx)
-	defer cancel()
-
-	conn, err := discovery.Discover(lctx)
+	conn, err := discovery.Discover(ctx)
 	if err != nil {
 		return err
 	}
 
-	cctx, cancel := globals.GetGlobalTimeouts().Agency().WithTimeout(ctx)
-	defer cancel()
-
-	cfg, err := GetAgencyConfig(cctx, conn)
+	cfg, err := GetAgencyConfig(ctx, conn)
 	if err != nil {
 		return err
 	}
@@ -122,10 +118,7 @@ func (s *simpleStateLoader[T]) Refresh(ctx context.Context, discovery agencyCach
 	if !s.valid || s.index != cfg.CommitIndex {
 		// Full reload
 
-		sctx, cancel := globals.GetGlobalTimeouts().Agency().WithTimeout(ctx)
-		defer cancel()
-
-		state, err := GetAgencyState[T](sctx, conn)
+		state, err := GetAgencyState[T](ctx, conn)
 		if err != nil {
 			return err
 		}

pkg/deployment/agency/loader_timeout.go

@@ -0,0 +1,68 @@
+//
+// 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 agency
+
+import (
+	"context"
+	"time"
+
+	agencyCache "github.com/arangodb/kube-arangodb/pkg/deployment/agency/cache"
+)
+
+func TimeoutLoader[T interface{}](loader agencyCache.StateLoader[T], timeout time.Duration) agencyCache.StateLoader[T] {
+	if timeout <= 0 {
+		return loader
+	}
+
+	return &timeoutLoader[T]{
+		parent:  loader,
+		timeout: timeout,
+	}
+}
+
+type timeoutLoader[T interface{}] struct {
+	parent agencyCache.StateLoader[T]
+
+	timeout time.Duration
+}
+
+func (i *timeoutLoader[T]) UpdateTime() time.Time {
+	return i.parent.UpdateTime()
+}
+
+func (i *timeoutLoader[T]) Valid() bool {
+	return i.parent.Valid()
+}
+
+func (i *timeoutLoader[T]) State() (*T, uint64, bool) {
+	return i.parent.State()
+}
+
+func (i *timeoutLoader[T]) Invalidate() {
+	i.parent.Invalidate()
+}
+
+func (i *timeoutLoader[T]) Refresh(ctx context.Context, discovery agencyCache.LeaderDiscovery) error {
+	nctx, c := context.WithTimeout(ctx, i.timeout)
+	defer c()
+
+	return i.parent.Refresh(nctx, discovery)
+}