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 {
+			// ref is the identifier used in data stream manifests: the Name qualifier
+			// when set (disambiguated same-type inputs), otherwise the Type.
+			ref := inputRef(input)
+			inputKey := fmt.Sprintf("%s-%s", pt.Name, ref)
+			if ref == 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(ref, 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}
 }
+
+// inputRef 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 inputRef(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 identifier, checked against the
+// input's Name qualifier first (when set) and then its Type. Use this when the
+// identifier may be a name qualifier (set on inputs that share a type) rather
+// than a bare type string.
+func (pt *PolicyTemplate) FindInput(identifier string) *Input {
+	for i := range pt.Inputs {
+		input := &pt.Inputs[i]
+		if (input.Name != "" && input.Name == identifier) || input.Type == identifier {
+			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

@@ -17,6 +17,7 @@ import (
 // inputPkgInfo holds the resolved metadata from an input package needed to
 // replace package: references in composable package manifests.
 type inputPkgInfo struct {
+	pkgName        string // manifest.name; used as the input name qualifier when disambiguation is needed
 	identifier     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 +28,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,11 +48,44 @@ func (r *RequiredInputsResolver) resolveStreamInputTypes(
 		return err
 	}
 
-	if err := applyInputTypesToComposableManifest(manifest, buildRoot, infoByPkg); err != nil {
+	streamInputRefs := buildStreamInputRefs(manifest, infoByPkg)
+
+	if err := applyInputTypesToComposableManifest(manifest, buildRoot, infoByPkg, streamInputRefs); err != nil {
 		return err
 	}
 
-	return applyInputTypesToDataStreamManifests(buildRoot, infoByPkg)
+	return applyInputTypesToDataStreamManifests(buildRoot, infoByPkg, streamInputRefs)
+}
+
+// buildStreamInputRefs computes what value to write to streams[].input per required package name.
+// When two or more inputs of the same type coexist in any single policy template, each conflicting
+// package is given its own package name as the qualifier (used as both the input name and the
+// stream input reference). Packages with a unique type within every policy template get their type
+// identifier. Once a package is marked as needing a qualifier it keeps that assignment even if it
+// appears in other policy templates without a type conflict.
+func buildStreamInputRefs(manifest *packages.PackageManifest, infoByPkg map[string]inputPkgInfo) map[string]string {
+	refs := make(map[string]string, len(infoByPkg))
+	for _, pt := range manifest.PolicyTemplates {
+		typeCounts := make(map[string]int)
+		for _, input := range pt.Inputs {
+			if input.Package == "" {
+				continue
+			}
+			typeCounts[infoByPkg[input.Package].identifier]++
+		}
+		for _, input := range pt.Inputs {
+			if input.Package == "" {
+				continue
+			}
+			info := infoByPkg[input.Package]
+			if typeCounts[info.identifier] > 1 {
+				refs[input.Package] = input.Package // package name as stable unique qualifier
+			} else if _, exists := refs[input.Package]; !exists {
+				refs[input.Package] = info.identifier
+			}
+		}
+	}
+	return refs
 }
 
 // buildInputPkgInfoByName loads inputPkgInfo for each downloaded required input package path.
@@ -61,12 +101,15 @@ func buildInputPkgInfoByName(inputPkgPaths map[string]string) (map[string]inputP
 	return infoByPkg, nil
 }
 
-// applyInputTypesToComposableManifest sets type (and optional title/description) on
+// applyInputTypesToComposableManifest sets type (and optional title/description/name) on
 // package-backed policy template inputs in manifest.yml and drops package:.
+// 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,
+	streamInputRefs map[string]string,
 ) error {
 	manifestBytes, err := buildRoot.ReadFile("manifest.yml")
 	if err != nil {
@@ -93,6 +136,9 @@ func applyInputTypesToComposableManifest(
 			}
 
 			upsertKey(inputNode, "type", strVal(info.identifier))
+			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 +162,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, infoByPkg map[string]inputPkgInfo, 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)
@@ -153,7 +202,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))
@@ -211,6 +260,7 @@ func loadInputPkgInfo(pkgPath string) (inputPkgInfo, error) {
 	}
 
 	return inputPkgInfo{
+		pkgName:        m.Name,
 		identifier:     pt.Input,
 		pkgTitle:       m.Title,
 		pkgDescription: m.Description,