Add support for inputs with qualified names

code/go/internal/validator/semantic/validate_integration_input_qualifier.go

@@ -0,0 +1,79 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"fmt"
+	"io/fs"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+	"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
+)
+
+type integrationInputQualifier struct {
+	Name string `yaml:"name"`
+	Type string `yaml:"type"`
+}
+
+type integrationPolicyTemplateQualifier struct {
+	Name   string                      `yaml:"name"`
+	Inputs []integrationInputQualifier `yaml:"inputs"`
+}
+
+type integrationPackageManifestQualifier struct {
+	Type            string                               `yaml:"type"`
+	PolicyTemplates []integrationPolicyTemplateQualifier `yaml:"policy_templates"`
+}
+
+// ValidateIntegrationInputQualifier checks that if a policy template contains
+// multiple inputs of the same type, all of them must have a name set. Without
+// names, Fleet cannot distinguish them, leading to the ambiguity this field
+// aims to solve.
+func ValidateIntegrationInputQualifier(fsys fspath.FS) specerrors.ValidationErrors {
+	manifestPath := "manifest.yml"
+	data, err := fs.ReadFile(fsys, manifestPath)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: failed to read manifest: %w", fsys.Path(manifestPath), err)}
+	}
+
+	var manifest integrationPackageManifestQualifier
+	if err := yaml.Unmarshal(data, &manifest); err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: failed to parse manifest: %w", fsys.Path(manifestPath), err)}
+	}
+
+	if manifest.Type != integrationPackageType {
+		return nil
+	}
+
+	return validateInputQualifiers(fsys, manifest, manifestPath)
+}
+
+func validateInputQualifiers(fsys fspath.FS, manifest integrationPackageManifestQualifier, manifestPath string) specerrors.ValidationErrors {
+	var errs specerrors.ValidationErrors
+
+	for _, policyTemplate := range manifest.PolicyTemplates {
+		typeCounts := make(map[string]int)
+		for _, input := range policyTemplate.Inputs {
+			typeCounts[input.Type]++
+		}
+
+		reported := make(map[string]bool)
+		for _, input := range policyTemplate.Inputs {
+			if typeCounts[input.Type] > 1 && input.Name == "" && !reported[input.Type] {
+				reported[input.Type] = true
+				errs = append(errs, specerrors.NewStructuredError(
+					fmt.Errorf("file \"%s\" is invalid: policy template \"%s\": input with type \"%s\" must have a name when multiple inputs of the same type are present",
+						fsys.Path(manifestPath), policyTemplate.Name, input.Type),
+					specerrors.CodeIntegrationInputQualifierRequired))
+			}
+		}
+	}
+
+	return errs
+}

code/go/internal/validator/semantic/validate_integration_input_qualifier_test.go

