Composable packages: input name qualifiers and stream input refs

Makefile

@@ -128,7 +128,7 @@ test-check-packages-with-custom-agent:
 	PACKAGE_TEST_TYPE=with-custom-agent ./scripts/test-check-packages.sh
 
 test-build-install-packages-composable:
-	./scripts/test-composable-packages.sh
+	PACKAGE_TEST_TYPE=composable ./scripts/test-composable-packages.sh
 
 test-build-zip:
 	./scripts/test-build-zip.sh

internal/builder/packages.go

@@ -110,6 +110,20 @@ func BuildPackagesDirectory(packageRoot string, buildDir string) (string, error)
 	return filepath.Join(buildDir, m.Name, m.Version), nil
 }
 
+// ReadBuiltPackageManifest locates the built package directory for packageRoot
+// and reads its manifest. Returns the built root path and parsed manifest.
+func ReadBuiltPackageManifest(packageRoot string) (string, *packages.PackageManifest, error) {
+	builtRoot, err := BuildPackagesDirectory(packageRoot, "")
+	if err != nil {
+		return "", nil, err
+	}
+	builtPkg, err := packages.ReadPackageManifestFromPackageRoot(builtRoot)
+	if err != nil {
+		return "", nil, err
+	}
+	return builtRoot, builtPkg, nil
+}
+
 // buildPackagesZipPath function returns the path to zipped built package.
 func buildPackagesZipPath(packageRoot string) (string, error) {
 	buildPackagesDir, err := buildPackagesRootDirectory()

internal/kibana/packagepolicy.go

@@ -48,17 +48,20 @@ func BuildIntegrationPackagePolicy(
 	inputs := make(map[string]PackagePolicyInput)
 	for _, pt := range manifest.PolicyTemplates {
 		for _, input := range pt.Inputs {
-			inputKey := fmt.Sprintf("%s-%s", pt.Name, input.Type)
-			if input.Type == streamInput && pt.Name == policyTemplate.Name {
+			// effectiveName is the identifier used in data stream manifests: the Name qualifier
+			// when set (disambiguated same-type inputs), otherwise the Type.
+			effectiveName := effectiveInputName(input)
+			inputKey := fmt.Sprintf("%s-%s", pt.Name, effectiveName)
+			if effectiveName == streamInput && pt.Name == policyTemplate.Name {
 				// The target input: enabled with user-provided vars.
 				streams := buildStreamsForInput(streamInput, manifest, dsManifest, enabled, dsVars, datastreams)
 				inputEntry := PackagePolicyInput{
 					Enabled:        enabled,
 					Streams:        streams,
-					inputType:      streamInput,
+					inputType:      input.Type, // Fleet agent input type (e.g. logfile, otelcol), not name qualifier or data_stream type (logs/metrics/traces).
 					policyTemplate: pt.Name,
 				}
-				if foundInput := policyTemplate.FindInputByType(streamInput); foundInput != nil {
+				if foundInput := policyTemplate.FindInput(streamInput); foundInput != nil {
 					iv := SetKibanaVariables(foundInput.Vars, allInputVars)
 					inputEntry.Vars = iv.ToMapStr()
 					inputEntry.legacyVars = iv
@@ -69,7 +72,7 @@ func BuildIntegrationPackagePolicy(
 				// so that sibling stream keys are correct even when multiple policy
 				// templates declare different data_streams lists.
 				ptDatastreams := packages.FilterDatastreamsForPolicyTemplate(allDatastreams, pt)
-				streams := buildStreamsForInput(input.Type, manifest, packages.DataStreamManifest{}, false, common.MapStr{}, ptDatastreams)
+				streams := buildStreamsForInput(effectiveName, manifest, packages.DataStreamManifest{}, false, common.MapStr{}, ptDatastreams)
 				entry := PackagePolicyInput{
 					Enabled:        false,
 					inputType:      input.Type,
@@ -310,3 +313,13 @@ func setVarFromUser(vars Vars, name, varType string, val packages.VarValue) {
 	}
 	vars[name] = Var{Type: varType, Value: val, fromUser: true}
 }
+
+// effectiveInputName returns the identifier used to reference an input from data stream
+// manifests: the Name qualifier when set (for inputs disambiguated by name), or
+// the Type when no qualifier is present.
+func effectiveInputName(input packages.Input) string {
+	if input.Name != "" {
+		return input.Name
+	}
+	return input.Type
+}

internal/kibana/policies.go

@@ -307,6 +307,9 @@ type PackagePolicyInput struct {
 	Streams map[string]PackagePolicyStream `json:"streams,omitempty"`
 
 	// Unexported fields carry metadata used only for legacy API conversion.
+	// inputType is the Elastic Agent / Fleet input type from the package manifest
+	// (e.g. logfile, httpjson, otelcol), not the policy-template or data_stream type
+	// (e.g. logs, metrics, traces).
 	inputType      string
 	policyTemplate string
 	legacyVars     Vars

internal/packages/packages.go

@@ -193,6 +193,7 @@ type Variable struct {
 
 // Input is a single input configuration.
 type Input struct {
+	Name          string     `config:"name,omitempty" json:"name,omitempty" yaml:"name,omitempty"`
 	Type          string     `config:"type" json:"type" yaml:"type"`
 	Package       string     `config:"package,omitempty" json:"package,omitempty" yaml:"package,omitempty"`
 	Vars          []Variable `config:"vars" json:"vars" yaml:"vars"`
@@ -841,11 +842,15 @@ func (dsm *DataStreamManifest) indexTemplateNamePrefix() string {
 	return ""
 }
 
-// FindInputByType returns the input for the provided type.
-func (pt *PolicyTemplate) FindInputByType(inputType string) *Input {
-	for _, input := range pt.Inputs {
-		if input.Type == inputType {
-			return &input
+// FindInput returns the first input matching effectiveName, checked against the
+// input's Name qualifier first (when set) and then its Type. Use this when the
+// effectiveName may be a name qualifier (set on inputs that share a type) rather
+// than a bare type string.
+func (pt *PolicyTemplate) FindInput(effectiveName string) *Input {
+	for i := range pt.Inputs {
+		input := &pt.Inputs[i]
+		if (input.Name != "" && input.Name == effectiveName) || input.Type == effectiveName {
+			return input
 		}
 	}
 	return nil
@@ -906,8 +911,8 @@ func findPolicyTemplateForDataStream(pkg PackageManifest, ds DataStreamManifest,
 
 	var matchedPolicyTemplates []string
 	for _, policyTemplate := range pkg.PolicyTemplates {
-		// Does this policy_template include this input type?
-		if policyTemplate.FindInputByType(inputName) == nil {
+		// Does this policy_template include an input for this identifier (name or type)?
+		if policyTemplate.FindInput(inputName) == nil {
 			continue
 		}
 

internal/requiredinputs/streamdefs.go

@@ -14,10 +14,10 @@ import (
 	"github.com/elastic/elastic-package/internal/packages"
 )
 
-// inputPkgInfo holds the resolved metadata from an input package needed to
-// replace package: references in composable package manifests.
-type inputPkgInfo struct {
-	identifier     string // policy_templates[0].input; if several templates exist, only the first is used
+// inputPolicyTemplateInfo holds the resolved metadata from an input package needed to
+// replace package: references in composable (integrations) package manifests.
+type inputPolicyTemplateInfo struct {
+	input          string // policy_templates[0].input; if several templates exist, only the first is used
 	pkgTitle       string // manifest.title (fallback title)
 	pkgDescription string // manifest.description (fallback description)
 }
@@ -27,6 +27,12 @@ type inputPkgInfo struct {
 // data_stream/*/manifest.yml (streams[]) with the actual input type identifier
 // from the referenced input package, then removes the package: key.
 //
+// When multiple inputs in the same policy template share the same type (e.g., two
+// otelcol inputs from different required packages), it also sets a unique name on
+// each such input (derived from the package name) so Fleet can distinguish them.
+// In that case, data stream manifests use the name as their input reference instead
+// of the type.
+//
 // This step must run last, after mergeVariables, because that step uses
 // stream.Package and input.Package to identify which entries to process.
 // It resolves metadata per required input via buildInputPkgInfoByName, then
@@ -41,32 +47,80 @@ func (r *RequiredInputsResolver) resolveStreamInputTypes(
 		return err
 	}
 
-	if err := applyInputTypesToComposableManifest(manifest, buildRoot, infoByPkg); err != nil {
+	streamInputEffectveNames := buildStreamInputRefs(manifest, infoByPkg)
+
+	if err := applyInputTypesToComposableManifest(manifest, buildRoot, infoByPkg, streamInputEffectveNames); err != nil {
 		return err
 	}
 
-	return applyInputTypesToDataStreamManifests(buildRoot, infoByPkg)
+	return applyInputTypesToDataStreamManifests(buildRoot, infoByPkg, streamInputEffectveNames)
 }
 
-// buildInputPkgInfoByName loads inputPkgInfo for each downloaded required input package path.
-func buildInputPkgInfoByName(inputPkgPaths map[string]string) (map[string]inputPkgInfo, error) {
-	infoByPkg := make(map[string]inputPkgInfo, len(inputPkgPaths))
+// buildStreamInputRefs computes the value to write to streams[].input for each
+// required input package referenced by the integration.
+//
+// By default the ref is the resolved input type (e.g. "logfile", "otelcol").
+// If a policy template contains multiple required input packages that resolve to
+// the same type, those packages use their package name as a stable qualifier;
+// streams[].input references that qualifier (and the corresponding input gets
+// name: <package>) so Fleet can distinguish same-type inputs.
+//
+// Duplicate detection is per policy template (not integration-wide) because only
+// inputs that coexist within a single policy template must be unique. When
+// building the Fleet package policy, inputs are keyed as
+// "{policyTemplate.Name}-{effectiveName}" where effectiveName is input Name when
+// set, otherwise Type.
+// Once a package is qualified, it keeps that ref across policy templates.
+func buildStreamInputRefs(manifest *packages.PackageManifest, infoByInputPkg map[string]inputPolicyTemplateInfo) map[string]string {
+	refs := make(map[string]string, len(infoByInputPkg))
+	// iterate over all policy templates from composable package manifest
+	for _, pt := range manifest.PolicyTemplates {
+		// first count the number of inputs of each type
+		typeCounts := make(map[string]int)
+		for _, input := range pt.Inputs {
+			if input.Package == "" {
+				continue
+			}
+			// integration input type is equivalent to the policy template input identifier in the required input package
+			typeCounts[infoByInputPkg[input.Package].input]++
+		}
+		for _, input := range pt.Inputs {
+			if input.Package == "" {
+				continue
+			}
+			importedInput := infoByInputPkg[input.Package]
+			if typeCounts[importedInput.input] > 1 {
+				refs[input.Package] = input.Package // package name as stable unique qualifier
+			} else if _, exists := refs[input.Package]; !exists {
+				refs[input.Package] = importedInput.input
+			}
+		}
+	}
+	return refs
+}
+
+// buildInputPkgInfoByName loads inputPolicyTemplateInfo for each downloaded required input package path.
+func buildInputPkgInfoByName(inputPkgPaths map[string]string) (map[string]inputPolicyTemplateInfo, error) {
+	infoByInputPkg := make(map[string]inputPolicyTemplateInfo, len(inputPkgPaths))
 	for pkgName, pkgPath := range inputPkgPaths {
 		info, err := loadInputPkgInfo(pkgPath)
 		if err != nil {
 			return nil, fmt.Errorf("loading input package info for %q: %w", pkgName, err)
 		}
-		infoByPkg[pkgName] = info
+		infoByInputPkg[pkgName] = *info
 	}
-	return infoByPkg, nil
+	return infoByInputPkg, nil
 }
 
-// applyInputTypesToComposableManifest sets type (and optional title/description) on
-// package-backed policy template inputs in manifest.yml and drops package:.
+// applyInputTypesToComposableManifest sets type (and optional title/description/name) on
+// package-backed policy template inputs in manifest.yml and drops package field.
+// When streamInputRefs maps a package to its own name (indicating a type conflict), it also
+// sets name on that input so Fleet can distinguish multiple inputs of the same type.
 func applyInputTypesToComposableManifest(
 	manifest *packages.PackageManifest,
 	buildRoot *os.Root,
-	infoByPkg map[string]inputPkgInfo,
+	infoByInputPkg map[string]inputPolicyTemplateInfo,
+	streamInputRefs map[string]string,
 ) error {
 	manifestBytes, err := buildRoot.ReadFile("manifest.yml")
 	if err != nil {
@@ -82,7 +136,7 @@ func applyInputTypesToComposableManifest(
 			if input.Package == "" {
 				continue
 			}
-			info, ok := infoByPkg[input.Package]
+			info, ok := infoByInputPkg[input.Package]
 			if !ok {
 				return fmt.Errorf("input package %q referenced in policy_templates[%d].inputs[%d] not found in required inputs", input.Package, ptIdx, inputIdx)
 			}
@@ -92,7 +146,10 @@ func applyInputTypesToComposableManifest(
 				return fmt.Errorf("getting input node at pt[%d].inputs[%d]: %w", ptIdx, inputIdx, err)
 			}
 
-			upsertKey(inputNode, "type", strVal(info.identifier))
+			upsertKey(inputNode, "type", strVal(info.input))
+			if streamInputRefs[input.Package] == input.Package {
+				upsertKey(inputNode, "name", strVal(input.Package))
+			}
 
 			if mappingValue(inputNode, "title") == nil && info.pkgTitle != "" {
 				upsertKey(inputNode, "title", strVal(info.pkgTitle))
@@ -116,8 +173,11 @@ func applyInputTypesToComposableManifest(
 }
 
 // applyInputTypesToDataStreamManifests sets input on package-backed streams in each
-// data_stream/*/manifest.yml and drops package:.
-func applyInputTypesToDataStreamManifests(buildRoot *os.Root, infoByPkg map[string]inputPkgInfo) error {
+// data_stream/*/manifest.yml and drops package:. The value written to streams[].input
+// is taken from streamInputRefs: the package name when disambiguation is required
+// (i.e. the corresponding policy template input carries a name qualifier), otherwise
+// the type identifier.
+func applyInputTypesToDataStreamManifests(buildRoot *os.Root, infoByInputPkg map[string]inputPolicyTemplateInfo, streamInputRefs map[string]string) error {
 	dsManifestPaths, err := fs.Glob(buildRoot.FS(), "data_stream/*/manifest.yml")
 	if err != nil {
 		return fmt.Errorf("globbing data stream manifests: %w", err)
@@ -143,7 +203,7 @@ func applyInputTypesToDataStreamManifests(buildRoot *os.Root, infoByPkg map[stri
 			if stream.Package == "" {
 				continue
 			}
-			info, ok := infoByPkg[stream.Package]
+			info, ok := infoByInputPkg[stream.Package]
 			if !ok {
 				return fmt.Errorf("input package %q referenced in %q streams[%d] not found in required inputs", stream.Package, path.Dir(manifestPath), streamIdx)
 			}
@@ -153,7 +213,7 @@ func applyInputTypesToDataStreamManifests(buildRoot *os.Root, infoByPkg map[stri
 				return fmt.Errorf("getting stream node at index %d in %q: %w", streamIdx, manifestPath, err)
 			}
 
-			upsertKey(streamNode, "input", strVal(info.identifier))
+			upsertKey(streamNode, "input", strVal(streamInputRefs[stream.Package]))
 
 			if stream.Title == "" && info.pkgTitle != "" {
 				upsertKey(streamNode, "title", strVal(info.pkgTitle))
@@ -181,37 +241,37 @@ func applyInputTypesToDataStreamManifests(buildRoot *os.Root, infoByPkg map[stri
 // needed to replace package: references in composable packages. When the input
 // package has several policy templates, only the first template's input id is
 // used and a warning is logged.
-func loadInputPkgInfo(pkgPath string) (inputPkgInfo, error) {
+func loadInputPkgInfo(pkgPath string) (*inputPolicyTemplateInfo, error) {
 	pkgFS, closeFn, err := openPackageFS(pkgPath)
 	if err != nil {
-		return inputPkgInfo{}, fmt.Errorf("opening package: %w", err)
+		return nil, fmt.Errorf("opening package: %w", err)
 	}
 	defer func() { _ = closeFn() }()
 
 	manifestBytes, err := fs.ReadFile(pkgFS, packages.PackageManifestFile)
 	if err != nil {
-		return inputPkgInfo{}, fmt.Errorf("reading manifest: %w", err)
+		return nil, fmt.Errorf("reading manifest: %w", err)
 	}
 
 	m, err := packages.ReadPackageManifestBytes(manifestBytes)
 	if err != nil {
-		return inputPkgInfo{}, fmt.Errorf("parsing manifest: %w", err)
+		return nil, fmt.Errorf("parsing manifest: %w", err)
 	}
 
 	if len(m.PolicyTemplates) == 0 {
-		return inputPkgInfo{}, fmt.Errorf("input package %q has no policy templates", m.Name)
+		return nil, fmt.Errorf("input package %q has no policy templates", m.Name)
 	}
 	if len(m.PolicyTemplates) > 1 {
 		logger.Warnf("Input package %q has multiple policy templates; using input identifier %q from first policy template only", m.Name, m.PolicyTemplates[0].Input)
 	}
 
 	pt := m.PolicyTemplates[0]
 	if pt.Input == "" {
-		return inputPkgInfo{}, fmt.Errorf("input package %q policy template %q has no input identifier", m.Name, pt.Name)
+		return nil, fmt.Errorf("input package %q policy template %q has no input identifier", m.Name, pt.Name)
 	}
 
-	return inputPkgInfo{
-		identifier:     pt.Input,
+	return &inputPolicyTemplateInfo{
+		input:          pt.Input,
 		pkgTitle:       m.Title,
 		pkgDescription: m.Description,
 	}, nil