chore: parallel manifest rendering (#1628)

* Parallelize manifest exporting

* Skip manifest export early if no resources to be exported

This saves not much but a couple hundred microseconds for these empty
environments.

* Group manifest generators again but without using a lexical scope

* Better environment attributes inside traces

* Cleanup

* Handle a canceled context properly

* Increase channel size to reduce wait

Author: zerok

internal/telemetry/attributes.go

@@ -1,8 +1,6 @@
 package telemetry
 
 import (
-	"fmt"
-
 	"github.com/grafana/tanka/pkg/spec/v1alpha1"
 	"go.opentelemetry.io/otel/attribute"
 )
@@ -21,6 +19,7 @@ func AttrNumEnvs(v int) attribute.KeyValue {
 
 func AttrEnv(v *v1alpha1.Environment) []attribute.KeyValue {
 	return []attribute.KeyValue{
-		attribute.String("tanka.env.id", fmt.Sprintf("%s@%s", v.Metadata.Name, v.Spec.APIServer)),
+		attribute.String("tanka.env.name", v.Metadata.Name),
+		attribute.String("tanka.env.namespace", v.Spec.Namespace),
 	}
 }

pkg/tanka/export.go

@@ -10,11 +10,14 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"text/template"
 
 	"github.com/Masterminds/sprig/v3"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
+	"go.opentelemetry.io/otel/attribute"
+	"golang.org/x/sync/errgroup"
 	"k8s.io/apimachinery/pkg/labels"
 
 	"github.com/grafana/tanka/internal/telemetry"
@@ -69,9 +72,6 @@ func ExportEnvironments(ctx context.Context, envs []*v1alpha1.Environment, to st
 
 	span.SetAttributes(telemetry.AttrNumEnvs(len(envs)))
 
-	// Keep track of which file maps to which environment
-	fileToEnv := map[string]string{}
-
 	// dir must be empty
 	empty, err := dirEmpty(to)
 	if err != nil {
@@ -93,79 +93,168 @@ func ExportEnvironments(ctx context.Context, envs []*v1alpha1.Environment, to st
 		return fmt.Errorf("deleting previously exported manifests from deleted environments: %w", err)
 	}
 
+	parallelism := opts.Parallelism
+
+	if parallelism <= 0 {
+		parallelism = defaultParallelism
+	}
+
+	if parallelism > len(envs) {
+		parallelism = len(envs)
+	}
+
 	// get all environments for paths
 	loadedEnvs, err := parallelLoadEnvironments(ctx, envs, parallelOpts{
 		Opts:        opts.Opts,
 		Selector:    opts.Selector,
-		Parallelism: opts.Parallelism,
+		Parallelism: parallelism,
 	})
 	if err != nil {
 		return err
 	}
 
-	{
-		ctx, span := tracer.Start(ctx, "generateManifests")
-		defer span.End()
+	fileToEnv, err := manifestEnvironments(ctx, loadedEnvs, parallelism, to, opts)
+	if err != nil {
+		return err
+	}
 
-		// FINDING: Generating the export files takes some time. Perhaps we should parallelize this.
+	return exportManifestFile(to, fileToEnv, nil)
+}
+
+func manifestEnvironments(ctx context.Context, loadedEnvs []*v1alpha1.Environment, parallelism int, to string, opts *ExportEnvOpts) (map[string]string, error) {
+	ctx, span := tracer.Start(ctx, "tanka.manifestEnvironments")
+	defer span.End()
+	span.SetAttributes(telemetry.AttrNumEnvs(len(loadedEnvs)))
+	fileToEnv := map[string]string{}
+	fileToEnvLock := sync.Mutex{}
+
+	// Similar to the parallel loader, we're going to split all that up
+	// into multiple routines that handle manifest loading and writing
+	grp, ctx := errgroup.WithContext(ctx)
+
+	// Create a work channel that holds enough for all workers * 2 so that a
+	// worker doesn't have to wait unnecessarily:
+	envsToManifest := make(chan *v1alpha1.Environment, parallelism*2)
+
+	for range parallelism {
+		grp.Go(func() error {
+			ctx, span := tracer.Start(ctx, "tanka.manifestsGenerateWorker")
+			defer span.End()
+
+			for {
+				select {
+				case <-ctx.Done():
+					telemetry.FailSpanWithError(span, ctx.Err())
+					return ctx.Err()
+				case work, ok := <-envsToManifest:
+					if !ok {
+						// Channel is empty and closed
+						return nil
+					}
+					localFileToEnv, err := manifestSingleEnv(ctx, work, to, opts)
+					if err != nil {
+						telemetry.FailSpanWithError(span, err)
+						return err
+					}
+
+					fileToEnvLock.Lock()
+					for name, namespace := range localFileToEnv {
+						fileToEnv[name] = namespace
+					}
+					fileToEnvLock.Unlock()
+				}
+			}
+		})
+	}
+
+	grp.Go(func() error {
 		for _, env := range loadedEnvs {
-			// get the manifests
-			loaded, err := LoadManifests(ctx, env, opts.Opts.Filters)
-			if err != nil {
-				return err
+			select {
+			case <-ctx.Done():
+				close(envsToManifest)
+				return ctx.Err()
+			case envsToManifest <- env:
 			}
+		}
+		close(envsToManifest)
+		return nil
+	})
 
-			env := loaded.Env
-			res := loaded.Resources
+	if err := grp.Wait(); err != nil {
+		telemetry.FailSpanWithError(span, err)
+		return nil, err
+	}
 
-			// create raw manifest version of env for templating
-			env.Data = nil
-			raw, err := json.Marshal(env)
-			if err != nil {
-				return err
-			}
-			var menv manifest.Manifest
-			if err := json.Unmarshal(raw, &menv); err != nil {
-				return err
-			}
+	return fileToEnv, nil
+}
 
-			// create template
-			manifestTemplate, err := createTemplate(opts.Format, menv)
-			if err != nil {
-				return fmt.Errorf("parsing format: %s", err)
-			}
+func manifestSingleEnv(ctx context.Context, work *v1alpha1.Environment, to string, opts *ExportEnvOpts) (map[string]string, error) {
+	ctx, span := tracer.Start(ctx, "tanka.manifestSingleEnv")
+	defer span.End()
+	span.SetAttributes(telemetry.AttrEnv(work)...)
 
-			// write each to a file
-			for _, m := range res {
-				// apply template
-				name, err := applyTemplate(manifestTemplate, m)
-				if err != nil {
-					return fmt.Errorf("executing name template: %w", err)
-				}
+	fileToEnv := make(map[string]string)
 
-				// Create all subfolders in path
-				relpath := name + "." + opts.Extension
-				path := filepath.Join(to, relpath)
+	loaded, err := LoadManifests(ctx, work, opts.Opts.Filters)
+	if err != nil {
+		return nil, err
+	}
 
-				fileToEnv[relpath] = env.Metadata.Namespace
+	env := loaded.Env
+	res := loaded.Resources
 
-				// Abort if already exists
-				if exists, err := fileExists(path); err != nil {
-					return err
-				} else if exists {
-					return fmt.Errorf("file '%s' already exists. Aborting", path)
-				}
+	span.SetAttributes(attribute.Int("tanka.env.num_resources", len(res)))
 
-				// Write manifest
-				data := m.String()
-				if err := writeExportFile(path, []byte(data)); err != nil {
-					return err
-				}
-			}
-		}
+	// If there is no thing to process then we can skip the rest.
+	if len(res) == 0 {
+		return fileToEnv, nil
 	}
 
-	return exportManifestFile(to, fileToEnv, nil)
+	// create raw manifest version of env for templating
+	env.Data = nil
+	raw, err := json.Marshal(env)
+	if err != nil {
+		return nil, err
+	}
+	var menv manifest.Manifest
+	if err := json.Unmarshal(raw, &menv); err != nil {
+		return nil, err
+	}
+
+	// create template
+	manifestTemplate, err := createTemplate(ctx, opts.Format, menv)
+	if err != nil {
+		return nil, fmt.Errorf("parsing format: %s", err)
+	}
+
+	// write each to a file
+	for _, m := range res {
+		// apply template
+		name, err := applyTemplate(ctx, manifestTemplate, m)
+		if err != nil {
+			return nil, fmt.Errorf("executing name template: %w", err)
+		}
+
+		// Create all subfolders in path
+		relpath := name + "." + opts.Extension
+		path := filepath.Join(to, relpath)
+
+		fileToEnv[relpath] = env.Metadata.Namespace
+
+		// Abort if already exists
+		if exists, err := fileExists(path); err != nil {
+			return nil, err
+		} else if exists {
+			return nil, fmt.Errorf("file '%s' already exists. Aborting", path)
+		}
+
+		// Write manifest
+		data := m.String()
+		if err := writeExportFile(path, []byte(data)); err != nil {
+			return nil, err
+		}
+	}
+	return fileToEnv, nil
 }
 
 func fileExists(name string) (bool, error) {
@@ -284,7 +373,7 @@ func writeExportFile(path string, data []byte) error {
 	return os.WriteFile(path, data, 0644)
 }
 
-func createTemplate(format string, env manifest.Manifest) (*template.Template, error) {
+func createTemplate(_ context.Context, format string, env manifest.Manifest) (*template.Template, error) {
 	// Replace all os.path separators in string with BelRune for creating subfolders
 	replaceFormat := replaceTmplText(format, string(os.PathSeparator), BelRune)
 
@@ -318,7 +407,7 @@ func replaceTmplText(s, oldString, newString string) string {
 	return strings.Join(parts, "")
 }
 
-func applyTemplate(template *template.Template, m manifest.Manifest) (path string, err error) {
+func applyTemplate(_ context.Context, template *template.Template, m manifest.Manifest) (path string, err error) {
 	buf := bytes.Buffer{}
 	if err := template.Execute(&buf, m); err != nil {
 		return "", err

pkg/tanka/load.go

@@ -76,11 +76,7 @@ func LoadEnvironment(ctx context.Context, path string, opts Opts) (*v1alpha1.Env
 	return env, nil
 }
 
-func LoadManifests(ctx context.Context, env *v1alpha1.Environment, filters process.Matchers) (*LoadResult, error) {
-	_, span := tracer.Start(ctx, "tanka.LoadManifests")
-	defer span.End()
-	span.SetAttributes(telemetry.AttrEnv(env)...)
-
+func LoadManifests(_ context.Context, env *v1alpha1.Environment, filters process.Matchers) (*LoadResult, error) {
 	if err := checkVersion(env.Spec.ExpectVersions.Tanka); err != nil {
 		return nil, err
 	}