[Fleet] Support data_stream.type variable in simplified policy API (elastic#269895)

## Summary

Closes elastic#266321

The simplified package policy API did not support `data_stream.type` as
an overridable variable. Input-only packages define a default data
stream type in their manifest (for example, `metrics`), but there was no
way to override it via the simplified API — any attempt threw a
`Variable not found` error because the variable was not synthesised into
the policy's stream vars. This blocked elastic-package policy tests that
rely on the simplified API to set a non-default `data_stream.type`.

This PR adds full support for `data_stream.type` as a top-level variable
in input-only package policies, following the same pattern as
`data_stream.dataset` and `use_apm`.

### What changed

**`policy_template.ts`**

- Adds `DATA_STREAM_TYPE_VAR` — a synthetic `RegistryVarsEntry`
definition for `data_stream.type`.
- Adds `addDataStreamTypeVarIfNotPresent` and
`shouldIncludeDataStreamTypeVar` helpers, mirroring the existing dataset
helpers.
- `getNormalizedDataStreams` now injects the synthetic
`data_stream.type` var into every input-only stream's vars (with the
package's default type as the default value), unless the policy template
declares `dynamic_signal_types: true` — in which case the type is
resolved at runtime and the var is intentionally omitted.

**`simplified_package_policy_helper.ts`**

- Adds `syncDataStreamTypeFromVar`: iterates all streams and, when
`stream.vars['data_stream.type'].value` differs from
`stream.data_stream.type`, updates the field in place. This is what
makes the compiled agent YAML and index template lookup use the
user-supplied type rather than the package default.
- `simplifiedPackagePolicytoNewPackagePolicy` calls
`syncDataStreamTypeFromVar` before returning, so the returned
`NewPackagePolicy` is consistent.

**`package_policy.ts` (server)**

- Calls `syncDataStreamTypeFromVar` before `preflightCheckPackagePolicy`
in both `create()` and `bulkCreate()`, covering the direct Fleet API
path in addition to the simplified API.

**`package_policies_to_agent_permissions.ts`**

- When computing index permissions, reads
`stream.vars['data_stream.type'].value` first and falls back to
`stream.data_stream.type`. This ensures the correct index prefix
(`logs-*` vs `metrics-*`) is granted when the type has been overridden.

**Rejection for `dynamic_signal_types` packages**

Packages with `dynamic_signal_types: true` (for example, OTel collector)
do not get the `data_stream.type` var injected. Because the var is
absent from `varsRecord`, `assignVariables` throws a
`PackagePolicyValidationError` if a caller attempts to set it,
preventing misuse.

### Testing

- Unit tests added and updated in `policy_template.test.ts`,
`simplified_package_policy_helper.test.ts`, and
`package_policies_to_agent_permissions.test.ts`.
- Manually validated end-to-end using the test fixture from
[elastic/elastic-package#3509](https://github.com/elastic/elastic-package/pull/3509/changes#diff-d01b43d83cadd491f2e69dca24265787b53b9513812615389d30bbb63c7b9664)
with the `policy_api_format` workaround removed from the test
configuration. The downloaded agent policy correctly showed
`data_stream.type: logs` in the stream and `logs-*-*` in the index
permissions.

---

### Checklist

- [ ] Any text added follows [EUI's writing
guidelines](https://elastic.github.io/eui/#/guidelines/writing), uses
sentence case text and includes [i18n
support](https://github.com/elastic/kibana/blob/main/src/platform/packages/shared/kbn-i18n/README.md)
- [ ]
[Documentation](https://www.elastic.co/guide/en/kibana/master/development-documentation.html)
was added for features that require explanation or tutorials
- [x] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios
- [ ] If a plugin configuration key changed, check if it needs to be
allowlisted in the cloud and added to the [docker
list](https://github.com/elastic/kibana/blob/main/src/dev/build/tasks/os_packages/docker_generator/resources/base/bin/kibana-docker)
- [ ] This was checked for breaking HTTP API changes, and any breaking
changes have been approved by the breaking-change committee. The
`release_note:breaking` label should be applied in these situations.
- [ ] [Flaky Test
Runner](https://ci-stats.kibana.dev/trigger_flaky_test_runner/1) was
used on any tests changed
- [ ] The PR description includes the appropriate Release Notes section,
and the correct `release_note:*` label is applied per the
[guidelines](https://www.elastic.co/guide/en/kibana/master/contributing.html#kibana-release-notes-process)
- [ ] Review the [backport
guidelines](https://docs.google.com/document/d/1VyN5k91e5OVumlc0Gb9RPa3h1ewuPE705nRtioPiTvY/edit?usp=sharing)
and apply applicable `backport:*` labels.

### Identify risks

- **Immutable type after creation** — like `data_stream.dataset`, the
`data_stream.type` var is immutable once the policy is saved. An attempt
to update it is rejected with a validation error (existing guard in
`package_policy.ts`). Low risk; consistent with existing dataset
immutability.
- **Input-only packages only** — the synthetic var is only injected for
`type: input` packages. Composable integration packages are unaffected.
- [ ] [See some risk
examples](https://github.com/elastic/kibana/blob/main/RISK_MATRIX.mdx)

### Release note

```
Fleet now supports overriding `data_stream.type` via the simplified package policy API for input-only packages. Setting this variable routes collected data to a different Elasticsearch index prefix (e.g. `logs-*` instead of `metrics-*`) and grants the appropriate index permissions to the agent.
```

Author: teresaromero

.buildkite/pipelines/storybooks_from_pr.yml

@@ -17,7 +17,6 @@ steps:
       diskType: hyperdisk-balanced
       preemptible: true
       spotZones: us-central1-a,us-central1-b,us-central1-c,us-central1-f,us-east1-b,us-east1-c,us-east1-d,us-west1-a,us-west1-b,us-west1-c
-      preemptible: true
       diskSizeGb: 105
     timeout_in_minutes: 50
     retry:

x-pack/platform/plugins/shared/fleet/common/index.ts

@@ -87,7 +87,9 @@ export {
   // Package policy helpers
   isValidNamespace,
   isValidDataset,
+  isValidDataStreamType,
   INVALID_NAMESPACE_CHARACTERS,
+  VALID_DATA_STREAM_TYPES,
   getFileMetadataIndexName,
   getFileDataIndexName,
   removeSOAttributes,

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

@@ -27,8 +27,10 @@ export { fullAgentPolicyToYaml } from './full_agent_policy_to_yaml';
 export { isPackageLimited, doesAgentPolicyAlreadyIncludePackage } from './limited_package';
 export {
   isValidDataset,
+  isValidDataStreamType,
   isValidNamespace,
   INVALID_NAMESPACE_CHARACTERS,
+  VALID_DATA_STREAM_TYPES,
 } from './is_valid_namespace';
 export { isDiffPathProtocol } from './is_diff_path_protocol';
 export { LicenseService } from './license';
@@ -47,10 +49,14 @@ export {
   MINIMUM_PRIVILEGE_LEVEL_CHANGE_AGENT_VERSION,
   isAgentEligibleForPrivilegeLevelChange,
 } from './agent_privilege_level_change_helpers';
+export { syncDataStreamTypeFromVar } from './simplified_package_policy_helper';
 export {
   addUseAPMVarIfNotPresent,
   DATA_STREAM_USE_APM_VAR,
   shouldIncludeUseAPMVar,
+  addDataStreamTypeVarIfNotPresent,
+  DATA_STREAM_TYPE_VAR,
+  shouldIncludeDataStreamTypeVar,
   isInputOnlyPolicyTemplate,
   isIntegrationPolicyTemplate,
   getNormalizedInputs,

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

@@ -5,7 +5,7 @@
  * 2.0.
  */
 
-import { isValidNamespace } from './is_valid_namespace';
+import { isValidNamespace, isValidDataStreamType } from './is_valid_namespace';
 
 describe('Fleet - isValidNamespace', () => {
   it('returns true for valid namespaces', () => {
@@ -51,3 +51,28 @@ describe('Fleet - isValidNamespace', () => {
     expect(isValidNamespace('qaenv', false, ['prod', 'qa']).valid).toBe(true);
   });
 });
+
+describe('Fleet - isValidDataStreamType', () => {
+  it('returns true for each valid type', () => {
+    expect(isValidDataStreamType('logs').valid).toBe(true);
+    expect(isValidDataStreamType('metrics').valid).toBe(true);
+    expect(isValidDataStreamType('traces').valid).toBe(true);
+    expect(isValidDataStreamType('synthetics').valid).toBe(true);
+    expect(isValidDataStreamType('profiling').valid).toBe(true);
+  });
+
+  it('returns false for unknown types', () => {
+    expect(isValidDataStreamType('bogus').valid).toBe(false);
+    expect(isValidDataStreamType('LOGS').valid).toBe(false);
+    expect(isValidDataStreamType('').valid).toBe(false);
+  });
+
+  it('returns true for blank when allowBlank is true', () => {
+    expect(isValidDataStreamType('', true).valid).toBe(true);
+    expect(isValidDataStreamType('  ', true).valid).toBe(true);
+  });
+
+  it('returns false for blank when allowBlank is false', () => {
+    expect(isValidDataStreamType('', false).valid).toBe(false);
+  });
+});

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

@@ -7,6 +7,8 @@
 
 import { i18n } from '@kbn/i18n';
 
+import type { PackageDataStreamTypes } from '../types';
+
 // Namespace string eventually becomes part of an index name. This method partially implements index name rules from
 // https://github.com/elastic/elasticsearch/blob/master/docs/reference/indices/create-index.asciidoc
 // and implements a limit based on https://github.com/elastic/kibana/issues/75846
@@ -118,3 +120,30 @@ function isValidEntity(
 }
 
 export const INVALID_NAMESPACE_CHARACTERS = /[\*\\/\?"<>|\s,#:-]+/;
+
+export const VALID_DATA_STREAM_TYPES: readonly PackageDataStreamTypes[] = [
+  'logs',
+  'metrics',
+  'traces',
+  'synthetics',
+  'profiling',
+];
+
+export function isValidDataStreamType(
+  type: string,
+  allowBlank?: boolean
+): { valid: boolean; error?: string } {
+  if (!type.trim() && allowBlank) {
+    return { valid: true };
+  }
+  if (!VALID_DATA_STREAM_TYPES.includes(type as PackageDataStreamTypes)) {
+    return {
+      valid: false,
+      error: i18n.translate('xpack.fleet.dataStreamTypeValidation.invalidValueErrorMessage', {
+        defaultMessage: 'Data stream type must be one of: {allowedTypes}',
+        values: { allowedTypes: VALID_DATA_STREAM_TYPES.join(', ') },
+      }),
+    };
+  }
+  return { valid: true };
+}

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

@@ -30,6 +30,8 @@ import {
   packagePolicyInputAllowsUndefinedDataStreamType,
   hasDynamicSignalTypes,
   shouldIncludeUseAPMVar,
+  shouldIncludeDataStreamTypeVar,
+  addDataStreamTypeVarIfNotPresent,
 } from './policy_template';
 
 describe('isInputOnlyPolicyTemplate', () => {
@@ -198,6 +200,54 @@ describe('shouldIncludeUseAPMVar', () => {
   });
 });
 
+describe('shouldIncludeDataStreamTypeVar', () => {
+  it('returns true when dynamic_signal_types is false', () => {
+    expect(shouldIncludeDataStreamTypeVar(false)).toBe(true);
+  });
+
+  it('returns false when dynamic_signal_types is true', () => {
+    expect(shouldIncludeDataStreamTypeVar(true)).toBe(false);
+  });
+});
+
+describe('addDataStreamTypeVarIfNotPresent', () => {
+  it('adds the data_stream.type var with the given default when not already present', () => {
+    const result = addDataStreamTypeVarIfNotPresent([], 'metrics');
+    expect(result).toHaveLength(1);
+    expect(result[0].name).toEqual('data_stream.type');
+    expect(result[0].default).toEqual('metrics');
+    expect(result[0].show_user).toEqual(false);
+  });
+
+  it('uses empty array when vars is undefined', () => {
+    const result = addDataStreamTypeVarIfNotPresent(undefined, 'logs');
+    expect(result).toHaveLength(1);
+    expect(result[0].name).toEqual('data_stream.type');
+  });
+
+  it('does not set a default when defaultType is not provided', () => {
+    const result = addDataStreamTypeVarIfNotPresent([]);
+    expect(result).toHaveLength(1);
+    expect(result[0].name).toEqual('data_stream.type');
+    expect(result[0].default).toBeUndefined();
+  });
+
+  it('does not add the var when data_stream.type is already present', () => {
+    const existing = {
+      name: 'data_stream.type',
+      type: 'text' as RegistryVarType,
+      title: 'existing',
+      description: 'existing',
+      multi: false,
+      required: false,
+      show_user: true,
+    };
+    const result = addDataStreamTypeVarIfNotPresent([existing], 'metrics');
+    expect(result).toHaveLength(1);
+    expect(result[0]).toBe(existing);
+  });
+});
+
 describe('getNormalizedDataStreams', () => {
   const integrationPkg: PackageInfo = {
     name: 'nginx',
@@ -324,7 +374,88 @@ describe('getNormalizedDataStreams', () => {
     });
     expect(result).toHaveLength(1);
     expect(result[0].streams).toHaveLength(1);
-    expect(result?.[0].streams?.[0]?.vars).toEqual([datasetVar]);
+    const vars = result?.[0].streams?.[0]?.vars;
+    // dataset var should not be duplicated
+    expect(vars?.filter((v) => v.name === 'data_stream.dataset')).toHaveLength(1);
+    expect(vars?.find((v) => v.name === 'data_stream.dataset')).toMatchObject(datasetVar);
+  });
+
+  it('should add synthetic data_stream.type var with policyTemplate.type as default', () => {
+    const result = getNormalizedDataStreams({
+      ...integrationPkg,
+      type: 'input',
+      policy_templates: [
+        {
+          input: 'logfile',
+          type: 'logs',
+          name: 'myinput',
+          template_path: 'some/path.hbl',
+          title: 'My Input',
+          description: 'My Input',
+          vars: [],
+        },
+      ],
+    });
+    expect(result).toHaveLength(1);
+    const vars = result[0].streams![0].vars;
+    const typeVar = vars?.find((v) => v.name === 'data_stream.type');
+    expect(typeVar).toBeDefined();
+    expect(typeVar?.default).toEqual('logs');
+    expect(typeVar?.show_user).toEqual(false);
+  });
+
+  it('should not add data_stream.type var when already declared by the package', () => {
+    const existingTypeVar = {
+      name: 'data_stream.type',
+      type: 'text' as RegistryVarType,
+      title: 'custom type',
+      description: 'custom',
+      multi: false,
+      required: false,
+      show_user: true,
+    };
+    const result = getNormalizedDataStreams({
+      ...integrationPkg,
+      type: 'input',
+      policy_templates: [
+        {
+          input: 'logfile',
+          type: 'logs',
+          name: 'myinput',
+          template_path: 'some/path.hbl',
+          title: 'My Input',
+          description: 'My Input',
+          vars: [existingTypeVar],
+        },
+      ],
+    });
+    expect(result).toHaveLength(1);
+    const vars = result[0].streams![0].vars;
+    const typeVars = vars?.filter((v) => v.name === 'data_stream.type');
+    expect(typeVars).toHaveLength(1);
+    expect(typeVars![0]).toMatchObject(existingTypeVar);
+  });
+
+  it('should not add data_stream.type var when dynamic_signal_types is true', () => {
+    const result = getNormalizedDataStreams({
+      ...integrationPkg,
+      type: 'input',
+      policy_templates: [
+        {
+          input: 'otelcol',
+          name: 'otel-dynamic',
+          template_path: 'some/path.hbl',
+          title: 'OTel Dynamic',
+          description: 'OTel with dynamic signal types',
+          dynamic_signal_types: true,
+          vars: [],
+        },
+      ],
+    } as any);
+    expect(result).toHaveLength(1);
+    const vars = result[0].streams![0].vars;
+    const typeVar = vars?.find((v) => v.name === 'data_stream.type');
+    expect(typeVar).toBeUndefined();
   });
 
   const inputPkg: PackageInfo = {

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

@@ -9,6 +9,7 @@ import { i18n } from '@kbn/i18n';
 
 import {
   DATASET_VAR_NAME,
+  DATA_STREAM_TYPE_VAR_NAME,
   dataTypes,
   OTEL_COLLECTOR_INPUT_TYPE,
   USE_APM_VAR_NAME,
@@ -43,6 +44,21 @@ const DATA_STREAM_DATASET_VAR: RegistryVarsEntry = {
   show_user: true,
 };
 
+export const DATA_STREAM_TYPE_VAR: RegistryVarsEntry = {
+  name: DATA_STREAM_TYPE_VAR_NAME,
+  type: 'text',
+  title: i18n.translate('xpack.fleet.policyTemplate.dataStreamTypeVar.title', {
+    defaultMessage: 'Data stream type',
+  }),
+  description: i18n.translate('xpack.fleet.policyTemplate.dataStreamTypeVar.description', {
+    defaultMessage:
+      'Set the type for your data stream. Valid values are logs, metrics, traces, and synthetics.',
+  }),
+  multi: false,
+  required: false,
+  show_user: false,
+};
+
 export const DATA_STREAM_USE_APM_VAR: RegistryVarsEntry = {
   name: USE_APM_VAR_NAME,
   type: 'bool',
@@ -202,6 +218,9 @@ export function getNormalizedDataStreams(
     const dataset = datasetName || createDefaultDatasetName(packageInfo, policyTemplate);
 
     let vars = addDatasetVarIfNotPresent(policyTemplate.vars, policyTemplate.name);
+    if (shouldIncludeDataStreamTypeVar(policyTemplate.dynamic_signal_types === true)) {
+      vars = addDataStreamTypeVarIfNotPresent(vars, policyTemplate.type);
+    }
     if (
       shouldIncludeUseAPMVar(
         policyTemplate.input,
@@ -288,6 +307,26 @@ export const addUseAPMVarIfNotPresent = (vars?: RegistryVarsEntry[]): RegistryVa
   }
 };
 
+export const shouldIncludeDataStreamTypeVar = (isDynamicSignalTypes: boolean): boolean =>
+  !isDynamicSignalTypes;
+
+export const addDataStreamTypeVarIfNotPresent = (
+  vars?: RegistryVarsEntry[],
+  defaultType?: string
+): RegistryVarsEntry[] => {
+  const newVars = vars ?? [];
+
+  const isDataStreamTypeVarAlreadyAdded = newVars.find(
+    (varEntry) => varEntry.name === DATA_STREAM_TYPE_VAR_NAME
+  );
+
+  if (isDataStreamTypeVarAlreadyAdded) {
+    return newVars;
+  } else {
+    return [...newVars, { ...DATA_STREAM_TYPE_VAR, ...(defaultType && { default: defaultType }) }];
+  }
+};
+
 const createDefaultDatasetName = (
   packageInfo: { name: string },
   policyTemplate: { name: string }