[Fleet] generate OTel config for integration packages with otelcol inputs

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

@@ -1230,4 +1230,393 @@ describe('generateOtelcolConfig', () => {
       });
     });
   });
+
+  describe('integration-type packages with otelcol input', () => {
+    const otelIntegrationInput: FullAgentPolicyInput = {
+      type: OTEL_COLLECTOR_INPUT_TYPE,
+      id: 'integration-otel',
+      name: 'integration-otel',
+      revision: 0,
+      data_stream: { namespace: 'default' },
+      use_output: 'default',
+      package_policy_id: 'integration-policy',
+      meta: {
+        package: { name: 'my_integration', version: '1.0.0' },
+      },
+      streams: [
+        {
+          id: 'stream-id-1',
+          data_stream: { dataset: 'my_integration.metrics', type: 'metrics' },
+          service: {
+            pipelines: {
+              'metrics/otlp': {
+                receivers: ['otlp/integration-otel-stream-id-1'],
+                processors: [],
+              },
+            },
+          },
+          receivers: { otlp: {} },
+        },
+      ],
+    };
+
+    it('generates OTel config for integration package with otelcol input', () => {
+      const result = generateOtelcolConfig([otelIntegrationInput], defaultOutput);
+
+      expect(result.receivers).toHaveProperty('otlp/integration-otel-stream-id-1');
+      expect(result.exporters).toHaveProperty('elasticsearch/default');
+      expect(result.service?.pipelines).toHaveProperty('metrics');
+    });
+
+    it('uses defaultPackageInfo for dynamic signal types when packageInfoCache has no meta match', () => {
+      // TemplateAgentPolicyInput has no meta.package — simulate via defaultPackageInfo
+      const templateInput: TemplateAgentPolicyInput = {
+        id: 'integration-otel',
+        type: OTEL_COLLECTOR_INPUT_TYPE,
+        streams: [
+          {
+            id: 'stream-id-1',
+            data_stream: { dataset: 'my_integration.data', type: 'metrics' },
+            service: {
+              pipelines: {
+                'metrics/otlp': {
+                  receivers: ['otlp/integration-otel-stream-id-1'],
+                  processors: [],
+                },
+                'logs/otlp': { receivers: ['otlp/integration-otel-stream-id-1'], processors: [] },
+              },
+            },
+            receivers: { otlp: {} },
+          },
+        ],
+      };
+
+      const integrationPackageInfo = {
+        type: 'integration',
+        name: 'my_integration',
+        version: '1.0.0',
+        policy_templates: [
+          {
+            name: 'my_policy',
+            title: 'My Policy',
+            description: 'My Policy',
+            inputs: [
+              {
+                type: OTEL_COLLECTOR_INPUT_TYPE,
+                title: 'OTel',
+                description: 'OTel',
+                dynamic_signal_types: true,
+              },
+            ],
+          },
+        ],
+      } as any;
+
+      const result = generateOtelcolConfig(
+        [templateInput],
+        undefined,
+        undefined,
+        integrationPackageInfo
+      );
+
+      // With dynamic_signal_types, routing transforms are generated per signal type
+      const routingKey = Object.keys(result.processors ?? {}).find((k) =>
+        k.startsWith('transform/')
+      );
+      expect(routingKey).toBeDefined();
+      const routingProcessor = result.processors?.[routingKey!];
+      // Should have both metric_statements and log_statements for dynamic signal types
+      expect(routingProcessor).toHaveProperty('metric_statements');
+      expect(routingProcessor).toHaveProperty('log_statements');
+    });
+
+    it('falls back to single signal type routing when integration package has no dynamic_signal_types', () => {
+      const templateInput: TemplateAgentPolicyInput = {
+        id: 'integration-otel',
+        type: OTEL_COLLECTOR_INPUT_TYPE,
+        streams: [
+          {
+            id: 'stream-id-1',
+            data_stream: { dataset: 'my_integration.metrics', type: 'metrics' },
+            service: {
+              pipelines: {
+                'metrics/otlp': {
+                  receivers: ['otlp/integration-otel-stream-id-1'],
+                  processors: [],
+                },
+              },
+            },
+            receivers: { otlp: {} },
+          },
+        ],
+      };
+
+      const integrationPackageInfo = {
+        type: 'integration',
+        name: 'my_integration',
+        version: '1.0.0',
+        policy_templates: [
+          {
+            name: 'my_policy',
+            title: 'My Policy',
+            description: 'My Policy',
+            inputs: [{ type: OTEL_COLLECTOR_INPUT_TYPE, title: 'OTel', description: 'OTel' }],
+          },
+        ],
+      } as any;
+
+      const result = generateOtelcolConfig(
+        [templateInput],
+        undefined,
+        undefined,
+        integrationPackageInfo
+      );
+
+      const routingKey = Object.keys(result.processors ?? {}).find((k) =>
+        k.startsWith('transform/')
+      );
+      expect(routingKey).toBeDefined();
+      const routingProcessor = result.processors?.[routingKey!];
+      // Without dynamic_signal_types, only metric_statements (from data_stream.type: metrics)
+      expect(routingProcessor).toHaveProperty('metric_statements');
+      expect(routingProcessor).not.toHaveProperty('log_statements');
+    });
+
+    it('defaultPackageInfo is overridden by packageInfoCache when meta.package is present', () => {
+      // When both are provided, cache takes precedence for inputs that have meta.package
+      const packageInfoCacheWithDynamic = new Map([
+        [
+          'my_integration-1.0.0',
+          {
+            type: 'integration',
+            name: 'my_integration',
+            version: '1.0.0',
+            policy_templates: [
+              {
+                name: 'my_policy',
+                title: 'My Policy',
+                description: 'My Policy',
+                inputs: [
+                  {
+                    type: OTEL_COLLECTOR_INPUT_TYPE,
+                    title: 'OTel',
+                    description: 'OTel',
+                    dynamic_signal_types: true,
+                  },
+                ],
+              },
+            ],
+          } as any,
+        ],
+      ]);
+
+      // defaultPackageInfo says no dynamic_signal_types
+      const defaultPkgInfoNoDynamic = {
+        type: 'integration',
+        name: 'my_integration',
+        version: '1.0.0',
+        policy_templates: [
+          {
+            name: 'my_policy',
+            title: 'My Policy',
+            description: 'My Policy',
+            inputs: [{ type: OTEL_COLLECTOR_INPUT_TYPE, title: 'OTel', description: 'OTel' }],
+          },
+        ],
+      } as any;
+
+      const inputWithMeta: FullAgentPolicyInput = {
+        ...otelIntegrationInput,
+        streams: [
+          {
+            id: 'stream-id-1',
+            data_stream: { dataset: 'my_integration.data', type: 'metrics' },
+            service: {
+              pipelines: {
+                'metrics/otlp': {
+                  receivers: ['otlp/integration-otel-stream-id-1'],
+                  processors: [],
+                },
+                'logs/otlp': { receivers: ['otlp/integration-otel-stream-id-1'], processors: [] },
+              },
+            },
+            receivers: { otlp: {} },
+          },
+        ],
+      };
+
+      const result = generateOtelcolConfig(
+        [inputWithMeta],
+        undefined,
+        packageInfoCacheWithDynamic,
+        defaultPkgInfoNoDynamic
+      );
+
+      const routingKey = Object.keys(result.processors ?? {}).find((k) =>
+        k.startsWith('transform/')
+      );
+      const routingProcessor = result.processors?.[routingKey!];
+      // Cache has dynamic_signal_types: true, so routing should have both metric and log statements
+      expect(routingProcessor).toHaveProperty('metric_statements');
+      expect(routingProcessor).toHaveProperty('log_statements');
+    });
+
+    it('uses single-type routing for a non-dynamic input even when another template in the same package has dynamic_signal_types', () => {
+      // Package has two policy templates: 'otel_policy' with dynamic_signal_types and
+      // 'metrics_policy' without it. The input below belongs to 'metrics_policy'.
+      const mixedPackageInfo = {
+        type: 'integration',
+        name: 'mixed_pkg',
+        version: '1.0.0',
+        policy_templates: [
+          {
+            name: 'otel_policy',
+            title: 'OTel',
+            description: 'OTel',
+            inputs: [
+              {
+                type: OTEL_COLLECTOR_INPUT_TYPE,
+                title: 'OTel',
+                description: 'OTel',
+                dynamic_signal_types: true,
+              },
+            ],
+          },
+          {
+            name: 'metrics_policy',
+            title: 'Metrics',
+            description: 'Metrics',
+            inputs: [
+              {
+                type: OTEL_COLLECTOR_INPUT_TYPE,
+                title: 'Metrics OTel',
+                description: 'Metrics only',
+                // No dynamic_signal_types
+              },
+            ],
+          },
+        ],
+      } as any;
+
+      const nonDynamicInput: FullAgentPolicyInput = {
+        type: OTEL_COLLECTOR_INPUT_TYPE,
+        id: 'non-dynamic-otel',
+        name: 'non-dynamic-otel',
+        revision: 0,
+        data_stream: { namespace: 'default' },
+        use_output: 'default',
+        package_policy_id: 'non-dynamic-policy',
+        meta: {
+          package: {
+            name: 'mixed_pkg',
+            version: '1.0.0',
+            policy_template: 'metrics_policy', // belongs to the non-dynamic template
+          },
+        },
+        streams: [
+          {
+            id: 'stream-id-1',
+            data_stream: { dataset: 'mixed_pkg.metrics', type: 'metrics' },
+            service: {
+              pipelines: {
+                'metrics/otlp': { receivers: ['otlp'] },
+                'logs/otlp': { receivers: ['otlp'] },
+              },
+            },
+          },
+        ],
+      };
+
+      const cache = new Map([['mixed_pkg-1.0.0', mixedPackageInfo]]);
+      const result = generateOtelcolConfig([nonDynamicInput], undefined, cache);
+
+      const routingKey = Object.keys(result.processors ?? {}).find((k) =>
+        k.startsWith('transform/')
+      );
+      expect(routingKey).toBeDefined();
+      const routingProcessor = result.processors?.[routingKey!];
+
+      // 'metrics_policy' input has no dynamic_signal_types — must produce only metric_statements
+      // (from data_stream.type: metrics), not log_statements from the other template.
+      expect(routingProcessor).toHaveProperty('metric_statements');
+      expect(routingProcessor).not.toHaveProperty('log_statements');
+    });
+
+    it('uses single-type routing for an otelcol input when the same template also has a different dynamic input type', () => {
+      // Single template 'combined_policy' with two inputs:
+      //   - 'otelcol'         — no dynamic_signal_types
+      //   - 'custom_receiver' — dynamic_signal_types: true
+      // Only the 'otelcol' input is processed here; it should produce single-type routing
+      // because its own definition has no dynamic_signal_types.
+      const mixedInputsPackageInfo = {
+        type: 'integration',
+        name: 'combined_pkg',
+        version: '1.0.0',
+        policy_templates: [
+          {
+            name: 'combined_policy',
+            title: 'Combined',
+            description: 'Combined inputs',
+            inputs: [
+              {
+                type: OTEL_COLLECTOR_INPUT_TYPE,
+                title: 'OTel',
+                description: 'Standard OTel collector',
+                // No dynamic_signal_types
+              },
+              {
+                type: 'custom_receiver',
+                title: 'Custom',
+                description: 'Custom dynamic receiver',
+                dynamic_signal_types: true,
+              },
+            ],
+          },
+        ],
+      } as any;
+
+      const nonDynamicOtelInput: FullAgentPolicyInput = {
+        type: OTEL_COLLECTOR_INPUT_TYPE,
+        id: 'combined-otel',
+        name: 'combined-otel',
+        revision: 0,
+        data_stream: { namespace: 'default' },
+        use_output: 'default',
+        package_policy_id: 'combined-policy',
+        meta: {
+          package: {
+            name: 'combined_pkg',
+            version: '1.0.0',
+            policy_template: 'combined_policy',
+          },
+        },
+        streams: [
+          {
+            id: 'stream-id-1',
+            data_stream: { dataset: 'combined_pkg.metrics', type: 'metrics' },
+            service: {
+              pipelines: {
+                'metrics/otlp': { receivers: ['otlp'] },
+                'logs/otlp': { receivers: ['otlp'] },
+              },
+            },
+          },
+        ],
+      };
+
+      const cache = new Map([['combined_pkg-1.0.0', mixedInputsPackageInfo]]);
+      const result = generateOtelcolConfig([nonDynamicOtelInput], undefined, cache);
+
+      const routingKey = Object.keys(result.processors ?? {}).find((k) =>
+        k.startsWith('transform/')
+      );
+      expect(routingKey).toBeDefined();
+      const routingProcessor = result.processors?.[routingKey!];
+
+      // otelcol in this template has no dynamic_signal_types — must produce only metric_statements
+      // (from data_stream.type: metrics), not log_statements from the sibling custom_receiver input.
+      expect(routingProcessor).toHaveProperty('metric_statements');
+      expect(routingProcessor).not.toHaveProperty('log_statements');
+    });
+  });
 });