Fleet Aggregate Player Tracking Logic (#1561)

Implementation of aggregation of Player Tracking values at the Fleet and
GameServerSet levels.

Includes both unit and e2e tests.

Work on #1033

Co-authored-by: Robert Bailey <[email protected]>

pkg/fleets/controller.go

@@ -531,6 +531,19 @@ func (c *Controller) updateFleetStatus(fleet *agonesv1.Fleet) error {
 		fCopy.Status.ReservedReplicas += gsSet.Status.ReservedReplicas
 		fCopy.Status.AllocatedReplicas += gsSet.Status.AllocatedReplicas
 	}
+	if runtime.FeatureEnabled(runtime.FeaturePlayerTracking) {
+		// to make this code simpler, while the feature gate is in place,
+		// we will loop around the gsSet list twice.
+		fCopy.Status.Players = &agonesv1.AggregatedPlayerStatus{}
+		// TODO: integrate this extra loop into the above for loop when PlayerTracking moves to GA
+		for _, gsSet := range list {
+			if gsSet.Status.Players != nil {
+				fCopy.Status.Players.Count += gsSet.Status.Players.Count
+				fCopy.Status.Players.Capacity += gsSet.Status.Players.Capacity
+			}
+		}
+	}
+
 	_, err = c.fleetGetter.Fleets(fCopy.ObjectMeta.Namespace).UpdateStatus(fCopy)
 	return errors.Wrapf(err, "error updating status of fleet %s", fCopy.ObjectMeta.Name)
 }

pkg/fleets/controller_test.go

@@ -24,6 +24,7 @@ import (
 	"agones.dev/agones/pkg/apis"
 	agonesv1 "agones.dev/agones/pkg/apis/agones/v1"
 	agtesting "agones.dev/agones/pkg/testing"
+	utilruntime "agones.dev/agones/pkg/util/runtime"
 	"agones.dev/agones/pkg/util/webhooks"
 	"github.com/heptiolabs/healthcheck"
 	"github.com/mattbaird/jsonpatch"
@@ -404,6 +405,57 @@ func TestControllerUpdateFleetStatus(t *testing.T) {
 	assert.True(t, updated)
 }
 
+func TestControllerUpdateFleetPlayerStatus(t *testing.T) {
+	t.Parallel()
+
+	utilruntime.FeatureTestMutex.Lock()
+	defer utilruntime.FeatureTestMutex.Unlock()
+
+	assert.NoError(t, utilruntime.ParseFeatures(string(utilruntime.FeaturePlayerTracking)+"=true"))
+
+	fleet := defaultFixture()
+	c, m := newFakeController()
+
+	gsSet1 := fleet.GameServerSet()
+	gsSet1.ObjectMeta.Name = "gsSet1"
+	gsSet1.Status.Players = &agonesv1.AggregatedPlayerStatus{
+		Count:    5,
+		Capacity: 10,
+	}
+
+	gsSet2 := fleet.GameServerSet()
+	gsSet2.ObjectMeta.Name = "gsSet2"
+	gsSet2.Status.Players = &agonesv1.AggregatedPlayerStatus{
+		Count:    10,
+		Capacity: 20,
+	}
+
+	m.AgonesClient.AddReactor("list", "gameserversets",
+		func(action k8stesting.Action) (bool, runtime.Object, error) {
+			return true, &agonesv1.GameServerSetList{Items: []agonesv1.GameServerSet{*gsSet1, *gsSet2}}, nil
+		})
+
+	updated := false
+	m.AgonesClient.AddReactor("update", "fleets",
+		func(action k8stesting.Action) (bool, runtime.Object, error) {
+			updated = true
+			ua := action.(k8stesting.UpdateAction)
+			fleet := ua.GetObject().(*agonesv1.Fleet)
+
+			assert.Equal(t, gsSet1.Status.Players.Count+gsSet2.Status.Players.Count, fleet.Status.Players.Count)
+			assert.Equal(t, gsSet1.Status.Players.Capacity+gsSet2.Status.Players.Capacity, fleet.Status.Players.Capacity)
+
+			return true, fleet, nil
+		})
+
+	_, cancel := agtesting.StartInformers(m, c.fleetSynced, c.gameServerSetSynced)
+	defer cancel()
+
+	err := c.updateFleetStatus(fleet)
+	assert.Nil(t, err)
+	assert.True(t, updated)
+}
+
 func TestControllerFilterGameServerSetByActive(t *testing.T) {
 	t.Parallel()
 

pkg/gameserversets/controller.go

@@ -606,5 +606,23 @@ func computeStatus(list []*agonesv1.GameServer) agonesv1.GameServerSetStatus {
 		}
 	}
 
+	if runtime.FeatureEnabled(runtime.FeaturePlayerTracking) {
+		// to make this code simpler, while the feature gate is in place,
+		// we will loop around the gs list twice.
+		status.Players = &agonesv1.AggregatedPlayerStatus{}
+		// TODO: integrate this extra loop into the above for loop when PlayerTracking moves to GA
+		for _, gs := range list {
+			if gs.ObjectMeta.DeletionTimestamp.IsZero() &&
+				(gs.Status.State == agonesv1.GameServerStateReady ||
+					gs.Status.State == agonesv1.GameServerStateReserved ||
+					gs.Status.State == agonesv1.GameServerStateAllocated) {
+				if gs.Status.Players != nil {
+					status.Players.Capacity += gs.Status.Players.Capacity
+					status.Players.Count += gs.Status.Players.Count
+				}
+			}
+		}
+	}
+
 	return status
 }

pkg/gameserversets/controller_test.go

