template_path validation for streams and template policies

code/go/internal/validator/semantic/validate_data_stream_streams_template_path.go

@@ -8,13 +8,18 @@ import (
 	"errors"
 	"io/fs"
 	"path"
+	"strings"
 
 	"gopkg.in/yaml.v3"
 
 	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
 	"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
 )
 
+const (
+	defaultTemplatePath = "stream.yml.hbs"
+)
+
 var (
 	errFailedToReadManifest  = errors.New("failed to read manifest")
 	errFailedToParseManifest = errors.New("failed to parse manifest")
@@ -60,17 +65,38 @@ func validateDataStreamManifestTemplates(fsys fspath.FS, dataStreamName string)
 	}
 
 	for _, stream := range manifest.Streams {
+		streamPath := stream.TemplatePath
 		if stream.TemplatePath == "" {
-			continue // template_path is optional
+			// When no template_path is specified, it defaults to "stream.yml.hbs"
+			streamPath = defaultTemplatePath
 		}
 
-		// Check if template file exists
-		templatePath := path.Join("data_stream", dataStreamName, "agent", "stream", stream.TemplatePath)
-		_, err := fs.Stat(fsys, templatePath)
-		if err != nil {
+		// Walk through the "data_stream/<dataStreamName>/agent/stream" directory
+		// This mirrors the logic in fleet where the assets are filtered based on the template_path
+		// https://github.com/elastic/kibana/blob/main/x-pack/platform/plugins/shared/fleet/server/services/package_policy.ts#L3317
+		streamDir := path.Join("data_stream", dataStreamName, "agent", "stream")
+		found := false
+		fs.WalkDir(fsys, streamDir, func(filePath string, d fs.DirEntry, walkErr error) error {
+			if walkErr != nil {
+				return walkErr
+			}
+			if !d.IsDir() && path.Base(filePath) != "" && strings.HasSuffix(filePath, streamPath) {
+				found = true
+				return fs.SkipDir // Stop walking once found
+			}
+			return nil
+		})
+		if !found {
+			if stream.TemplatePath == "" {
+				errs = append(errs, specerrors.NewStructuredErrorf(
+					"file \"%s\" is invalid: stream \"%s\" is missing template file \"%s\": %w",
+					fsys.Path(manifestPath), stream.Input, streamPath, errTemplateNotFound))
+				continue
+			}
 			errs = append(errs, specerrors.NewStructuredErrorf(
 				"file \"%s\" is invalid: stream \"%s\" references template_path \"%s\": %w",
-				fsys.Path(manifestPath), stream.Input, stream.TemplatePath, errTemplateNotFound))
+				fsys.Path(manifestPath), stream.Input, streamPath, errTemplateNotFound))
+			continue
 		}
 	}
 

code/go/internal/validator/semantic/validate_data_stream_streams_template_path_test.go

@@ -33,6 +33,45 @@ streams:
 		err = os.WriteFile(filepath.Join(d, "data_stream", "test", "agent", "stream", "udp.yml.hbs"), []byte("# UDP template"), 0o644)
 		require.NoError(t, err)
 
+		errs := ValidateStreamTemplates(fspath.DirFS(d))
+		require.Empty(t, errs, "expected no validation errors")
+	})
+
+	t.Run("valid_data_stream_with_default_templates", func(t *testing.T) {
+		d := t.TempDir()
+		err := os.MkdirAll(filepath.Join(d, "data_stream", "test", "agent", "stream"), 0o755)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "data_stream", "test", "manifest.yml"), []byte(`
+streams:
+  - input: udp
+    title: Test UDP
+    description: Test UDP stream
+`), 0o644)
+		require.NoError(t, err)
+
+		err = os.WriteFile(filepath.Join(d, "data_stream", "test", "agent", "stream", "stream.yml.hbs"), []byte("# UDP template"), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateStreamTemplates(fspath.DirFS(d))
+		require.Empty(t, errs, "expected no validation errors")
+
+	})
+
+	t.Run("valid_data_stream_with_default_templates_endsWith_stream", func(t *testing.T) {
+		d := t.TempDir()
+		err := os.MkdirAll(filepath.Join(d, "data_stream", "test", "agent", "stream"), 0o755)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "data_stream", "test", "manifest.yml"), []byte(`
+streams:
+  - input: udp
+    title: Test UDP
+    description: Test UDP stream
+`), 0o644)
+		require.NoError(t, err)
+
+		err = os.WriteFile(filepath.Join(d, "data_stream", "test", "agent", "stream", "filestream.yml.hbs"), []byte("# UDP template"), 0o644)
+		require.NoError(t, err)
+
 		errs := ValidateStreamTemplates(fspath.DirFS(d))
 		require.Empty(t, errs, "expected no validation errors")
 
