Added validate mutator to surface additional bundle warnings

bundle/config/validate/files_to_sync.go

@@ -0,0 +1,52 @@
+package validate
+
+import (
+	"context"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/deploy/files"
+	"github.com/databricks/cli/libs/diag"
+)
+
+func FilesToSync() bundle.Mutator {
+	return &filesToSync{}
+}
+
+type filesToSync struct {
+}
+
+func (v *filesToSync) Name() string {
+	return "validate:files_to_sync"
+}
+
+func (v *filesToSync) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	sync, err := files.GetSync(ctx, b)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	diags := diag.Diagnostics{}
+	fl, err := sync.GetFileList(ctx)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+
+	if len(fl) == 0 {
+		if len(b.Config.Sync.Exclude) == 0 {
+			diags = diags.Append(diag.Diagnostic{
+				Severity: diag.Warning,
+				Summary:  "There are no files to sync, please check your .gitignore",
+			})
+		} else {
+			loc := location{path: "sync.exclude", b: b}
+			diags = diags.Append(diag.Diagnostic{
+				Severity: diag.Warning,
+				Summary:  "There are no files to sync, please check your .gitignore and sync.exclude configuration",
+				Location: loc.Location(),
+				Path:     loc.Path(),
+			})
+		}
+	}
+
+	return diags
+}

bundle/config/validate/job_cluster_key_defined.go

@@ -0,0 +1,66 @@
+package validate
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+)
+
+func JobClusterKeyDefined() bundle.Mutator {
+	return &jobClusterKeyDefined{}
+}
+
+type jobClusterKeyDefined struct {
+}
+
+func (v *jobClusterKeyDefined) Name() string {
+	return "validate:job_cluster_key_defined"
+}
+
+func (v *jobClusterKeyDefined) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	// Collect all job_cluster_key references from defined tasks
+	jobClusterKeys := make(map[string]location)
+	for i, job := range b.Config.Resources.Jobs {
+		for j, task := range job.Tasks {
+			if task.JobClusterKey != "" {
+				jobClusterKeys[task.JobClusterKey] = location{
+					path: fmt.Sprintf("resources.jobs.%s.tasks[%d].job_cluster_key", i, j),
+					b:    b,
+				}
+			}
+		}
+	}
+
+	if len(jobClusterKeys) == 0 {
+		return nil
+	}
+
+	diags := diag.Diagnostics{}
+
+	// Check if all job_cluster_keys are defined
+	for key, loc := range jobClusterKeys {
+		if !isJobClusterKeyDefined(key, b) {
+			diags = diags.Append(diag.Diagnostic{
+				Severity: diag.Warning,
+				Summary:  fmt.Sprintf("job_cluster_key %s is not defined", key),
+				Location: loc.Location(),
+				Path:     loc.Path(),
+			})
+		}
+	}
+
+	return diags
+}
+
+func isJobClusterKeyDefined(key string, b *bundle.Bundle) bool {
+	for _, job := range b.Config.Resources.Jobs {
+		for _, cluster := range job.JobClusters {
+			if cluster.JobClusterKey == key {
+				return true
+			}
+		}
+	}
+	return false
+}

bundle/config/validate/validate.go

@@ -0,0 +1,42 @@
+package validate
+
+import (
+	"context"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/dyn"
+)
+
+type validate struct {
+}
+
+type location struct {
+	path string
+	b    *bundle.Bundle
+}
+
+func (l location) Location() dyn.Location {
+	return l.b.Config.GetLocation(l.path)
+}
+
+func (l location) Path() dyn.Path {
+	return dyn.MustPathFromString(l.path)
+}
+
+// Apply implements bundle.Mutator.
+func (v *validate) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	return bundle.Apply(ctx, b, bundle.Parallel(
+		JobClusterKeyDefined(),
+		FilesToSync(),
+	))
+}
+
+// Name implements bundle.Mutator.
+func (v *validate) Name() string {
+	return "validate"
+}
+
+func Validate() bundle.Mutator {
+	return &validate{}
+}

bundle/parallel.go

