[9.4] [Fleet] Resolve streams and secrets by effective input name (#264366)

x-pack/platform/plugins/shared/fleet/common/services/validate_package_policy.ts

@@ -481,7 +481,7 @@ export const validatePackagePolicy = (
     if (input.streams.length) {
       input.streams.forEach((stream) => {
         const streamValidationResults: PackagePolicyConfigValidationResults = {};
-        const streamKey = `${stream.data_stream.dataset}-${input.type}`;
+        const streamKey = `${stream.data_stream.dataset}-${getInputEffectiveName(input)}`;
 
         const streamVarDefs = streamVarDefsByDatasetAndInput[streamKey];
         const streamVarGroups = streamVarGroupsByDatasetAndInput[streamKey];

x-pack/platform/plugins/shared/fleet/server/services/epm/packages/get_template_inputs.ts

@@ -14,6 +14,7 @@ import { getNormalizedInputs, isIntegrationPolicyTemplate } from '../../../../co
 
 import {
   getStreamsForInputType,
+  getInputEffectiveName,
   packageToPackagePolicy,
 } from '../../../../common/services/package_to_package_policy';
 import { _compilePackagePolicyInputs } from '../../package_policy';
@@ -64,9 +65,10 @@ export const templatePackagePolicyToFullInputStreams = (
   packagePolicyInputs.forEach((input) => {
     const streamsIdsMap = new Map();
 
+    const inputEffectiveName = getInputEffectiveName(input);
     const inputId = input.policy_template
-      ? `${input.policy_template}-${input.type}`
-      : `${input.type}`;
+      ? `${input.policy_template}-${inputEffectiveName}`
+      : inputEffectiveName;
     const fullInputStream = {
       // @ts-ignore-next-line the following id is actually one level above the one in fullInputStream, but the linter thinks it gets overwritten
       id: inputId,
@@ -142,9 +144,10 @@ export async function getTemplateInputs(
   if (format === 'yml') {
     // Add a placeholder <VAR_NAME> to all variables without default value
     for (const inputWithStreamIds of inputsWithStreamIds) {
+      const inputEffectiveName = getInputEffectiveName(inputWithStreamIds);
       const inputId = inputWithStreamIds.policy_template
-        ? `${inputWithStreamIds.policy_template}-${inputWithStreamIds.type}`
-        : inputWithStreamIds.type;
+        ? `${inputWithStreamIds.policy_template}-${inputEffectiveName}`
+        : inputEffectiveName;
 
       const packageInput = indexedInputsAndStreams[inputId];
       if (!packageInput) {
@@ -270,10 +273,11 @@ function buildIndexedPackage(packageInfo: PackageInfo): PackageWithInputAndStrea
         const inputs = getNormalizedInputs(policyTemplate);
 
         inputs.forEach((packageInput) => {
-          const inputId = `${policyTemplate.name}-${packageInput.type}`;
+          const inputEffectiveName = getInputEffectiveName(packageInput);
+          const inputId = `${policyTemplate.name}-${inputEffectiveName}`;
 
           const streams = getStreamsForInputType(
-            packageInput.type,
+            inputEffectiveName,
             packageInfo,
             isIntegrationPolicyTemplate(policyTemplate) && policyTemplate.data_streams
               ? policyTemplate.data_streams
@@ -286,7 +290,7 @@ function buildIndexedPackage(packageInfo: PackageInfo): PackageWithInputAndStrea
               }
             >
           >((acc, stream) => {
-            const streamId = `${packageInput.type}-${stream.data_stream.dataset}`;
+            const streamId = `${inputEffectiveName}-${stream.data_stream.dataset}`;
             acc[streamId] = {
               ...stream,
             };

x-pack/platform/plugins/shared/fleet/server/services/epm/packages/get_templates_inputs.test.ts

@@ -221,6 +221,34 @@ describe('Fleet - templatePackagePolicyToFullInputStreams', () => {
     ]);
   });
 
+  it('uses getInputEffectiveName for template input id when name differs from type', async () => {
+    expect(
+      await templatePackagePolicyToFullInputStreams([
+        {
+          ...mockInput2,
+          name: 'custom-metrics',
+          type: 'test-metrics',
+        },
+      ])
+    ).toEqual([
+      {
+        id: 'some-template-custom-metrics',
+        type: 'test-metrics',
+        streams: [
+          {
+            data_stream: {
+              dataset: 'foo',
+              type: 'metrics',
+            },
+            fooKey: 'fooValue1',
+            fooKey2: ['fooValue2'],
+            id: 'test-metrics-foo',
+          },
+        ],
+      },
+    ]);
+  });
+
   it('returns agent inputs without disabled streams', async () => {
     expect(
       await templatePackagePolicyToFullInputStreams([

x-pack/platform/plugins/shared/fleet/server/services/package_policy.test.ts

@@ -2143,6 +2143,55 @@ describe('Package policy service', () => {
       ]);
     });
 
+    it('should compile stream templates when stream.input references a named input (name ?? type)', async () => {
+      const inputs = await _compilePackagePolicyInputs(
+        {
+          name: 'test',
+          version: '1.0.0',
+          data_streams: [
+            {
+              type: 'logs',
+              dataset: 'package.dataset1',
+              streams: [{ input: 'named_logfile', template_path: 'some_template_path.yml' }],
+              path: 'dataset1',
+            },
+          ],
+          policy_templates: [
+            {
+              inputs: [{ type: 'log', name: 'named_logfile' }],
+            },
+          ],
+        } as unknown as PackageInfo,
+        {},
+        [
+          {
+            type: 'log',
+            name: 'named_logfile',
+            enabled: true,
+            streams: [
+              {
+                id: 'datastream01',
+                data_stream: { dataset: 'package.dataset1', type: 'logs' },
+                enabled: true,
+                vars: {
+                  paths: {
+                    value: ['/var/log/set.log'],
+                  },
+                },
+              },
+            ],
+          },
+        ],
+        ASSETS_MAP_FIXTURES
+      );
+
+      expect(inputs[0].streams[0].compiled_stream).toEqual({
+        metricset: ['dataset1'],
+        paths: ['/var/log/set.log'],
+        type: 'log',
+      });
+    });
+
     it('should compile integration stream data_stream.dataset from package stream var default value', async () => {
       const pkgInfo = {
         name: 'dsvarpkg',

x-pack/platform/plugins/shared/fleet/server/services/package_policy.ts

@@ -3808,12 +3808,13 @@ function _compilePackageStream(
 
   stream = _applyIndexPrivileges(packageDataStream, streamIn);
 
+  const effectiveInputName = getInputEffectiveName(input);
   const streamFromPkg = (packageDataStream.streams || []).find(
-    (pkgStream) => pkgStream.input === input.type
+    (pkgStream) => pkgStream.input === effectiveInputName
   );
   if (!streamFromPkg) {
     throw new StreamNotFoundError(
-      `Stream template not found, unable to find stream for input ${input.type}`
+      `Stream template not found, unable to find stream for input ${effectiveInputName} (type: ${input.type})`
     );
   }
 

x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.ts

@@ -13,6 +13,7 @@ import type { NewPackagePolicy, RegistryStream, UpdatePackagePolicy } from '../.
 import { SO_SEARCH_LIMIT } from '../../../common';
 import {
   doesPackageHaveIntegrations,
+  getInputEffectiveName,
   getNormalizedDataStreams,
   getNormalizedInputs,
 } from '../../../common/services';
@@ -391,7 +392,9 @@ function _getInputSecretPaths(
     if (input.streams.length) {
       input.streams.forEach((stream, streamIndex) => {
         const streamVarDefs =
-          streamSecretVarDefsByDatasetAndInput[`${stream.data_stream.dataset}-${input.type}`];
+          streamSecretVarDefsByDatasetAndInput[
+            `${stream.data_stream.dataset}-${getInputEffectiveName(input)}`
+          ];
         if (streamVarDefs && Object.keys(streamVarDefs).length) {
           Object.entries(stream.vars || {}).forEach(([name, configEntry]) => {
             if (streamVarDefs[name]) {