[Logs+] Implement Data Source APIs

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;
+  pageAfter?: 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 coreContext = await context.core;
+    const savedObjectsClient = coreContext.savedObjects.client;
+    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;
+    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,14 @@ export const registerRoutes = (router: FleetAuthzRouter) => {
     getListHandler
   );
 
+  router.get(
+    {
+      path: EPM_API_ROUTES.INSTALLED_LIST_PATTERN,
+      validate: GetInstalledPackagesRequestSchema,
+    },
+    getInstalledListHandler
+  );
+
   router.get(
     {
       path: EPM_API_ROUTES.LIMITED_LIST_PATTERN,
@@ -201,6 +213,14 @@ export const registerRoutes = (router: FleetAuthzRouter) => {
     getVerificationKeyIdHandler
   );
 
+  router.get(
+    {
+      path: EPM_API_ROUTES.DATA_STREAMS_PATTERN,
+      validate: GetDataStreamsRequestSchema,
+    },
+    getDataStreamsHandler
+  );
+
   // deprecated since 8.0
   router.get(
     {

x-pack/plugins/fleet/server/services/epm/data_streams/get.ts

@@ -0,0 +1,47 @@
+/*
+ * 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 type { ElasticsearchClient } from '@kbn/core/server';
+
+import type { PackageDataStreamTypes } from '../../../../common/types';
+
+import { dataStreamService } from '../../data_streams';
+
+export async function getDataStreams(options: {
+  esClient: ElasticsearchClient;
+  type?: PackageDataStreamTypes;
+  datasetQuery?: string;
+  sortDirection: 'asc' | 'desc';
+  uncategorisedOnly: boolean;
+}) {
+  const { esClient, type, datasetQuery, uncategorisedOnly, sortDirection } = options;
+
+  const allDataStreams = await dataStreamService.getMatchingDataStreams(esClient, {
+    type: type ? type : '*',
+    dataset: datasetQuery ? `*${datasetQuery}*` : '*',
+  });
+
+  const filteredDataStreams = uncategorisedOnly
+    ? allDataStreams.filter((stream) => {
+        return !stream._meta || !stream._meta.managed_by || stream._meta.managed_by !== 'fleet';
+      })
+    : allDataStreams;
+
+  const mappedDataStreams = filteredDataStreams.map((dataStream) => {
+    return { name: dataStream.name };
+  });
+
+  const sortedDataStreams = mappedDataStreams.sort((a, b) => {
+    return a.name.localeCompare(b.name);
+  });
+
+  const dataStreams = sortDirection === 'asc' ? sortedDataStreams : sortedDataStreams.reverse();
+
+  return {
+    items: dataStreams,
+  };
+}

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

@@ -0,0 +1,8 @@
+/*
+ * 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.
+ */
+
+export { getDataStreams } from './get';

x-pack/plugins/fleet/server/services/epm/packages/get.ts

@@ -10,6 +10,13 @@ import semverGte from 'semver/functions/gte';
 import type { Logger } from '@kbn/core/server';
 import { withSpan } from '@kbn/apm-utils';
 
+import type { SortResults } from '@elastic/elasticsearch/lib/api/types';
+
+import { nodeBuilder } from '@kbn/es-query';
+
+import { buildNode as buildFunctionNode } from '@kbn/es-query/src/kuery/node_types/function';
+import { buildNode as buildWildcardNode } from '@kbn/es-query/src/kuery/node_types/wildcard';
+
 import {
   installationStatuses,
   PACKAGE_POLICY_SAVED_OBJECT_TYPE,
@@ -20,6 +27,7 @@ import type {
   PackageUsageStats,
   PackagePolicySOAttributes,
   Installable,
+  PackageDataStreamTypes,
 } from '../../../../common/types';
 import { PACKAGES_SAVED_OBJECT_TYPE } from '../../../constants';
 import type {
@@ -142,6 +150,90 @@ export async function getPackages(
   return packageListWithoutStatus;
 }
 
+export async function getInstalledPackages(options: {
+  savedObjectsClient: SavedObjectsClientContract;
+  type?: PackageDataStreamTypes;
+  nameQuery?: string;
+  pageAfter?: SortResults;
+  pageSize: number;
+  sortDirection: 'asc' | 'desc';
+}) {
+  const { savedObjectsClient, pageAfter, pageSize, nameQuery, sortDirection, type } = options;
+
+  const packageSavedObjects = await savedObjectsClient.find<Installation>({
+    type: PACKAGES_SAVED_OBJECT_TYPE,
+    // Pagination
+    perPage: pageSize,
+    ...(pageAfter && { searchAfter: pageAfter }),
+    // Sort
+    sortField: 'name',
+    sortOrder: sortDirection,
+    // Name filter
+    ...(nameQuery && { searchFields: ['name'] }),
+    ...(nameQuery && { search: `${nameQuery}* | ${nameQuery}` }),
+    filter: nodeBuilder.and([
+      // Filter to installed packages only
+      nodeBuilder.is(
+        `${PACKAGES_SAVED_OBJECT_TYPE}.attributes.install_status`,
+        installationStatuses.Installed
+      ),
+      // Filter for a "queryable" marker
+      buildFunctionNode(
+        'nested',
+        `${PACKAGES_SAVED_OBJECT_TYPE}.attributes.installed_es`,
+        nodeBuilder.is('type', 'index_template')
+      ),
+      // "Type" filter
+      ...(type
+        ? [
+            buildFunctionNode(
+              'nested',
+              `${PACKAGES_SAVED_OBJECT_TYPE}.attributes.installed_es`,
+              nodeBuilder.is('id', buildWildcardNode(`${type}-*`))
+            ),
+          ]
+        : []),
+    ]),
+  });
+
+  const integrations = packageSavedObjects.saved_objects.map((integrationSavedObject) => {
+    const {
+      name,
+      version,
+      install_status: installStatus,
+      es_index_patterns: esIndexPatterns,
+    } = integrationSavedObject.attributes;
+
+    const dataStreams = Object.entries(esIndexPatterns)
+      .map(([key, value]) => {
+        return {
+          name: key,
+          title: value,
+        };
+      })
+      .filter((stream) => {
+        if (!type) {
+          return true;
+        } else {
+          return stream.title.startsWith(`${type}-`);
+        }
+      });
+
+    return {
+      name,
+      version,
+      status: installStatus,
+      dataStreams,
+    };
+  });
+
+  return {
+    items: integrations,
+    total: packageSavedObjects.total,
+    pageAfter: packageSavedObjects.saved_objects.at(-1)?.sort, // Enable ability to use searchAfter in subsequent queries
+  };
+}
+
 // Get package names for packages which cannot have more than one package policy on an agent policy
 export async function getLimitedPackages(options: {
   savedObjectsClient: SavedObjectsClientContract;

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

@@ -22,6 +22,7 @@ export {
   getInstallations,
   getPackageInfo,
   getPackages,
+  getInstalledPackages,
   getLimitedPackages,
 } from './get';