@@ -46,7 +85,7 @@ streams:
 		errs := ValidateStreamTemplates(fspath.DirFS(d))
 		require.NotEmpty(t, errs, "expected validation errors")
 		assert.Len(t, errs, 1)
-		assert.ErrorAs(t, errs[0], &errFailedToReadManifest)
+		assert.ErrorIs(t, errs[0], errFailedToReadManifest)
 	})
 
 	t.Run("err_parse_manifest", func(t *testing.T) {
@@ -68,7 +107,7 @@ streams:
 		errs := ValidateStreamTemplates(fspath.DirFS(d))
 		require.NotEmpty(t, errs, "expected validation errors")
 		assert.Len(t, errs, 1)
-		assert.ErrorAs(t, errs[0], &errFailedToParseManifest)
+		assert.ErrorIs(t, errs[0], errFailedToParseManifest)
 
 	})
 
@@ -88,7 +127,27 @@ streams:
 		errs := ValidateStreamTemplates(fspath.DirFS(d))
 		require.NotEmpty(t, errs, "expected validation errors")
 		assert.Len(t, errs, 1)
-		assert.ErrorAs(t, errs[0], &errTemplateNotFound)
+		assert.ErrorIs(t, errs[0], errTemplateNotFound)
+		assert.ErrorContains(t, errs[0], "stream \"udp\" references template_path \"missing.yml.hbs\": template file not found")
+	})
+
+	t.Run("err_missing_default_template_file", func(t *testing.T) {
+		d := t.TempDir()
+		err := os.MkdirAll(filepath.Join(d, "data_stream", "test", "agent", "stream"), 0o755)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "data_stream", "test", "manifest.yml"), []byte(`
+streams:
+  - input: udp
+    title: Test UDP
+    description: Test UDP stream
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateStreamTemplates(fspath.DirFS(d))
+		require.NotEmpty(t, errs, "expected validation errors")
+		assert.Len(t, errs, 1)
+		assert.ErrorIs(t, errs[0], errTemplateNotFound)
+		assert.ErrorContains(t, errs[0], "stream \"udp\" is missing template file \"stream.yml.hbs\": template file not found")
 	})
 
 }

code/go/internal/validator/semantic/validate_input_policy_template_path.go

@@ -5,7 +5,10 @@
 package semantic
 
 import (
+	"errors"
+	"fmt"
 	"io/fs"
+	"os"
 	"path"
 
 	"gopkg.in/yaml.v3"
@@ -30,10 +33,10 @@ func ValidateInputPolicyTemplates(fsys fspath.FS) specerrors.ValidationErrors {
 
 		PolicyTemplates []struct {
 			Name         string `yaml:"name"`
-			TemplatePath string `yaml:"template_path"` // optional, input type packages
+			TemplatePath string `yaml:"template_path"` // input type packages require this field
 			Inputs       []struct {
 				Title        string `yaml:"title"`
-				TemplatePath string `yaml:"template_path"` // optional, integration type packages
+				TemplatePath string `yaml:"template_path"` // optional for integration packages
 			} `yaml:"inputs"`
 		} `yaml:"policy_templates"`
 	}
@@ -50,7 +53,7 @@ func ValidateInputPolicyTemplates(fsys fspath.FS) specerrors.ValidationErrors {
 				if input.TemplatePath == "" {
 					continue // template_path is optional
 				}
-				err := validateTemplatePath(fsys, input.TemplatePath)
+				err := validateAgentInputTemplatePath(fsys, input.TemplatePath)
 				if err != nil {
 					errs = append(errs, specerrors.NewStructuredErrorf(
 						"file \"%s\" is invalid: policy template \"%s\" references template_path \"%s\": %w",
@@ -60,9 +63,12 @@ func ValidateInputPolicyTemplates(fsys fspath.FS) specerrors.ValidationErrors {
 
 		case "input":
 			if policyTemplate.TemplatePath == "" {
-				continue // template_path is optional
+				errs = append(errs, specerrors.NewStructuredErrorf(
+					"file \"%s\" is invalid: policy template \"%s\" is missing required field \"template_path\"",
+					fsys.Path(manifestPath), policyTemplate.Name))
+				continue
 			}
-			err := validateTemplatePath(fsys, policyTemplate.TemplatePath)
+			err := validateAgentInputTemplatePath(fsys, policyTemplate.TemplatePath)
 			if err != nil {
 				errs = append(errs, specerrors.NewStructuredErrorf(
 					"file \"%s\" is invalid: policy template \"%s\" references template_path \"%s\": %w",
@@ -74,11 +80,14 @@ func ValidateInputPolicyTemplates(fsys fspath.FS) specerrors.ValidationErrors {
 	return errs
 }
 
-func validateTemplatePath(fsys fspath.FS, tmplPath string) error {
+func validateAgentInputTemplatePath(fsys fspath.FS, tmplPath string) error {
 	templatePath := path.Join("agent", "input", tmplPath)
 	_, err := fs.Stat(fsys, templatePath)
 	if err != nil {
-		return errTemplateNotFound
+		if errors.Is(err, os.ErrNotExist) {
+			return errTemplateNotFound
+		}
+		return fmt.Errorf("failed to stat template file %s: %w", fsys.Path(templatePath), err)
 	}
 
 	return nil

code/go/internal/validator/semantic/validate_input_policy_template_path_test.go

@@ -37,6 +37,25 @@ policy_templates:
 
 	})
 
+	t.Run("input_manifest_with_policy_template_missing_template_path", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.MkdirAll(filepath.Join(d, "agent", "input"), 0o755)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: input
+policy_templates:
+  - name: udp
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateInputPolicyTemplates(fspath.DirFS(d))
+		require.NotEmpty(t, errs, "expected no validation errors")
+
+		assert.Len(t, errs, 1)
+		assert.Contains(t, errs[0].Error(), "is missing required field \"template_path\"")
+	})
+
 	t.Run("input_manifest_with_policy_template_missing_template_file", func(t *testing.T) {
 		d := t.TempDir()
 
@@ -53,7 +72,7 @@ policy_templates:
 		errs := ValidateInputPolicyTemplates(fspath.DirFS(d))
 		require.NotEmpty(t, errs, "expected validation errors")
 		assert.Len(t, errs, 1)
-		assert.ErrorAs(t, errs[0], &errTemplateNotFound)
+		assert.ErrorIs(t, errs[0], errTemplateNotFound)
 	})
 	t.Run("integration_manifest_with_policy_template_success", func(t *testing.T) {
 		d := t.TempDir()
@@ -92,6 +111,6 @@ policy_templates:
 		errs := ValidateInputPolicyTemplates(fspath.DirFS(d))
 		require.NotEmpty(t, errs, "expected validation errors")
 		assert.Len(t, errs, 1)
-		assert.ErrorAs(t, errs[0], &errTemplateNotFound)
+		assert.ErrorIs(t, errs[0], errTemplateNotFound)
 	})
 }

code/go/pkg/validator/limits_test.go

@@ -163,6 +163,7 @@ func (fs *mockFS) Good() *mockFS {
 		newMockFile("_dev/deploy/docker/docker-compose.yml").WithContent("version: 2.3"),
 		newMockFile("data_stream/foo/manifest.yml").WithContent(datastreamManifestYml),
 		newMockFile("data_stream/foo/fields/base-fields.yml").WithContent(fieldsYml),
+		newMockFile("data_stream/foo/agent/stream/stream.yml.hbs").WithContent("# Stream template"),
 	)
 }
 

spec/changelog.yml

@@ -4,9 +4,6 @@
 ##
 - version: 3.6.0-next
   changes:
-    - description: Add validation for template_path in policy templates and data streams.
-      type: enhancement
-      link: https://github.com/elastic/package-spec/pull/986
     # Pending on https://github.com/elastic/kibana/pull/186974
     - description: Add support for `slo` assets.
       type: enhancement
@@ -15,6 +12,9 @@
     - description: Add support for semantic_text field definition.
       type: enhancement
       link: https://github.com/elastic/package-spec/pull/807
+    - description: Add validation for template_path in policy templates and data streams.
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/986
 - version: 3.5.1-next
   changes:
     - description: Input packages don't require to define fields.

spec/input/manifest.spec.yml

@@ -94,6 +94,7 @@ spec:
           - description
           - type
           - input
+          - template_path
     icons:
       $ref: "../integration/manifest.spec.yml#/definitions/icons"
     screenshots:

spec/integration/data_stream/manifest.spec.yml

@@ -586,6 +586,7 @@ spec:
           template_path:
             description: "Path to Elasticsearch index template for stream."
             type: string
+            default: "stream.yml.hbs"
           required_vars:
             $ref: "#/definitions/required_vars"
           vars:

test/packages/bad_duplicated_fields_input/manifest.yml

@@ -27,6 +27,7 @@ policy_templates:
     type: logs
     input: log_file
     description: Collect sample logs
+    template_path: input.yml.hbs
     vars:
       - name: paths
         required: true

test/packages/good_v3/manifest.yml

@@ -94,6 +94,7 @@ policy_templates:
               - http://127.0.0.1
             hide_in_deployment_modes:
               - agentless
+        template_path: apache_metrics.yml.hbs
       - type: httpjson
         title: Collect data via HTTP JSON API
         description: Collecting data from HTTP JSON API (default only)
@@ -110,6 +111,7 @@ policy_templates:
             min_duration: 10s
             max_duration: 4h3m2s1ms
             default: 1m
+        template_path: httpjson.yml.hbs
   - name: apache-agentless
     title: Apache logs and metrics in agentless
     description: Collect logs and metrics from Apache instances in agentless
@@ -143,6 +145,7 @@ policy_templates:
             show_user: true
             default:
               - http://127.0.0.1
+        template_path: apache_metrics.yml.hbs
       - type: aws/s3
         title: Collect S3 logs (agentless only)
         description: Collecting logs from AWS S3 in agentless mode
@@ -154,6 +157,7 @@ policy_templates:
             title: S3 Bucket Name
             show_user: true
             required: true
+        template_path: aws_s3.yml.hbs
 owner:
   github: elastic/foobar
   type: elastic

test/packages/input_policy_template_invalid/manifest.yml

@@ -1,4 +1,4 @@
-format_version: 3.4.1
+format_version: 3.6.0
 name: input_policy_template_valid
 title: "New Package"
 version: 0.0.1

test/packages/input_policy_template_valid/manifest.yml

@@ -1,4 +1,4 @@
-format_version: 3.4.1
+format_version: 3.6.0
 name: input_policy_template_valid
 title: "New Package"
 version: 0.0.1

test/packages/integration_policy_template_invalid/manifest.yml

@@ -1,4 +1,4 @@
-format_version: 3.4.1
+format_version: 3.6.0
 name: integration_policy_template_valid
 title: "New Package"
 version: 0.0.1

test/packages/integration_policy_template_valid/manifest.yml

@@ -1,4 +1,4 @@
-format_version: 3.4.1
+format_version: 3.6.0
 name: integration_policy_template_valid
 title: "New Package"
 version: 0.0.1

test/packages/missing_required_fields_input/manifest.yml

@@ -27,6 +27,7 @@ policy_templates:
     input: log_file
     title: Sample logs
     description: Collect sample logs
+    template_path: input.yml.hbs
     vars:
       - name: paths
         required: true

test/packages/stream_templates_invalid/manifest.yml

@@ -1,4 +1,4 @@
-format_version: 3.4.1
+format_version: 3.6.0
 name: stream_templates_valid
 title: "Stream Templates Valid Test"
 version: 0.0.1
@@ -31,6 +31,7 @@ policy_templates:
       - type: logfile
         title: Collect sample logs from instances
         description: Collecting sample logs
+        template_path: template.yml.hbs
 owner:
   github: elastic/integrations
   type: elastic