Composable packages: input `name` qualifiers and stream `input` refs #3480

mrodm · 2026-04-22T09:36:16Z

Q: Should this loop duplicated to first look for input.Name and if it does not find any , then look for input.Type ?

Is it ok to search at the same time for both ?

as defiened on spec https://github.com/elastic/package-spec/blob/61cee23beb811b6c93ddb71caf822914ee30afba/code/go/internal/validator/semantic/validate_integration_input_qualifier.go#L68 the propoerty name is required when inputs share the same type. so i think is safe to make this comparison, so when name is found, its due to the fact that type is the same. when name is not found (empty) then we can trust type

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up @@
     	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()
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand All / @@ -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,
@@ Expand Down Expand Up @@
     	}
     	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
+    }

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -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"`
@@ Expand Down Expand Up @@
     	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
@@ Expand Down Expand Up @@
     	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
     		}
@@ Expand Down @@

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Composable packages: input `name` qualifiers and stream `input` refs #3480

Uh oh!

Diff view

Diff view

There are no files selected for viewing

mrodm Apr 22, 2026

Uh oh!

teresaromero Apr 22, 2026

Uh oh!

Uh oh!

-Original file line number
+Diff line change
@@ Expand Up / @@ -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)
     }
@@ Expand All / @@ -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
@@ Expand All / @@ -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 {
@@ Expand All / @@ -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)
     			}
@@ Expand All / @@ -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))
@@ Expand All / @@ -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)
@@ Expand All @@
     			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)
     			}
@@ Expand All @@
     				return fmt.Errorf("getting stream node at index %d in %q: %w", streamIdx, manifestPath, err)
     			}
-    			upsertKey(streamNode, "input", strVal(info.identifier))
+    			streamInputRef, ok := streamInputRefs[stream.Package]
+    			if !ok {
+    				return fmt.Errorf("stream input ref for package %q not found in streamInputRefs", stream.Package)
+    			}
+    			upsertKey(streamNode, "input", strVal(streamInputRef))
     			if stream.Title == "" && info.pkgTitle != "" {
     				upsertKey(streamNode, "title", strVal(info.pkgTitle))
@@ Expand Down Expand Up @@
     // 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
@@ Expand Down @@

Composable packages: input name qualifiers and stream input refs #3480

Uh oh!

Composable packages: input name qualifiers and stream input refs #3480

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

mrodm Apr 22, 2026

Choose a reason for hiding this comment

Uh oh!

teresaromero Apr 22, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Composable packages: input `name` qualifiers and stream `input` refs #3480

Composable packages: input `name` qualifiers and stream `input` refs #3480