diff --git a/pkg/gameserverallocations/allocation_cache.go b/pkg/gameserverallocations/allocation_cache.go index 34bf898ac6..ef4468c764 100644 --- a/pkg/gameserverallocations/allocation_cache.go +++ b/pkg/gameserverallocations/allocation_cache.go @@ -18,6 +18,7 @@ import ( "context" "sort" + "agones.dev/agones/pkg/apis" "agones.dev/agones/pkg/apis/agones" agonesv1 "agones.dev/agones/pkg/apis/agones/v1" allocationv1 "agones.dev/agones/pkg/apis/allocation/v1" @@ -151,6 +152,23 @@ func (c *AllocationCache) AddGameServer(gs *agonesv1.GameServer) { c.cache.Store(key, gs) } +// refreshGameServer updates the cache with the live version of the GameServer read from the API server: +// it is stored if it is still allocatable, and removed from the cache otherwise +func (c *AllocationCache) refreshGameServer(gs *agonesv1.GameServer) { + key, _ := cache.MetaNamespaceKeyFunc(gs) + + if !gs.IsBeingDeleted() && c.matcher(gs) { + c.cache.Store(key, gs) + return + } + c.cache.Delete(key) +} + +// GetGameServer returns the cached GameServer for the given key, along with a boolean indicating if it was found +func (c *AllocationCache) GetGameServer(key string) (*agonesv1.GameServer, bool) { + return c.cache.Load(key) +} + // getGameServers returns a list of game servers in the cache. func (c *AllocationCache) getGameServers() []*agonesv1.GameServer { length := c.cache.Len() @@ -175,50 +193,12 @@ func (c *AllocationCache) ListSortedGameServers(gsa *allocationv1.GameServerAllo } counts := c.counter.Counts() + var priorities []agonesv1.Priority + if gsa != nil { + priorities = gsa.Spec.Priorities + } sort.Slice(list, func(i, j int) bool { - gs1 := list[i] - gs2 := list[j] - - // Search Allocated GameServers first. - if gs1.Status.State != gs2.Status.State { - return gs1.Status.State == agonesv1.GameServerStateAllocated - } - - c1, ok := counts[gs1.Status.NodeName] - if !ok { - return false - } - - c2, ok := counts[gs2.Status.NodeName] - if !ok { - return true - } - - if c1.Allocated > c2.Allocated { - return true - } - if c1.Allocated < c2.Allocated { - return false - } - - // prefer nodes that have the most Ready gameservers on them - they are most likely to be - // completely filled and least likely target for scale down. - if c1.Ready < c2.Ready { - return false - } - if c1.Ready > c2.Ready { - return true - } - - // if we end up here, then break the tie with Counter or List Priority. - if gsa != nil { - if res := gs1.CompareCountAndListPriorities(gsa.Spec.Priorities, gs2); res != nil { - return *res - } - } - - // finally sort lexicographically, so we have a stable order - return gs1.GetObjectMeta().GetName() < gs2.GetObjectMeta().GetName() + return compareGameServersForPackedStrategy(list[i], list[j], priorities, counts) }) return list @@ -232,18 +212,12 @@ func (c *AllocationCache) ListSortedGameServersPriorities(gsa *allocationv1.Game return []*agonesv1.GameServer{} } + var priorities []agonesv1.Priority + if gsa != nil { + priorities = gsa.Spec.Priorities + } sort.Slice(list, func(i, j int) bool { - gs1 := list[i] - gs2 := list[j] - - if gsa != nil { - if res := gs1.CompareCountAndListPriorities(gsa.Spec.Priorities, gs2); res != nil { - return *res - } - } - - // finally sort lexicographically, so we have a stable order - return gs1.GetObjectMeta().GetName() < gs2.GetObjectMeta().GetName() + return compareGameServersForDistributedStrategy(list[i], list[j], priorities) }) return list @@ -314,3 +288,155 @@ func (c *AllocationCache) getKey(gs *agonesv1.GameServer) (string, bool) { } return key, ok } + +// ReorderGameServerAfterAllocation repositions gsAfterAllocation in gsList after a local allocation +// has been applied, using gsIndexBeforeAllocation as a search hint. Used by the batch allocator +// to maintain sort order without a full re-sort. +func (c *AllocationCache) ReorderGameServerAfterAllocation( + gsList []*agonesv1.GameServer, + gsIndexBeforeAllocation int, gsAfterAllocation *agonesv1.GameServer, + priorities []agonesv1.Priority, strategy apis.SchedulingStrategy) { + if len(gsList) == 0 || gsIndexBeforeAllocation < 0 || gsIndexBeforeAllocation >= len(gsList) || + gsList[gsIndexBeforeAllocation] == nil || gsAfterAllocation == nil { + c.baseLogger.WithField("gsIndexBeforeAllocation", gsIndexBeforeAllocation). + WithField("gsAfterAllocation", gsAfterAllocation). + WithField("gsListLength", len(gsList)). + Warn("ReorderGameServerAfterAllocation called with invalid parameters, reordering is skipped") + return + } + + newIndex := gsIndexBeforeAllocation + gsToReorderOriginal := gsList[gsIndexBeforeAllocation] + + optimizeList := func(greater bool) []*agonesv1.GameServer { + if greater { + return gsList[gsIndexBeforeAllocation+1:] + } + return gsList[:gsIndexBeforeAllocation] + } + + switch strategy { + case apis.Packed: + counts := c.counter.Counts() + greater, equal := compareGameServersAfterAllocationForPackedStrategy(gsToReorderOriginal, gsAfterAllocation, priorities, counts) + if !equal { + newIndex = findIndexAfterAllocationForPackedStrategy(optimizeList(greater), gsAfterAllocation, priorities, counts) + if greater { + newIndex += gsIndexBeforeAllocation + } + } + case apis.Distributed: + greater, equal := compareGameServersAfterAllocationForDistributedStrategy(gsToReorderOriginal, gsAfterAllocation, priorities) + if !equal { + newIndex = findIndexAfterAllocationForDistributedStrategy(optimizeList(greater), gsAfterAllocation, priorities) + if greater { + newIndex += gsIndexBeforeAllocation + } + } + default: + c.baseLogger.WithField("strategy", strategy). + Warn("scheduling strategy not supported, reordering is skipped") + } + + if newIndex != gsIndexBeforeAllocation { + gsList = append(gsList[:gsIndexBeforeAllocation], gsList[gsIndexBeforeAllocation+1:]...) + gsList = append(gsList[:newIndex], append([]*agonesv1.GameServer{gsAfterAllocation}, gsList[newIndex:]...)...) + } else { + gsList[gsIndexBeforeAllocation] = gsAfterAllocation + } +} + +// compareGameServersAfterAllocationForPackedStrategy compares a game server before and after a local +// allocation is applied. Returns (greater, equal) where greater means before has higher priority than after. +func compareGameServersAfterAllocationForPackedStrategy( + before, after *agonesv1.GameServer, + priorities []agonesv1.Priority, + counts map[string]gameservers.NodeCount) (bool, bool) { + if before.Status.State != after.Status.State { + return before.Status.State == agonesv1.GameServerStateAllocated, false + } + + c1, ok := counts[before.Status.NodeName] + if !ok { + return false, false + } + c2, ok := counts[after.Status.NodeName] + if !ok { + return true, false + } + + if c1.Allocated > c2.Allocated { + return true, false + } + if c1.Allocated < c2.Allocated { + return false, false + } + + if c1.Ready < c2.Ready { + return false, false + } + if c1.Ready > c2.Ready { + return true, false + } + + if priorities != nil { + if res := before.CompareCountAndListPriorities(priorities, after); res != nil { + return *res, false + } + } + + return false, true +} + +// compareGameServersAfterAllocationForDistributedStrategy compares a game server before and after a local +// allocation is applied using the distributed strategy. Returns (greater, equal). +func compareGameServersAfterAllocationForDistributedStrategy( + before, after *agonesv1.GameServer, + priorities []agonesv1.Priority) (bool, bool) { + if priorities != nil { + if res := before.CompareCountAndListPriorities(priorities, after); res != nil { + return *res, false + } + } + return false, true +} + +// compareGameServersForPackedStrategy compares two game servers for the packed strategy with +// lexicographic tie-breaking for a stable sort order. +func compareGameServersForPackedStrategy(gs0, gs1 *agonesv1.GameServer, priorities []agonesv1.Priority, counts map[string]gameservers.NodeCount) bool { + if gs1 == nil { + return true + } + greater, equal := compareGameServersAfterAllocationForPackedStrategy(gs0, gs1, priorities, counts) + if !equal { + return greater + } + return gs0.GetObjectMeta().GetName() < gs1.GetObjectMeta().GetName() +} + +// compareGameServersForDistributedStrategy compares two game servers for the distributed strategy with +// lexicographic tie-breaking for a stable sort order. +func compareGameServersForDistributedStrategy(gs0, gs1 *agonesv1.GameServer, priorities []agonesv1.Priority) bool { + if gs1 == nil { + return true + } + greater, equal := compareGameServersAfterAllocationForDistributedStrategy(gs0, gs1, priorities) + if !equal { + return greater + } + return gs0.GetObjectMeta().GetName() < gs1.GetObjectMeta().GetName() +} + +// findIndexAfterAllocationForPackedStrategy binary-searches for the insertion position that maintains sort order. +func findIndexAfterAllocationForPackedStrategy(gsList []*agonesv1.GameServer, gs *agonesv1.GameServer, priorities []agonesv1.Priority, counts map[string]gameservers.NodeCount) int { + return sort.Search(len(gsList), func(i int) bool { + return compareGameServersForPackedStrategy(gs, gsList[i], priorities, counts) + }) +} + +// findIndexAfterAllocationForDistributedStrategy binary-searches for the insertion position that maintains sort order. +func findIndexAfterAllocationForDistributedStrategy(gsList []*agonesv1.GameServer, gs *agonesv1.GameServer, priorities []agonesv1.Priority) int { + return sort.Search(len(gsList), func(i int) bool { + return compareGameServersForDistributedStrategy(gs, gsList[i], priorities) + }) +} diff --git a/pkg/gameserverallocations/allocation_cache_test.go b/pkg/gameserverallocations/allocation_cache_test.go index 3a33d001fe..e4fbd0e8e8 100644 --- a/pkg/gameserverallocations/allocation_cache_test.go +++ b/pkg/gameserverallocations/allocation_cache_test.go @@ -16,9 +16,12 @@ package gameserverallocations import ( "context" + "fmt" + "strconv" "testing" "time" + "agones.dev/agones/pkg/apis" agonesv1 "agones.dev/agones/pkg/apis/agones/v1" allocationv1 "agones.dev/agones/pkg/apis/allocation/v1" "agones.dev/agones/pkg/gameservers" @@ -30,6 +33,7 @@ import ( "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" k8sruntime "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" k8stesting "k8s.io/client-go/testing" @@ -579,6 +583,249 @@ func TestAllocatorRunCacheSync(t *testing.T) { assertCacheEntries(0) } +func TestAllocationCacheReorderGameServerAfterAllocation(t *testing.T) { + t.Parallel() + + newGs := func(i int, node string, state agonesv1.GameServerState) agonesv1.GameServer { + name := fmt.Sprintf("gs%d", i) + return agonesv1.GameServer{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: defaultNs, UID: types.UID(strconv.Itoa(i))}, + Status: agonesv1.GameServerStatus{NodeName: node, State: state}} + } + // withPlayers sets the same number of players in the Players status, the "players" counter and the "players" list + withPlayers := func(gs agonesv1.GameServer, count int) agonesv1.GameServer { + values := make([]string, count) + for i := range values { + values[i] = fmt.Sprintf("player%d", i) + } + gs.Status.Players = &agonesv1.PlayerStatus{Count: int64(count), Capacity: 10} + gs.Status.Counters = map[string]agonesv1.CounterStatus{"players": {Count: int64(count), Capacity: 10}} + gs.Status.Lists = map[string]agonesv1.ListStatus{"players": {Values: values, Capacity: 10}} + return gs + } + + ready, allocated := agonesv1.GameServerStateReady, agonesv1.GameServerStateAllocated + gs0Allocated := newGs(0, "node0", allocated) + gs1, gs1Allocated := newGs(1, "node0", ready), newGs(1, "node0", allocated) + gs2, gs2Allocated := newGs(2, "node0", ready), newGs(2, "node0", allocated) + gs3 := newGs(3, "node1", ready) + gs4 := withPlayers(newGs(4, "node1", allocated), 3) + gs5, gs5Allocated := withPlayers(newGs(5, "node1", allocated), 2), withPlayers(newGs(5, "node1", allocated), 5) + gs6 := withPlayers(newGs(6, "node1", allocated), 1) + + counterPriority := []agonesv1.Priority{{Type: "Counter", Key: "players", Order: "Ascending"}} + listPriority := []agonesv1.Priority{{Type: "List", Key: "players", Order: "Descending"}} + + fixtures := map[string]struct { + list []*agonesv1.GameServer + priorities []agonesv1.Priority + packingStrategy apis.SchedulingStrategy + gsToReorder *agonesv1.GameServer + gsToReorderIndex int + want []*agonesv1.GameServer + }{ + "packed (no change)": { + list: []*agonesv1.GameServer{&gs0Allocated, &gs1, &gs2, &gs3}, + gsToReorder: &gs1Allocated, + gsToReorderIndex: 1, + want: []*agonesv1.GameServer{&gs0Allocated, &gs1Allocated, &gs2, &gs3}, + }, + "packed": { + list: []*agonesv1.GameServer{&gs0Allocated, &gs1, &gs2, &gs3}, + gsToReorder: &gs2Allocated, + gsToReorderIndex: 2, + want: []*agonesv1.GameServer{&gs0Allocated, &gs2Allocated, &gs1, &gs3}, + }, + "packed (sort by name)": { + list: []*agonesv1.GameServer{&gs0Allocated, &gs2Allocated, &gs1, &gs3}, + gsToReorder: &gs1Allocated, + gsToReorderIndex: 2, + want: []*agonesv1.GameServer{&gs0Allocated, &gs1Allocated, &gs2Allocated, &gs3}, + }, + "packed (all ready)": { + list: []*agonesv1.GameServer{&gs1, &gs2, &gs3}, + gsToReorder: &gs1Allocated, + gsToReorderIndex: 0, + want: []*agonesv1.GameServer{&gs1Allocated, &gs2, &gs3}, + }, + "packed (only one)": { + list: []*agonesv1.GameServer{&gs1}, + gsToReorder: &gs1Allocated, + gsToReorderIndex: 0, + want: []*agonesv1.GameServer{&gs1Allocated}, + }, + "packed (priority counter)": { + list: []*agonesv1.GameServer{&gs4, &gs5, &gs6}, + priorities: counterPriority, + gsToReorder: &gs5Allocated, + gsToReorderIndex: 1, + want: []*agonesv1.GameServer{&gs5Allocated, &gs4, &gs6}, + }, + "packed (priority list)": { + list: []*agonesv1.GameServer{&gs6, &gs5, &gs4}, + priorities: listPriority, + gsToReorder: &gs5Allocated, + gsToReorderIndex: 1, + want: []*agonesv1.GameServer{&gs6, &gs4, &gs5Allocated}, + }, + "distributed (no change)": { + list: []*agonesv1.GameServer{&gs0Allocated, &gs1, &gs2, &gs3}, + packingStrategy: apis.Distributed, + gsToReorder: &gs2Allocated, + gsToReorderIndex: 2, + want: []*agonesv1.GameServer{&gs0Allocated, &gs1, &gs2Allocated, &gs3}, + }, + "distributed (only one)": { + list: []*agonesv1.GameServer{&gs1}, + packingStrategy: apis.Distributed, + gsToReorder: &gs1Allocated, + gsToReorderIndex: 0, + want: []*agonesv1.GameServer{&gs1Allocated}, + }, + "distributed (priority counter)": { + list: []*agonesv1.GameServer{&gs4, &gs5, &gs6}, + priorities: counterPriority, + packingStrategy: apis.Distributed, + gsToReorder: &gs5Allocated, + gsToReorderIndex: 1, + want: []*agonesv1.GameServer{&gs5Allocated, &gs4, &gs6}, + }, + "distributed (priority list)": { + list: []*agonesv1.GameServer{&gs6, &gs5, &gs4}, + priorities: listPriority, + packingStrategy: apis.Distributed, + gsToReorder: &gs5Allocated, + gsToReorderIndex: 1, + want: []*agonesv1.GameServer{&gs6, &gs4, &gs5Allocated}, + }, + } + + for testName, testScenario := range fixtures { + t.Run(testName, func(t *testing.T) { + cache, m := newFakeAllocationCache() + + items := make([]agonesv1.GameServer, len(testScenario.list)) + for i, gs := range testScenario.list { + items[i] = *gs + } + m.AgonesClient.AddReactor("list", "gameservers", func(_ k8stesting.Action) (bool, k8sruntime.Object, error) { + return true, &agonesv1.GameServerList{Items: items}, nil + }) + + ctx, cancel := agtesting.StartInformers(m, cache.gameServerSynced) + defer cancel() + + err := cache.syncCache() + assert.NoError(t, err) + + err = cache.counter.Run(ctx, 0) + assert.NoError(t, err) + + strategy := apis.Packed + if testScenario.packingStrategy != "" { + strategy = testScenario.packingStrategy + } + + cache.ReorderGameServerAfterAllocation( + testScenario.list, + testScenario.gsToReorderIndex, testScenario.gsToReorder, + testScenario.priorities, strategy) + + if !assert.Equal(t, testScenario.want, testScenario.list, "reordered list should match expected") { + for _, gs := range testScenario.list { + t.Logf("%s, ", gs.Name) + } + } + }) + } +} + +func TestAllocationCacheGetGameServer(t *testing.T) { + t.Parallel() + + a, _ := newFakeAllocator() + gs := &agonesv1.GameServer{ + ObjectMeta: metav1.ObjectMeta{Name: "gs1", Namespace: defaultNs, UID: "1", ResourceVersion: "1"}, + Status: agonesv1.GameServerStatus{State: agonesv1.GameServerStateReady}, + } + + _, ok := a.allocationCache.GetGameServer("default/gs1") + assert.False(t, ok) + + a.allocationCache.AddGameServer(gs) + + got, ok := a.allocationCache.GetGameServer("default/gs1") + require.True(t, ok) + assert.Equal(t, gs, got) + + _, ok = a.allocationCache.GetGameServer("default/gs2") + assert.False(t, ok) +} + +func TestAllocationCacheRefreshGameServer(t *testing.T) { + t.Parallel() + + const key = "default/gs1" + now := metav1.Now() + newGs := func(resourceVersion string, state agonesv1.GameServerState) *agonesv1.GameServer { + return &agonesv1.GameServer{ + ObjectMeta: metav1.ObjectMeta{Name: "gs1", Namespace: defaultNs, UID: "1", ResourceVersion: resourceVersion}, + Status: agonesv1.GameServerStatus{State: state}, + } + } + + testCases := map[string]struct { + cached *agonesv1.GameServer + live *agonesv1.GameServer + wantInRV string + }{ + "an allocatable GameServer replaces the cached one": { + cached: newGs("1", agonesv1.GameServerStateAllocated), + live: newGs("2", agonesv1.GameServerStateAllocated), + wantInRV: "2", + }, + "an allocatable GameServer is added when not cached": { + live: newGs("2", agonesv1.GameServerStateReady), + wantInRV: "2", + }, + "a GameServer that is no longer allocatable is removed": { + cached: newGs("1", agonesv1.GameServerStateAllocated), + live: newGs("2", agonesv1.GameServerStateShutdown), + }, + "a GameServer being deleted is removed": { + cached: newGs("1", agonesv1.GameServerStateReady), + live: func() *agonesv1.GameServer { + gs := newGs("2", agonesv1.GameServerStateReady) + gs.ObjectMeta.DeletionTimestamp = &now + return gs + }(), + }, + "a GameServer that is not allocatable and not cached is a no-op": { + live: newGs("2", agonesv1.GameServerStateShutdown), + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + t.Parallel() + + a, _ := newFakeAllocator() + if tc.cached != nil { + a.allocationCache.AddGameServer(tc.cached) + } + + a.allocationCache.refreshGameServer(tc.live) + + got, ok := a.allocationCache.GetGameServer(key) + if tc.wantInRV == "" { + assert.False(t, ok) + return + } + require.True(t, ok) + assert.Equal(t, tc.wantInRV, got.ObjectMeta.ResourceVersion) + }) + } +} + func newFakeAllocationCache() (*AllocationCache, agtesting.Mocks) { m := agtesting.NewMocks() cache := NewAllocationCache(m.AgonesInformerFactory.Agones().V1().GameServers(), gameservers.NewPerNodeCounter(m.KubeInformerFactory, m.AgonesInformerFactory), healthcheck.NewHandler()) diff --git a/pkg/gameserverallocations/allocator.go b/pkg/gameserverallocations/allocator.go index 79175c078c..960c992a0f 100644 --- a/pkg/gameserverallocations/allocator.go +++ b/pkg/gameserverallocations/allocator.go @@ -184,7 +184,11 @@ func (c *Allocator) Run(ctx context.Context) error { } // workers and logic for batching allocations - go c.ListenAndAllocate(ctx, maxBatchQueue) + if runtime.FeatureEnabled(runtime.FeatureProcessorAllocator) { + go c.ListenAndBatchAllocate(ctx, maxBatchQueue) + } else { + go c.ListenAndAllocate(ctx, maxBatchQueue) + } return nil } @@ -278,34 +282,38 @@ func (c *Allocator) allocateFromLocalCluster(ctx context.Context, gsa *allocatio return err }) - if err != nil && !goErrors.Is(err, ErrNoGameServer) && !goErrors.Is(err, ErrConflictInGameServerSelection) { - c.allocationCache.Resync() - return nil, err - } - switch { - case goErrors.Is(err, ErrNoGameServer), goErrors.Is(err, ErrGameServerUpdateConflict): + case err == nil: + // success - fall through to fill allocation result below + case goErrors.Is(err, ErrNoGameServer): gsa.Status.State = allocationv1.GameServerAllocationUnAllocated + return gsa, nil case goErrors.Is(err, ErrConflictInGameServerSelection): gsa.Status.State = allocationv1.GameServerAllocationContention + return gsa, nil + case goErrors.Is(err, ErrGameServerUpdateConflict): + c.allocationCache.Resync() + gsa.Status.State = allocationv1.GameServerAllocationUnAllocated + return gsa, nil default: - gsa.ObjectMeta.Name = gs.ObjectMeta.Name - gsa.Status.State = allocationv1.GameServerAllocationAllocated - gsa.Status.GameServerName = gs.ObjectMeta.Name - gsa.Status.Ports = gs.Status.Ports - gsa.Status.Address = gs.Status.Address - gsa.Status.Addresses = append(gsa.Status.Addresses, gs.Status.Addresses...) - gsa.Status.NodeName = gs.Status.NodeName - gsa.Status.Source = localAllocationSource - gsa.Status.Metadata = &allocationv1.GameServerMetadata{ - Labels: gs.ObjectMeta.Labels, - Annotations: gs.ObjectMeta.Annotations, - } - - gsa.Status.Counters = gs.Status.Counters - gsa.Status.Lists = gs.Status.Lists + c.allocationCache.Resync() + return nil, err + } + gsa.ObjectMeta.Name = gs.ObjectMeta.Name + gsa.Status.State = allocationv1.GameServerAllocationAllocated + gsa.Status.GameServerName = gs.ObjectMeta.Name + gsa.Status.Ports = gs.Status.Ports + gsa.Status.Address = gs.Status.Address + gsa.Status.Addresses = append(gsa.Status.Addresses, gs.Status.Addresses...) + gsa.Status.NodeName = gs.Status.NodeName + gsa.Status.Source = localAllocationSource + gsa.Status.Metadata = &allocationv1.GameServerMetadata{ + Labels: gs.ObjectMeta.Labels, + Annotations: gs.ObjectMeta.Annotations, } + gsa.Status.Counters = gs.Status.Counters + gsa.Status.Lists = gs.Status.Lists c.loggerForGameServerAllocation(gsa).Debug("Game server allocation") return gsa, nil @@ -620,7 +628,7 @@ func (c *Allocator) allocationUpdateWorkers(ctx context.Context, workerCount int // we should wait for it to get updated with fresh info. c.allocationCache.AddGameServer(gs) } - res.err = ErrGameServerUpdateConflict + res.err = goErrors.Join(ErrGameServerUpdateConflict, err) } else { // put the GameServer back into the cache, so it's immediately around for re-allocation c.allocationCache.AddGameServer(gs) @@ -720,7 +728,7 @@ func Retry(backoff wait.Backoff, fn func() error) error { switch { case err == nil: return true, nil - case goErrors.Is(err, ErrNoGameServer), goErrors.Is(err, ErrTotalTimeoutExceeded): + case goErrors.Is(err, ErrTotalTimeoutExceeded): return true, err default: lastConflictErr = err diff --git a/pkg/gameserverallocations/allocator_test.go b/pkg/gameserverallocations/allocator_test.go index f6a059fd89..27bc6f94f5 100644 --- a/pkg/gameserverallocations/allocator_test.go +++ b/pkg/gameserverallocations/allocator_test.go @@ -17,6 +17,7 @@ package gameserverallocations import ( "context" "errors" + "sync" "testing" "time" @@ -26,12 +27,14 @@ import ( multiclusterv1 "agones.dev/agones/pkg/apis/multicluster/v1" "agones.dev/agones/pkg/gameservers" agtesting "agones.dev/agones/pkg/testing" + "agones.dev/agones/pkg/util/runtime" "agones.dev/agones/test/e2e/framework" "github.com/heptiolabs/healthcheck" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" + k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" k8sruntime "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/wait" @@ -479,7 +482,7 @@ func TestAllocatorAllocateOnGameServerUpdateError(t *testing.T) { _, err := a.allocate(ctx, gsa.DeepCopy()) log.WithError(err).Info("allocate (private): failed allocation") require.NotEqual(t, ErrNoGameServer, err) - require.EqualError(t, err, ErrGameServerUpdateConflict.Error()) + require.ErrorIs(t, err, ErrGameServerUpdateConflict) // make sure we aren't in the same batch! time.Sleep(2 * a.batchWaitTime) @@ -492,9 +495,11 @@ func TestAllocatorAllocateOnGameServerUpdateError(t *testing.T) { // try the public method result, err := a.Allocate(ctx, gsa.DeepCopy()) log.WithField("result", result).WithError(err).Info("Allocate (public): failed allocation") - require.Nil(t, result) - require.NotEqual(t, ErrNoGameServer, err) - require.EqualError(t, err, ErrGameServerUpdateConflict.Error()) + require.NoError(t, err) + require.NotNil(t, result) + gsa2, ok := result.(*allocationv1.GameServerAllocation) + require.True(t, ok) + require.Equal(t, allocationv1.GameServerAllocationUnAllocated, gsa2.Status.State) } func TestAllocatorRunLocalAllocations(t *testing.T) { @@ -980,7 +985,8 @@ func TestControllerAllocationUpdateWorkers(t *testing.T) { r = <-r.request.response assert.True(t, updated) - assert.EqualError(t, r.err, ErrGameServerUpdateConflict.Error()) + assert.ErrorIs(t, r.err, ErrGameServerUpdateConflict) + assert.ErrorContains(t, r.err, "something went wrong") assert.Equal(t, gs1, r.gs) agtesting.AssertNoEvent(t, m.FakeRecorder.Events) @@ -1096,8 +1102,239 @@ func TestAllocatorCreateRestClientError(t *testing.T) { }) } -// newFakeAllocator returns a fake allocator. +// setFeatures enables the given feature gates for the duration of the test, then resets them. +func setFeatures(t *testing.T, features string) { + t.Helper() + runtime.FeatureTestMutex.Lock() + t.Cleanup(func() { + require.NoError(t, runtime.ParseFeatures("")) + runtime.FeatureTestMutex.Unlock() + }) + require.NoError(t, runtime.ParseFeatures(features)) +} + +// fleetGSA returns a GameServerAllocation selecting the GameServers of the given fleet +func fleetGSA(fleet string, edits ...func(*allocationv1.GameServerAllocation)) *allocationv1.GameServerAllocation { + gsa := &allocationv1.GameServerAllocation{ + ObjectMeta: metav1.ObjectMeta{Namespace: defaultNs}, + Spec: allocationv1.GameServerAllocationSpec{ + Selectors: []allocationv1.GameServerSelector{{ + LabelSelector: metav1.LabelSelector{MatchLabels: map[string]string{agonesv1.FleetNameLabel: fleet}}, + }}, + }, + } + for _, edit := range edits { + edit(gsa) + } + gsa.ApplyDefaults() + return gsa +} + +// runBatch serves gsList through a fake allocator, queues one request per gsa in the same batch, +// and returns their responses (in order) together with the number of GameServer updates made +func runBatch(t *testing.T, gsList []agonesv1.GameServer, updateErr error, gsas ...*allocationv1.GameServerAllocation) ([]response, int) { + t.Helper() + + a, m := newFakeAllocator() + m.AgonesClient.AddReactor("list", "gameservers", func(_ k8stesting.Action) (bool, k8sruntime.Object, error) { + return true, &agonesv1.GameServerList{Items: gsList}, nil + }) + updates := 0 + m.AgonesClient.AddReactor("update", "gameservers", func(action k8stesting.Action) (bool, k8sruntime.Object, error) { + updates++ + if updateErr != nil { + return true, nil, updateErr + } + return true, action.(k8stesting.UpdateAction).GetObject(), nil + }) + + ctx, cancel := agtesting.StartInformers(m, a.allocationCache.gameServerSynced) + t.Cleanup(cancel) + require.NoError(t, a.allocationCache.syncCache()) + require.NoError(t, a.allocationCache.counter.Run(ctx, 0)) + + requests := make([]request, len(gsas)) + for i, gsa := range gsas { + requests[i] = request{gsa: gsa, response: make(chan response, 1), ctx: context.Background()} + a.pendingRequests <- requests[i] + } + go a.ListenAndBatchAllocate(ctx, 3) + + responses := make([]response, len(requests)) + for i, r := range requests { + responses[i] = <-r.response + } + return responses, updates +} + +func TestAllocatorListenAndBatchAllocate(t *testing.T) { + setFeatures(t, string(runtime.FeatureProcessorAllocator)+"=true") + + t.Run("single allocation succeeds", func(t *testing.T) { + f, gsList := defaultFixtures(3) + + res, updates := runBatch(t, gsList, nil, fleetGSA(f.Name)) + + require.NoError(t, res[0].err) + assert.Equal(t, agonesv1.GameServerStateAllocated, res[0].gs.Status.State) + assert.Equal(t, 1, updates) + }) + + t.Run("multiple allocations in one batch are all allocated", func(t *testing.T) { + f, gsList := defaultFixtures(5) + + res, updates := runBatch(t, gsList, nil, fleetGSA(f.Name), fleetGSA(f.Name), fleetGSA(f.Name)) + + for _, r := range res { + require.NoError(t, r.err) + assert.Equal(t, agonesv1.GameServerStateAllocated, r.gs.Status.State) + } + assert.NotEqual(t, res[0].gs.Name, res[1].gs.Name) + // Packed strategy may consolidate all 3 onto one GS (batch collapsing) or spread them, + // either way at least 1 K8s update must have happened. + assert.GreaterOrEqual(t, updates, 1) + }) + + t.Run("k8s conflict is returned to the caller", func(t *testing.T) { + f, gsList := defaultFixtures(3) + conflict := k8serrors.NewConflict(agonesv1.Resource("gameservers"), "gs1", errors.New("conflict")) + + res, _ := runBatch(t, gsList, conflict, fleetGSA(f.Name)) + + assert.ErrorIs(t, res[0].err, ErrGameServerUpdateConflict) + }) + + t.Run("no game servers returns error", func(t *testing.T) { + res, _ := runBatch(t, nil, nil, fleetGSA("thereisnofleet")) + + assert.Nil(t, res[0].gs) + assert.ErrorIs(t, res[0].err, ErrNoGameServer) + }) +} + +func TestAllocatorListenAndBatchAllocateCountsAndLists(t *testing.T) { + setFeatures(t, string(runtime.FeatureProcessorAllocator)+"=true") + + allocated := agonesv1.GameServerStateAllocated + + selectAllocated := func(gsa *allocationv1.GameServerAllocation) { + gsa.Spec.Selectors[0].GameServerState = &allocated + } + // joinList returns an edit that adds value to the "players" list, only on GameServers with at + // least minAvailable room in it (if > 0). + joinList := func(value string, minAvailable int64) func(*allocationv1.GameServerAllocation) { + return func(gsa *allocationv1.GameServerAllocation) { + if minAvailable > 0 { + gsa.Spec.Selectors[0].Lists = map[string]allocationv1.ListSelector{"players": {MinAvailable: minAvailable}} + } + gsa.Spec.Lists = map[string]allocationv1.ListAction{"players": {AddValues: []string{value}}} + } + } + // incrementRooms is the same for the "rooms" counter, incrementing it by 1. + incrementRooms := func(minAvailable int64) func(*allocationv1.GameServerAllocation) { + action, one := "Increment", int64(1) + return func(gsa *allocationv1.GameServerAllocation) { + if minAvailable > 0 { + gsa.Spec.Selectors[0].Counters = map[string]allocationv1.CounterSelector{"rooms": {MinAvailable: minAvailable}} + } + gsa.Spec.Counters = map[string]allocationv1.CounterAction{"rooms": {Action: &action, Amount: &one}} + } + } + + // A single Allocated GameServer, with a "players" list holding alice out of a capacity of 2, + // and a "rooms" counter at 9 out of 10: both have room for exactly one more entry. + // Two requests land in the same batch, so the second one finds the GameServer full. + type want struct { + err error // expected error, nil if the request is allocated + players []string // expected "players" list values, if checked + rooms int64 // expected "rooms" counter, if checked + } + testCases := map[string]struct { + edits []func(*allocationv1.GameServerAllocation) // one per request + want []want + }{ + "list: a full list is skipped but the request is still allocated": { + edits: []func(*allocationv1.GameServerAllocation){joinList("bob", 0), joinList("carol", 0)}, + want: []want{{players: []string{"alice", "bob"}}, {players: []string{"alice", "bob"}}}, + }, + "list: a full list is rejected by a minAvailable selector": { + edits: []func(*allocationv1.GameServerAllocation){joinList("bob", 1), joinList("carol", 1)}, + want: []want{{players: []string{"alice", "bob"}}, {err: ErrNoGameServer}}, + }, + "counter: a full counter is skipped but the request is still allocated": { + edits: []func(*allocationv1.GameServerAllocation){incrementRooms(0), incrementRooms(0)}, + want: []want{{rooms: 10}, {rooms: 10}}, + }, + "counter: a full counter is rejected by a minAvailable selector": { + edits: []func(*allocationv1.GameServerAllocation){incrementRooms(1), incrementRooms(1)}, + want: []want{{rooms: 10}, {err: ErrNoGameServer}}, + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + f, gsList := defaultFixtures(1) + // Mark the first GameServer as allocated and set its initial list and counter status + gsList[0].Status.State = allocated + gsList[0].Status.Lists = map[string]agonesv1.ListStatus{"players": {Values: []string{"alice"}, Capacity: 2}} + gsList[0].Status.Counters = map[string]agonesv1.CounterStatus{"rooms": {Count: 9, Capacity: 10}} + + // Create GameServerAllocation requests based on the edits specified in the test case + gsas := make([]*allocationv1.GameServerAllocation, len(tc.edits)) + for i, edit := range tc.edits { + gsas[i] = fleetGSA(f.Name, selectAllocated, edit) + } + + // Run the batch allocation and capture the results and the number of updates made + res, updates := runBatch(t, gsList, nil, gsas...) + + // Iterate over the expected results and compare them with the actual allocation result + for i, w := range tc.want { + // Check if the allocation resulted in an error as expected + if w.err != nil { + assert.Nil(t, res[i].gs) + assert.ErrorIs(t, res[i].err, w.err) + continue + } + require.NoError(t, res[i].err) + // Check if the allocation resulted in the expected players status + if w.players != nil { + assert.ElementsMatch(t, w.players, res[i].gs.Status.Lists["players"].Values) + } + // Check if the allocation resulted in the expected rooms status + if w.rooms != 0 { + assert.Equal(t, w.rooms, res[i].gs.Status.Counters["rooms"].Count) + } + } + + // both requests share the one GameServer, so they are written with a single update + assert.Equal(t, 1, updates) + }) + } + + t.Run("a game server without room is skipped by a minAvailable selector", func(t *testing.T) { + f, gsList := defaultFixtures(2) + // Set up the initial players status for the GameServers + gsList[0].Status.Lists = map[string]agonesv1.ListStatus{"players": {Values: []string{"alice", "bob"}, Capacity: 2}} + gsList[1].Status.Lists = map[string]agonesv1.ListStatus{"players": {Values: []string{}, Capacity: 2}} + + // Run the batch allocation with a request to join the list "players" with "carol" + res, updates := runBatch(t, gsList, nil, fleetGSA(f.Name, joinList("carol", 1))) + + require.NoError(t, res[0].err) + assert.Equal(t, gsList[1].Name, res[0].gs.Name) + assert.Equal(t, []string{"carol"}, res[0].gs.Status.Lists["players"].Values) + assert.Equal(t, 1, updates) + }) +} + +// newFakeAllocator returns a fake allocator func newFakeAllocator() (*Allocator, agtesting.Mocks) { + return newFakeAllocatorWithCustomBatchWaitTime(500 * time.Millisecond) +} + +// newFakeAllocatorWithCustomBatchWaitTime returns a fake allocator with a configurable batchWaitTime +func newFakeAllocatorWithCustomBatchWaitTime(batchWaitTime time.Duration) (*Allocator, agtesting.Mocks) { m := agtesting.NewMocks() counter := gameservers.NewPerNodeCounter(m.KubeInformerFactory, m.AgonesInformerFactory) @@ -1109,9 +1346,62 @@ func newFakeAllocator() (*Allocator, agtesting.Mocks) { NewAllocationCache(m.AgonesInformerFactory.Agones().V1().GameServers(), counter, healthcheck.NewHandler()), time.Second, 5*time.Second, - 500*time.Millisecond, + batchWaitTime, defaultTestListMaxCapacity) a.recorder = m.FakeRecorder return a, m } + +func TestAllocatorAllocateNoQuickNoGameServerError(t *testing.T) { + setFeatures(t, string(runtime.FeatureProcessorAllocator)+"=true") + + a, m := newFakeAllocatorWithCustomBatchWaitTime(2 * time.Second) + + gs1 := agonesv1.GameServer{ObjectMeta: metav1.ObjectMeta{Name: "gs1", Namespace: defaultNs, UID: "1"}, + Status: agonesv1.GameServerStatus{NodeName: "node1", State: agonesv1.GameServerStateAllocated, + Counters: map[string]agonesv1.CounterStatus{"capacity": {Count: 1, Capacity: 1000}}}} + m.AgonesClient.AddReactor("list", "gameservers", func(_ k8stesting.Action) (bool, k8sruntime.Object, error) { + return true, &agonesv1.GameServerList{Items: []agonesv1.GameServer{gs1}}, nil + }) + m.AgonesClient.AddReactor("update", "gameservers", func(action k8stesting.Action) (bool, k8sruntime.Object, error) { + return true, action.(k8stesting.UpdateAction).GetObject(), nil + }) + + ctx, cancel := agtesting.StartInformers(m, a.allocationCache.gameServerSynced) + defer cancel() + + require.NoError(t, a.Run(ctx)) + require.Eventually(t, func() bool { + return a.allocationCache.cache.Len() == 1 + }, 10*time.Second, time.Second, "the GameServer should be in the cache") + + allocated := agonesv1.GameServerStateAllocated + gsa := &allocationv1.GameServerAllocation{ + ObjectMeta: metav1.ObjectMeta{Namespace: defaultNs}, + Spec: allocationv1.GameServerAllocationSpec{ + Scheduling: apis.Packed, + Selectors: []allocationv1.GameServerSelector{{ + GameServerState: &allocated, + Counters: map[string]allocationv1.CounterSelector{"capacity": {MinAvailable: 1}}, + }}, + }, + } + + // Two concurrent allocations must both land on the only GameServer, even though it is + // already taken by the first one in the same batch. + var wg sync.WaitGroup + for range 2 { + wg.Go(func() { + result, err := a.Allocate(ctx, gsa.DeepCopy()) + assert.NoError(t, err) + out, ok := result.(*allocationv1.GameServerAllocation) + if !assert.True(t, ok) { + return + } + assert.Equal(t, allocationv1.GameServerAllocationAllocated, out.Status.State) + assert.Equal(t, gs1.Name, out.Status.GameServerName) + }) + } + wg.Wait() +} diff --git a/pkg/gameserverallocations/batch_allocator.go b/pkg/gameserverallocations/batch_allocator.go new file mode 100644 index 0000000000..2a84f1af74 --- /dev/null +++ b/pkg/gameserverallocations/batch_allocator.go @@ -0,0 +1,327 @@ +// Copyright Contributors to Agones a Series of LF Projects, LLC. +// +// 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. + +package gameserverallocations + +import ( + "context" + goErrors "errors" + "maps" + "slices" + "time" + + "agones.dev/agones/pkg/apis" + agonesv1 "agones.dev/agones/pkg/apis/agones/v1" + allocationv1 "agones.dev/agones/pkg/apis/allocation/v1" + corev1 "k8s.io/api/core/v1" + k8serrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" +) + +// Batch allocation flow +// +// requests (c.pendingRequests) +// │ +// ▼ +// ┌───────────────────────────────┐ reads ┌──────────────────────┐ +// │ LISTENER (single goroutine) │◀─────────│ ALLOCATION CACHE │ +// │ ListenAndBatchAllocate │ │ sorted GameServers │ +// │ 1. sortedGameServers │ └──────────▲───────────┘ +// │ 2. findGameServerForBatch... │ │ +// │ 3. plan on a local copy │ │ AddGameServer +// │ 4. keep the list sorted │ │ (or refresh on conflict) +// └───────┬───────────────┬───────┘ │ +// │ pending.add │ submit() when idle, │ +// │ (key, req) │ or after N requests │ +// ▼ ▼ │ +// ┌──────────────┐ ┌───────────┐ Get ┌─────────────┴─────────────────┐ +// │ PENDING │ │ WORKQUEUE │───────▶│ UPDATE WORKERS (N goroutines) │ +// │ key → reqs │ │ one key │ │ runUpdateWorker │ +// │ + projection │◀─│ at a time │ │ take(key) → updateGameServer │ +// └──────────────┘ └───────────┘ │ applyRequests on cached GS │ +// take(key) │ ONE Update() for all of them │ +// └───────────────┬───────────────┘ +// │ +// ▼ +// req.response <- GameServer or error +// (conflict: error, the caller retries) +// +// Requests planned on the same GameServer while a worker updates it are merged into its next update. + +// ListenAndBatchAllocate listens for incoming allocation requests and processes them in batches +func (c *Allocator) ListenAndBatchAllocate(ctx context.Context, updateWorkerCount int) { + pending := newPendingAllocations() + queue := c.batchAllocationUpdateWorkers(ctx, updateWorkerCount, pending) + + // candidates are the sorted GameServers the requests are planned onto. They are refreshed from the cache + // when the loop was idle, after maxBatchBeforeRefresh requests, or when the allocation spec changes + var candidates []*agonesv1.GameServer + // candidatesSortKey is used to determine if the candidates need to be refreshed based on the allocation request's sort key + var candidatesSortKey uint64 + // plannedKeys are the GameServers planned onto since the candidates were refreshed, handed to the update workers + // all at once, so that each gets a single update for all of them (duplicates are collapsed by the queue) + var plannedKeys []string + + // submit hands the planned GameServers to the update workers and clears the plannedKeys slice + submit := func() { + for _, key := range plannedKeys { + queue.Add(key) + } + plannedKeys = plannedKeys[:0] + } + + for { + select { + case req := <-c.pendingRequests: + if req.ctx.Err() != nil { + c.tryRespondWithError(req, ErrTotalTimeoutExceeded) + continue + } + + // determine if we need to refresh the candidates based on the sort key, the number of planned requests, or if there are no candidates yet + reqSortKey := c.batchSortKey(req.gsa) + if candidates == nil || len(plannedKeys) >= maxBatchBeforeRefresh || reqSortKey != candidatesSortKey { + submit() + candidates = c.sortedGameServers(req.gsa, pending) + candidatesSortKey = reqSortKey + } + + // find a suitable GameServer for the current allocation request + gs, index, err := findGameServerForAllocation(req.gsa, candidates) + if err != nil { + req.response <- response{request: req, gs: nil, err: err} + continue + } + + // make a deep copy of the GameServer to apply the allocation locally + allocated := gs.DeepCopy() + applyErr, _, _ := c.applyAllocationToLocalGameServer(req.gsa.Spec.MetaPatch, allocated, req.gsa) + if applyErr != nil { + req.response <- response{request: req, gs: nil, err: applyErr} + continue + } + + // add the planned GameServer to the pending allocations and record its key for the update workers + key, _ := cache.MetaNamespaceKeyFunc(gs) + pending.add(key, req, allocated) + plannedKeys = append(plannedKeys, key) + + // reorder the candidates after applying the allocation to keep them sorted for the next request + c.allocationCache.ReorderGameServerAfterAllocation(candidates, index, allocated, req.gsa.Spec.Priorities, req.gsa.Spec.Scheduling) + + case <-ctx.Done(): + return + + default: + // idle: hand the planned GameServers to the update workers, clear the candidates, and wait for the next batch + submit() + candidates = nil + time.Sleep(c.batchWaitTime) + } + } +} + +// batchAllocationUpdateWorkers starts the specified number of worker goroutines that process updates for the planned GameServers +func (c *Allocator) batchAllocationUpdateWorkers(ctx context.Context, workerCount int, pending *pendingAllocations) workqueue.TypedInterface[string] { + queue := workqueue.NewTyped[string]() + go func() { + <-ctx.Done() + queue.ShutDown() + }() + + for range workerCount { + // start a new update worker goroutine + go c.runUpdateWorker(ctx, queue, pending) + } + + return queue +} + +// runUpdateWorker continuously processes keys from the queue, applying the pending requests to the corresponding GameServers until the queue is shut down +func (c *Allocator) runUpdateWorker(ctx context.Context, queue workqueue.TypedInterface[string], pending *pendingAllocations) { + for { + // get the next key from the queue, along with a flag indicating if the queue is shutting down + key, shutdown := queue.Get() + if shutdown { + return + } + + // process the next key from the queue by applying the pending requests to the corresponding GameServer + c.updateGameServer(ctx, key, pending.take(key)) + // mark the pending requests for this key as finished and signal the queue that processing is done + pending.finish(key) + // signal the queue that processing for this key is done + queue.Done(key) + } +} + +// updateGameServer applies the pending requests to the cached GameServer identified by the key and persists the changes +// If the update fails due to a version conflict, the cache is refreshed with the live GameServer, and the requests receive an error, making them available for retries +func (c *Allocator) updateGameServer(ctx context.Context, key string, reqs []request) { + // retrieve the cached GameServer by its key from the allocation cache + gs, ok := c.allocationCache.GetGameServer(key) + if !ok { + for _, req := range reqs { + req.response <- response{request: req, gs: nil, err: ErrNoGameServer} + } + return + } + + // apply the pending requests to a copy of the cached GameServer and collect the results + toUpdate, applied, counterErrors, listErrors := c.applyRequests(gs, reqs) + + // if no requests were successfully applied, there is nothing to update + if len(applied) == 0 { + return + } + + // attempt to persist the changes to the GameServer in the API server + updatedGs, err := c.gameServerGetter.GameServers(toUpdate.ObjectMeta.Namespace).Update(ctx, toUpdate, metav1.UpdateOptions{}) + switch { + // update succeeded without errors + case err == nil: + // update the cache with the successfully updated GameServer + c.allocationCache.AddGameServer(updatedGs) + c.recordAllocated(updatedGs, counterErrors, listErrors) + + // update failed due to a version conflict + case k8serrors.IsConflict(err): + // refresh the cache with the live GameServer if it can be retrieved + live, getErr := c.gameServerGetter.GameServers(toUpdate.ObjectMeta.Namespace).Get(ctx, toUpdate.ObjectMeta.Name, metav1.GetOptions{}) + if getErr == nil { + // update the cache with the live GameServer + c.allocationCache.refreshGameServer(live) + } + err = goErrors.Join(ErrGameServerUpdateConflict, err) + } + + // notify all successfully applied requests of the result, including any errors encountered during the update + for _, res := range applied { + res.err = err + res.request.response <- res + } +} + +// applyRequests applies each allocation request to a copy of the given GameServer +// It returns the updated GameServer, a list of successfully applied responses, and any errors encountered while applying counters or lists +func (c *Allocator) applyRequests(gs *agonesv1.GameServer, reqs []request) (toUpdate *agonesv1.GameServer, applied []response, counterErrors, listErrors error) { + toUpdate = gs.DeepCopy() + for _, req := range reqs { + switch { + // check if the request context has expired before applying it + case req.ctx.Err() != nil: + req.response <- response{request: req, gs: nil, err: ErrTotalTimeoutExceeded} + + // the GameServer is no longer allocatable, so the request is rejected and can be retried + case toUpdate.IsBeingDeleted() || !readyOrAllocatedGameServerMatcher(toUpdate): + req.response <- response{request: req, gs: nil, err: ErrConflictInGameServerSelection} + + // apply the request, counter and list action errors do not prevent the allocation + default: + applyErr, cErr, lErr := c.applyAllocationToLocalGameServer(req.gsa.Spec.MetaPatch, toUpdate, req.gsa) + if applyErr != nil { + req.response <- response{request: req, gs: nil, err: applyErr} + continue + } + counterErrors = goErrors.Join(counterErrors, cErr) + listErrors = goErrors.Join(listErrors, lErr) + applied = append(applied, response{request: req, gs: toUpdate.DeepCopy(), err: nil}) + } + } + + return toUpdate, applied, counterErrors, listErrors +} + +// applyAllocationToLocalGameServer patches the GameServer with allocation metadata and sets +// it to Allocated state without persisting to Kubernetes. Counter/List actions are applied +func (c *Allocator) applyAllocationToLocalGameServer(mp allocationv1.MetaPatch, gs *agonesv1.GameServer, gsa *allocationv1.GameServerAllocation) (applyErr, counterErrors, listErrors error) { + ts, err := time.Now().MarshalText() + if err != nil { + return err, nil, nil + } + if gs.ObjectMeta.Annotations == nil { + gs.ObjectMeta.Annotations = make(map[string]string, 1+len(mp.Annotations)) + } + gs.ObjectMeta.Annotations[LastAllocatedAnnotationKey] = string(ts) + gs.Status.State = agonesv1.GameServerStateAllocated + + if mp.Labels != nil { + if gs.ObjectMeta.Labels == nil { + gs.ObjectMeta.Labels = make(map[string]string, len(mp.Labels)) + } + maps.Copy(gs.ObjectMeta.Labels, mp.Labels) + } + + maps.Copy(gs.ObjectMeta.Annotations, mp.Annotations) + + for counter, ca := range gsa.Spec.Counters { + counterErrors = goErrors.Join(counterErrors, ca.CounterActions(counter, gs)) + } + for list, la := range gsa.Spec.Lists { + listErrors = goErrors.Join(listErrors, la.ListActions(list, gs, c.listMaxCapacity)) + } + + return nil, counterErrors, listErrors +} + +// recordAllocated records the Allocated event of the GameServer, and any Counter or List action errors as warnings +func (c *Allocator) recordAllocated(gs *agonesv1.GameServer, counterErrors, listErrors error) { + if counterErrors != nil { + c.recorder.Event(gs, corev1.EventTypeWarning, "CounterActionError", counterErrors.Error()) + } + if listErrors != nil { + c.recorder.Event(gs, corev1.EventTypeWarning, "ListActionError", listErrors.Error()) + } + c.recorder.Event(gs, corev1.EventTypeNormal, string(gs.Status.State), "Allocated") +} + +// batchSortKey returns the sort key for the given GameServerAllocation +func (c *Allocator) batchSortKey(gsa *allocationv1.GameServerAllocation) uint64 { + // Get the sort key for the GameServerAllocation, which determines the order in which it should be considered for allocation + sortKey, err := gsa.SortKey() + if err != nil { + c.baseLogger.WithError(err).Warn("error getting sortKey for GameServerAllocationSpec") + } + + return sortKey +} + +// sortedGameServers returns the list of GameServers sorted according to the allocation strategy, with pending allocations taken into account +func (c *Allocator) sortedGameServers(gsa *allocationv1.GameServerAllocation, pending *pendingAllocations) []*agonesv1.GameServer { + var list []*agonesv1.GameServer + if gsa.Spec.Scheduling == apis.Packed { + list = c.allocationCache.ListSortedGameServers(gsa) + } else { + list = c.allocationCache.ListSortedGameServersPriorities(gsa) + } + + for _, projected := range pending.projections() { + // find the index of the projected GameServer in the sorted list + i := slices.IndexFunc(list, func(gs *agonesv1.GameServer) bool { + return gs.ObjectMeta.Name == projected.ObjectMeta.Name && gs.ObjectMeta.Namespace == projected.ObjectMeta.Namespace + }) + + // if the projected GameServer is not found in the list, skip it + if i < 0 { + continue + } + + // reorder the GameServer in the list based on the allocation projection + c.allocationCache.ReorderGameServerAfterAllocation(list, i, projected, gsa.Spec.Priorities, gsa.Spec.Scheduling) + } + + return list +} diff --git a/pkg/gameserverallocations/find.go b/pkg/gameserverallocations/find.go index 0c5b96f1b5..7115f28238 100644 --- a/pkg/gameserverallocations/find.go +++ b/pkg/gameserverallocations/find.go @@ -79,6 +79,10 @@ func findGameServerForAllocation(gsa *allocationv1.GameServerAllocation, list [] } loop(list, func(i int, gs *agonesv1.GameServer) { + if gs == nil { + return + } + // only search the same namespace if gs.ObjectMeta.Namespace != gsa.ObjectMeta.Namespace { return diff --git a/pkg/gameserverallocations/pending_allocations.go b/pkg/gameserverallocations/pending_allocations.go new file mode 100644 index 0000000000..801c730153 --- /dev/null +++ b/pkg/gameserverallocations/pending_allocations.go @@ -0,0 +1,91 @@ +// Copyright Contributors to Agones a Series of LF Projects, LLC. +// +// 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. + +package gameserverallocations + +import ( + "sync" + + agonesv1 "agones.dev/agones/pkg/apis/agones/v1" +) + +// pendingAllocations is keyed by GameServer. The workqueue only carries the keys, and each worker gets the +// requests of its key from here. A key is handed to one worker at a time +// +// key reqs(key) projected(key) workqueue +// gs-1 r1 r2 r3 gs-1 after r1 r2 r3 gs-1 -> worker A: take(gs-1) -> r1 r2 r3 +// gs-2 r4 gs-2 after r4 gs-2 -> worker B: take(gs-2) -> r4 +// +// add(key, req, projected) appends to reqs[key] and replaces projected[key] +// take(key) returns reqs[key] and clears it +// finish(key) drops projected[key], unless add() was called again since take() +// projections() returns every projected[key] for all GameServers with pending or in progress allocations + +// pendingAllocations manages the requests and projected GameServer states for the batch allocator +type pendingAllocations struct { + reqs map[string][]request // pending requests for each GameServer key + projected map[string]*agonesv1.GameServer // projected GameServer state for each GameServer key + mu sync.Mutex +} + +// newPendingAllocations creates and returns a new instance of pendingAllocations +func newPendingAllocations() *pendingAllocations { + return &pendingAllocations{ + reqs: make(map[string][]request), + projected: make(map[string]*agonesv1.GameServer), + } +} + +// add records a request planned onto the GameServer, and its projected version +func (p *pendingAllocations) add(key string, req request, projected *agonesv1.GameServer) { + p.mu.Lock() + defer p.mu.Unlock() + + p.reqs[key] = append(p.reqs[key], req) + p.projected[key] = projected +} + +// take returns and clears the requests pending on the GameServer +func (p *pendingAllocations) take(key string) []request { + p.mu.Lock() + defer p.mu.Unlock() + + reqs := p.reqs[key] + delete(p.reqs, key) + + return reqs +} + +// finish drops the projected version of the GameServer once it is updated, unless requests were planned onto it in the meantime +func (p *pendingAllocations) finish(key string) { + p.mu.Lock() + defer p.mu.Unlock() + + if len(p.reqs[key]) == 0 { + delete(p.projected, key) + } +} + +// projections returns a list of the projected GameServer states for all GameServers with pending or in progress allocations +func (p *pendingAllocations) projections() []*agonesv1.GameServer { + p.mu.Lock() + defer p.mu.Unlock() + + list := make([]*agonesv1.GameServer, 0, len(p.projected)) + for _, gs := range p.projected { + list = append(list, gs) + } + + return list +} diff --git a/pkg/gameserverallocations/pending_allocations_test.go b/pkg/gameserverallocations/pending_allocations_test.go new file mode 100644 index 0000000000..c5c381c41b --- /dev/null +++ b/pkg/gameserverallocations/pending_allocations_test.go @@ -0,0 +1,91 @@ +// Copyright Contributors to Agones a Series of LF Projects, LLC. +// +// 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. + +package gameserverallocations + +import ( + "testing" + + agonesv1 "agones.dev/agones/pkg/apis/agones/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestPendingAllocations(t *testing.T) { + t.Parallel() + + newGs := func(resourceVersion string) *agonesv1.GameServer { + return &agonesv1.GameServer{ObjectMeta: metav1.ObjectMeta{Name: "gs1", Namespace: defaultNs, ResourceVersion: resourceVersion}} + } + key := defaultNs + "/gs1" + + t.Run("take returns the requests in planning order and keeps the last projection", func(t *testing.T) { + t.Parallel() + p := newPendingAllocations() + r1 := request{response: make(chan response, 1)} + r2 := request{response: make(chan response, 1)} + + p.add(key, r1, newGs("1")) + p.add(key, r2, newGs("2")) + + projections := p.projections() + require.Len(t, projections, 1) + assert.Equal(t, "2", projections[0].ObjectMeta.ResourceVersion) + + reqs := p.take(key) + require.Len(t, reqs, 2) + assert.Equal(t, r1.response, reqs[0].response) + assert.Equal(t, r2.response, reqs[1].response) + assert.Empty(t, p.take(key)) + + // the projection stays visible to the planner while the update is in progress + assert.Len(t, p.projections(), 1) + }) + + t.Run("finish drops the projection once nothing is pending", func(t *testing.T) { + t.Parallel() + p := newPendingAllocations() + p.add(key, request{}, newGs("1")) + + p.take(key) + p.finish(key) + + assert.Empty(t, p.projections()) + }) + + t.Run("finish keeps the projection of requests planned during the update", func(t *testing.T) { + t.Parallel() + p := newPendingAllocations() + p.add(key, request{}, newGs("1")) + + p.take(key) + p.add(key, request{}, newGs("2")) + p.finish(key) + + projections := p.projections() + require.Len(t, projections, 1) + assert.Equal(t, "2", projections[0].ObjectMeta.ResourceVersion) + assert.Len(t, p.take(key), 1) + }) + + t.Run("take of an unknown key returns nothing", func(t *testing.T) { + t.Parallel() + p := newPendingAllocations() + + assert.Empty(t, p.take(key)) + p.finish(key) + assert.Empty(t, p.projections()) + }) +}