feat: add concurrency to release rendering

nestoca · Aug 30, 2024 · 05f5eb8 · 05f5eb8
1 parent db542d9
commit 05f5eb8
Show file tree

Hide file tree

Showing 2 changed files with 92 additions and 65 deletions.
diff --git a/cmd/server/api_get_params_test.go b/cmd/server/api_get_params_test.go
@@ -12,6 +12,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/davidmdm/conf"
@@ -51,7 +52,10 @@ func TestGetParamsE2E(t *testing.T) {
 	repo, err := user.NewRepo(catalog)
 	require.NoError(t, err, "failed to create repo for user: %s", user.Name)
 
-	logs := &TestLogOutputs{}
+	logs := &TestLogOutputs{
+		Records: []map[string]any{},
+		Mutex:   &sync.Mutex{},
+	}
 	logger := zerolog.New(logs)
 
 	repo = repo.WithLogger(logger)
@@ -115,13 +119,18 @@ func TestGetParamsE2E(t *testing.T) {
 
 type TestLogOutputs struct {
 	Records []map[string]any
+	Mutex   *sync.Mutex
 }
 
 func (output *TestLogOutputs) Write(data []byte) (int, error) {
 	var record map[string]any
 	if err := json.Unmarshal(data, &record); err != nil {
 		return 0, fmt.Errorf("invalid record: %w", err)
 	}
+
+	output.Mutex.Lock()
+	defer output.Mutex.Unlock()
+
 	output.Records = append(output.Records, record)
 	return len(data), nil
 }

diff --git a/internal/generator/generator.go b/internal/generator/generator.go
@@ -135,82 +135,100 @@ func (generator *Generator) Run(ctx context.Context) ([]Result, error) {
 		return nil, fmt.Errorf("loading joy context: %w", err)
 	}
 
-	span.SetAttributes(
-		attribute.Int("release_count", len(joyctx.Catalog.Releases.Items)),
-	)
-
 	cache := helm.ChartCache{
 		Root:            generator.CacheRoot,
 		Puller:          generator.ChartPuller,
 		Refs:            joyctx.Config.Charts,
 		DefaultChartRef: joyctx.Config.DefaultChartRef,
 	}
 
-	var reconciledReleases []Result
-	for _, crossRelease := range joyctx.Catalog.Releases.Items {
-		for _, release := range crossRelease.Releases {
-			func() {
-				if release == nil {
-					return
-				}
+	var releases []*v1alpha1.Release
+	for _, cross := range joyctx.Catalog.Releases.Items {
+		for _, release := range cross.Releases {
+			if release == nil {
+				continue
+			}
+			releases = append(releases, release)
+		}
+	}
+
+	span.SetAttributes(
+		attribute.Int("release_count", len(joyctx.Catalog.Releases.Items)),
+	)
+
+	var (
+		wg                 sync.WaitGroup
+		reconciledReleases = make([]Result, len(releases))
+		semaphore          = make(chan struct{}, 8)
+	)
+
+	for i, release := range releases {
+		semaphore <- struct{}{}
+
+    wg.Add(1)
+
+		go func() {
+			defer wg.Done()
+			defer func() { <-semaphore }()
 
-				ctx, span := observability.StartTrace(ctx, "release_render")
-				defer span.End()
+			ctx, span := observability.StartTrace(ctx, "release_render")
+			defer span.End()
 
-				span.SetAttributes(
-					attribute.String("release", release.Name),
-					attribute.String("env", release.Environment.Name),
-				)
+			span.SetAttributes(
+				attribute.String("release", release.Name),
+				attribute.String("env", release.Environment.Name),
+			)
 
+			generator.Logger.
+				Debug().
+				Str("release", release.Name).
+				Str("environment", release.Environment.Name).
+				Msg("processing release")
+
+			chart, err := cache.GetReleaseChartFS(ctx, release)
+			if err != nil {
 				generator.Logger.
-					Debug().
-					Str("release", release.Name).
-					Str("environment", release.Environment.Name).
-					Msg("processing release")
-
-				chart, err := cache.GetReleaseChartFS(ctx, release)
-				if err != nil {
-					generator.Logger.
-						Error().
-						Err(err).
-						Str("release", release.Name).Str("environment", release.Environment.Name).
-						Msgf("error getting chart for release %s", release.Name)
-					return
-				}
-
-				release.Spec.Chart.RepoUrl = chart.RepoURL
-				release.Spec.Chart.Name = chart.Name
-				release.Spec.Chart.Version = chart.Version
-
-				values, err := joy.ComputeReleaseValues(release, chart)
-				if err != nil {
-					generator.Logger.
-						Error().
-						Err(err).
-						Str("release", release.Name).Str("environment", release.Environment.Name).
-						Msgf("error computing values for release %s", release.Name)
-					return
-				}
-
-				renderedValues, err := yaml.Marshal(values)
-				if err != nil {
-					generator.Logger.
-						Error().
-						Err(err).
-						Str("release", release.Name).Str("environment", release.Environment.Name).
-						Msgf("error marshaling values for release %s", release.Name)
-					return
-				}
-
-				reconciledReleases = append(reconciledReleases, Result{
-					Release:     release,
-					Environment: release.Environment,
-					Project:     release.Project,
-					Values:      string(renderedValues),
-				})
-			}()
-		}
+					Error().
+					Err(err).
+					Str("release", release.Name).Str("environment", release.Environment.Name).
+					Msgf("error getting chart for release %s", release.Name)
+				return
+			}
+
+			release.Spec.Chart.RepoUrl = chart.RepoURL
+			release.Spec.Chart.Name = chart.Name
+			release.Spec.Chart.Version = chart.Version
+
+			values, err := joy.ComputeReleaseValues(release, chart)
+			if err != nil {
+				generator.Logger.
+					Error().
+					Err(err).
+					Str("release", release.Name).Str("environment", release.Environment.Name).
+					Msgf("error computing values for release %s", release.Name)
+				return
+			}
+
+			renderedValues, err := yaml.Marshal(values)
+			if err != nil {
+				generator.Logger.
+					Error().
+					Err(err).
+					Str("release", release.Name).Str("environment", release.Environment.Name).
+					Msgf("error marshaling values for release %s", release.Name)
+				return
+			}
+
+			reconciledReleases[i] = Result{
+				Release:     release,
+				Environment: release.Environment,
+				Project:     release.Project,
+				Values:      string(renderedValues),
+			}
+		}()
 	}
 
+	wg.Wait()
+
 	return reconciledReleases, nil
 }