[Logs+] Implement Data Source APIs (elastic#156413)

## Summary

Research: https://github.com/elastic/observability-dev/issues/2620
Requirements: https://github.com/elastic/observability-dev/issues/2639
Implementation: https://github.com/elastic/observability-dev/issues/2653

This implements two APIs ("Enumerate integrations" and "Enumerate data
streams") under the Fleet plugin, and specifically the `/epm` namespace
given the input here:
https://github.com/elastic/observability-dev/issues/2639#issuecomment-1521689096

The Enumerate Integrations API can be queried like so (example with all
parameters):

`GET
/api/fleet/epm/packages/installed?nameQuery=system&pageSize=5&type=logs&pageAfter=["system"]&sortDirection=asc`

The Enumerate Data Streams API can be queried like so (example with all
parameters):

`GET
/api/fleet/epm/data_streams?uncategorisedOnly=true&type=logs&sortDirection=desc&datasetQuery=beat`

x-pack/plugins/fleet/common/constants/routes.ts

@@ -21,14 +21,17 @@ export const LIMITED_CONCURRENCY_ROUTE_TAG = 'ingest:limited-concurrency';
 
 // EPM API routes
 const EPM_PACKAGES_MANY = `${EPM_API_ROOT}/packages`;
+const EPM_PACKAGES_INSTALLED = `${EPM_API_ROOT}/packages/installed`;
 const EPM_PACKAGES_BULK = `${EPM_PACKAGES_MANY}/_bulk`;
 const EPM_PACKAGES_ONE_DEPRECATED = `${EPM_PACKAGES_MANY}/{pkgkey}`;
 const EPM_PACKAGES_ONE = `${EPM_PACKAGES_MANY}/{pkgName}/{pkgVersion}`;
 export const EPM_API_ROUTES = {
   BULK_INSTALL_PATTERN: EPM_PACKAGES_BULK,
   LIST_PATTERN: EPM_PACKAGES_MANY,
+  INSTALLED_LIST_PATTERN: EPM_PACKAGES_INSTALLED,
   LIMITED_LIST_PATTERN: `${EPM_PACKAGES_MANY}/limited`,
   INFO_PATTERN: EPM_PACKAGES_ONE,
+  DATA_STREAMS_PATTERN: `${EPM_API_ROOT}/data_streams`,
   INSTALL_FROM_REGISTRY_PATTERN: EPM_PACKAGES_ONE,
   INSTALL_BY_UPLOAD_PATTERN: EPM_PACKAGES_MANY,
   DELETE_PATTERN: EPM_PACKAGES_ONE,

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

@@ -24,3 +24,5 @@ export interface DataStream {
     serviceName: string;
   } | null;
 }
+
+export type PackageDataStreamTypes = 'logs' | 'metrics' | 'traces' | 'synthetics' | 'profiling';

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

@@ -5,6 +5,8 @@
  * 2.0.
  */
 
+import type { SortResults } from '@elastic/elasticsearch/lib/api/types';
+
 import type {
   AssetReference,
   CategorySummaryList,
@@ -13,6 +15,7 @@ import type {
   PackageUsageStats,
   InstallType,
   InstallSource,
+  EpmPackageInstallStatus,
 } from '../models/epm';
 
 export interface GetCategoriesRequest {
@@ -46,6 +49,26 @@ export interface GetPackagesResponse {
   response?: PackageList;
 }
 
+interface InstalledPackage {
+  name: string;
+  version: string;
+  status: EpmPackageInstallStatus;
+  dataStreams: Array<{
+    name: string;
+    title: string;
+  }>;
+}
+export interface GetInstalledPackagesResponse {
+  items: InstalledPackage[];
+  total: number;
+  searchAfter?: SortResults;
+}
+
+export interface GetEpmDataStreamsResponse {
+  items: Array<{
+    name: string;
+  }>;
+}
 export interface GetLimitedPackagesResponse {
   items: string[];
   // deprecated in 8.0

x-pack/plugins/fleet/server/routes/epm/handlers.ts

@@ -30,10 +30,14 @@ import type {
   GetStatsResponse,
   UpdatePackageResponse,
   GetVerificationKeyIdResponse,
+  GetInstalledPackagesResponse,
+  GetEpmDataStreamsResponse,
 } from '../../../common/types';
 import type {
   GetCategoriesRequestSchema,
   GetPackagesRequestSchema,
+  GetInstalledPackagesRequestSchema,
+  GetDataStreamsRequestSchema,
   GetFileRequestSchema,
   GetInfoRequestSchema,
   InstallPackageFromRegistryRequestSchema,
@@ -49,6 +53,7 @@ import {
   bulkInstallPackages,
   getCategories,
   getPackages,
+  getInstalledPackages,
   getFile,
   getPackageInfo,
   isBulkInstallError,
@@ -66,6 +71,7 @@ import { getPackageUsageStats } from '../../services/epm/packages/get';
 import { updatePackage } from '../../services/epm/packages/update';
 import { getGpgKeyIdOrUndefined } from '../../services/epm/packages/package_verification';
 import type { ReauthorizeTransformRequestSchema } from '../../types';
+import { getDataStreams } from '../../services/epm/data_streams';
 
 const CACHE_CONTROL_10_MINUTES_HEADER: HttpResponseOptions['headers'] = {
   'cache-control': 'max-age=600',
@@ -115,6 +121,52 @@ export const getListHandler: FleetRequestHandler<
   }
 };
 
+export const getInstalledListHandler: FleetRequestHandler<
+  undefined,
+  TypeOf<typeof GetInstalledPackagesRequestSchema.query>
+> = async (context, request, response) => {
+  try {
+    const savedObjectsClient = (await context.fleet).internalSoClient;
+    const res = await getInstalledPackages({
+      savedObjectsClient,
+      ...request.query,
+    });
+
+    const body: GetInstalledPackagesResponse = { ...res };
+
+    return response.ok({
+      body,
+    });
+  } catch (error) {
+    return defaultFleetErrorHandler({ error, response });
+  }
+};
+
+export const getDataStreamsHandler: FleetRequestHandler<
+  undefined,
+  TypeOf<typeof GetDataStreamsRequestSchema.query>
+> = async (context, request, response) => {
+  try {
+    const coreContext = await context.core;
+    // Query datastreams as the current user as the Kibana internal user may not have all the required permissions
+    const esClient = coreContext.elasticsearch.client.asCurrentUser;
+    const res = await getDataStreams({
+      esClient,
+      ...request.query,
+    });
+
+    const body: GetEpmDataStreamsResponse = {
+      ...res,
+    };
+
+    return response.ok({
+      body,
+    });
+  } catch (error) {
+    return defaultFleetErrorHandler({ error, response });
+  }
+};
+
 export const getLimitedListHandler: FleetRequestHandler<
   undefined,
   TypeOf<typeof GetLimitedPackagesRequestSchema.query>,

x-pack/plugins/fleet/server/routes/epm/index.ts

@@ -27,6 +27,7 @@ import { splitPkgKey } from '../../services/epm/registry';
 import {
   GetCategoriesRequestSchema,
   GetPackagesRequestSchema,
+  GetInstalledPackagesRequestSchema,
   GetFileRequestSchema,
   GetInfoRequestSchema,
   GetInfoRequestSchemaDeprecated,
@@ -40,11 +41,13 @@ import {
   UpdatePackageRequestSchema,
   UpdatePackageRequestSchemaDeprecated,
   ReauthorizeTransformRequestSchema,
+  GetDataStreamsRequestSchema,
 } from '../../types';
 
 import {
   getCategoriesHandler,
   getListHandler,
+  getInstalledListHandler,
   getLimitedListHandler,
   getFileHandler,
   getInfoHandler,
@@ -56,6 +59,7 @@ import {
   updatePackageHandler,
   getVerificationKeyIdHandler,
   reauthorizeTransformsHandler,
+  getDataStreamsHandler,
 } from './handlers';
 
 const MAX_FILE_SIZE_BYTES = 104857600; // 100MB
@@ -83,6 +87,17 @@ export const registerRoutes = (router: FleetAuthzRouter) => {
     getListHandler
   );
 
+  router.get(
+    {
+      path: EPM_API_ROUTES.INSTALLED_LIST_PATTERN,
+      validate: GetInstalledPackagesRequestSchema,
+      fleetAuthz: {
+        integrations: { readPackageInfo: true },
+      },
+    },
+    getInstalledListHandler
+  );
+
   router.get(
     {
       path: EPM_API_ROUTES.LIMITED_LIST_PATTERN,
@@ -201,6 +216,17 @@ export const registerRoutes = (router: FleetAuthzRouter) => {
     getVerificationKeyIdHandler
   );
 
+  router.get(
+    {
+      path: EPM_API_ROUTES.DATA_STREAMS_PATTERN,
+      validate: GetDataStreamsRequestSchema,
+      fleetAuthz: {
+        integrations: { readPackageInfo: true },
+      },
+    },
+    getDataStreamsHandler
+  );
+
   // deprecated since 8.0
   router.get(
     {