@@ -25,6 +25,7 @@ import (
 	agonesv1 "agones.dev/agones/pkg/apis/agones/v1"
 	"agones.dev/agones/pkg/gameservers"
 	agtesting "agones.dev/agones/pkg/testing"
+	utilruntime "agones.dev/agones/pkg/util/runtime"
 	"agones.dev/agones/pkg/util/webhooks"
 	"github.com/heptiolabs/healthcheck"
 	"github.com/sirupsen/logrus"
@@ -265,6 +266,8 @@ func TestComputeReconciliationAction(t *testing.T) {
 }
 
 func TestComputeStatus(t *testing.T) {
+	t.Parallel()
+
 	cases := []struct {
 		list       []*agonesv1.GameServer
 		wantStatus agonesv1.GameServerSetStatus
@@ -293,6 +296,37 @@ func TestComputeStatus(t *testing.T) {
 	for _, tc := range cases {
 		assert.Equal(t, tc.wantStatus, computeStatus(tc.list))
 	}
+
+	t.Run("player tracking", func(t *testing.T) {
+		utilruntime.FeatureTestMutex.Lock()
+		defer utilruntime.FeatureTestMutex.Unlock()
+
+		assert.NoError(t, utilruntime.ParseFeatures(string(utilruntime.FeaturePlayerTracking)+"=true"))
+
+		var list []*agonesv1.GameServer
+		gs1 := gsWithState(agonesv1.GameServerStateAllocated)
+		gs1.Status.Players = &agonesv1.PlayerStatus{Count: 5, Capacity: 10}
+		gs2 := gsWithState(agonesv1.GameServerStateReserved)
+		gs2.Status.Players = &agonesv1.PlayerStatus{Count: 10, Capacity: 15}
+		gs3 := gsWithState(agonesv1.GameServerStateCreating)
+		gs3.Status.Players = &agonesv1.PlayerStatus{Count: 20, Capacity: 30}
+		gs4 := gsWithState(agonesv1.GameServerStateReady)
+		gs4.Status.Players = &agonesv1.PlayerStatus{Count: 15, Capacity: 30}
+		list = append(list, gs1, gs2, gs3, gs4)
+
+		expected := agonesv1.GameServerSetStatus{
+			Replicas:          4,
+			ReadyReplicas:     1,
+			ReservedReplicas:  1,
+			AllocatedReplicas: 1,
+			Players: &agonesv1.AggregatedPlayerStatus{
+				Count:    30,
+				Capacity: 55,
+			},
+		}
+
+		assert.Equal(t, expected, computeStatus(list))
+	})
 }
 
 func TestControllerWatchGameServers(t *testing.T) {

test/e2e/fleet_test.go

@@ -40,6 +40,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/retry"
@@ -1216,6 +1217,69 @@ func TestFleetResourceValidation(t *testing.T) {
 	}
 }
 
+func TestFleetAggregatedPlayerStatus(t *testing.T) {
+	if !runtime.FeatureEnabled(runtime.FeaturePlayerTracking) {
+		t.SkipNow()
+	}
+	t.Parallel()
+	client := framework.AgonesClient.AgonesV1()
+
+	flt := defaultFleet(defaultNs)
+	flt.Spec.Template.Spec.Players = &agonesv1.PlayersSpec{
+		InitialCapacity: 10,
+	}
+
+	flt, err := client.Fleets(defaultNs).Create(flt.DeepCopy())
+	assert.NoError(t, err)
+
+	framework.AssertFleetCondition(t, flt, func(fleet *agonesv1.Fleet) bool {
+		if fleet.Status.Players == nil {
+			logrus.WithField("status", fleet.Status).Info("No Players")
+			return false
+		}
+
+		logrus.WithField("status", fleet.Status).Info("Checking Capacity")
+		return fleet.Status.Players.Capacity == 30
+	})
+
+	list, err := framework.ListGameServersFromFleet(flt)
+	assert.NoError(t, err)
+	// set 3 random capacities, and connect a random number of players
+	totalCapacity := 0
+	totalPlayers := 0
+	for i := range list {
+		// Do this, otherwise scopelint complains about "using a reference for the variable on range scope"
+		gs := &list[i]
+		capacity := rand.IntnRange(1, 100)
+		totalCapacity += capacity
+
+		msg := fmt.Sprintf("PLAYER_CAPACITY %d", capacity)
+		reply, err := e2e.SendGameServerUDP(gs, msg)
+		if err != nil {
+			t.Fatalf("Could not message GameServer: %v", err)
+		}
+		assert.Equal(t, fmt.Sprintf("ACK: %s\n", msg), reply)
+
+		players := rand.IntnRange(1, 5)
+		totalPlayers += players
+		for i := 1; i <= players; i++ {
+			msg := "PLAYER_CONNECT " + fmt.Sprintf("%d", i)
+			logrus.WithField("msg", msg).WithField("gs", gs.ObjectMeta.Name).Info("Sending Player Connect")
+			reply, err := e2e.SendGameServerUDP(gs, msg)
+			if err != nil {
+				t.Fatalf("Could not message GameServer: %v", err)
+			}
+			assert.Equal(t, fmt.Sprintf("ACK: %s\n", msg), reply)
+		}
+	}
+
+	framework.AssertFleetCondition(t, flt, func(fleet *agonesv1.Fleet) bool {
+		logrus.WithField("players", fleet.Status.Players).WithField("totalCapacity", totalCapacity).
+			WithField("totalPlayers", totalPlayers).Info("Checking Capacity")
+		return (fleet.Status.Players.Capacity == int64(totalCapacity)) && (fleet.Status.Players.Count == int64(totalPlayers))
+	})
+}
+
 func assertCausesContainsString(t *testing.T, causes []metav1.StatusCause, expected string) {
 	found := false
 	for _, v := range causes {