@@ -0,0 +1,134 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+	"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
+)
+
+func TestValidateIntegrationInputQualifier(t *testing.T) {
+	tests := map[string]struct {
+		manifest      string
+		expectedErrs  []string
+		expectedCodes []string
+	}{
+		"single_input_per_type_no_name": {
+			manifest: `
+type: integration
+policy_templates:
+  - name: nginx
+    inputs:
+      - type: logfile
+        title: Logs
+        description: Collect logs
+      - type: httpjson
+        title: Metrics
+        description: Collect metrics
+`,
+		},
+		"multiple_inputs_same_type_with_names": {
+			manifest: `
+type: integration
+policy_templates:
+  - name: nginx
+    inputs:
+      - name: filelog_otel
+        type: otelcol
+        title: Logs
+        description: Collect logs
+      - name: nginx_otel
+        type: otelcol
+        title: Metrics
+        description: Collect metrics
+`,
+		},
+		"multiple_inputs_same_type_no_names": {
+			manifest: `
+type: integration
+policy_templates:
+  - name: nginx
+    inputs:
+      - type: otelcol
+        title: Logs
+        description: Collect logs
+      - type: otelcol
+        title: Metrics
+        description: Collect metrics
+`,
+			expectedErrs: []string{
+				`policy template "nginx": input with type "otelcol" must have a name when multiple inputs of the same type are present`,
+			},
+			expectedCodes: []string{
+				specerrors.CodeIntegrationInputQualifierRequired,
+			},
+		},
+		"multiple_inputs_same_type_partial_names": {
+			manifest: `
+type: integration
+policy_templates:
+  - name: nginx
+    inputs:
+      - name: filelog_otel
+        type: otelcol
+        title: Logs
+        description: Collect logs
+      - type: otelcol
+        title: Metrics
+        description: Collect metrics
+`,
+			expectedErrs: []string{
+				`policy template "nginx": input with type "otelcol" must have a name when multiple inputs of the same type are present`,
+			},
+			expectedCodes: []string{
+				specerrors.CodeIntegrationInputQualifierRequired,
+			},
+		},
+		"non_integration_package": {
+			manifest: `
+type: input
+policy_templates:
+  - name: sample
+    inputs:
+      - type: logfile
+        title: Logs
+        description: Collect logs
+      - type: logfile
+        title: More logs
+        description: Collect more logs
+`,
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			d := t.TempDir()
+			err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(tc.manifest), 0o644)
+			require.NoError(t, err)
+
+			errs := ValidateIntegrationInputQualifier(fspath.DirFS(d))
+
+			if len(tc.expectedErrs) == 0 {
+				require.Empty(t, errs)
+				return
+			}
+
+			require.Len(t, errs, len(tc.expectedErrs))
+			for i, expected := range tc.expectedErrs {
+				assert.True(t, strings.Contains(errs[i].Error(), expected),
+					"error %q does not contain %q", errs[i].Error(), expected)
+				assert.Equal(t, tc.expectedCodes[i], errs[i].Code())
+			}
+		})
+	}
+}

code/go/internal/validator/spec.go

