Merge pull request #250 from stefanprodan/runtime-clusters

Add multi-cluster deployments capability to bundles

api/v1alpha1/runtime.go

@@ -27,6 +27,9 @@ const (
 	// RuntimeKind is the name of the Timoni runtime CUE attributes.
 	RuntimeKind string = "runtime"
 
+	// RuntimeDefaultName is the name of the default Timoni runtime.
+	RuntimeDefaultName string = "_default"
+
 	// RuntimeDelimiter is the delimiter used in Timoni runtime CUE attributes.
 	RuntimeDelimiter string = ":"
 
@@ -36,6 +39,9 @@ const (
 	// RuntimeName is the CUE path for the Timoni's bundle name.
 	RuntimeName Selector = "runtime.name"
 
+	// 	RuntimeClustersSelector is the CUE path for the Timoni's runtime clusters.
+	RuntimeClustersSelector Selector = "runtime.clusters"
+
 	// RuntimeValuesSelector is the CUE path for the Timoni's runtime values.
 	RuntimeValuesSelector Selector = "runtime.values"
 )
@@ -53,7 +59,13 @@ import "strings"
 #Runtime: {
 	apiVersion: string & =~"^v1alpha1$"
 	name:       string & =~"^(([A-Za-z0-9][-A-Za-z0-9_]*)?[A-Za-z0-9])?$" & strings.MaxRunes(63) & strings.MinRunes(1)
-	values: [...#RuntimeValue]
+
+	clusters?: [string & =~"^(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?$" & strings.MaxRunes(63) & strings.MinRunes(1)]: {
+		group!:       string
+		kubeContext!: string
+	}
+	
+	values?: [...#RuntimeValue]
 }
 `
 
@@ -99,10 +111,76 @@ type Runtime struct {
 	// Name of the runtime.
 	Name string `json:"name"`
 
+	// Clusters is the list of Kubernetes
+	// clusters belonging to this runtime.
+	Clusters []RuntimeCluster `json:"clusters"`
+
 	// Refs is the list of in-cluster resource references.
 	Refs []RuntimeResourceRef `json:"refs"`
 }
 
+// DefaultRuntime returns an empty Runtime with an unnamed
+// cluster set to the specified context.
+func DefaultRuntime(kubeContext string) *Runtime {
+	defaultCluster := RuntimeCluster{
+		Name:        RuntimeDefaultName,
+		Group:       RuntimeDefaultName,
+		KubeContext: kubeContext,
+	}
+
+	return &Runtime{
+		Name:     RuntimeDefaultName,
+		Clusters: []RuntimeCluster{defaultCluster},
+		Refs:     []RuntimeResourceRef{},
+	}
+}
+
+// RuntimeCluster holds the reference to a Kubernetes cluster.
+type RuntimeCluster struct {
+	// Name of the cluster.
+	Name string `json:"name"`
+
+	// Group name of the cluster.
+	Group string `json:"group"`
+
+	// KubeContext is the name of kubeconfig context for this cluster.
+	KubeContext string `json:"kubeContext"`
+}
+
+// IsDefault returns true if the given cluster
+// was initialised by a Runtime with no target clusters.
+func (rt *RuntimeCluster) IsDefault() bool {
+	return rt.Name == RuntimeDefaultName
+}
+
+// NameGroupValues returns the cluster name and group variables
+// as specified in the Runtime definition. If the given cluster
+// was initialised by an empty Runtime, the returned map is empty.
+func (rt *RuntimeCluster) NameGroupValues() map[string]string {
+	result := make(map[string]string)
+	if !rt.IsDefault() {
+		result["TIMONI_CLUSTER_NAME"] = rt.Name
+		result["TIMONI_CLUSTER_GROUP"] = rt.Group
+	}
+	return result
+}
+
+// SelectClusters returns the clusters matching the specified name and group.
+// Both the name and group support the '*' wildcard.
+func (r *Runtime) SelectClusters(name, group string) []RuntimeCluster {
+	var result []RuntimeCluster
+	for _, cluster := range r.Clusters {
+		if name != "" && name != "*" && !strings.EqualFold(cluster.Name, name) {
+			continue
+		}
+		if group != "" && group != "*" && !strings.EqualFold(cluster.Group, group) {
+			continue
+		}
+		result = append(result, cluster)
+	}
+	return result
+}
+
 // RuntimeResourceRef holds the data needed to query the fields
 // of a Kubernetes resource using CUE expressions.
 type RuntimeResourceRef struct {

cmd/timoni/bundle.go

@@ -20,11 +20,28 @@ import (
 	"github.com/spf13/cobra"
 )
 
+type bundleFlags struct {
+	runtimeFromEnv      bool
+	runtimeFiles        []string
+	runtimeCluster      string
+	runtimeClusterGroup string
+}
+
+var bundleArgs bundleFlags
+
 var bundleCmd = &cobra.Command{
 	Use:   "bundle",
 	Short: "Commands for managing bundles",
 }
 
 func init() {
+	bundleCmd.PersistentFlags().BoolVar(&bundleArgs.runtimeFromEnv, "runtime-from-env", false,
+		"Inject runtime values from the environment.")
+	bundleCmd.PersistentFlags().StringSliceVarP(&bundleArgs.runtimeFiles, "runtime", "r", nil,
+		"The local path to runtime.cue files.")
+	bundleCmd.PersistentFlags().StringVar(&bundleArgs.runtimeCluster, "runtime-cluster", "*",
+		"Filter runtime cluster by name.")
+	bundleCmd.PersistentFlags().StringVar(&bundleArgs.runtimeClusterGroup, "runtime-group", "*",
+		"Filter runtime clusters by group.")
 	rootCmd.AddCommand(bundleCmd)
 }

cmd/timoni/bundle_apply.go

@@ -71,8 +71,6 @@ type bundleApplyFlags struct {
 	wait               bool
 	force              bool
 	overwriteOwnership bool
-	runtimeFromEnv     bool
-	runtimeFiles       []string
 	creds              flags.Credentials
 }
 
@@ -92,10 +90,6 @@ func init() {
 		"Perform a server-side apply dry run and prints the diff.")
 	bundleApplyCmd.Flags().BoolVar(&bundleApplyArgs.wait, "wait", true,
 		"Wait for the applied Kubernetes objects to become ready.")
-	bundleApplyCmd.Flags().StringSliceVarP(&bundleApplyArgs.runtimeFiles, "runtime", "r", nil,
-		"The local path to runtime.cue files.")
-	bundleApplyCmd.Flags().BoolVar(&bundleApplyArgs.runtimeFromEnv, "runtime-from-env", false,
-		"Inject runtime values from the environment.")
 	bundleApplyCmd.Flags().Var(&bundleApplyArgs.creds, bundleApplyArgs.creds.Type(), bundleApplyArgs.creds.Description())
 	bundleCmd.AddCommand(bundleApplyCmd)
 }
@@ -135,92 +129,115 @@ func runBundleApplyCmd(cmd *cobra.Command, _ []string) error {
 
 	runtimeValues := make(map[string]string)
 
-	if bundleApplyArgs.runtimeFromEnv {
+	if bundleArgs.runtimeFromEnv {
 		maps.Copy(runtimeValues, engine.GetEnv())
 	}
 
-	if len(bundleApplyArgs.runtimeFiles) > 0 {
-		rt, err := buildRuntime(bundleApplyArgs.runtimeFiles)
-		if err != nil {
-			return err
-		}
+	rt, err := buildRuntime(bundleArgs.runtimeFiles)
+	if err != nil {
+		return err
+	}
+
+	clusters := rt.SelectClusters(bundleArgs.runtimeCluster, bundleArgs.runtimeClusterGroup)
+	if len(clusters) == 0 {
+		return fmt.Errorf("no cluster found")
+	}
 
+	ctxPull, cancel := context.WithTimeout(ctx, rootArgs.timeout)
+	defer cancel()
+
+	for _, cluster := range clusters {
+		kubeconfigArgs.Context = &cluster.KubeContext
+
+		clusterValues := make(map[string]string)
+
+		// add values from env
+		maps.Copy(clusterValues, runtimeValues)
+
+		// add values from cluster
 		rm, err := runtime.NewResourceManager(kubeconfigArgs)
 		if err != nil {
 			return err
 		}
-
 		reader := runtime.NewResourceReader(rm)
 		rv, err := reader.Read(ctx, rt.Refs)
 		if err != nil {
 			return err
 		}
+		maps.Copy(clusterValues, rv)
 
-		maps.Copy(runtimeValues, rv)
-	}
+		// add cluster info
+		maps.Copy(clusterValues, cluster.NameGroupValues())
 
-	if err := bm.InitWorkspace(tmpDir, runtimeValues); err != nil {
-		return err
-	}
-
-	v, err := bm.Build()
-	if err != nil {
-		return describeErr(tmpDir, "failed to build bundle", err)
-	}
+		// create cluster workspace
+		workspace := path.Join(tmpDir, cluster.Name)
+		if err := os.MkdirAll(workspace, os.ModePerm); err != nil {
+			return err
+		}
 
-	bundle, err := bm.GetBundle(v)
-	if err != nil {
-		return err
-	}
+		if err := bm.InitWorkspace(workspace, clusterValues); err != nil {
+			return describeErr(workspace, "failed to parse bundle", err)
+		}
 
-	log := LoggerBundle(cmd.Context(), bundle.Name)
+		v, err := bm.Build()
+		if err != nil {
+			return describeErr(tmpDir, "failed to build bundle", err)
+		}
 
-	if !bundleApplyArgs.overwriteOwnership {
-		err = bundleInstancesOwnershipConflicts(bundle.Instances)
+		bundle, err := bm.GetBundle(v)
 		if err != nil {
 			return err
 		}
-	}
 
-	ctxPull, cancel := context.WithTimeout(ctx, rootArgs.timeout)
-	defer cancel()
+		log := LoggerBundle(cmd.Context(), bundle.Name, cluster.Name)
 
-	for _, instance := range bundle.Instances {
-		spin := StartSpinner(fmt.Sprintf("pulling %s", instance.Module.Repository))
-		pullErr := fetchBundleInstanceModule(ctxPull, instance, tmpDir)
-		spin.Stop()
-		if pullErr != nil {
-			return pullErr
+		if !bundleApplyArgs.overwriteOwnership {
+			err = bundleInstancesOwnershipConflicts(bundle.Instances)
+			if err != nil {
+				return err
+			}
 		}
-	}
 
-	kubeVersion, err := runtime.ServerVersion(kubeconfigArgs)
-	if err != nil {
-		return err
-	}
-
-	if bundleApplyArgs.dryrun || bundleApplyArgs.diff {
-		log.Info(fmt.Sprintf("applying %v instance(s) %s",
-			len(bundle.Instances), colorizeDryRun("(server dry run)")))
-	} else {
-		log.Info(fmt.Sprintf("applying %v instance(s)",
-			len(bundle.Instances)))
-	}
+		for _, instance := range bundle.Instances {
+			spin := StartSpinner(fmt.Sprintf("pulling %s", instance.Module.Repository))
+			pullErr := fetchBundleInstanceModule(ctxPull, instance, tmpDir)
+			spin.Stop()
+			if pullErr != nil {
+				return pullErr
+			}
+		}
 
-	for _, instance := range bundle.Instances {
-		if err := applyBundleInstance(logr.NewContext(ctx, log), cuectx, instance, kubeVersion, tmpDir); err != nil {
+		kubeVersion, err := runtime.ServerVersion(kubeconfigArgs)
+		if err != nil {
 			return err
 		}
-	}
 
-	elapsed := time.Since(start)
-	if bundleApplyArgs.dryrun || bundleApplyArgs.diff {
-		log.Info(fmt.Sprintf("applied successfully %s",
-			colorizeDryRun("(server dry run)")))
-	} else {
-		log.Info(fmt.Sprintf("applied successfully in %s", elapsed.Round(time.Second)))
-	}
+		startMsg := fmt.Sprintf("applying %v instance(s)", len(bundle.Instances))
+		if !cluster.IsDefault() {
+			startMsg = fmt.Sprintf("%s on %s", startMsg, colorizeSubject(cluster.Group))
+		}
+
+		if bundleApplyArgs.dryrun || bundleApplyArgs.diff {
+			log.Info(fmt.Sprintf("%s %s", startMsg, colorizeDryRun("(server dry run)")))
+		} else {
+			log.Info(startMsg)
+		}
 
+		for _, instance := range bundle.Instances {
+			instance.Cluster = cluster.Name
+			if err := applyBundleInstance(logr.NewContext(ctx, log), cuectx, instance, kubeVersion, tmpDir); err != nil {
+				return err
+			}
+		}
+
+		elapsed := time.Since(start)
+		if bundleApplyArgs.dryrun || bundleApplyArgs.diff {
+			log.Info(fmt.Sprintf("applied successfully %s",
+				colorizeDryRun("(server dry run)")))
+		} else {
+			log.Info(fmt.Sprintf("applied successfully in %s", elapsed.Round(time.Second)))
+		}
+	}
 	return nil
 }
 
@@ -258,7 +275,7 @@ func fetchBundleInstanceModule(ctx context.Context, instance *engine.BundleInsta
 }
 
 func applyBundleInstance(ctx context.Context, cuectx *cue.Context, instance *engine.BundleInstance, kubeVersion string, rootDir string) error {
-	log := LoggerBundleInstance(ctx, instance.Bundle, instance.Name)
+	log := LoggerBundleInstance(ctx, instance.Bundle, instance.Cluster, instance.Name)
 
 	modDir := path.Join(rootDir, instance.Name, "module")
 	builder := engine.NewModuleBuilder(
@@ -408,7 +425,7 @@ func applyBundleInstance(ctx context.Context, cuectx *cue.Context, instance *eng
 			if err != nil {
 				return err
 			}
-			log.Info("resources are ready")
+			log.Info(fmt.Sprintf("%s resources %s", set.Name, colorizeReady("ready")))
 		}
 	}
 
@@ -439,10 +456,8 @@ func applyBundleInstance(ctx context.Context, cuectx *cue.Context, instance *eng
 			err = rm.WaitForTermination(deletedObjects, waitOptions)
 			spin.Stop()
 			if err != nil {
-				return fmt.Errorf("wating for termination failed: %w", err)
+				return fmt.Errorf("waiting for termination failed: %w", err)
 			}
-
-			log.Info("all resources are ready")
 		}
 	}