[asset manager] use metrics data plugin (#166756)

## Summary

Closes https://github.com/elastic/kibana/issues/166636

- use metrics_data_access plugin to dynamically retrieve indices where
needed
- add basic `/assets/hosts` api test

---------

Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>
Co-authored-by: Jason Rhodes <jason.rhodes@elastic.co>

x-pack/plugins/asset_manager/kibana.jsonc

@@ -12,7 +12,8 @@
     "optionalPlugins": [
     ],
     "requiredPlugins": [
-      "apmDataAccess"
+      "apmDataAccess",
+      "metricsDataAccess"
     ],
     "browser": false,
     "server": true,

x-pack/plugins/asset_manager/server/lib/accessors/hosts/get_hosts_by_signals.ts

@@ -12,11 +12,18 @@ import { collectHosts } from '../../collectors/hosts';
 export async function getHostsBySignals(
   options: GetHostsOptionsInjected
 ): Promise<{ hosts: Asset[] }> {
+  const metricsIndices = await options.metricsClient.getMetricIndices({
+    savedObjectsClient: options.soClient,
+  });
+
   const { assets } = await collectHosts({
     client: options.esClient,
     from: options.from,
     to: options.to,
-    sourceIndices: options.sourceIndices,
+    sourceIndices: {
+      metrics: metricsIndices,
+      logs: options.sourceIndices.logs,
+    },
   });
   return {
     hosts: assets,

x-pack/plugins/asset_manager/server/lib/accessors/index.ts

@@ -7,12 +7,14 @@
 
 import { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
 import { APMDataAccessConfig } from '@kbn/apm-data-access-plugin/server';
+import { MetricsDataClient } from '@kbn/metrics-data-access-plugin/server';
 import { SavedObjectsClientContract } from '@kbn/core/server';
 import { AssetManagerConfig } from '../../types';
 
 export interface InjectedValues {
   sourceIndices: AssetManagerConfig['sourceIndices'];
   getApmIndices: (soClient: SavedObjectsClientContract) => Promise<APMDataAccessConfig['indices']>;
+  metricsClient: MetricsDataClient;
 }
 
 export type OptionsWithInjectedValues<T extends object> = T & InjectedValues;

x-pack/plugins/asset_manager/server/lib/accessors/services/get_services_by_signals.ts

@@ -31,7 +31,9 @@ export async function getServicesBySignals(
     client: options.esClient,
     from: options.from,
     to: options.to,
-    apmIndices,
+    sourceIndices: {
+      apm: apmIndices,
+    },
     filters,
   });
 

x-pack/plugins/asset_manager/server/lib/asset_accessor.ts

@@ -6,6 +6,7 @@
  */
 
 import { APMDataAccessConfig } from '@kbn/apm-data-access-plugin/server';
+import { MetricsDataClient } from '@kbn/metrics-data-access-plugin/server';
 import { SavedObjectsClientContract } from '@kbn/core/server';
 import { Asset } from '../../common/types_api';
 import { AssetManagerConfig } from '../types';
@@ -21,6 +22,7 @@ interface AssetAccessorClassOptions {
   sourceIndices: AssetManagerConfig['sourceIndices'];
   source: AssetManagerConfig['lockedSource'];
   getApmIndices: (soClient: SavedObjectsClientContract) => Promise<APMDataAccessConfig['indices']>;
+  metricsClient: MetricsDataClient;
 }
 
 export class AssetAccessor {
@@ -31,6 +33,7 @@ export class AssetAccessor {
       ...options,
       sourceIndices: this.options.sourceIndices,
       getApmIndices: this.options.getApmIndices,
+      metricsClient: this.options.metricsClient,
     };
   }
 

x-pack/plugins/asset_manager/server/lib/collectors/containers.ts

@@ -16,9 +16,13 @@ export async function collectContainers({
   sourceIndices,
   afterKey,
 }: CollectorOptions) {
+  if (!sourceIndices?.metrics || !sourceIndices?.logs) {
+    throw new Error('missing required metrics/logs indices');
+  }
+
   const { metrics, logs } = sourceIndices;
   const dsl: estypes.SearchRequest = {
-    index: [logs, metrics],
+    index: [metrics, logs],
     size: QUERY_MAX_SIZE,
     collapse: {
       field: 'container.id',

x-pack/plugins/asset_manager/server/lib/collectors/hosts.ts

@@ -16,6 +16,10 @@ export async function collectHosts({
   sourceIndices,
   afterKey,
 }: CollectorOptions) {
+  if (!sourceIndices?.metrics || !sourceIndices?.logs) {
+    throw new Error('missing required metrics/logs indices');
+  }
+
   const { metrics, logs } = sourceIndices;
   const dsl: estypes.SearchRequest = {
     index: [metrics, logs],

x-pack/plugins/asset_manager/server/lib/collectors/index.ts

@@ -8,7 +8,6 @@
 import { estypes } from '@elastic/elasticsearch';
 import type { APMIndices } from '@kbn/apm-data-access-plugin/server';
 import { ElasticsearchClient } from '@kbn/core/server';
-import { AssetManagerConfig } from '../../types';
 import { Asset } from '../../../common/types_api';
 
 export const QUERY_MAX_SIZE = 10000;
@@ -19,17 +18,15 @@ export interface CollectorOptions {
   client: ElasticsearchClient;
   from: string;
   to: string;
-  sourceIndices: AssetManagerConfig['sourceIndices'];
+  sourceIndices?: {
+    apm?: APMIndices;
+    metrics?: string;
+    logs?: string;
+  };
   afterKey?: estypes.SortResults;
   filters?: estypes.QueryDslQueryContainer[];
 }
 
-type OmitSourceIndices<T> = Omit<T, 'sourceIndices'>;
-
-export type ServicesCollectorOptions = OmitSourceIndices<CollectorOptions> & {
-  apmIndices: APMIndices;
-};
-
 export interface CollectorResult {
   assets: Asset[];
   afterKey?: estypes.SortResults;

x-pack/plugins/asset_manager/server/lib/collectors/pods.ts

@@ -10,6 +10,10 @@ import { Asset } from '../../../common/types_api';
 import { CollectorOptions, QUERY_MAX_SIZE } from '.';
 
 export async function collectPods({ client, from, to, sourceIndices, afterKey }: CollectorOptions) {
+  if (!sourceIndices?.metrics || !sourceIndices?.logs) {
+    throw new Error('missing required metrics/logs indices');
+  }
+
   const { metrics, logs } = sourceIndices;
   const dsl: estypes.SearchRequest = {
     index: [metrics, logs],

x-pack/plugins/asset_manager/server/lib/collectors/services.ts

@@ -7,17 +7,21 @@
 
 import { estypes } from '@elastic/elasticsearch';
 import { Asset } from '../../../common/types_api';
-import { ServicesCollectorOptions, QUERY_MAX_SIZE } from '.';
+import { CollectorOptions, QUERY_MAX_SIZE } from '.';
 
 export async function collectServices({
   client,
   from,
   to,
-  apmIndices,
+  sourceIndices,
   afterKey,
   filters = [],
-}: ServicesCollectorOptions) {
-  const { transaction, error, metric } = apmIndices;
+}: CollectorOptions) {
+  if (!sourceIndices?.apm) {
+    throw new Error('missing required apm indices');
+  }
+
+  const { transaction, error, metric } = sourceIndices.apm;
   const musts: estypes.QueryDslQueryContainer[] = [
     ...filters,
     {

x-pack/plugins/asset_manager/server/plugin.ts

@@ -59,6 +59,7 @@ export class AssetManagerServerPlugin
       source: this.config.lockedSource,
       sourceIndices: this.config.sourceIndices,
       getApmIndices: plugins.apmDataAccess.getApmIndices,
+      metricsClient: plugins.metricsDataAccess.client,
     });
 
     const router = core.http.createRouter();

x-pack/plugins/asset_manager/server/routes/assets/hosts.ts

@@ -36,7 +36,6 @@ export function hostsRoutes<T extends RequestHandlerContext>({
   router,
   assetAccessor,
 }: SetupRouteOptions<T>) {
-  // GET /assets/hosts
   router.get<unknown, GetHostAssetsQueryOptions, unknown>(
     {
       path: `${ASSET_MANAGER_API_BASE}/assets/hosts`,

x-pack/plugins/asset_manager/server/types.ts

@@ -11,13 +11,13 @@ import {
   ApmDataAccessPluginSetup,
   ApmDataAccessPluginStart,
 } from '@kbn/apm-data-access-plugin/server';
+import { MetricsDataPluginSetup } from '@kbn/metrics-data-access-plugin/server';
 
 export interface ElasticsearchAccessorOptions {
   esClient: ElasticsearchClient;
 }
 
 export const INDEX_DEFAULTS = {
-  metrics: 'metricbeat-*,metrics-*',
   logs: 'filebeat-*,logs-*',
 };
 
@@ -29,7 +29,6 @@ export const configSchema = schema.object({
   // that value is propagated everywhere. For now, we duplicate the value here.
   sourceIndices: schema.object(
     {
-      metrics: schema.string({ defaultValue: INDEX_DEFAULTS.metrics }),
       logs: schema.string({ defaultValue: INDEX_DEFAULTS.logs }),
     },
     { defaultValue: INDEX_DEFAULTS }
@@ -48,6 +47,7 @@ export type AssetManagerConfig = TypeOf<typeof configSchema>;
 
 export interface AssetManagerPluginSetupDependencies {
   apmDataAccess: ApmDataAccessPluginSetup;
+  metricsDataAccess: MetricsDataPluginSetup;
 }
 export interface AssetManagerPluginStartDependencies {
   apmDataAccess: ApmDataAccessPluginStart;

x-pack/plugins/asset_manager/tsconfig.json

@@ -20,6 +20,7 @@
     "@kbn/core-elasticsearch-server",
     "@kbn/core-http-request-handler-context-server",
     "@kbn/datemath",
-    "@kbn/apm-data-access-plugin"
+    "@kbn/apm-data-access-plugin",
+    "@kbn/metrics-data-access-plugin"
   ]
 }

x-pack/test/api_integration/apis/asset_manager/tests/with_signals_source/hosts.ts

@@ -0,0 +1,54 @@
+/*
+ * 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 { timerange, infra } from '@kbn/apm-synthtrace-client';
+import expect from '@kbn/expect';
+import { ASSETS_ENDPOINT } from '../constants';
+import { FtrProviderContext } from '../../types';
+
+const HOSTS_ASSETS_ENDPOINT = `${ASSETS_ENDPOINT}/hosts`;
+
+export default function ({ getService }: FtrProviderContext) {
+  const supertest = getService('supertest');
+  const synthtrace = getService('infraSynthtraceEsClient');
+
+  describe('GET /assets/hosts', () => {
+    beforeEach(async () => {
+      await synthtrace.clean();
+    });
+
+    it('should return hosts', async () => {
+      const from = new Date(Date.now() - 1000 * 60 * 2).toISOString();
+      const to = new Date().toISOString();
+      await synthtrace.index(generateHostsData({ from, to, count: 5 }));
+
+      const response = await supertest
+        .get(HOSTS_ASSETS_ENDPOINT)
+        .query({
+          from,
+          to,
+        })
+        .expect(200);
+
+      expect(response.body).to.have.property('hosts');
+      expect(response.body.hosts.length).to.equal(5);
+    });
+  });
+}
+
+function generateHostsData({ from, to, count = 1 }: { from: string; to: string; count: number }) {
+  const range = timerange(from, to);
+
+  const hosts = Array(count)
+    .fill(0)
+    .map((_, idx) => infra.host(`my-host-${idx}`));
+
+  return range
+    .interval('1m')
+    .rate(1)
+    .generator((timestamp, index) => hosts.map((host) => host.metrics().timestamp(timestamp)));
+}

x-pack/test/api_integration/apis/asset_manager/tests/with_signals_source/index.ts

@@ -9,5 +9,7 @@ import { FtrProviderContext } from '../../../../ftr_provider_context';
 export default function ({ loadTestFile }: FtrProviderContext) {
   describe('Asset Manager API Endpoints - with signals source', () => {
     loadTestFile(require.resolve('./basics'));
+    loadTestFile(require.resolve('./hosts'));
+    loadTestFile(require.resolve('./services'));
   });
 }

x-pack/test/api_integration/apis/asset_manager/tests/with_signals_source/services.ts

@@ -17,53 +17,51 @@ export default function ({ getService }: FtrProviderContext) {
   const supertest = getService('supertest');
   const synthtrace = getService('apmSynthtraceEsClient');
 
-  describe('asset management', () => {
+  describe('GET /assets/services', () => {
     beforeEach(async () => {
       await synthtrace.clean();
     });
 
-    describe('GET /assets/services', () => {
-      it('should return services', async () => {
-        const from = new Date(Date.now() - 1000 * 60 * 2).toISOString();
-        const to = new Date().toISOString();
-        await synthtrace.index(generateServicesData({ from, to, count: 2 }));
+    it('should return services', async () => {
+      const from = new Date(Date.now() - 1000 * 60 * 2).toISOString();
+      const to = new Date().toISOString();
+      await synthtrace.index(generateServicesData({ from, to, count: 2 }));
 
-        const response = await supertest
-          .get(SERVICES_ASSETS_ENDPOINT)
-          .query({
-            from,
-            to,
-          })
-          .expect(200);
+      const response = await supertest
+        .get(SERVICES_ASSETS_ENDPOINT)
+        .query({
+          from,
+          to,
+        })
+        .expect(200);
 
-        expect(response.body).to.have.property('services');
-        expect(response.body.services.length).to.equal(2);
-      });
+      expect(response.body).to.have.property('services');
+      expect(response.body.services.length).to.equal(2);
+    });
 
-      it('should return services running on specified host', async () => {
-        const from = new Date(Date.now() - 1000 * 60 * 2).toISOString();
-        const to = new Date().toISOString();
-        await synthtrace.index(generateServicesData({ from, to, count: 5 }));
+    it('should return services running on specified host', async () => {
+      const from = new Date(Date.now() - 1000 * 60 * 2).toISOString();
+      const to = new Date().toISOString();
+      await synthtrace.index(generateServicesData({ from, to, count: 5 }));
 
-        const response = await supertest
-          .get(SERVICES_ASSETS_ENDPOINT)
-          .query({
-            from,
-            to,
-            parent: 'my-host-1',
-          })
-          .expect(200);
+      const response = await supertest
+        .get(SERVICES_ASSETS_ENDPOINT)
+        .query({
+          from,
+          to,
+          parent: 'my-host-1',
+        })
+        .expect(200);
 
-        expect(response.body).to.have.property('services');
-        expect(response.body.services.length).to.equal(1);
-        expect(omit(response.body.services[0], ['@timestamp'])).to.eql({
-          'asset.kind': 'service',
-          'asset.id': 'service-1',
-          'asset.ean': 'service:service-1',
-          'asset.references': [],
-          'asset.parents': [],
-          'service.environment': 'production',
-        });
+      expect(response.body).to.have.property('services');
+      expect(response.body.services.length).to.equal(1);
+      expect(omit(response.body.services[0], ['@timestamp'])).to.eql({
+        'asset.kind': 'service',
+        'asset.id': 'service-1',
+        'asset.ean': 'service:service-1',
+        'asset.references': [],
+        'asset.parents': [],
+        'service.environment': 'production',
       });
     });
   });