@@ -0,0 +1,37 @@
+package bundle
+
+import (
+	"context"
+	"sync"
+
+	"github.com/databricks/cli/libs/diag"
+)
+
+type parallel struct {
+	mutators []Mutator
+}
+
+func (m *parallel) Name() string {
+	return "parallel"
+}
+
+func (m *parallel) Apply(ctx context.Context, b *Bundle) diag.Diagnostics {
+	var wg sync.WaitGroup
+	diags := diag.Diagnostics{}
+	wg.Add(len(m.mutators))
+	for _, mutator := range m.mutators {
+		go func(mutator Mutator) {
+			defer wg.Done()
+			diags = diags.Extend(mutator.Apply(ctx, b))
+		}(mutator)
+	}
+	wg.Wait()
+	return diags
+}
+
+// Parallel runs the given mutators in parallel.
+func Parallel(mutators ...Mutator) Mutator {
+	return &parallel{
+		mutators: mutators,
+	}
+}

bundle/parallel_test.go

@@ -0,0 +1,67 @@
+package bundle
+
+import (
+	"context"
+	"testing"
+
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/stretchr/testify/require"
+)
+
+type addToContainer struct {
+	container *[]int
+	err       bool
+}
+
+func (m *addToContainer) Apply(ctx context.Context, b *Bundle) diag.Diagnostics {
+	if m.err {
+		return diag.Errorf("error")
+	}
+
+	c := *m.container
+	c = append(c, 1)
+	*m.container = c
+	return nil
+}
+
+func (m *addToContainer) Name() string {
+	return "addToContainer"
+}
+
+func TestParallelMutatorWork(t *testing.T) {
+	b := &Bundle{
+		Config: config.Root{},
+	}
+
+	container := []int{}
+	m1 := &addToContainer{container: &container}
+	m2 := &addToContainer{container: &container}
+	m3 := &addToContainer{container: &container}
+
+	m := Parallel(m1, m2, m3)
+
+	// Apply the mutator
+	diags := m.Apply(context.Background(), b)
+	require.Empty(t, diags)
+	require.Len(t, container, 3)
+}
+
+func TestParallelMutatorWorkWithErrors(t *testing.T) {
+	b := &Bundle{
+		Config: config.Root{},
+	}
+
+	container := []int{}
+	m1 := &addToContainer{container: &container}
+	m2 := &addToContainer{container: &container, err: true}
+	m3 := &addToContainer{container: &container}
+
+	m := Parallel(m1, m2, m3)
+
+	// Apply the mutator
+	diags := m.Apply(context.Background(), b)
+	require.Len(t, diags, 1)
+	require.Equal(t, "error", diags[0].Summary)
+	require.Len(t, container, 2)
+}

bundle/tests/job_cluster_key/databricks.yml

@@ -0,0 +1,27 @@
+bundle:
+  name: job_cluster_key
+
+workspace:
+  host: https://acme.cloud.databricks.com/
+
+targets:
+  default:
+    resources:
+      jobs:
+        foo:
+          name: job
+          tasks:
+          - task_key: test
+            job_cluster_key: key
+  development:
+    resources:
+      jobs:
+        foo:
+          job_clusters:
+            - job_cluster_key: key
+              new_cluster:
+                node_type_id: i3.xlarge
+                num_workers: 1
+          tasks:
+          - task_key: test
+            job_cluster_key: key

bundle/tests/job_cluster_key_test.go

@@ -0,0 +1,28 @@
+package config_tests
+
+import (
+	"context"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/validate"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/stretchr/testify/require"
+)
+
+func TestJobClusterKeyNotDefinedTest(t *testing.T) {
+	b := loadTarget(t, "./job_cluster_key", "default")
+
+	diags := bundle.Apply(context.Background(), b, validate.JobClusterKeyDefined())
+	require.Len(t, diags, 1)
+	require.NoError(t, diags.Error())
+	require.Equal(t, diags[0].Severity, diag.Warning)
+	require.Equal(t, diags[0].Summary, "job_cluster_key key is not defined")
+}
+
+func TestJobClusterKeyDefinedTest(t *testing.T) {
+	b := loadTarget(t, "./job_cluster_key", "development")
+
+	diags := bundle.Apply(context.Background(), b, validate.JobClusterKeyDefined())
+	require.Len(t, diags, 0)
+}

cmd/bundle/validate.go

@@ -8,6 +8,7 @@ import (
 	"text/template"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config/validate"
 	"github.com/databricks/cli/bundle/phases"
 	"github.com/databricks/cli/cmd/bundle/utils"
 	"github.com/databricks/cli/cmd/root"
@@ -140,6 +141,7 @@ func newValidateCommand() *cobra.Command {
 		}
 
 		diags = diags.Extend(bundle.Apply(ctx, b, phases.Initialize()))
+		diags = diags.Extend(bundle.Apply(ctx, b, validate.Validate()))
 		if err := diags.Error(); err != nil {
 			return err
 		}