feat(cd-service): do env release trains in parallel for env groups (#…

…2189)

Ref: SRX-AC1K9Y

docker-compose.yml

@@ -26,7 +26,7 @@ services:
       - KUBERPULT_DB_OPTION=postgreSQL
       - KUBERPULT_DB_WRITE_ESL_TABLE_ONLY=false
       - KUBERPULT_GIT_BRANCH=master
-      - KUBERPULT_GIT_NETWORK_TIMEOUT=3s
+      - KUBERPULT_GIT_NETWORK_TIMEOUT=120s
       - KUBERPULT_DEX_MOCK=false
       - KUBERPULT_DEX_ENABLED=false
       - KUBERPULT_DEX_RBAC_POLICY_PATH=/etc/policy.csv

pkg/db/db.go

@@ -2230,7 +2230,6 @@ func (h *DBHandler) DBSelectAllActiveAppLocksForSliceApps(ctx context.Context, t
 	for _, app := range appNames {
 		args = append(args, app)
 	}
-
 	rows, err = tx.QueryContext(
 		ctx,
 		selectQuery,

services/cd-service/pkg/repository/transformer.go

@@ -34,6 +34,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	time2 "github.com/freiheit-com/kuberpult/pkg/time"
@@ -3714,30 +3715,87 @@ func (c *ReleaseTrain) Transform(
 	}
 	sort.Strings(envNames)
 	span.SetTag("environments", len(envNames))
+	if state.DBHandler.ShouldUseOtherTables() && isEnvGroup {
+		var wg sync.WaitGroup
+		wg.Add(len(envNames))
+		errChan := make(chan error, len(envNames))
+		for _, envName := range envNames {
+			var trainGroup *string
+			if isEnvGroup {
+				trainGroup = conversion.FromString(targetGroupName)
+			}
+			go c.runEnvReleaseTrainBackground(ctx, state, t, envName, trainGroup, envGroupConfigs, configs, &wg, errChan)
+		}
+		wg.Wait()
+		close(errChan)
+		allErrorsMessage := ""
+		for err := range errChan {
+			if err != nil {
+				allErrorsMessage += err.Error()
+			}
+		}
 
-	for _, envName := range envNames {
-		var trainGroup *string
-		if isEnvGroup {
-			trainGroup = conversion.FromString(targetGroupName)
+		if allErrorsMessage != "" {
+			return "", grpc.PublicError(ctx, fmt.Errorf("Error in env Release Trains: %s", allErrorsMessage))
+		}
+	} else {
+		for _, envName := range envNames {
+			var trainGroup *string
+			if isEnvGroup {
+				trainGroup = conversion.FromString(targetGroupName)
+			}
+			err = t.Execute(ctx, &envReleaseTrain{
+				Parent:                c,
+				Env:                   envName,
+				EnvConfigs:            configs,
+				EnvGroupConfigs:       envGroupConfigs,
+				WriteCommitData:       c.WriteCommitData,
+				TrainGroup:            trainGroup,
+				TransformerEslVersion: c.TransformerEslVersion,
+				CiLink:                c.CiLink,
+			}, transaction)
+			if err != nil {
+				return "", err
+			}
 		}
+	}
+
+	return fmt.Sprintf(
+		"Release Train to environment/environment group '%s':\n",
+		targetGroupName), nil
+}
 
-		if err := t.Execute(ctx, &envReleaseTrain{
+func (c *ReleaseTrain) runEnvReleaseTrainBackground(ctx context.Context,
+	state *State,
+	t TransformerContext,
+	envName string,
+	trainGroup *string,
+	envGroupConfigs map[string]config.EnvironmentConfig,
+	configs map[string]config.EnvironmentConfig,
+	workerGroup *sync.WaitGroup,
+	errChan chan error) {
+	defer workerGroup.Done()
+	err := state.DBHandler.WithTransactionR(ctx, 2, false, func(ctx context.Context, transaction2 *sql.Tx) error {
+		internal, err := state.DBHandler.DBReadEslEventInternal(ctx, transaction2, false)
+		if err != nil {
+			return err
+		}
+		if internal == nil {
+			return fmt.Errorf("could not find esl event that was just inserted")
+		}
+		err = t.Execute(ctx, &envReleaseTrain{
 			Parent:                c,
 			Env:                   envName,
 			EnvConfigs:            configs,
 			EnvGroupConfigs:       envGroupConfigs,
 			WriteCommitData:       c.WriteCommitData,
 			TrainGroup:            trainGroup,
-			TransformerEslVersion: c.TransformerEslVersion,
+			TransformerEslVersion: db.TransformerID(internal.EslVersion),
 			CiLink:                c.CiLink,
-		}, transaction); err != nil {
-			return "", err
-		}
-	}
-
-	return fmt.Sprintf(
-		"Release Train to environment/environment group '%s':\n",
-		targetGroupName), nil
+		}, transaction2)
+		return err
+	})
+	errChan <- err
 }
 
 type envReleaseTrain struct {