pipelines: warn about unknown attributes

Author: kanterov

acceptance/pipelines/generate/unknown-attribute/out.test.toml

@@ -0,0 +1,5 @@
+Warning: unknown field: unknown_attribute
+  in src/my_pipeline/spark-pipeline.yml:2:1
+
+Generated pipeline configuration: resources/my_pipeline.pipeline.yml
+

acceptance/pipelines/generate/unknown-attribute/output.txt

@@ -0,0 +1,12 @@
+resources:
+  pipelines:
+    my_pipeline:
+      name: My Pipeline
+      catalog: ${var.catalog}
+      schema: ${var.schema}
+      root_path: ../src/my_pipeline
+      serverless: true
+      libraries: []
+      environment:
+        dependencies:
+          - --editable ${workspace.file_path}

acceptance/pipelines/generate/unknown-attribute/output/my_pipeline.pipeline.yml

@@ -0,0 +1,5 @@
+export ENABLE_PIPELINES_CLI=1
+
+$CLI pipelines generate --existing-pipeline-dir src/my_pipeline
+
+mv resources output/

acceptance/pipelines/generate/unknown-attribute/script

@@ -0,0 +1,2 @@
+name: My Pipeline
+unknown_attribute: foo

acceptance/pipelines/generate/unknown-attribute/src/my_pipeline/spark-pipeline.yml

@@ -1,6 +1,7 @@
 package pipelines
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -13,6 +14,7 @@ import (
 	"github.com/databricks/cli/libs/dyn/convert"
 	"github.com/databricks/cli/libs/dyn/yamlloader"
 	"github.com/databricks/cli/libs/dyn/yamlsaver"
+	"github.com/databricks/cli/libs/logdiag"
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
 	"github.com/spf13/cobra"
 )
@@ -50,8 +52,10 @@ Use --existing-pipeline-dir to generate pipeline configuration from spark-pipeli
 	}
 
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		ctx := logdiag.InitContext(cmd.Context())
+		cmd.SetContext(ctx)
+
 		folderPath := existingPipelineDir
-		ctx := cmd.Context()
 
 		info, err := validateAndParsePath(folderPath)
 		if err != nil {
@@ -66,7 +70,7 @@ Use --existing-pipeline-dir to generate pipeline configuration from spark-pipeli
 			}
 		}
 
-		spec, err := parseSparkPipelineYAML(sparkPipelineFile)
+		spec, err := parseSparkPipelineYAML(ctx, sparkPipelineFile)
 		if err != nil {
 			return fmt.Errorf("failed to parse %s: %w", sparkPipelineFile, err)
 		}
@@ -181,6 +185,7 @@ type sdpPipeline struct {
 	Catalog       string               `json:"catalog,omitempty"`
 	Database      string               `json:"database,omitempty"`
 	Libraries     []sdpPipelineLibrary `json:"libraries,omitempty"`
+	Storage       string               `json:"storage,omitempty"`
 	Configuration map[string]string    `json:"configuration,omitempty"`
 }
 
@@ -195,7 +200,7 @@ type sdpPipelineLibraryGlob struct {
 }
 
 // parseSparkPipelineYAML parses a spark-pipeline.yml file.
-func parseSparkPipelineYAML(filePath string) (*sdpPipeline, error) {
+func parseSparkPipelineYAML(ctx context.Context, filePath string) (*sdpPipeline, error) {
 	file, err := os.Open(filePath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open %s: %w", filePath, err)
@@ -208,9 +213,18 @@ func parseSparkPipelineYAML(filePath string) (*sdpPipeline, error) {
 	}
 
 	out := sdpPipeline{}
-	err = convert.ToTyped(&out, dv)
+	normalized, diags := convert.Normalize(&out, dv)
+	if diags.HasError() {
+		return nil, fmt.Errorf("failed to parse %s: %w", filePath, diags.Error())
+	}
+
+	for _, diag := range diags {
+		logdiag.LogDiag(ctx, diag)
+	}
+
+	err = convert.ToTyped(&out, normalized)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse %s: %w", filePath, err)
+		return nil, fmt.Errorf("failed to parse %s: %w", filePath, diags.Error())
 	}
 
 	return &out, nil
@@ -261,6 +275,9 @@ func convertToResources(spec *sdpPipeline, resourceName, srcFolder string) (map[
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert libraries into dyn.Value: %w", err)
 	}
+	if librariesDyn.Kind() == dyn.KindNil {
+		librariesDyn = dyn.V([]dyn.Value{})
+	}
 
 	// maps are unordered, and saver is sorting keys by dyn.Location
 	// this is helper function to monotonically assign locations as keys are created