[Fleet] Add showOnlyActiveDatastream to /packages/installed endpoint (#192631)

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

@@ -142,9 +142,12 @@ export const getInstalledListHandler: FleetRequestHandler<
   TypeOf<typeof GetInstalledPackagesRequestSchema.query>
 > = async (context, request, response) => {
   try {
-    const savedObjectsClient = (await context.fleet).internalSoClient;
+    const [fleetContext, coreContext] = await Promise.all([context.fleet, context.core]);
+    const savedObjectsClient = fleetContext.internalSoClient;
+    const esClient = coreContext.elasticsearch.client.asCurrentUser;
     const res = await getInstalledPackages({
       savedObjectsClient,
+      esClient,
       ...request.query,
     });
 

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

@@ -8,7 +8,7 @@
 import type { SavedObjectsClientContract, SavedObjectsFindResult } from '@kbn/core/server';
 
 import { SavedObjectsErrorHelpers } from '@kbn/core/server';
-import { savedObjectsClientMock } from '@kbn/core/server/mocks';
+import { savedObjectsClientMock, elasticsearchServiceMock } from '@kbn/core/server/mocks';
 
 import {
   ASSETS_SAVED_OBJECT_TYPE,
@@ -17,7 +17,7 @@ import {
 } from '../../../../common';
 import type { RegistryPackage } from '../../../../common/types';
 import type { PackagePolicySOAttributes } from '../../../types';
-
+import { dataStreamService } from '../../data_streams';
 import { createAppContextStartContractMock } from '../../../mocks';
 import { appContextService } from '../../app_context';
 import { PackageNotFoundError } from '../../../errors';
@@ -32,6 +32,7 @@ import { getInstalledPackages, getPackageInfo, getPackages, getPackageUsageStats
 jest.mock('../registry');
 jest.mock('../../settings');
 jest.mock('../../audit_logging');
+jest.mock('../../data_streams');
 
 const MockRegistry = jest.mocked(Registry);
 
@@ -643,6 +644,7 @@ owner: elastic`,
       });
 
       await getInstalledPackages({
+        esClient: elasticsearchServiceMock.createInternalClient(),
         savedObjectsClient: soClient,
         dataStreamType: 'logs',
         nameQuery: 'nginx',
@@ -785,6 +787,7 @@ owner: elastic`,
       });
 
       const results = await getInstalledPackages({
+        esClient: elasticsearchServiceMock.createInternalClient(),
         savedObjectsClient: soClient,
         dataStreamType: 'logs',
         nameQuery: 'nginx',
@@ -816,6 +819,84 @@ owner: elastic`,
         total: 5,
       });
     });
+    it('filter non active datastreams if flag is true', async () => {
+      const soClient = savedObjectsClientMock.create();
+
+      jest.mocked(dataStreamService.getAllFleetDataStreams).mockResolvedValue([
+        {
+          name: `logs-elastic_agent.apm_server-production`,
+        },
+        {
+          name: `metrics-elastic_agent.apm_server-production`,
+        },
+      ] as any);
+
+      soClient.find.mockImplementation(async (options) => {
+        if (options.type === PACKAGES_SAVED_OBJECT_TYPE) {
+          return {
+            total: 5,
+            saved_objects: [
+              {
+                type: 'epm-packages',
+                id: 'elastic_agent',
+                attributes: {
+                  es_index_patterns: {
+                    fleet_server_logs: 'logs-elastic_agent.fleet_server-*',
+                    apm_server_logs: 'logs-elastic_agent.apm_server-*',
+                    apm_server_metrics: 'metrics-elastic_agent.apm_server-*',
+                  },
+                  name: 'elastic_agent',
+                  version: '1.8.0',
+                  install_status: 'installed',
+                },
+                references: [],
+                sort: ['elastic_agent'],
+              },
+            ],
+          } as any;
+        } else if (options.type === ASSETS_SAVED_OBJECT_TYPE) {
+          return {
+            total: 5,
+            saved_objects: [
+              {
+                type: 'epm-packages-assets',
+                id: '338b6f9e-e126-5f1e-abb9-afe017d4788b',
+                attributes: {
+                  package_name: 'elastic_agent',
+                  package_version: '1.8.0',
+                  install_source: 'upload',
+                  asset_path: 'elastic_agent-1.8.0/manifest.yml',
+                  media_type: 'text/yaml; charset=utf-8',
+                  data_utf8:
+                    'name: elastic_agent\ntitle: Elastic Agent\nversion: 1.8.0\ndescription: Collect logs and metrics from Elastic Agents.\ntype: integration\nformat_version: 1.0.0\nlicense: basic\ncategories: ["elastic_stack"]\nconditions:\n  kibana.version: "^8.7.1"\nowner:\n  github: elastic/elastic-agent\nicons:\n  - src: /img/logo_elastic_agent.svg\n    title: logo Elastic Agent\n    size: 64x64\n    type: image/svg+xml\nscreenshots:\n  - src: /img/elastic_agent_overview.png\n    title: Elastic Agent Overview\n    size: 2560×1234\n    type: image/png\n  - src: /img/elastic_agent_metrics.png\n    title: Elastic Agent Metrics\n    size: 2560×1234\n    type: image/png\n  - src: /img/elastic_agent_info.png\n    title: Elastic Agent Information\n    size: 2560×1234\n    type: image/png\n  - src: /img/elastic_agent_integrations.png\n    title: Elastic Agent Integrations\n    size: 2560×1234\n    type: image/png\n',
+                  data_base64: '',
+                },
+                references: [],
+              },
+            ],
+          } as any;
+        }
+      });
+
+      const results = await getInstalledPackages({
+        savedObjectsClient: soClient,
+        esClient: elasticsearchServiceMock.createInternalClient(),
+        perPage: 10,
+        sortOrder: 'asc',
+        showOnlyActiveDataStreams: true,
+      });
+
+      expect(results.items[0].dataStreams).toEqual([
+        {
+          name: 'logs-elastic_agent.apm_server-*',
+          title: 'apm_server_logs',
+        },
+        {
+          name: 'metrics-elastic_agent.apm_server-*',
+          title: 'apm_server_metrics',
+        },
+      ]);
+    });
   });
 
   describe('getPackageInfo', () => {

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

@@ -7,12 +7,17 @@
 
 import { safeLoad } from 'js-yaml';
 import pMap from 'p-map';
-import type { SavedObjectsClientContract, SavedObjectsFindOptions } from '@kbn/core/server';
+import minimatch from 'minimatch';
+import type {
+  ElasticsearchClient,
+  SavedObjectsClientContract,
+  SavedObjectsFindOptions,
+} from '@kbn/core/server';
 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 type { IndicesDataStream, SortResults } from '@elastic/elasticsearch/lib/api/types';
 
 import { nodeBuilder } from '@kbn/es-query';
 
@@ -50,6 +55,7 @@ import {
   PackageInvalidArchiveError,
 } from '../../../errors';
 import { appContextService } from '../..';
+import { dataStreamService } from '../../data_streams';
 import * as Registry from '../registry';
 import type { PackageAsset } from '../archive/storage';
 import { getEsPackage } from '../archive/storage';
@@ -180,20 +186,22 @@ export async function getPackages(
 
 interface GetInstalledPackagesOptions {
   savedObjectsClient: SavedObjectsClientContract;
+  esClient: ElasticsearchClient;
   dataStreamType?: PackageDataStreamTypes;
   nameQuery?: string;
   searchAfter?: SortResults;
   perPage: number;
   sortOrder: 'asc' | 'desc';
+  showOnlyActiveDataStreams?: boolean;
 }
 export async function getInstalledPackages(options: GetInstalledPackagesOptions) {
-  const { savedObjectsClient, ...otherOptions } = options;
+  const { savedObjectsClient, esClient, showOnlyActiveDataStreams, ...otherOptions } = options;
   const { dataStreamType } = otherOptions;
 
-  const packageSavedObjects = await getInstalledPackageSavedObjects(
-    savedObjectsClient,
-    otherOptions
-  );
+  const [packageSavedObjects, allFleetDataStreams] = await Promise.all([
+    getInstalledPackageSavedObjects(savedObjectsClient, otherOptions),
+    showOnlyActiveDataStreams ? dataStreamService.getAllFleetDataStreams(esClient) : undefined,
+  ]);
 
   const integrations = packageSavedObjects.saved_objects.map((integrationSavedObject) => {
     const {
@@ -203,7 +211,11 @@ export async function getInstalledPackages(options: GetInstalledPackagesOptions)
       es_index_patterns: esIndexPatterns,
     } = integrationSavedObject.attributes;
 
-    const dataStreams = getInstalledPackageSavedObjectDataStreams(esIndexPatterns, dataStreamType);
+    const dataStreams = getInstalledPackageSavedObjectDataStreams(
+      esIndexPatterns,
+      dataStreamType,
+      allFleetDataStreams
+    );
 
     return {
       name,
@@ -296,7 +308,7 @@ export async function getPackageSavedObjects(
 
 async function getInstalledPackageSavedObjects(
   savedObjectsClient: SavedObjectsClientContract,
-  options: Omit<GetInstalledPackagesOptions, 'savedObjectsClient'>
+  options: Omit<GetInstalledPackagesOptions, 'savedObjectsClient' | 'esClient'>
 ) {
   const { searchAfter, sortOrder, perPage, nameQuery, dataStreamType } = options;
 
@@ -385,8 +397,13 @@ export async function getInstalledPackageManifests(
 
 function getInstalledPackageSavedObjectDataStreams(
   indexPatterns: Record<string, string>,
-  dataStreamType?: string
+  dataStreamType?: string,
+  filterActiveDatastreams?: IndicesDataStream[]
 ) {
+  const filterActiveDatastreamsName = filterActiveDatastreams
+    ? filterActiveDatastreams.map((ds) => ds.name)
+    : undefined;
+
   return Object.entries(indexPatterns)
     .map(([key, value]) => {
       return {
@@ -395,11 +412,22 @@ function getInstalledPackageSavedObjectDataStreams(
       };
     })
     .filter((stream) => {
-      if (!dataStreamType) {
-        return true;
-      } else {
-        return stream.name.startsWith(`${dataStreamType}-`);
+      if (dataStreamType && !stream.name.startsWith(`${dataStreamType}-`)) {
+        return false;
       }
+
+      if (filterActiveDatastreamsName) {
+        const patternRegex = new minimatch.Minimatch(stream.name, {
+          noglobstar: true,
+          nonegate: true,
+        }).makeRe();
+
+        return filterActiveDatastreamsName.some((dataStreamName) =>
+          dataStreamName.match(patternRegex)
+        );
+      }
+
+      return true;
     });
 }
 

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

@@ -35,6 +35,7 @@ export const GetInstalledPackagesRequestSchema = {
         schema.literal('profiling'),
       ])
     ),
+    showOnlyActiveDataStreams: schema.maybe(schema.boolean()),
     nameQuery: schema.maybe(schema.string()),
     searchAfter: schema.maybe(schema.arrayOf(schema.oneOf([schema.string(), schema.number()]))),
     perPage: schema.number({ defaultValue: 15 }),

x-pack/test/fleet_api_integration/apis/epm/get.ts

@@ -18,6 +18,7 @@ export default function (providerContext: FtrProviderContext) {
   const { getService } = providerContext;
 
   const supertest = getService('supertest');
+  const es = getService('es');
   const supertestWithoutAuth = getService('supertestWithoutAuth');
   const fleetAndAgents = getService('fleetAndAgents');
 
@@ -117,12 +118,22 @@ export default function (providerContext: FtrProviderContext) {
         await installPackage('experimental', '0.1.0');
         await bundlePackage('endpoint-8.6.1');
         await installPackage('endpoint', '8.6.1');
+        await es.index({
+          index: 'logs-apache.access-default',
+          document: {
+            '@timestamp': new Date().toISOString(),
+          },
+          refresh: 'wait_for',
+        });
       });
       after(async () => {
         await uninstallPackage(testPkgName, testPkgVersion);
         await uninstallPackage('experimental', '0.1.0');
         await uninstallPackage('endpoint', '8.6.1');
         await removeBundledPackages(log);
+        await es.indices.deleteDataStream({
+          name: 'logs-apache.access-default',
+        });
       });
       it('Allows the fetching of installed packages', async () => {
         const res = await supertest.get(`/api/fleet/epm/packages/installed`).expect(200);
@@ -173,6 +184,16 @@ export default function (providerContext: FtrProviderContext) {
         expect(packages.length).to.be(1);
         expect(packages[0].name).to.be('experimental');
       });
+      it('Can be to only return active datastreams', async () => {
+        const res = await supertest
+          .get(`/api/fleet/epm/packages/installed?nameQuery=apache&showOnlyActiveDataStreams=true`)
+          .expect(200);
+        const packages = res.body.items;
+        expect(packages.length).to.be(1);
+        expect(packages[0].name).to.be('apache');
+        expect(packages[0].dataStreams.length).to.be(1);
+        expect(packages[0].dataStreams[0].name).to.be('logs-apache.access-*');
+      });
     });
     it('returns a 404 for a package that do not exists', async function () {
       await supertest.get('/api/fleet/epm/packages/notexists/99.99.99').expect(404);