@@ -237,6 +237,7 @@ func (s Spec) rules(pkgType string, rootSpec spectypes.ItemSpec) validationRules
 		{fn: semantic.ValidateKibanaTagDuplicates},
 		{fn: semantic.ValidatePipelineOnFailure, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
 		{fn: semantic.ValidateIntegrationInputsDeprecation, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
+		{fn: semantic.ValidateIntegrationInputQualifier, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
 		{fn: semantic.ValidateDeprecatedReplacedBy, since: semver.MustParse("3.6.0")},
 		{fn: semantic.ValidatePackageReferences, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
 		{fn: semantic.ValidateTestPackageRequirements, types: []string{"integration"}, since: semver.MustParse("3.6.0")},

code/go/pkg/specerrors/constants.go

@@ -25,4 +25,5 @@ const (
 	CodeKibanaTagDuplicates                 = "SVR00007"
 	CodePipelineOnFailureEventKind          = "SVR00008"
 	CodePipelineOnFailureMessage            = "SVR00009"
+	CodeIntegrationInputQualifierRequired   = "SVR00010"
 )

code/go/pkg/validator/validator_test.go

@@ -36,6 +36,7 @@ func TestValidateFile(t *testing.T) {
 		"good_var_groups_input":                  {},
 		"good_input":                             {},
 		"good_input_otel":                        {},
+		"good_input_qualifier":                   {},
 		"good_input_dynamic_signal_type":         {},
 		"good_input_profiles":                    {},
 		"good_input_template_paths":              {},
@@ -421,6 +422,19 @@ func TestValidateFile(t *testing.T) {
 				"field policy_templates.0.inputs.0.type: Must not be present",
 			},
 		},
+		"bad_input_qualifier_ambiguous": {
+			"manifest.yml",
+			[]string{
+				`policy template "nginx": input with type "otelcol" must have a name when multiple inputs of the same type are present (SVR00010)`,
+			},
+		},
+		"bad_input_qualifier_old_version": {
+			"manifest.yml",
+			[]string{
+				"field policy_templates.0.inputs.0.type: Must not be present",
+				"field policy_templates.0.inputs.0: Additional property name is not allowed",
+			},
+		},
 		"bad_input_dynamic_signal_types_old_version": {
 			"manifest.yml",
 			[]string{

spec/changelog.yml

@@ -13,6 +13,10 @@
     - description: Add var_groups support to policy template and input levels in integration packages, and to policy template and package root levels in input packages.
       type: enhancement
       link: https://github.com/elastic/package-spec/pull/1120
+    # Pending on https://github.com/elastic/kibana/pull/262138
+    - description: Add support for named inputs in policy templates to disambiguate multiple inputs of the same type.
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1135
 - version: 3.6.0
   changes:
     - description: Add pipeline tag validations.

spec/integration/manifest.spec.yml

@@ -779,6 +779,12 @@ spec:
               type: object
               additionalProperties: false
               properties:
+                name:
+                  description: >
+                    Unique name for this input within the policy template. When set, data streams
+                    reference this input by name instead of type, allowing multiple inputs of the
+                    same type to coexist in the same policy template.
+                  type: string
                 type:
                   description: Type of input.
                   type: string
@@ -933,6 +939,9 @@ spec:
 versions:
   - before: 3.6.0
     patch:
+      # Input qualifier (named inputs).
+      - op: remove
+        path: "/properties/policy_templates/items/properties/inputs/items/properties/name"
       # Support for otelcol and dynamic_signal_types in integration packages.
       - op: add
         path: "/properties/policy_templates/items/properties/inputs/items/properties/type"

test/packages/bad_input_qualifier_ambiguous/changelog.yml

@@ -0,0 +1,5 @@
+- version: 0.0.1
+  changes:
+    - description: Initial release
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1234

test/packages/bad_input_qualifier_ambiguous/data_stream/access/agent/stream/stream.yml.hbs

@@ -0,0 +1,15 @@
+receivers:
+  filelog:
+    include:
+{{#each paths}}
+      - {{this}}
+{{/each}}
+exporters:
+  {{#each exporters}}
+  {{this}}
+  {{/each}}
+service:
+  pipelines:
+    {{#each pipelines}}
+    {{this}}
+    {{/each}}

test/packages/bad_input_qualifier_ambiguous/data_stream/access/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,18 @@
+---
+description: Pipeline for processing Nginx access logs
+processors:
+  - set:
+      tag: set_data_stream_type
+      field: data_stream.type
+      value: logs
+on_failure:
+  - set:
+      field: event.kind
+      value: pipeline_error
+  - set:
+      field: error.message
+      value: >-
+        Processor '{{{ _ingest.on_failure_processor_type }}}'
+        with tag '{{{ _ingest.on_failure_processor_tag }}}'
+        in pipeline '{{{ _ingest.pipeline }}}'
+        failed with message '{{{ _ingest.on_failure_message }}}'

test/packages/bad_input_qualifier_ambiguous/data_stream/access/fields/fields.yml

@@ -0,0 +1,12 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: Event timestamp.

test/packages/bad_input_qualifier_ambiguous/data_stream/access/manifest.yml

@@ -0,0 +1,15 @@
+title: Nginx Access Logs
+type: logs
+streams:
+  - input: filelog_otel
+    title: Collect Nginx access logs via filelog OTel receiver
+    description: Collecting Nginx access logs via filelog OpenTelemetry receiver.
+    vars:
+      - name: paths
+        type: text
+        title: Log file paths
+        multi: true
+        required: true
+        show_user: true
+        default:
+          - /var/log/nginx/access.log

test/packages/bad_input_qualifier_ambiguous/data_stream/stubstatus/agent/stream/stream.yml.hbs

@@ -0,0 +1,13 @@
+receivers:
+  nginx:
+    endpoint: http://127.0.0.1/nginx_status
+    collection_interval: {{period}}
+exporters:
+  {{#each exporters}}
+  {{this}}
+  {{/each}}
+service:
+  pipelines:
+    {{#each pipelines}}
+    {{this}}
+    {{/each}}

test/packages/bad_input_qualifier_ambiguous/data_stream/stubstatus/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,18 @@
+---
+description: Pipeline for processing Nginx stub status metrics
+processors:
+  - set:
+      tag: set_data_stream_type
+      field: data_stream.type
+      value: metrics
+on_failure:
+  - set:
+      field: event.kind
+      value: pipeline_error
+  - set:
+      field: error.message
+      value: >-
+        Processor '{{{ _ingest.on_failure_processor_type }}}'
+        with tag '{{{ _ingest.on_failure_processor_tag }}}'
+        in pipeline '{{{ _ingest.pipeline }}}'
+        failed with message '{{{ _ingest.on_failure_message }}}'