[Fleet] Make upload and registry package info consistent

x-pack/plugins/fleet/common/types/models/epm.ts

@@ -305,6 +305,7 @@ export enum RegistryDataStreamKeys {
 }
 
 export interface RegistryDataStream {
+  [key: string]: any;
   [RegistryDataStreamKeys.type]: string;
   [RegistryDataStreamKeys.ilm_policy]?: string;
   [RegistryDataStreamKeys.hidden]?: boolean;
@@ -323,6 +324,7 @@ export interface RegistryElasticsearch {
   privileges?: RegistryDataStreamPrivileges;
   'index_template.settings'?: estypes.IndicesIndexSettings;
   'index_template.mappings'?: estypes.MappingTypeMapping;
+  'ingest_pipeline.name'?: string;
 }
 
 export interface RegistryDataStreamPrivileges {

x-pack/plugins/fleet/common/types/models/package_spec.ts

@@ -70,4 +70,5 @@ export interface PackageSpecScreenshot {
   title: string;
   size?: string;
   type?: string;
+  path?: string;
 }

x-pack/plugins/fleet/server/integration_tests/helpers/docker_registry_helper.ts

@@ -24,7 +24,7 @@ export function useDockerRegistry() {
 
   let dockerProcess: ChildProcess | undefined;
   async function startDockerRegistryServer() {
-    const dockerImage = `docker.elastic.co/package-registry/distribution@sha256:8b4ce36ecdf86e6cfdf781d9df8d564a014add9afc9aec21cf2c5a68ff82d3ab`;
+    const dockerImage = `docker.elastic.co/package-registry/distribution@sha256:b3dfc6a11ff7dce82ba8689ea9eeb54e353c6b4bfd2d28127b20ef72fd8883e9`;
 
     const args = ['run', '--rm', '-p', `${packageRegistryPort}:8080`, dockerImage];
 

x-pack/plugins/fleet/server/integration_tests/validate_bundled_packages.test.ts

@@ -0,0 +1,77 @@
+/*
+ * 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 path from 'path';
+import fs from 'fs/promises';
+
+import JSON5 from 'json5';
+import { REPO_ROOT } from '@kbn/utils';
+
+import * as Registry from '../services/epm/registry';
+import { parseAndVerifyArchiveEntries } from '../services/epm/archive';
+
+import { createAppContextStartContractMock } from '../mocks';
+import { appContextService } from '../services';
+
+import { useDockerRegistry } from './helpers';
+
+describe('validate bundled packages', () => {
+  const registryUrl = useDockerRegistry();
+  let mockContract: ReturnType<typeof createAppContextStartContractMock>;
+
+  beforeEach(() => {
+    mockContract = createAppContextStartContractMock({ registryUrl });
+    appContextService.start(mockContract);
+  });
+
+  async function getBundledPackageEntries() {
+    const configFilePath = path.resolve(REPO_ROOT, 'fleet_packages.json');
+    const configFile = await fs.readFile(configFilePath, 'utf8');
+    const bundledPackages = JSON5.parse(configFile);
+
+    return bundledPackages as Array<{ name: string; version: string }>;
+  }
+
+  async function setupPackageObjects() {
+    const bundledPackages = await getBundledPackageEntries();
+
+    const packageObjects = await Promise.all(
+      bundledPackages.map(async (bundledPackage) => {
+        const registryPackage = await Registry.getRegistryPackage(
+          bundledPackage.name,
+          bundledPackage.version
+        );
+
+        const packageArchive = await Registry.fetchArchiveBuffer(
+          bundledPackage.name,
+          bundledPackage.version
+        );
+
+        return { registryPackage, packageArchive };
+      })
+    );
+
+    return packageObjects;
+  }
+
+  it('generates matching package info objects for uploaded and registry packages', async () => {
+    const packageObjects = await setupPackageObjects();
+
+    for (const packageObject of packageObjects) {
+      const { registryPackage, packageArchive } = packageObject;
+
+      const archivePackageInfo = await parseAndVerifyArchiveEntries(
+        packageArchive.archiveBuffer,
+        'application/zip'
+      );
+
+      expect(archivePackageInfo.packageInfo.data_streams).toEqual(
+        registryPackage.packageInfo.data_streams
+      );
+    }
+  });
+});

x-pack/plugins/fleet/server/mocks/index.ts

@@ -21,6 +21,7 @@ import type { PackagePolicyServiceInterface } from '../services/package_policy';
 import type { AgentPolicyServiceInterface } from '../services';
 import type { FleetAppContext } from '../plugin';
 import { createMockTelemetryEventsSender } from '../telemetry/__mocks__';
+import type { FleetConfigType } from '../../common';
 import { createFleetAuthzMock } from '../../common';
 import { agentServiceMock } from '../services/agents/agent_service.mock';
 import type { FleetRequestHandlerContext } from '../types';
@@ -39,11 +40,14 @@ export interface MockedFleetAppContext extends FleetAppContext {
   logger: ReturnType<ReturnType<typeof loggingSystemMock.create>['get']>;
 }
 
-export const createAppContextStartContractMock = (): MockedFleetAppContext => {
+export const createAppContextStartContractMock = (
+  configOverrides: Partial<FleetConfigType> = {}
+): MockedFleetAppContext => {
   const config = {
     agents: { enabled: true, elasticsearch: {} },
     enabled: true,
     agentIdVerificationEnabled: true,
+    ...configOverrides,
   };
 
   const config$ = of(config);

x-pack/plugins/fleet/server/services/epm/archive/validation.ts

@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import { merge } from '@kbn/std';
 import yaml from 'js-yaml';
 import { pick, uniq } from 'lodash';
 
@@ -32,6 +33,42 @@ import { unpackBufferEntries } from './index';
 const MANIFESTS: Record<string, Buffer> = {};
 const MANIFEST_NAME = 'manifest.yml';
 
+const DEFAULT_RELEASE_VALUE = 'ga';
+
+const DEFAULT_INGEST_PIPELINE_VALUE = 'default';
+const DEFAULT_INGEST_PIPELINE_FILE_NAME_YML = 'default.yml';
+const DEFAULT_INGEST_PIPELINE_FILE_NAME_JSON = 'default.json';
+
+// Borrowed from https://github.com/elastic/kibana/blob/main/x-pack/plugins/security_solution/common/utils/expand_dotted.ts
+// With some alterations around non-object values
+const expandDottedField = (dottedFieldName: string, val: unknown): object => {
+  const parts = dottedFieldName.split('.');
+
+  if (parts.length === 1) {
+    return { [parts[0]]: val };
+  } else {
+    return { [parts[0]]: expandDottedField(parts.slice(1).join('.'), val) };
+  }
+};
+
+export const expandDottedObject = (dottedObj: object) => {
+  if (typeof dottedObj !== 'object' || Array.isArray(dottedObj)) {
+    return dottedObj;
+  }
+  return Object.entries(dottedObj).reduce(
+    (acc, [key, val]) => merge(acc, expandDottedField(key, val)),
+    {}
+  );
+};
+
+export const expandDottedEntries = (obj: object) => {
+  return Object.entries<any>(obj).reduce<any>((acc, [key, value]) => {
+    acc[key] = expandDottedObject(value);
+
+    return acc;
+  }, {} as Record<string, any>);
+};
+
 // not sure these are 100% correct but they do the job here
 // keeping them local until others need them
 type OptionalPropertyOf<T extends object> = Exclude<
@@ -144,8 +181,13 @@ function parseAndVerifyArchive(paths: string[]): ArchivePackage {
     );
   }
 
-  parsed.data_streams = parseAndVerifyDataStreams(paths, parsed.name, parsed.version);
+  const parsedDataStreams = parseAndVerifyDataStreams(paths, parsed.name, parsed.version);
+  if (parsedDataStreams.length) {
+    parsed.data_streams = parsedDataStreams;
+  }
+
   parsed.policy_templates = parseAndVerifyPolicyTemplates(manifest);
+
   // add readme if exists
   const readme = parseAndVerifyReadme(paths, parsed.name, parsed.version);
   if (readme) {
@@ -202,40 +244,87 @@ export function parseAndVerifyDataStreams(
 
     const {
       title: dataStreamTitle,
-      release,
+      release = DEFAULT_RELEASE_VALUE,
       type,
       dataset,
-      ingest_pipeline: ingestPipeline,
       streams: manifestStreams,
+      elasticsearch,
       ...restOfProps
     } = manifest;
     if (!(dataStreamTitle && type)) {
       throw new PackageInvalidArchiveError(
         `Invalid manifest for data stream '${dataStreamPath}': one or more fields missing of 'title', 'type'`
       );
     }
+
+    let ingestPipeline;
+    const ingestPipelinePaths = paths.filter((path) =>
+      path.startsWith(`${pkgKey}/data_stream/${dataStreamPath}/elasticsearch/ingest_pipeline`)
+    );
+
+    if (
+      ingestPipelinePaths.length &&
+      (ingestPipelinePaths.some((ingestPipelinePath) =>
+        ingestPipelinePath.endsWith(DEFAULT_INGEST_PIPELINE_FILE_NAME_YML)
+      ) ||
+        ingestPipelinePaths.some((ingestPipelinePath) =>
+          ingestPipelinePath.endsWith(DEFAULT_INGEST_PIPELINE_FILE_NAME_JSON)
+        ))
+    ) {
+      ingestPipeline = DEFAULT_INGEST_PIPELINE_VALUE;
+    }
+
     const streams = parseAndVerifyStreams(manifestStreams, dataStreamPath);
 
+    const parsedElasticsearchEntry: Record<string, any> = {};
+
+    if (ingestPipeline) {
+      parsedElasticsearchEntry['ingest_pipeline.name'] = DEFAULT_INGEST_PIPELINE_VALUE;
+    }
+
+    if (elasticsearch?.privileges) {
+      parsedElasticsearchEntry.privileges = elasticsearch.privileges;
+    }
+
+    if (elasticsearch?.index_template?.mappings) {
+      parsedElasticsearchEntry['index_template.mappings'] = expandDottedEntries(
+        elasticsearch.index_template.mappings
+      );
+    }
+
+    if (elasticsearch?.index_template?.settings) {
+      parsedElasticsearchEntry['index_template.settings'] = expandDottedEntries(
+        elasticsearch.index_template.settings
+      );
+    }
+
+    // Build up the stream object here so we can conditionally insert nullable fields. The package registry omits undefined
+    // fields, so we're mimicking that behavior here.
+    const dataStreamObject: RegistryDataStream = {
+      title: dataStreamTitle,
+      release,
+      type,
+      package: pkgName,
+      dataset: dataset || `${pkgName}.${dataStreamPath}`,
+      path: dataStreamPath,
+      elasticsearch: parsedElasticsearchEntry,
+    };
+
+    if (ingestPipeline) {
+      dataStreamObject.ingest_pipeline = ingestPipeline;
+    }
+
+    if (streams.length) {
+      dataStreamObject.streams = streams;
+    }
+
     dataStreams.push(
-      Object.entries(restOfProps).reduce(
-        (validatedDataStream, [key, value]) => {
-          if (registryDataStreamProps.includes(key as RegistryDataStreamKeys)) {
-            // @ts-expect-error
-            validatedDataStream[key] = value;
-          }
-          return validatedDataStream;
-        },
-        {
-          title: dataStreamTitle,
-          release,
-          type,
-          package: pkgName,
-          dataset: dataset || `${pkgName}.${dataStreamPath}`,
-          ingest_pipeline: ingestPipeline,
-          path: dataStreamPath,
-          streams,
+      Object.entries(restOfProps).reduce((validatedDataStream, [key, value]) => {
+        if (registryDataStreamProps.includes(key as RegistryDataStreamKeys)) {
+          validatedDataStream[key] = value;
         }
-      )
+        return validatedDataStream;
+      }, dataStreamObject)
     );
   });
 
@@ -261,25 +350,28 @@ export function parseAndVerifyStreams(
           `Invalid manifest for data stream ${dataStreamPath}: stream is missing one or more fields of: input, title`
         );
       }
+
       const vars = parseAndVerifyVars(manifestVars, `data stream ${dataStreamPath}`);
 
+      const streamObject: RegistryStream = {
+        input,
+        title: streamTitle,
+        template_path: templatePath || 'stream.yml.hbs',
+      };
+
+      if (vars.length) {
+        streamObject.vars = vars;
+      }
+
       // default template path name see https://github.com/elastic/package-registry/blob/master/util/dataset.go#L143
       streams.push(
-        Object.entries(restOfProps).reduce(
-          (validatedStream, [key, value]) => {
-            if (registryStreamProps.includes(key as RegistryStreamKeys)) {
-              // @ts-expect-error
-              validatedStream[key] = value;
-            }
-            return validatedStream;
-          },
-          {
-            input,
-            title: streamTitle,
-            vars,
-            template_path: templatePath || 'stream.yml.hbs',
-          } as RegistryStream
-        )
+        Object.entries(restOfProps).reduce((validatedStream, [key, value]) => {
+          if (registryStreamProps.includes(key as RegistryStreamKeys)) {
+            // @ts-expect-error
+            validatedStream[key] = value;
+          }
+          return validatedStream;
+        }, streamObject)
       );
     });
   }

x-pack/plugins/fleet/server/services/epm/registry/index.ts

@@ -261,7 +261,7 @@ export async function ensureCachedArchiveInfo(
   }
 }
 
-async function fetchArchiveBuffer(
+export async function fetchArchiveBuffer(
   pkgName: string,
   pkgVersion: string
 ): Promise<{ archiveBuffer: Buffer; archivePath: string }> {

x-pack/test/fleet_api_integration/apis/epm/index.js

@@ -14,7 +14,7 @@ export default function loadTests({ loadTestFile }) {
     loadTestFile(require.resolve('./file'));
     loadTestFile(require.resolve('./template'));
     loadTestFile(require.resolve('./ilm'));
-    // loadTestFile(require.resolve('./install_bundled'));
+    loadTestFile(require.resolve('./install_bundled'));
     loadTestFile(require.resolve('./install_by_upload'));
     loadTestFile(require.resolve('./install_endpoint'));
     loadTestFile(require.resolve('./install_overrides'));
@@ -29,5 +29,6 @@ export default function loadTests({ loadTestFile }) {
     loadTestFile(require.resolve('./package_install_complete'));
     loadTestFile(require.resolve('./install_error_rollback'));
     loadTestFile(require.resolve('./final_pipeline'));
+    loadTestFile(require.resolve('./validate_bundled_packages'));
   });
 }

x-pack/test/fleet_api_integration/config.ts

@@ -15,7 +15,7 @@ import { defineDockerServersConfig } from '@kbn/test';
 // example: https://beats-ci.elastic.co/blue/organizations/jenkins/Ingest-manager%2Fpackage-storage/detail/snapshot/74/pipeline/257#step-302-log-1.
 // It should be updated any time there is a new Docker image published for the Snapshot Distribution of the Package Registry.
 export const dockerImage =
-  'docker.elastic.co/package-registry/distribution@sha256:8b4ce36ecdf86e6cfdf781d9df8d564a014add9afc9aec21cf2c5a68ff82d3ab';
+  'docker.elastic.co/package-registry/distribution@sha256:b3dfc6a11ff7dce82ba8689ea9eeb54e353c6b4bfd2d28127b20ef72fd8883e9';
 
 export default async function ({ readConfigFile }: FtrConfigProviderContext) {
   const xPackAPITestsConfig = await readConfigFile(require.resolve('../api_integration/config.ts'));

x-pack/test/functional/config.js

@@ -15,7 +15,7 @@ import { pageObjects } from './page_objects';
 // example: https://beats-ci.elastic.co/blue/organizations/jenkins/Ingest-manager%2Fpackage-storage/detail/snapshot/74/pipeline/257#step-302-log-1.
 // It should be updated any time there is a new Docker image published for the Snapshot Distribution of the Package Registry.
 export const dockerImage =
-  'docker.elastic.co/package-registry/distribution@sha256:8b4ce36ecdf86e6cfdf781d9df8d564a014add9afc9aec21cf2c5a68ff82d3ab';
+  'docker.elastic.co/package-registry/distribution@sha256:b3dfc6a11ff7dce82ba8689ea9eeb54e353c6b4bfd2d28127b20ef72fd8883e9';
 
 // the default export of config files must be a config provider
 // that returns an object with the projects config values