[Fleet] Ensure custom data stream types and data sets are included in policies (#236016)

Customization of data stream type and dataset relies on having the
`data_stream` object in the compiled stream. In the case of OTel inputs
the compiled stream doesn't contain this object, as it is not
supported by the OTel collector. Get these values directly from the
user-defined variables.

Author: jsoriano

x-pack/platform/plugins/shared/fleet/cypress/e2e/input_packages_real.cy.ts

@@ -20,6 +20,7 @@ import { login } from '../tasks/login';
 import { request } from '../tasks/common';
 
 const INPUT_TEST_PACKAGE = 'input_package-1.0.0';
+const OTEL_INPUT_TEST_PACKAGE = 'otel_input_package-1.0.0';
 
 describe('Input package with custom data stream type', () => {
   beforeEach(() => {
@@ -121,3 +122,127 @@ describe('Input package with custom data stream type', () => {
     cy.get('[data-test-subj="templateDetails"').contains(`${dataStreamType}-${datasetName}-*`);
   });
 });
+
+describe('OTel input package with custom data stream type', () => {
+  beforeEach(() => {
+    login();
+  });
+
+  const agentPolicyId = 'test-otel-input-package-policy';
+  const agentPolicyName = 'Test input package policy';
+  const packagePolicyName = 'input-package-policy';
+  const datasetName = 'check'; // Default from the package.
+  const dataStreamType = 'logs';
+
+  before(() => {
+    cy.task('installTestPackage', OTEL_INPUT_TEST_PACKAGE);
+
+    request({
+      method: 'POST',
+      url: `/api/fleet/agent_policies`,
+      body: {
+        id: agentPolicyId,
+        name: agentPolicyName,
+        description: 'desc',
+        namespace: 'default',
+        monitoring_enabled: [],
+      },
+      headers: { 'kbn-xsrf': 'cypress', 'Elastic-Api-Version': `${API_VERSIONS.public.v1}` },
+    });
+  });
+
+  after(() => {
+    cleanupAgentPolicies();
+    cy.task('uninstallTestPackage', OTEL_INPUT_TEST_PACKAGE);
+  });
+
+  it('should successfully create a package policy', () => {
+    cy.visit(`/app/integrations/detail/${OTEL_INPUT_TEST_PACKAGE}/overview`);
+    cy.getBySel(ADD_INTEGRATION_POLICY_BTN).click();
+
+    cy.getBySel(POLICY_EDITOR.POLICY_NAME_INPUT).click().clear().type(packagePolicyName);
+    cy.getBySel('multiTextInput-http-endpoints-to-check')
+      .find('[data-test-subj="multiTextInputRow-0"]')
+      .click()
+      .type('https://www.elastic.co/integrations');
+
+    // Select logs data stream type.
+    cy.get('[data-test-subj^="advancedStreamOptionsToggle"]').click();
+    cy.get('[data-test-subj="packagePolicyDataStreamType"')
+      .find(`label[for="${dataStreamType}"]`)
+      .click();
+
+    cy.getBySel(EXISTING_HOSTS_TAB).click();
+
+    cy.getBySel(POLICY_EDITOR.AGENT_POLICY_SELECT).click();
+    cy.getBySel('agentPolicyMultiItem').each(($el) => {
+      if ($el.text() === agentPolicyName) {
+        $el.trigger('click');
+      }
+    });
+    cy.wait(1000); // wait for policy id to be set
+    cy.getBySel(CREATE_PACKAGE_POLICY_SAVE_BTN).click();
+
+    cy.getBySel(CONFIRM_MODAL.CANCEL_BUTTON).click();
+  });
+
+  it(`agent policy configures the dataset and type`, () => {
+    cy.request({
+      method: 'GET',
+      url: `/api/fleet/agent_policies/${agentPolicyId}/full`,
+    }).then(({ body }) => {
+      expect(body.item).to.have.property('processors');
+      let routingTransform;
+      Object.entries(body.item.processors).forEach(([name, config]) => {
+        if (!name.match(/transform\/.*-routing/)) {
+          return;
+        }
+        routingTransform = config;
+      });
+      expect(routingTransform).to.deep.equal({
+        log_statements: [
+          {
+            context: 'log',
+            statements: [
+              `set(attributes["data_stream.type"], "${dataStreamType}")`,
+              `set(attributes["data_stream.dataset"], "${datasetName}")`,
+              'set(attributes["data_stream.namespace"], "default")',
+            ],
+          },
+        ],
+      });
+    });
+  });
+
+  it(`${dataStreamType} checkbox should be checked`, () => {
+    cy.visit(`/app/integrations/detail/${OTEL_INPUT_TEST_PACKAGE}/policies`);
+
+    cy.getBySel(INTEGRATION_NAME_LINK).contains(packagePolicyName).click();
+
+    cy.get('button').contains('Change defaults').click();
+    cy.get('[data-test-subj^="advancedStreamOptionsToggle"]').click();
+    cy.get('[data-test-subj="packagePolicyDataStreamType"')
+      .find(`input#${dataStreamType}`)
+      .should('be.checked');
+  });
+
+  it('should not allow to edit data stream type', () => {
+    cy.visit(`/app/integrations/detail/${OTEL_INPUT_TEST_PACKAGE}/policies`);
+
+    cy.getBySel(INTEGRATION_NAME_LINK).contains(packagePolicyName).click();
+
+    cy.get('button').contains('Change defaults').click();
+    cy.get('[data-test-subj^="advancedStreamOptionsToggle"]').click();
+    cy.get('[data-test-subj="packagePolicyDataStreamType"')
+      .find('input')
+      .should('have.length', 3)
+      .each(($el) => cy.wrap($el).should('be.disabled'));
+  });
+
+  it('has an index template', () => {
+    cy.visit(`app/management/data/index_management/templates/${dataStreamType}-${datasetName}`);
+
+    // Check that the index pattern appears in the view.
+    cy.get('[data-test-subj="templateDetails"').contains(`${dataStreamType}-${datasetName}.otel-*`);
+  });
+});

x-pack/platform/plugins/shared/fleet/cypress/packages/otel_input_package-1.0.0.zip

@@ -93,7 +93,7 @@ function generateOTelAttributesTransform(
       context = 'span';
       break;
     default:
-      throw new Error(`unexpected data stream type ${type}`);
+      throw new FleetError(`unexpected data stream type ${type}`);
   }
   return {
     [`transform/${suffix}-routing`]: {

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

@@ -20,7 +20,12 @@ import type {
 } from '../../types';
 import { DEFAULT_OUTPUT } from '../../constants';
 import { pkgToPkgKey } from '../epm/registry';
-import { GLOBAL_DATA_TAG_EXCLUDED_INPUTS } from '../../../common/constants/epm';
+import {
+  DATASET_VAR_NAME,
+  DATA_STREAM_TYPE_VAR_NAME,
+  GLOBAL_DATA_TAG_EXCLUDED_INPUTS,
+  OTEL_COLLECTOR_INPUT_TYPE,
+} from '../../../common/constants/epm';
 
 const isPolicyEnabled = (packagePolicy: PackagePolicy) => {
   return packagePolicy.enabled && packagePolicy.inputs && packagePolicy.inputs.length;
@@ -145,6 +150,18 @@ export const getFullInputStreams = (
                   return acc;
                 }, {} as { [k: string]: any }),
               };
+              if (input.type === OTEL_COLLECTOR_INPUT_TYPE) {
+                // otelcol inputs are not going to have the data_stream type and dataset in
+                // the compiled stream, get them directly from the user-defined variables.
+                const dsTypeVar = stream.vars?.[DATA_STREAM_TYPE_VAR_NAME]?.value;
+                const datasetVar = stream.vars?.[DATASET_VAR_NAME]?.value;
+                fullStream.data_stream = {
+                  ...fullStream.data_stream,
+                  ...(dsTypeVar ? { type: dsTypeVar } : {}),
+                  ...(datasetVar ? { dataset: datasetVar } : {}),
+                };
+              }
+
               streamsOriginalIdsMap?.set(fullStream.id, streamId);
 
               return fullStream;

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

@@ -43,6 +43,7 @@ export default async function ({ readConfigFile }: FtrConfigProviderContext) {
         // add feature flags here
         `--xpack.fleet.enableExperimental=${JSON.stringify([
           'agentTamperProtectionEnabled',
+          'enableOtelIntegrations',
           'subfeaturePrivileges',
         ])}`,