diff --git a/x-pack/platform/plugins/shared/fleet/common/services/policy_template.test.ts b/x-pack/platform/plugins/shared/fleet/common/services/policy_template.test.ts index acf0e862c8399..0f1e434c2614e 100644 --- a/x-pack/platform/plugins/shared/fleet/common/services/policy_template.test.ts +++ b/x-pack/platform/plugins/shared/fleet/common/services/policy_template.test.ts @@ -379,6 +379,31 @@ describe('getNormalizedDataStreams', () => { expect(useApmVar?.default).toEqual(true); expect(useApmVar?.title).toEqual('Enable Elastic APM Enrichment'); }); + + it('should add use_apm var when otel input has dynamic_signal_types true', () => { + const result = getNormalizedDataStreams({ + ...integrationPkg, + type: 'input', + policy_templates: [ + { + input: 'otelcol', + type: 'logs', + name: 'otel-dynamic', + template_path: 'some/path.hbl', + title: 'OTel Dynamic', + description: 'OTel with dynamic signal types', + dynamic_signal_types: true, + vars: [], + }, + ], + }); + expect(result).toHaveLength(1); + expect(result[0].streams).toHaveLength(1); + const vars = result[0].streams![0].vars; + const useApmVar = vars?.find((v) => v.name === 'use_apm'); + expect(useApmVar).toBeDefined(); + expect(useApmVar?.default).toEqual(true); + }); }); describe('filterPolicyTemplatesTiles', () => { diff --git a/x-pack/platform/plugins/shared/fleet/common/services/policy_template.ts b/x-pack/platform/plugins/shared/fleet/common/services/policy_template.ts index 089b7cca49dc9..3d04b5df36c0b 100644 --- a/x-pack/platform/plugins/shared/fleet/common/services/policy_template.ts +++ b/x-pack/platform/plugins/shared/fleet/common/services/policy_template.ts @@ -113,9 +113,11 @@ export const getNormalizedDataStreams = ( const dataset = datasetName || createDefaultDatasetName(packageInfo, policyTemplate); let vars = addDatasetVarIfNotPresent(policyTemplate.vars, policyTemplate.name); + const isOtelTraces = (dataStreamType || policyTemplate.type) === dataTypes.Traces; + const isOtelDynamicSignalTypes = policyTemplate.dynamic_signal_types === true; if ( policyTemplate.input === OTEL_COLLECTOR_INPUT_TYPE && - (dataStreamType || policyTemplate.type) === dataTypes.Traces + (isOtelTraces || isOtelDynamicSignalTypes) ) { vars = addUseAPMVarIfNotPresent(vars); } diff --git a/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.test.ts b/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.test.ts index 17ae3d2ac0b40..2c79a51aab5c0 100644 --- a/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.test.ts +++ b/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.test.ts @@ -787,6 +787,36 @@ describe('generateOtelcolConfig', () => { ], ]); + it('should add elasticapm connector and processor when stream has traces pipeline and use_apm enabled even if data_stream.type is not traces', () => { + const inputWithUseApm: FullAgentPolicyInput = { + ...otelInputWithMultipleSignalTypes, + streams: + otelInputWithMultipleSignalTypes.streams?.map((stream) => ({ + ...stream, + use_apm: true, + })) ?? [], + }; + const inputs: FullAgentPolicyInput[] = [inputWithUseApm]; + const result = generateOtelcolConfig(inputs, defaultOutput, packageInfoCache); + + expect(result.connectors?.elasticapm).toEqual({}); + expect(result.processors?.elasticapm).toEqual({}); + expect(result.service?.pipelines?.['metrics/aggregated-otel-metrics']).toEqual({ + receivers: ['elasticapm'], + exporters: ['forward'], + }); + const tracesPipelineKey = 'traces/otlp/test-multi-signal-stream-id-1'; + const tracesPipeline = result.service?.pipelines?.[tracesPipelineKey]; + expect(tracesPipeline).toBeDefined(); + expect(tracesPipeline?.exporters).toContain('elasticapm'); + expect(tracesPipeline?.processors).toContain('elasticapm'); + const metricsPipelineKey = 'metrics/otlp/test-multi-signal-stream-id-1'; + const metricsPipeline = result.service?.pipelines?.[metricsPipelineKey]; + expect(metricsPipeline).toBeDefined(); + expect(metricsPipeline?.exporters).not.toContain('elasticapm'); + expect(metricsPipeline?.processors).not.toContain('elasticapm'); + }); + it('should generate transform with multiple signal type statements when dynamic_signal_types is true', () => { const inputs: FullAgentPolicyInput[] = [otelInputWithMultipleSignalTypes]; const result = generateOtelcolConfig(inputs, defaultOutput, packageInfoCache); diff --git a/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.ts b/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.ts index c386dd8d25945..f0b6bda225182 100644 --- a/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.ts +++ b/x-pack/platform/plugins/shared/fleet/server/services/agent_policies/otel_collector.ts @@ -50,6 +50,10 @@ export function generateOtelcolConfig( const otelInputs: OTelCollectorConfig[] = (input?.streams ?? []).map((stream) => { // Avoid dots in keys, as they can create subobjects in agent config. const suffix = (input.id + '-' + stream.id).replaceAll('.', '-'); + // Extract signal types from pipeline IDs + const signalTypes = stream.service?.pipelines + ? extractSignalTypesFromPipelines(stream.service?.pipelines) + : []; const attributesTransform = generateOTelAttributesTransform( stream.data_stream.type ? stream.data_stream.type : 'logs', stream.data_stream.dataset, @@ -58,11 +62,13 @@ export function generateOtelcolConfig( : 'default', suffix, packageInfo, - stream.service?.pipelines + signalTypes ); + const hasTracesPipeline = signalTypes.includes('traces'); const shouldAddAPMConfig = - stream.data_stream.type === dataTypes.Traces && stream[USE_APM_VAR_NAME] === true; + (stream.data_stream.type === dataTypes.Traces || hasTracesPipeline) && + stream[USE_APM_VAR_NAME] === true; let otelConfig: OTelCollectorConfig = { ...addSuffixToOtelcolComponentsConfig('extensions', suffix, stream?.extensions), @@ -89,7 +95,7 @@ export function generateOtelcolConfig( otelConfig = appendOtelComponents(otelConfig, 'processors', [attributesTransform]); - if (stream.data_stream.type === dataTypes.Traces && stream[USE_APM_VAR_NAME] === true) { + if (shouldAddAPMConfig) { if (!otelConfig?.connectors) { otelConfig.connectors = {}; } @@ -208,17 +214,15 @@ function generateOTelAttributesTransform( namespace: string, suffix: string, packageInfo?: PackageInfo, - streamPipelines?: Record + signalTypes?: string[] ): Record { const dynamicSignalTypes = hasDynamicSignalTypes(packageInfo); let transformStatements: Record = {}; - if (dynamicSignalTypes && streamPipelines) { - // When dynamic_signal_types is true, extract signal types from pipeline IDs - // and generate transforms for each. This allows the collector to route data + if (dynamicSignalTypes && signalTypes) { + // When dynamic_signal_types is true, generate transforms for each signal type. This allows the collector to route data // to the appropriate datastreams based on the pipelines configured in the policy. - const signalTypes = extractSignalTypesFromPipelines(streamPipelines); // Generate transforms for each signal type found in pipelines signalTypes.forEach((signalType) => { const typeTransforms = generateOtelTypeTransforms(signalType, dataset, namespace); @@ -281,24 +285,18 @@ function conditionallyAddApmToPipelines( if (!shouldAddAPMConfig) { return pipelines; } - pipelines = addCompomentToPipelines(pipelines, 'elasticapm', 'exporters'); - pipelines = addCompomentToPipelines(pipelines, 'elasticapm', 'processors'); - return pipelines; -} - -function addCompomentToPipelines( - pipelines: any, - componentId: string, - type: string -): Record { - for (const pipelineId in pipelines) { - if (pipelines[pipelineId][type]) { - pipelines[pipelineId][type] = pipelines[pipelineId][type].concat([componentId]); - } else { - pipelines[pipelineId][type] = [componentId]; + const result: Record = {}; + Object.entries(pipelines as Record>).forEach( + ([pipelineID, pipeline]) => { + const signalType = getSignalType(pipelineID); + if (signalType === 'traces') { + pipeline.exporters = [...(pipeline.exporters || []), 'elasticapm']; + pipeline.processors = [...(pipeline.processors || []), 'elasticapm']; + } + result[pipelineID] = pipeline; } - } - return pipelines; + ); + return result; } function addSuffixToOtelcolPipelinesComponents(