Merge pull request #309 from topfreegames/chore/improve-watcher-event-processing

chore(runtime watcher, rooms adapters): troubleshooting improvements

Author: caiordjesus

config/rooms-api.local.yaml

@@ -1,12 +1,12 @@
 internalApi:
-  port: 8081
+  port: 8091
   gracefulShutdownTimeout: 10s
   metrics:
     enabled: true
   healthcheck:
     enabled: true
 roomsApi:
-  port: 8080
+  port: 8090
   gracefulShutdownTimeout: 10s
 adapters:
   portAllocator:

config/runtime-watcher.local.yaml

@@ -1,5 +1,5 @@
 internalApi:
-  port: 8081
+  port: 8083
   gracefulShutdownTimeout: 10s
   metrics:
     enabled: true

config/worker.local.yaml

@@ -1,5 +1,5 @@
 internalApi:
-  port: 8081
+  port: 8082
   gracefulShutdownTimeout: 10s
   metrics:
     enabled: true

internal/adapters/instance_storage/redis/redis.go

@@ -60,6 +60,9 @@ func (r redisInstanceStorage) GetInstance(ctx context.Context, scheduler string,
 }
 
 func (r redisInstanceStorage) UpsertInstance(ctx context.Context, instance *game_room.Instance) error {
+	if instance == nil {
+		return errors.NewErrUnexpected("Cannot upsert nil instance")
+	}
 	instanceJson, err := json.Marshal(instance)
 	if err != nil {
 		return errors.NewErrUnexpected("error marshalling room %s json", instance.ID).WithError(err)

internal/adapters/instance_storage/redis/redis_test.go

@@ -62,31 +62,38 @@ func assertInstanceRedis(t *testing.T, client *redis.Client, expectedInstance *g
 func TestRedisInstanceStorage_UpsertInstance(t *testing.T) {
 	client := test.GetRedisConnection(t, redisAddress)
 	storage := NewRedisInstanceStorage(client, 0)
-	instance := &game_room.Instance{
-		ID:          "1",
-		SchedulerID: "game",
-		Status: game_room.InstanceStatus{
-			Type: game_room.InstancePending,
-		},
-	}
+	t.Run("should succeed", func(t *testing.T) {
+		instance := &game_room.Instance{
+			ID:          "1",
+			SchedulerID: "game",
+			Status: game_room.InstanceStatus{
+				Type: game_room.InstancePending,
+			},
+		}
+
+		require.NoError(t, storage.UpsertInstance(context.Background(), instance))
+		assertInstanceRedis(t, client, instance)
 
-	require.NoError(t, storage.UpsertInstance(context.Background(), instance))
-	assertInstanceRedis(t, client, instance)
-
-	instance.Status.Type = game_room.InstanceReady
-	instance.Address = &game_room.Address{
-		Host: "host",
-		Ports: []game_room.Port{
-			{
-				Name:     "game",
-				Port:     7000,
-				Protocol: "udp",
+		instance.Status.Type = game_room.InstanceReady
+		instance.Address = &game_room.Address{
+			Host: "host",
+			Ports: []game_room.Port{
+				{
+					Name:     "game",
+					Port:     7000,
+					Protocol: "udp",
+				},
 			},
-		},
-	}
+		}
 
-	require.NoError(t, storage.UpsertInstance(context.Background(), instance))
-	assertInstanceRedis(t, client, instance)
+		require.NoError(t, storage.UpsertInstance(context.Background(), instance))
+		assertInstanceRedis(t, client, instance)
+	})
+
+	t.Run("should fail - instance is nil", func(t *testing.T) {
+		err := storage.UpsertInstance(context.Background(), nil)
+		require.Error(t, err)
+	})
 }
 
 func TestRedisInstanceStorage_GetInstance(t *testing.T) {

internal/core/entities/game_room/game_room.go

@@ -145,6 +145,7 @@ var validStatusTransitions = map[GameRoomStatus]map[GameRoomStatus]struct{}{
 		GameStatusTerminating: struct{}{},
 		GameStatusReady:       struct{}{},
 		GameStatusError:       struct{}{},
+		GameStatusPending:     struct{}{},
 	},
 	GameStatusOccupied: {
 		GameStatusReady:       struct{}{},
@@ -183,6 +184,7 @@ var roomStatusComposition = []struct {
 
 	// Unready
 	{GameRoomPingStatusUnknown, InstanceReady, GameStatusUnready},
+	{GameRoomPingStatusUnknown, InstancePending, GameStatusUnready},
 
 	// Terminating
 	{GameRoomPingStatusUnknown, InstanceTerminating, GameStatusTerminating},

internal/core/services/room_manager/room_manager.go

@@ -172,6 +172,10 @@ func (m *RoomManager) UpdateRoom(ctx context.Context, gameRoom *game_room.GameRo
 }
 
 func (m *RoomManager) UpdateRoomInstance(ctx context.Context, gameRoomInstance *game_room.Instance) error {
+	if gameRoomInstance == nil {
+		return fmt.Errorf("cannot update room instance since it is nil")
+	}
+	m.Logger.Sugar().Infof("Updating room instance. ID: %v", gameRoomInstance.ID)
 	err := m.InstanceStorage.UpsertInstance(ctx, gameRoomInstance)
 	if err != nil {
 		return fmt.Errorf("failed when updating the game room instance on storage: %w", err)
@@ -182,10 +186,12 @@ func (m *RoomManager) UpdateRoomInstance(ctx context.Context, gameRoomInstance *
 		return fmt.Errorf("failed to update game room status: %w", err)
 	}
 
+	m.Logger.Info("Updating room success")
 	return nil
 }
 
 func (m *RoomManager) CleanRoomState(ctx context.Context, schedulerName, roomId string) error {
+	m.Logger.Sugar().Infof("Cleaning room \"%v\", scheduler \"%v\"", roomId, schedulerName)
 	err := m.RoomStorage.DeleteRoom(ctx, schedulerName, roomId)
 	if err != nil && !errors.Is(porterrors.ErrNotFound, err) {
 		return fmt.Errorf("failed to delete room state: %w", err)
@@ -196,6 +202,7 @@ func (m *RoomManager) CleanRoomState(ctx context.Context, schedulerName, roomId
 		return fmt.Errorf("failed to delete room state: %w", err)
 	}
 
+	m.Logger.Info("cleaning room success")
 	return nil
 }
 

internal/core/services/room_manager/room_manager_test.go

@@ -557,6 +557,11 @@ func TestRoomManager_UpdateRoomInstance(t *testing.T) {
 		err := roomManager.UpdateRoomInstance(context.Background(), newGameRoomInstance)
 		require.Error(t, err)
 	})
+
+	t.Run("should fail - room instance is nil => returns error", func(t *testing.T) {
+		err := roomManager.UpdateRoomInstance(context.Background(), nil)
+		require.Error(t, err)
+	})
 }
 
 func TestRoomManager_CleanRoomState(t *testing.T) {

internal/core/workers/runtime_watcher_worker/runtime_watcher_worker.go

@@ -93,13 +93,16 @@ func (w *runtimeWatcherWorker) IsRunning() bool {
 }
 
 func (w *runtimeWatcherWorker) processEvent(ctx context.Context, event game_room.InstanceEvent) error {
+	w.logger.Sugar().Infof("processing event: %++v", event)
 	switch event.Type {
 	case game_room.InstanceEventTypeAdded, game_room.InstanceEventTypeUpdated:
+		w.logger.Info("processing event. Updating rooms instance")
 		err := w.roomManager.UpdateRoomInstance(ctx, event.Instance)
 		if err != nil {
 			return fmt.Errorf("failed to update room instance %s: %w", event.Instance.ID, err)
 		}
 	case game_room.InstanceEventTypeDeleted:
+		w.logger.Info("processing event. Cleaning Room state")
 		err := w.roomManager.CleanRoomState(ctx, event.Instance.SchedulerID, event.Instance.ID)
 		if err != nil {
 			return fmt.Errorf("failed to clean room %s state: %w", event.Instance.ID, err)