[Feature] Agency cache generic (#1329)

ajanikow · web-flow · commit 7b9c0b4fedfc · 2023-06-09T13:18:14.000+02:00
diff --git a/pkg/deployment/agency/cache.go b/pkg/deployment/agency/cache.go
@@ -167,7 +167,7 @@ func NewAgencyCache(namespace, name string) Cache {
 	}
 
 	c.log = logger.WrapObj(c)
-	c.loader = getLoader()
+	c.loader = getLoader[state.Root]()
 
 	return c
 }
@@ -211,7 +211,7 @@ type cache struct {
 
 	lock sync.RWMutex
 
-	loader StateLoader
+	loader StateLoader[state.Root]
 
 	health Health
 
@@ -382,6 +382,10 @@ func (c *cache) getLeader(ctx context.Context, size int, clients Connections) (c
 	h.election = make(map[string]int, len(clients))
 
 	for id := range configs {
+		if err := errs[id]; err != nil {
+			c.log.Err(err).Str("agent", names[id]).Warn("Agent config request failed")
+		}
+
 		if config := configs[id]; config != nil {
 			name := config.Configuration.ID
 			if name == h.names[id] {
@@ -390,6 +394,8 @@ func (c *cache) getLeader(ctx context.Context, size int, clients Connections) (c
 					h.leaders[name] = config.LeaderId
 					h.election[config.LeaderId]++
 					h.leaderID = config.LeaderId
+				} else {
+					c.log.Str("agent", names[id]).Warn("Agent does not have leader")
 				}
 			}
 		}
diff --git a/pkg/deployment/agency/loader.go b/pkg/deployment/agency/loader.go
@@ -26,30 +26,29 @@ import (
 	"time"
 
 	agencyCecheConfig "github.com/arangodb/kube-arangodb/pkg/deployment/agency/cache"
-	"github.com/arangodb/kube-arangodb/pkg/deployment/agency/state"
 )
 
-func getLoader() StateLoader {
-	loader := getLoaderBase()
+func getLoader[T interface{}]() StateLoader[T] {
+	loader := getLoaderBase[T]()
 
-	loader = InvalidateOnErrorLoader(loader)
+	loader = InvalidateOnErrorLoader[T](loader)
 
-	loader = DelayLoader(loader, agencyCecheConfig.GlobalConfig().RefreshDelay)
-	loader = RefreshLoader(loader, agencyCecheConfig.GlobalConfig().RefreshInterval)
+	loader = DelayLoader[T](loader, agencyCecheConfig.GlobalConfig().RefreshDelay)
+	loader = RefreshLoader[T](loader, agencyCecheConfig.GlobalConfig().RefreshInterval)
 
 	return loader
 }
 
-func getLoaderBase() StateLoader {
+func getLoaderBase[T interface{}]() StateLoader[T] {
 	if agencyCecheConfig.GlobalConfig().PollEnabled {
-		return NewSimpleStateLoader()
+		return NewSimpleStateLoader[T]()
 	} else {
-		return NewSimpleStateLoader()
+		return NewSimpleStateLoader[T]()
 	}
 }
 
-type StateLoader interface {
-	State() (*state.Root, uint64, bool)
+type StateLoader[T interface{}] interface {
+	State() (*T, uint64, bool)
 
 	Invalidate()
 	Valid() bool
@@ -59,35 +58,35 @@ type StateLoader interface {
 	Refresh(ctx context.Context, discovery LeaderDiscovery) error
 }
 
-func NewSimpleStateLoader() StateLoader {
-	return &simpleStateLoader{}
+func NewSimpleStateLoader[T interface{}]() StateLoader[T] {
+	return &simpleStateLoader[T]{}
 }
 
-type simpleStateLoader struct {
+type simpleStateLoader[T interface{}] struct {
 	lock sync.Mutex
 
-	state *state.Root
+	state *T
 	index uint64
 	valid bool
 
 	updateTime time.Time
 }
 
-func (s *simpleStateLoader) UpdateTime() time.Time {
+func (s *simpleStateLoader[T]) UpdateTime() time.Time {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
 	return s.updateTime
 }
 
-func (s *simpleStateLoader) Valid() bool {
+func (s *simpleStateLoader[T]) Valid() bool {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
 	return s.valid
 }
 
-func (s *simpleStateLoader) State() (*state.Root, uint64, bool) {
+func (s *simpleStateLoader[T]) State() (*T, uint64, bool) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -98,14 +97,14 @@ func (s *simpleStateLoader) State() (*state.Root, uint64, bool) {
 	return s.state, s.index, true
 }
 
-func (s *simpleStateLoader) Invalidate() {
+func (s *simpleStateLoader[T]) Invalidate() {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
 	s.valid = false
 }
 
-func (s *simpleStateLoader) Refresh(ctx context.Context, discovery LeaderDiscovery) error {
+func (s *simpleStateLoader[T]) Refresh(ctx context.Context, discovery LeaderDiscovery) error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -121,7 +120,7 @@ func (s *simpleStateLoader) Refresh(ctx context.Context, discovery LeaderDiscove
 
 	if s.index != cfg.CommitIndex {
 		// Full reload
-		state, err := GetAgencyState(ctx, conn)
+		state, err := GetAgencyState[T](ctx, conn)
 		if err != nil {
 			return err
 		}
diff --git a/pkg/deployment/agency/loader_delayer.go b/pkg/deployment/agency/loader_delayer.go
@@ -24,59 +24,57 @@ import (
 	"context"
 	"sync"
 	"time"
-
-	"github.com/arangodb/kube-arangodb/pkg/deployment/agency/state"
 )
 
-func DelayLoader(loader StateLoader, delay time.Duration) StateLoader {
+func DelayLoader[T interface{}](loader StateLoader[T], delay time.Duration) StateLoader[T] {
 	if delay <= 0 {
 		return loader
 	}
 
-	return &delayerLoader{
+	return &delayerLoader[T]{
 		parent: loader,
 		delay:  delay,
 	}
 }
 
-type delayerLoader struct {
+type delayerLoader[T interface{}] struct {
 	lock sync.Mutex
 
 	last  time.Time
 	delay time.Duration
 
-	parent StateLoader
+	parent StateLoader[T]
 }
 
-func (i *delayerLoader) UpdateTime() time.Time {
+func (i *delayerLoader[T]) UpdateTime() time.Time {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.UpdateTime()
 }
 
-func (i *delayerLoader) Valid() bool {
+func (i *delayerLoader[T]) Valid() bool {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.Valid()
 }
 
-func (i *delayerLoader) State() (*state.Root, uint64, bool) {
+func (i *delayerLoader[T]) State() (*T, uint64, bool) {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.State()
 }
 
-func (i *delayerLoader) Invalidate() {
+func (i *delayerLoader[T]) Invalidate() {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	i.parent.Invalidate()
 }
 
-func (i *delayerLoader) Refresh(ctx context.Context, discovery LeaderDiscovery) error {
+func (i *delayerLoader[T]) Refresh(ctx context.Context, discovery LeaderDiscovery) error {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
diff --git a/pkg/deployment/agency/loader_invalidate.go b/pkg/deployment/agency/loader_invalidate.go
@@ -24,51 +24,49 @@ import (
 	"context"
 	"sync"
 	"time"
-
-	"github.com/arangodb/kube-arangodb/pkg/deployment/agency/state"
 )
 
-func InvalidateOnErrorLoader(loader StateLoader) StateLoader {
-	return &invalidateOnErrorLoader{
+func InvalidateOnErrorLoader[T interface{}](loader StateLoader[T]) StateLoader[T] {
+	return &invalidateOnErrorLoader[T]{
 		parent: loader,
 	}
 }
 
-type invalidateOnErrorLoader struct {
+type invalidateOnErrorLoader[T interface{}] struct {
 	lock sync.Mutex
 
-	parent StateLoader
+	parent StateLoader[T]
 }
 
-func (i *invalidateOnErrorLoader) UpdateTime() time.Time {
+func (i *invalidateOnErrorLoader[T]) UpdateTime() time.Time {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.UpdateTime()
 }
 
-func (i *invalidateOnErrorLoader) Valid() bool {
+func (i *invalidateOnErrorLoader[T]) Valid() bool {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.Valid()
 }
 
-func (i *invalidateOnErrorLoader) State() (*state.Root, uint64, bool) {
+func (i *invalidateOnErrorLoader[T]) State() (*T, uint64, bool) {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.State()
 }
 
-func (i *invalidateOnErrorLoader) Invalidate() {
+func (i *invalidateOnErrorLoader[T]) Invalidate() {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	i.parent.Invalidate()
 }
 
-func (i *invalidateOnErrorLoader) Refresh(ctx context.Context, discovery LeaderDiscovery) (err error) {
+func (i *invalidateOnErrorLoader[T]) Refresh(ctx context.Context, discovery LeaderDiscovery) (err error) {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
diff --git a/pkg/deployment/agency/loader_refresher.go b/pkg/deployment/agency/loader_refresher.go
@@ -24,59 +24,57 @@ import (
 	"context"
 	"sync"
 	"time"
-
-	"github.com/arangodb/kube-arangodb/pkg/deployment/agency/state"
 )
 
-func RefreshLoader(loader StateLoader, delay time.Duration) StateLoader {
+func RefreshLoader[T interface{}](loader StateLoader[T], delay time.Duration) StateLoader[T] {
 	if delay <= 0 {
 		return loader
 	}
 
-	return &refresherLoader{
+	return &refresherLoader[T]{
 		parent: loader,
 		delay:  delay,
 	}
 }
 
-type refresherLoader struct {
+type refresherLoader[T interface{}] struct {
 	lock sync.Mutex
 
 	last  time.Time
 	delay time.Duration
 
-	parent StateLoader
+	parent StateLoader[T]
 }
 
-func (i *refresherLoader) UpdateTime() time.Time {
+func (i *refresherLoader[T]) UpdateTime() time.Time {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.UpdateTime()
 }
 
-func (i *refresherLoader) Valid() bool {
+func (i *refresherLoader[T]) Valid() bool {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.Valid()
 }
 
-func (i *refresherLoader) State() (*state.Root, uint64, bool) {
+func (i *refresherLoader[T]) State() (*T, uint64, bool) {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	return i.parent.State()
 }
 
-func (i *refresherLoader) Invalidate() {
+func (i *refresherLoader[T]) Invalidate() {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
 	i.parent.Invalidate()
 }
 
-func (i *refresherLoader) Refresh(ctx context.Context, discovery LeaderDiscovery) error {
+func (i *refresherLoader[T]) Refresh(ctx context.Context, discovery LeaderDiscovery) error {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 
diff --git a/pkg/deployment/agency/state.go b/pkg/deployment/agency/state.go
@@ -24,27 +24,28 @@ import (
 	"context"
 	"net/http"
 
-	"github.com/arangodb/kube-arangodb/pkg/deployment/agency/state"
 	"github.com/arangodb/kube-arangodb/pkg/util/arangod/conn"
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 )
 
-func GetAgencyState(ctx context.Context, connection conn.Connection) (state.Root, error) {
-	resp, code, err := conn.NewExecutor[ReadRequest, state.Roots](connection).Execute(ctx, http.MethodPost, "/_api/agency/read", GetAgencyReadRequestFields())
+func GetAgencyState[T interface{}](ctx context.Context, connection conn.Connection) (T, error) {
+	var def T
+
+	resp, code, err := conn.NewExecutor[ReadRequest, []T](connection).Execute(ctx, http.MethodPost, "/_api/agency/read", GetAgencyReadRequestFields())
 	if err != nil {
-		return state.Root{}, err
+		return def, err
 	}
 
 	if code != http.StatusOK {
-		return state.Root{}, errors.Newf("Unknown response code %d", code)
+		return def, errors.Newf("Unknown response code %d", code)
 	}
 
 	if resp == nil {
-		return state.Root{}, errors.Newf("Missing response body")
+		return def, errors.Newf("Missing response body")
 	}
 
 	if len(*resp) != 1 {
-		return state.Root{}, errors.Newf("Invalid response size")
+		return def, errors.Newf("Invalid response size")
 	}
 
 	return (*resp)[0], nil
diff --git a/pkg/deployment/agency/state/state.go b/pkg/deployment/agency/state/state.go
@@ -20,8 +20,6 @@
 
 package state
 
-type Roots []Root
-
 type Root struct {
 	Arango   State `json:"arango"`
 	ArangoDB DB    `json:"arangodb,omitempty"`
diff --git a/pkg/deployment/agency/state/state_test.go b/pkg/deployment/agency/state/state_test.go

Original file line number	Diff line number	Diff line change
`@@ -167,7 +167,7 @@ func NewAgencyCache(namespace, name string) Cache {`
`167`	`167`	`}`
`168`	`168`
`169`	`169`	`c.log = logger.WrapObj(c)`
`170`		`- c.loader = getLoader()`
	`170`	`+ c.loader = getLoader[state.Root]()`
`171`	`171`
`172`	`172`	`return c`
`173`	`173`	`}`
`@@ -211,7 +211,7 @@ type cache struct {`
`211`	`211`
`212`	`212`	`lock sync.RWMutex`
`213`	`213`
`214`		`- loader StateLoader`
	`214`	`+ loader StateLoader[state.Root]`
`215`	`215`
`216`	`216`	`health Health`
`217`	`217`
`@@ -382,6 +382,10 @@ func (c *cache) getLeader(ctx context.Context, size int, clients Connections) (c`
`382`	`382`	`h.election = make(map[string]int, len(clients))`
`383`	`383`
`384`	`384`	`for id := range configs {`
	`385`	`+ if err := errs[id]; err != nil {`
	`386`	`+ c.log.Err(err).Str("agent", names[id]).Warn("Agent config request failed")`
	`387`	`+ }`
	`388`	`+`
`385`	`389`	`if config := configs[id]; config != nil {`
`386`	`390`	`name := config.Configuration.ID`
`387`	`391`	`if name == h.names[id] {`
`@@ -390,6 +394,8 @@ func (c *cache) getLeader(ctx context.Context, size int, clients Connections) (c`
`390`	`394`	`h.leaders[name] = config.LeaderId`
`391`	`395`	`h.election[config.LeaderId]++`
`392`	`396`	`h.leaderID = config.LeaderId`
	`397`	`+ } else {`
	`398`	`+ c.log.Str("agent", names[id]).Warn("Agent does not have leader")`
`393`	`399`	`}`
`394`	`400`	`}`
`395`	`401`	`}`