elastic · kibanamachine · Jun 9, 2025 · May 26, 2025
@@ -49,6 +49,7 @@ export const PACKAGE_TEMPLATE_SUFFIX = '@package';
 export const USER_SETTINGS_TEMPLATE_SUFFIX = '@custom';
 
 export const DATASET_VAR_NAME = 'data_stream.dataset';
+export const DATA_STREAM_TYPE_VAR_NAME = 'data_stream.type';
 
 export const CUSTOM_INTEGRATION_PACKAGE_SPEC_VERSION = '2.9.0';
 

@@ -11,6 +11,7 @@ import type {
   PackageInfo,
   RegistryVarType,
   PackageListItem,
+  RegistryDataStream,
 } from '../types';
 
 import {
@@ -281,6 +282,75 @@ describe('getNormalizedDataStreams', () => {
     expect(result[0].streams).toHaveLength(1);
     expect(result?.[0].streams?.[0]?.vars).toEqual([datasetVar]);
   });
+
+  const inputPkg: PackageInfo = {
+    name: 'log',
+    type: 'input',
+    title: 'Custom logs',
+    version: '2.4.0',
+    description: 'Collect custom logs with Elastic Agent.',
+    format_version: '3.1.5',
+    owner: { github: '' },
+    assets: {} as any,
+    data_streams: [],
+    policy_templates: [
+      {
+        name: 'logs',
+        type: 'logs',
+        title: 'Custom log file',
+        description: 'Collect logs from custom files.',
+        input: 'logfile',
+        template_path: 'input.yml.hbs',
+        vars: [
+          {
+            name: 'paths',
+            type: 'text',
+            title: 'Paths',
+            multi: true,
+            required: true,
+            show_user: true,
+            default: ['/var/log/nginx/access.log*'],
+          },
+        ],
+      },
+    ],
+    latestVersion: '1.3.0',
+    keepPoliciesUpToDate: false,
+    status: 'not_installed',
+  };
+  const expectedInputPackageDataStream: RegistryDataStream = {
+    type: 'logs',
+    dataset: 'log.logs',
+    elasticsearch: {
+      dynamic_dataset: true,
+      dynamic_namespace: true,
+    },
+    title: expect.any(String),
+    release: 'ga',
+    package: 'log',
+    path: 'log.logs',
+    streams: [
+      {
+        input: 'logfile',
+        vars: expect.any(Array),
+        template_path: 'input.yml.hbs',
+        title: 'Custom log file',
+        description: 'Custom log file',
+        enabled: true,
+      },
+    ],
+  };
+  it('should build data streams for input package', () => {
+    expect(getNormalizedDataStreams(inputPkg)).toEqual([expectedInputPackageDataStream]);
+  });
+  it('should use user-defined data stream type in input package', () => {
+    expect(getNormalizedDataStreams(inputPkg, undefined, 'metrics')).toEqual([
+      {
+        ...expectedInputPackageDataStream,
+        type: 'metrics',
+      },
+    ]);
+  });
 });
 
 describe('filterPolicyTemplatesTiles', () => {

@@ -68,7 +68,8 @@ export const getNormalizedInputs = (policyTemplate: RegistryPolicyTemplate): Reg
 
 export const getNormalizedDataStreams = (
   packageInfo: PackageInfo | InstallablePackage,
-  datasetName?: string
+  datasetName?: string,
+  dataStreamType?: string
 ): RegistryDataStream[] => {
   if (packageInfo.type !== 'input') {
     return packageInfo.data_streams || [];
@@ -84,7 +85,7 @@ export const getNormalizedDataStreams = (
     const dataset = datasetName || createDefaultDatasetName(packageInfo, policyTemplate);
 
     const dataStream: RegistryDataStream = {
-      type: policyTemplate.type,
+      type: dataStreamType || policyTemplate.type,
       dataset,
       title: policyTemplate.title + ' Dataset',
       release: packageInfo.release || 'ga',

@@ -0,0 +1,123 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import {
+  ADD_INTEGRATION_POLICY_BTN,
+  CREATE_PACKAGE_POLICY_SAVE_BTN,
+  INTEGRATION_NAME_LINK,
+  POLICY_EDITOR,
+} from '../screens/integrations';
+import { EXISTING_HOSTS_TAB } from '../screens/fleet';
+import { CONFIRM_MODAL } from '../screens/navigation';
+
+import { API_VERSIONS } from '../../common';
+import { cleanupAgentPolicies } from '../tasks/cleanup';
+import { login } from '../tasks/login';
+import { request } from '../tasks/common';
+
+const INPUT_TEST_PACKAGE = 'input_package-1.0.0';
+
+describe('Input package with custom data stream type', () => {
+  beforeEach(() => {
+    login();
+  });
+
+  const agentPolicyId = 'test-input-package-policy';
+  const agentPolicyName = 'Test input package policy';
+  const packagePolicyName = 'input-package-policy';
+  const datasetName = 'logs'; // Default from the package.
+  const dataStreamType = 'metrics';
+
+  before(() => {
+    cy.task('installTestPackage', 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', INPUT_TEST_PACKAGE);
+  });
+
+  it('should successfully create a package policy', () => {
+    cy.visit(`/app/integrations/detail/${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-paths')
+      .find('[data-test-subj="multiTextInputRow-0"]')
+      .click()
+      .type('/var/log/test.log');
+
+    cy.getBySel('multiTextInput-tags')
+      .find('[data-test-subj="multiTextInputRow-0"]')
+      .click()
+      .type('tag1');
+
+    // Select metrics 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(`${dataStreamType} checkbox should be checked`, () => {
+    cy.visit(`/app/integrations/detail/${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/${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}-*`);
+  });
+});
@@ -10,10 +10,14 @@ import ReactMarkdown from 'react-markdown';
 import styled from 'styled-components';
 import { uniq } from 'lodash';
 import { FormattedMessage } from '@kbn/i18n-react';
+import { i18n } from '@kbn/i18n';
 import {
   EuiFlexGrid,
   EuiFlexGroup,
   EuiFlexItem,
+  EuiFormRow,
+  EuiLink,
+  EuiRadioGroup,
   EuiSwitch,
   EuiText,
   EuiSpacer,
@@ -24,9 +28,12 @@ import { useRouteMatch } from 'react-router-dom';
 
 import { useQuery } from '@tanstack/react-query';
 
-import { DATASET_VAR_NAME } from '../../../../../../../../../common/constants';
+import {
+  DATASET_VAR_NAME,
+  DATA_STREAM_TYPE_VAR_NAME,
+} from '../../../../../../../../../common/constants';
 
-import { useConfig, sendGetDataStreams } from '../../../../../../../../hooks';
+import { useConfig, sendGetDataStreams, useStartServices } from '../../../../../../../../hooks';
 
 import {
   getRegistryDataStreamAssetBaseName,
@@ -76,6 +83,8 @@ export const PackagePolicyInputStreamConfig = memo<Props>(
     forceShowErrors,
     isEditPage,
   }) => {
+    const { docLinks } = useStartServices();
+
     const config = useConfig();
     const isExperimentalDataStreamSettingsEnabled =
       config.enableExperimental?.includes('experimentalDataStreamSettings') ?? false;
@@ -95,6 +104,10 @@ export const PackagePolicyInputStreamConfig = memo<Props>(
     const customDatasetVar = packagePolicyInputStream.vars?.[DATASET_VAR_NAME];
     const customDatasetVarValue = customDatasetVar?.value?.dataset || customDatasetVar?.value;
 
+    const customDataStreamTypeVar = packagePolicyInputStream.vars?.[DATA_STREAM_TYPE_VAR_NAME];
+    const customDataStreamTypeVarValue =
+      customDataStreamTypeVar?.value || packagePolicyInputStream.data_stream.type || 'logs';
+
     const { exists: indexTemplateExists, isLoading: isLoadingIndexTemplate } =
       useIndexTemplateExists(
         getRegistryDataStreamAssetBaseName({
@@ -255,7 +268,7 @@ export const PackagePolicyInputStreamConfig = memo<Props>(
               })}
 
               {/* Advanced section */}
-              {hasAdvancedOptions && (
+              {(hasAdvancedOptions || packageInfo.type === 'input') && (
                 <Fragment>
                   <EuiFlexItem>
                     <EuiFlexGroup justifyContent="spaceBetween" alignItems="center">
@@ -288,6 +301,75 @@ export const PackagePolicyInputStreamConfig = memo<Props>(
                   </EuiFlexItem>
                   {isShowingAdvanced ? (
                     <>
+                      {packageInfo.type === 'input' && (
+                        <EuiFlexItem>
+                          <EuiFormRow
+                            label={
+                              <FormattedMessage
+                                id="xpack.fleet.createPackagePolicy.stepConfigure.packagePolicyDataStreamTypeInputLabel"
+                                defaultMessage="Data Stream Type"
+                              />
+                            }
+                            helpText={
+                              isEditPage ? (
+                                <FormattedMessage
+                                  id="xpack.fleet.createPackagePolicy.stepConfigure.packagePolicyInputOnlyEditDataStreamTypeHelpLabel"
+                                  defaultMessage="The data stream type cannot be changed for this integration. Create a new integration policy to use a different input type."
+                                />
+                              ) : (
+                                <FormattedMessage
+                                  id="xpack.fleet.createPackagePolicy.stepConfigure.packagePolicyDataStreamTypeHelpLabel"
+                                  defaultMessage="Select a data stream type for this policy. This setting changes the name of the integration's data stream. {learnMore}."
+                                  values={{
+                                    learnMore: (
+                                      <EuiLink
+                                        href={docLinks.links.fleet.datastreamsNamingScheme}
+                                        target="_blank"
+                                      >
+                                        {i18n.translate(
+                                          'xpack.fleet.createPackagePolicy.stepConfigure.packagePolicyNamespaceHelpLearnMoreLabel',
+                                          { defaultMessage: 'Learn more' }
+                                        )}
+                                      </EuiLink>
+                                    ),
+                                  }}
+                                />
+                              )
+                            }
+                          >
+                            <EuiRadioGroup
+                              data-test-subj="packagePolicyDataStreamType"
+                              disabled={isEditPage}
+                              idSelected={customDataStreamTypeVarValue}
+                              options={[
+                                {
+                                  id: 'logs',
+                                  label: 'Logs',
+                                },
+                                {
+                                  id: 'metrics',
+                                  label: 'Metrics',
+                                },
+                                {
+                                  id: 'traces',
+                                  label: 'Traces',
+                                },
+                              ]}
+                              onChange={(type: string) => {
+                                updatePackagePolicyInputStream({
+                                  vars: {
+                                    ...packagePolicyInputStream.vars,
+                                    [DATA_STREAM_TYPE_VAR_NAME]: {
+                                      type: 'string',
+                                      value: type,
+                                    },
+                                  },
+                                });
+                              }}
+                            />
+                          </EuiFormRow>
+                        </EuiFlexItem>
+                      )}
                       {advancedVars.map((varDef) => {
                         if (!packagePolicyInputStream.vars) return null;
                         const { name: varName, type: varType } = varDef;

@@ -72,6 +72,7 @@ describe('Fleet - storedPackagePoliciesToAgentInputs', () => {
         compiled_stream: {
           fooKey: 'fooValue1',
           fooKey2: ['fooValue2'],
+          data_stream: { dataset: 'foo' }, // data_stream.dataset can be set in the compiled stream, ensure that rest of data_stream object is properly merged.
         },
       },
       {
@@ -124,6 +125,7 @@ describe('Fleet - storedPackagePoliciesToAgentInputs', () => {
         compiled_stream: {
           fooKey: 'fooValue1',
           fooKey2: ['fooValue2'],
+          data_stream: { dataset: 'foo' }, // data_stream.dataset can be set in the compiled stream, ensure that rest of data_stream object is properly merged.
         },
       },
     ],