add validation for foreach task clusters as well

bundle/config/validate/single_node_cluster.go

@@ -67,31 +67,31 @@ func showSingleNodeClusterWarning(ctx context.Context, v dyn.Value) bool {
 
 	profile, ok := conf.SparkConf["spark.databricks.cluster.profile"]
 	if !ok {
-		log.Warnf(ctx, "spark_conf spark.databricks.cluster.profile not found in single-node cluster spec")
+		log.Debugf(ctx, "spark_conf spark.databricks.cluster.profile not found in single-node cluster spec")
 		return true
 	}
 	if profile != "singleNode" {
-		log.Warnf(ctx, "spark_conf spark.databricks.cluster.profile is not singleNode in single-node cluster spec: %s", profile)
+		log.Debugf(ctx, "spark_conf spark.databricks.cluster.profile is not singleNode in single-node cluster spec: %s", profile)
 		return true
 	}
 
 	master, ok := conf.SparkConf["spark.master"]
 	if !ok {
-		log.Warnf(ctx, "spark_conf spark.master not found in single-node cluster spec")
+		log.Debugf(ctx, "spark_conf spark.master not found in single-node cluster spec")
 		return true
 	}
 	if !strings.HasPrefix(master, "local") {
-		log.Warnf(ctx, "spark_conf spark.master does not start with local in single-node cluster spec: %s", master)
+		log.Debugf(ctx, "spark_conf spark.master does not start with local in single-node cluster spec: %s", master)
 		return true
 	}
 
 	resourceClass, ok := conf.CustomTags["ResourceClass"]
 	if !ok {
-		log.Warnf(ctx, "custom_tag ResourceClass not found in single-node cluster spec")
+		log.Debugf(ctx, "custom_tag ResourceClass not found in single-node cluster spec")
 		return true
 	}
 	if resourceClass != "SingleNode" {
-		log.Warnf(ctx, "custom_tag ResourceClass is not SingleNode in single-node cluster spec: %s", resourceClass)
+		log.Debugf(ctx, "custom_tag ResourceClass is not SingleNode in single-node cluster spec: %s", resourceClass)
 		return true
 	}
 
@@ -108,6 +108,8 @@ func (m *singleNodeCluster) Apply(ctx context.Context, rb bundle.ReadOnlyBundle)
 		dyn.NewPattern(dyn.Key("resources"), dyn.Key("jobs"), dyn.AnyKey(), dyn.Key("job_clusters"), dyn.AnyIndex(), dyn.Key("new_cluster")),
 		// Job task clusters
 		dyn.NewPattern(dyn.Key("resources"), dyn.Key("jobs"), dyn.AnyKey(), dyn.Key("tasks"), dyn.AnyIndex(), dyn.Key("new_cluster")),
+		// Job for each task clusters
+		dyn.NewPattern(dyn.Key("resources"), dyn.Key("jobs"), dyn.AnyKey(), dyn.Key("tasks"), dyn.AnyIndex(), dyn.Key("for_each_task"), dyn.Key("task"), dyn.Key("new_cluster")),
 		// Pipeline clusters
 		dyn.NewPattern(dyn.Key("resources"), dyn.Key("pipelines"), dyn.AnyKey(), dyn.Key("clusters"), dyn.AnyIndex()),
 	}

bundle/config/validate/single_node_cluster_test.go

@@ -16,8 +16,12 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestValidateSingleNodeClusterFail(t *testing.T) {
-	failCases := []struct {
+func failCases() []struct {
+	name       string
+	sparkConf  map[string]string
+	customTags map[string]string
+} {
+	return []struct {
 		name       string
 		sparkConf  map[string]string
 		customTags map[string]string
@@ -83,12 +87,13 @@ func TestValidateSingleNodeClusterFail(t *testing.T) {
 			customTags: map[string]string{"ResourceClass": "SingleNode"},
 		},
 	}
+}
 
+func TestValidateSingleNodeClusterFailForInteractiveClusters(t *testing.T) {
 	ctx := context.Background()
 
-	// Interactive clusters.
-	for _, tc := range failCases {
-		t.Run("interactive_"+tc.name, func(t *testing.T) {
+	for _, tc := range failCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -123,10 +128,13 @@ func TestValidateSingleNodeClusterFail(t *testing.T) {
 			}, diags)
 		})
 	}
+}
 
-	// Job clusters.
-	for _, tc := range failCases {
-		t.Run("job_"+tc.name, func(t *testing.T) {
+func TestValidateSingleNodeClusterFailForJobClusters(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range failCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -170,10 +178,13 @@ func TestValidateSingleNodeClusterFail(t *testing.T) {
 
 		})
 	}
+}
 
-	// Job task clusters.
-	for _, tc := range failCases {
-		t.Run("task_"+tc.name, func(t *testing.T) {
+func TestValidateSingleNodeClusterFailForJobTaskClusters(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range failCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -216,10 +227,13 @@ func TestValidateSingleNodeClusterFail(t *testing.T) {
 			}, diags)
 		})
 	}
+}
 
-	// Pipeline clusters.
-	for _, tc := range failCases {
-		t.Run("pipeline_"+tc.name, func(t *testing.T) {
+func TestValidateSingleNodeClusterFailForPipelineClusters(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range failCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -259,14 +273,72 @@ func TestValidateSingleNodeClusterFail(t *testing.T) {
 			}, diags)
 		})
 	}
-
 }
 
