pipelines: warn about unknown attributes (#4325)

## Changes
Warn about unknown attributes in spark-pipeline.yml

## Why
It avoids cases when spark-pipeline.yml is incorrectly formatted and
"pipelines generate" command produces incorrect output.

## Tests
Acceptance tests

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
@@ -235,29 +249,18 @@ func convertToResources(spec *sdpPipeline, resourceName, srcFolder string) (map[
 		schema = spec.Database
 	}
 
-	var libraries []pipelines.PipelineLibrary
 	environment := pipelines.PipelinesEnvironment{
 		Dependencies: []string{
 			"--editable ${workspace.file_path}",
 		},
 	}
 
-	for _, lib := range spec.Libraries {
-		if lib.Glob.Include != "" {
-			relativeIncludePath := filepath.ToSlash(filepath.Join(relativePath, lib.Glob.Include))
-
-			libraries = append(libraries, pipelines.PipelineLibrary{
-				Glob: &pipelines.PathPattern{Include: relativeIncludePath},
-			})
-		}
-	}
-
 	environmentDyn, err := convert.FromTyped(environment, dyn.NilValue)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert environments into dyn.Value: %w", err)
 	}
 
-	librariesDyn, err := convert.FromTyped(libraries, dyn.NilValue)
+	librariesDyn, err := convertLibraries(relativePath, spec.Libraries)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert libraries into dyn.Value: %w", err)
 	}
@@ -307,3 +310,34 @@ func convertToResources(spec *sdpPipeline, resourceName, srcFolder string) (map[
 
 	return resourcesMap, nil
 }
+
+// convertLibraries converts SDP libraries into DABs YAML format
+//
+// relativePath contains a path to append into SDP libraries path to make
+// them relative to generated DABs YAML
+func convertLibraries(relativePath string, specLibraries []sdpPipelineLibrary) (dyn.Value, error) {
+	var libraries []pipelines.PipelineLibrary
+
+	for _, lib := range specLibraries {
+		if lib.Glob.Include != "" {
+			relativeIncludePath := filepath.ToSlash(filepath.Join(relativePath, lib.Glob.Include))
+
+			libraries = append(libraries, pipelines.PipelineLibrary{
+				Glob: &pipelines.PathPattern{Include: relativeIncludePath},
+			})
+		}
+	}
+
+	librariesDyn, err := convert.FromTyped(libraries, dyn.NilValue)
+	if err != nil {
+		return dyn.InvalidValue, fmt.Errorf("failed to convert libraries into dyn.Value: %w", err)
+	}
+
+	// FromTyped returns NilValue if libraries is an empty array
+	if librariesDyn.Kind() == dyn.KindNil {
+		// we always want to leave empty array as a placeholder in generated YAML
+		return dyn.V([]dyn.Value{}), nil
+	}
+
+	return librariesDyn, nil
+}