Added maestro room resilience (#668)

* Added maestro room resilience

* Fixed docker and log message

* Fixed code review

* fix errAllErrors

* add context cancel

* Fixed code review

* Update err handle cha

Co-authored-by: Pedro Soares <pedro.soares@wildlifestudios.com>

* changed erro handle

---------

Co-authored-by: Luiz Felipe Limao <luizfelipe.limao@MacBook-Pro-de-Luiz.local>
Co-authored-by: Pedro Soares <pedro.soares@wildlifestudios.com>

Author: 3 people

e2e/framework/maestro/Dockerfile

@@ -47,4 +47,4 @@ COPY --from=builder /app/maestro .
 COPY ./config ./config
 
 ENTRYPOINT ["/app/maestro"]
-CMD ["--help"]
+CMD ["--help"]

e2e/framework/maestro/dependencies.go

@@ -48,7 +48,7 @@ func provideDependencies(ctx context.Context, maestroPath string, compose tc.Com
 		return nil, fmt.Errorf("failed to tear down running containers: %w", err)
 	}
 
-	services := []string{"postgres", "redis", "k3s_agent", "k3s_server"}
+	services := []string{"postgres", "redis"}
 	err = compose.Up(context.Background(), tc.RunServices(services...), tc.RemoveOrphans(true), tc.Wait(true))
 	if err != nil {
 		return nil, fmt.Errorf("failed to start dependencies: %w", err)

internal/core/operations/rooms/add/executor.go

@@ -25,13 +25,14 @@ package add
 import (
 	"context"
 	"fmt"
+	"sync"
+	"time"
 
 	"github.com/topfreegames/maestro/internal/core/logs"
 
 	"github.com/topfreegames/maestro/internal/core/ports"
 
 	"github.com/topfreegames/maestro/internal/core/entities"
-	"golang.org/x/sync/errgroup"
 
 	"go.uber.org/zap"
 
@@ -88,24 +89,51 @@ func (ex *Executor) Execute(ctx context.Context, op *operation.Operation, defini
 		amount = ex.config.AmountLimit
 	}
 
-	errGroup, errContext := errgroup.WithContext(ctx)
-	executionLogger.Info("start adding rooms", zap.Int32("amount", amount))
+	var wg sync.WaitGroup
+	errsCh := make(chan error, amount)
+
 	for i := int32(1); i <= amount; i++ {
-		errGroup.Go(func() error {
-			err := ex.createRoom(errContext, scheduler, executionLogger)
-			return err
-		})
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
+
+			errsCh <- ex.createRoom(ctx, scheduler, executionLogger)
+		}()
 	}
 
-	if executionErr := errGroup.Wait(); executionErr != nil {
-		executionLogger.Error("Error creating rooms", zap.Error(executionErr))
-		ex.operationManager.AppendOperationEventToExecutionHistory(ctx, op, executionErr.Error())
-		return executionErr
+	wg.Wait()
+	close(errsCh)
+
+	var collectedErrors []error
+	for err := range errsCh {
+		if err != nil {
+			collectedErrors = append(collectedErrors, err)
+		}
 	}
 
-	ex.operationManager.AppendOperationEventToExecutionHistory(ctx, op, fmt.Sprintf("added %d rooms", amount))
-	executionLogger.Info("finished adding rooms")
-	return nil
+	errCount := int32(len(collectedErrors))
+	successCount := amount - errCount
+
+	switch {
+	case errCount > successCount:
+		ErrMajorityRooms := fmt.Errorf("more rooms failed than succeeded, errors: %d and successes: %d of amount: %d", errCount, successCount, amount)
+
+		executionLogger.Error(ErrMajorityRooms.Error(),
+			zap.Error(ErrMajorityRooms),
+			zap.Int32("failedRoomsCount", errCount),
+			zap.Int32("successRoomsCount", successCount),
+			zap.Any("allErrors", collectedErrors))
+
+		ex.operationManager.AppendOperationEventToExecutionHistory(ctx, op, ErrMajorityRooms.Error())
+		return ErrMajorityRooms
+	default:
+		executionLogger.Sugar().Infof("added rooms successfully with errors: %d and success: %d of amount: %d", errCount, successCount, amount)
+		ex.operationManager.AppendOperationEventToExecutionHistory(ctx, op, fmt.Sprintf("added %d rooms", amount))
+		return nil
+	}
 }
 
 func (ex *Executor) Rollback(ctx context.Context, op *operation.Operation, definition operations.Definition, executeErr error) error {

internal/core/operations/rooms/add/executor_test.go

@@ -26,14 +26,13 @@
 package add
 
 import (
+	"context"
 	"fmt"
+	"testing"
 	"time"
 
 	clock_mock "github.com/topfreegames/maestro/internal/core/ports/clock_mock.go"
 
-	"context"
-	"testing"
-
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
 	"github.com/topfreegames/maestro/internal/core/entities"
@@ -108,7 +107,7 @@ func TestExecutor_Execute(t *testing.T) {
 		require.Nil(t, err)
 	})
 
-	t.Run("should fail - some room creation fail, others succeed => returns unexpected error", func(t *testing.T) {
+	t.Run("should succeed - some room creation fail, others succeed => returns success", func(t *testing.T) {
 		_, roomsManager, schedulerStorage, operationsManager := testSetup(t, config)
 
 		schedulerStorage.EXPECT().GetScheduler(gomock.Any(), op.SchedulerName).Return(&scheduler, nil)
@@ -117,14 +116,15 @@ func TestExecutor_Execute(t *testing.T) {
 		gameRoomReady.Status = game_room.GameStatusReady
 
 		roomsManager.EXPECT().CreateRoom(gomock.Any(), gomock.Any(), false).Return(&gameRoom, &gameRoomInstance, nil).Times(9)
+
 		roomsManager.EXPECT().CreateRoom(gomock.Any(), gomock.Any(), false).Return(nil, nil, porterrors.NewErrUnexpected("error"))
-		operationsManager.EXPECT().AppendOperationEventToExecutionHistory(gomock.Any(), &op, "error while creating room: error")
+
+		operationsManager.EXPECT().AppendOperationEventToExecutionHistory(gomock.Any(), &op, fmt.Sprintf("added %d rooms", definition.Amount))
 
 		executor := NewExecutor(roomsManager, schedulerStorage, operationsManager, config)
 		err := executor.Execute(context.Background(), &op, &definition)
 
-		require.NotNil(t, err)
-		require.ErrorContains(t, err, "error while creating room: error")
+		require.Nil(t, err)
 	})
 
 	t.Run("should fail - no scheduler found => returns error", func(t *testing.T) {
@@ -165,10 +165,29 @@ func TestExecutor_Execute(t *testing.T) {
 
 		require.Nil(t, err)
 	})
+
+	t.Run("should fail - majority of room creation fail, others succeed => returns error", func(t *testing.T) {
+		_, roomsManager, schedulerStorage, operationsManager := testSetup(t, config)
+
+		schedulerStorage.EXPECT().GetScheduler(gomock.Any(), op.SchedulerName).Return(&scheduler, nil)
+
+		gameRoomReady := gameRoom
+		gameRoomReady.Status = game_room.GameStatusReady
+
+		roomsManager.EXPECT().CreateRoom(gomock.Any(), gomock.Any(), false).Return(&gameRoom, &gameRoomInstance, nil).Times(4)
+		roomsManager.EXPECT().CreateRoom(gomock.Any(), gomock.Any(), false).Return(nil, nil, porterrors.NewErrUnexpected("error")).Times(6)
+
+		operationsManager.EXPECT().AppendOperationEventToExecutionHistory(gomock.Any(), &op, "more rooms failed than succeeded, errors: 6 and successes: 4 of amount: 10")
+
+		executor := NewExecutor(roomsManager, schedulerStorage, operationsManager, config)
+		err := executor.Execute(context.Background(), &op, &definition)
+
+		require.NotNil(t, err)
+		require.ErrorContains(t, err, "more rooms failed than succeeded, errors: 6 and successes: 4 of amount: 10")
+	})
 }
 
 func TestExecutor_Rollback(t *testing.T) {
-
 	definition := Definition{Amount: 10}
 
 	op := operation.Operation{
@@ -188,7 +207,6 @@ func TestExecutor_Rollback(t *testing.T) {
 		rollbackErr := NewExecutor(roomsManager, schedulerStorage, operationsManager, config).Rollback(context.Background(), &op, &definition, nil)
 		require.NoError(t, rollbackErr)
 	})
-
 }
 
 func testSetup(t *testing.T, cfg Config) (*Executor, *mockports.MockRoomManager, *mockports.MockSchedulerStorage, *mockports.MockOperationManager) {

internal/service/adapters.go

@@ -170,7 +170,9 @@ func NewRuntimeKubernetes(c config.Config) (ports.Runtime, error) {
 
 // NewOperationStorageRedis instantiates redis as operation storage.
 func NewOperationStorageRedis(clock ports.Clock, operationDefinitionProviders map[string]operations.DefinitionConstructor, c config.Config) (ports.OperationStorage, error) {
-	client, err := createRedisClient(c, c.GetString(operationStorageRedisURLPath))
+	client, err := createRedisClient(c, c.GetString(operationStorageRedisURLPath), func(opts *redis.Options) {
+		opts.MaxRetries = 3
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Redis operation storage: %w", err)
 	}
@@ -187,7 +189,9 @@ func NewOperationStorageRedis(clock ports.Clock, operationDefinitionProviders ma
 
 // NewOperationLeaseStorageRedis instantiates redis as operation lease storage.
 func NewOperationLeaseStorageRedis(clock ports.Clock, c config.Config) (ports.OperationLeaseStorage, error) {
-	client, err := createRedisClient(c, c.GetString(operationLeaseStorageRedisURLPath))
+	client, err := createRedisClient(c, c.GetString(operationLeaseStorageRedisURLPath), func(opts *redis.Options) {
+		opts.MaxRetries = 3
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Redis operation lease storage: %w", err)
 	}
@@ -197,7 +201,9 @@ func NewOperationLeaseStorageRedis(clock ports.Clock, c config.Config) (ports.Op
 
 // NewRoomStorageRedis instantiates redis as room storage.
 func NewRoomStorageRedis(c config.Config) (ports.RoomStorage, error) {
-	client, err := createRedisClient(c, c.GetString(roomStorageRedisURLPath))
+	client, err := createRedisClient(c, c.GetString(roomStorageRedisURLPath), func(opts *redis.Options) {
+		opts.MaxRetries = -1
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Redis room storage: %w", err)
 	}
@@ -207,7 +213,9 @@ func NewRoomStorageRedis(c config.Config) (ports.RoomStorage, error) {
 
 // NewGameRoomInstanceStorageRedis instantiates redis as instance storage.
 func NewGameRoomInstanceStorageRedis(c config.Config) (ports.GameRoomInstanceStorage, error) {
-	client, err := createRedisClient(c, c.GetString(instanceStorageRedisURLPath))
+	client, err := createRedisClient(c, c.GetString(instanceStorageRedisURLPath), func(opts *redis.Options) {
+		opts.MaxRetries = 3
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Redis instance storage: %w", err)
 	}
@@ -217,7 +225,9 @@ func NewGameRoomInstanceStorageRedis(c config.Config) (ports.GameRoomInstanceSto
 
 // NewSchedulerCacheRedis instantiates redis as scheduler cache.
 func NewSchedulerCacheRedis(c config.Config) (ports.SchedulerCache, error) {
-	client, err := createRedisClient(c, c.GetString(schedulerCacheRedisURLPath))
+	client, err := createRedisClient(c, c.GetString(schedulerCacheRedisURLPath), func(opts *redis.Options) {
+		opts.MaxRetries = 3
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Redis scheduler cache: %w", err)
 	}
@@ -261,7 +271,7 @@ func GetSchedulerStoragePostgresURL(c config.Config) string {
 	return c.GetString(schedulerStoragePostgresURLPath)
 }
 
-func createRedisClient(c config.Config, url string) (*redis.Client, error) {
+func createRedisClient(c config.Config, url string, customizers ...func(*redis.Options)) (*redis.Client, error) {
 	opts, err := redis.ParseURL(url)
 	if err != nil {
 		return nil, fmt.Errorf("invalid redis URL: %w", err)
@@ -273,6 +283,10 @@ func createRedisClient(c config.Config, url string) (*redis.Client, error) {
 
 	hostPort := strings.Split(opts.Addr, ":")
 
+	for _, customizer := range customizers {
+		customizer(opts)
+	}
+
 	client := redis.NewClient(opts)
 
 	if tracing.IsTracingEnabled(c) {
@@ -299,7 +313,9 @@ func NewAutoscaler(policies autoscaler.PolicyMap) ports.Autoscaler {
 
 // NewOperationFlowRedis instantiates a new operation flow using redis as backend.
 func NewOperationFlowRedis(c config.Config) (ports.OperationFlow, error) {
-	client, err := createRedisClient(c, c.GetString(operationFlowRedisURLPath))
+	client, err := createRedisClient(c, c.GetString(operationFlowRedisURLPath), func(opts *redis.Options) {
+		opts.MaxRetries = 3
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Redis operation storage: %w", err)
 	}