-func TestValidateSingleNodeClusterPass(t *testing.T) {
+func TestValidateSingleNodeClusterFailForJobForEachTaskCluster(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range failCases() {
+		t.Run(tc.name, func(t *testing.T) {
+			b := &bundle.Bundle{
+				Config: config.Root{
+					Resources: config.Resources{
+						Jobs: map[string]*resources.Job{
+							"foo": {
+								JobSettings: &jobs.JobSettings{
+									Tasks: []jobs.Task{
+										{
+											ForEachTask: &jobs.ForEachTask{
+												Task: jobs.Task{
+													NewCluster: &compute.ClusterSpec{
+														ClusterName: "my_cluster",
+														SparkConf:   tc.sparkConf,
+														CustomTags:  tc.customTags,
+													},
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			}
+
+			bundletest.SetLocation(b, "resources.jobs.foo.tasks[0].for_each_task.task.new_cluster", []dyn.Location{{File: "e.yml", Line: 1, Column: 1}})
+
+			// We can't set num_workers to 0 explicitly in the typed configuration.
+			// Do it on the dyn.Value directly.
+			bundletest.Mutate(t, b, func(v dyn.Value) (dyn.Value, error) {
+				return dyn.Set(v, "resources.jobs.foo.tasks[0].for_each_task.task.new_cluster.num_workers", dyn.V(0))
+			})
+
+			diags := bundle.ApplyReadOnly(ctx, bundle.ReadOnly(b), SingleNodeCluster())
+			assert.Equal(t, diag.Diagnostics{
+				{
+					Severity:  diag.Warning,
+					Summary:   singleNodeWarningSummary,
+					Detail:    singleNodeWarningDetail,
+					Locations: []dyn.Location{{File: "e.yml", Line: 1, Column: 1}},
+					Paths:     []dyn.Path{dyn.MustPathFromString("resources.jobs.foo.tasks[0].for_each_task.task.new_cluster")},
+				},
+			}, diags)
+		})
+	}
+}
+
+func passCases() []struct {
+	name       string
+	numWorkers *int
+	sparkConf  map[string]string
+	customTags map[string]string
+	policyId   string
+} {
 	zero := 0
 	one := 1
 
-	passCases := []struct {
+	return []struct {
 		name       string
 		numWorkers *int
 		sparkConf  map[string]string
@@ -297,12 +369,13 @@ func TestValidateSingleNodeClusterPass(t *testing.T) {
 			numWorkers: &zero,
 		},
 	}
+}
 
+func TestValidateSingleNodeClusterPassInteractiveClusters(t *testing.T) {
 	ctx := context.Background()
 
-	// Interactive clusters.
-	for _, tc := range passCases {
-		t.Run("interactive_"+tc.name, func(t *testing.T) {
+	for _, tc := range passCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -329,10 +402,13 @@ func TestValidateSingleNodeClusterPass(t *testing.T) {
 			assert.Empty(t, diags)
 		})
 	}
+}
 
-	// Job clusters.
-	for _, tc := range passCases {
-		t.Run("job_"+tc.name, func(t *testing.T) {
+func TestValidateSingleNodeClusterPassJobClusters(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range passCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -366,10 +442,13 @@ func TestValidateSingleNodeClusterPass(t *testing.T) {
 			assert.Empty(t, diags)
 		})
 	}
+}
 
-	// Job task clusters.
-	for _, tc := range passCases {
-		t.Run("task_"+tc.name, func(t *testing.T) {
+func TestValidateSingleNodeClusterPassJobTaskClusters(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range passCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -403,10 +482,13 @@ func TestValidateSingleNodeClusterPass(t *testing.T) {
 			assert.Empty(t, diags)
 		})
 	}
+}
 
-	// Pipeline clusters.
-	for _, tc := range passCases {
-		t.Run("pipeline_"+tc.name, func(t *testing.T) {
+func TestValidateSingleNodeClusterPassPipelineClusters(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range passCases() {
+		t.Run(tc.name, func(t *testing.T) {
 			b := &bundle.Bundle{
 				Config: config.Root{
 					Resources: config.Resources{
@@ -438,3 +520,47 @@ func TestValidateSingleNodeClusterPass(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateSingleNodeClusterPassJobForEachTaskCluster(t *testing.T) {
+	ctx := context.Background()
+
+	for _, tc := range passCases() {
+		t.Run(tc.name, func(t *testing.T) {
+			b := &bundle.Bundle{
+				Config: config.Root{
+					Resources: config.Resources{
+						Jobs: map[string]*resources.Job{
+							"foo": {
+								JobSettings: &jobs.JobSettings{
+									Tasks: []jobs.Task{
+										{
+											ForEachTask: &jobs.ForEachTask{
+												Task: jobs.Task{
+													NewCluster: &compute.ClusterSpec{
+														ClusterName: "my_cluster",
+														SparkConf:   tc.sparkConf,
+														CustomTags:  tc.customTags,
+														PolicyId:    tc.policyId,
+													},
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			}
+
+			if tc.numWorkers != nil {
+				bundletest.Mutate(t, b, func(v dyn.Value) (dyn.Value, error) {
+					return dyn.Set(v, "resources.jobs.foo.tasks[0].for_each_task.task.new_cluster.num_workers", dyn.V(*tc.numWorkers))
+				})
+			}
+
+			diags := bundle.ApplyReadOnly(ctx, bundle.ReadOnly(b), SingleNodeCluster())
+			assert.Empty(t, diags)
+		})
+	}
+}