[Asset Manager] services endpoint (#164181)

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

Implements `/assets/services` endpoint that returns service assets found
in the configured source (signals or assets indices). Consumer can
provide a `parent` query to filter the returned services. While the
_assets_ mode supports any kind of parent/depth thanks to its common
interface, the _signals_ mode only supports host parent for the moment.

1. pull this branch and point it at an oblt-cli created cluster that
uses cross-cluster search to read from the edge cluster
2. add the following[1] to your kibana.yml file
3. hit
`/api/asset-manager/assets/services?from=<from>&to=<to>&(parent=<host>)?`.
services should be returned. Add/remove parent query string to filter
services only running on specific host.

[1]
```
xpack.assetManager:
  alphaEnabled: true
  sourceIndices:
    metrics: remote_cluster:metricbeat*,remote_cluster:metrics-*
    logs: remote_cluster:filebeat*,remote_cluster:logs-*
    traces: remote_cluster:traces-*
    serviceMetrics: remote_cluster:metrics-apm*
    serviceLogs: remote_cluster:logs-apm*
  lockedSource: signals
```

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

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

@@ -8,8 +8,8 @@
 import { AccessorOptions, OptionsWithInjectedValues } from '..';
 
 export interface GetHostsOptions extends AccessorOptions {
-  from: number;
+  from: string;
-  to: number;
+  to: string;
 }
 export type GetHostsOptionsInjected = OptionsWithInjectedValues<GetHostsOptions>;
 

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

@@ -0,0 +1,46 @@
+/*
+ * 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 { Asset } from '../../../../common/types_api';
+import { GetServicesOptionsInjected } from '.';
+import { getAssets } from '../../get_assets';
+import { getAllRelatedAssets } from '../../get_all_related_assets';
+
+export async function getServicesByAssets(
+  options: GetServicesOptionsInjected
+): Promise<{ services: Asset[] }> {
+  if (options.parent) {
+    return getServicesByParent(options);
+  }
+
+  const services = await getAssets({
+    esClient: options.esClient,
+    filters: {
+      kind: 'service',
+      from: options.from,
+      to: options.to,
+    },
+  });
+
+  return { services };
+}
+
+async function getServicesByParent(
+  options: GetServicesOptionsInjected
+): Promise<{ services: Asset[] }> {
+  const { descendants } = await getAllRelatedAssets(options.esClient, {
+    from: options.from,
+    to: options.to,
+    maxDistance: 5,
+    kind: ['service'],
+    size: 100,
+    relation: 'descendants',
+    ean: options.parent!,
+  });
+
+  return { services: descendants as Asset[] };
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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 { Asset } from '../../../../common/types_api';
+import { GetServicesOptionsInjected } from '.';
+import { collectServices } from '../../collectors/services';
+
+export async function getServicesBySignals(
+  options: GetServicesOptionsInjected
+): Promise<{ services: Asset[] }> {
+  const filters = [];
+
+  if (options.parent) {
+    filters.push({
+      bool: {
+        should: [
+          { term: { 'host.name': options.parent } },
+          { term: { 'host.hostname': options.parent } },
+        ],
+        minimum_should_match: 1,
+      },
+    });
+  }
+
+  const { assets } = await collectServices({
+    client: options.esClient,
+    from: options.from,
+    to: options.to,
+    sourceIndices: options.sourceIndices,
+    filters,
+  });
+
+  return { services: assets };
+}

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

@@ -0,0 +1,22 @@
+/*
+ * 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 { AccessorOptions, OptionsWithInjectedValues } from '..';
+
+export interface GetServicesOptions extends AccessorOptions {
+  from: string;
+  to: string;
+  parent?: string;
+}
+export type GetServicesOptionsInjected = OptionsWithInjectedValues<GetServicesOptions>;
+
+export interface ServiceIdentifier {
+  'asset.ean': string;
+  'asset.id': string;
+  'asset.name'?: string;
+  'service.environment'?: string;
+}

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

@@ -9,8 +9,11 @@ import { Asset } from '../../common/types_api';
 import { AssetManagerConfig } from '../types';
 import { OptionsWithInjectedValues } from './accessors';
 import { GetHostsOptions } from './accessors/hosts';
+import { GetServicesOptions } from './accessors/services';
 import { getHostsByAssets } from './accessors/hosts/get_hosts_by_assets';
 import { getHostsBySignals } from './accessors/hosts/get_hosts_by_signals';
+import { getServicesByAssets } from './accessors/services/get_services_by_assets';
+import { getServicesBySignals } from './accessors/services/get_services_by_signals';
 
 interface AssetAccessorClassOptions {
   sourceIndices: AssetManagerConfig['sourceIndices'];
@@ -35,4 +38,13 @@ export class AssetAccessor {
       return await getHostsBySignals(withInjected);
     }
   }
+
+  async getServices(options: GetServicesOptions): Promise<{ services: Asset[] }> {
+    const withInjected = this.injectOptions(options);
+    if (this.options.source === 'assets') {
+      return await getServicesByAssets(withInjected);
+    } else {
+      return await getServicesBySignals(withInjected);
+    }
+  }
 }

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

@@ -16,10 +16,11 @@ export type Collector = (opts: CollectorOptions) => Promise<CollectorResult>;
 
 export interface CollectorOptions {
   client: ElasticsearchClient;
-  from: number;
+  from: string;
-  to: number;
+  to: string;
   sourceIndices: AssetManagerConfig['sourceIndices'];
   afterKey?: estypes.SortResults;
+  filters?: estypes.QueryDslQueryContainer[];
 }
 
 export interface CollectorResult {

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

@@ -15,8 +15,18 @@ export async function collectServices({
   to,
   sourceIndices,
   afterKey,
+  filters = [],
 }: CollectorOptions) {
   const { traces, serviceMetrics, serviceLogs } = sourceIndices;
+  const musts: estypes.QueryDslQueryContainer[] = [
+    ...filters,
+    {
+      exists: {
+        field: 'service.name',
+      },
+    },
+  ];
+
   const dsl: estypes.SearchRequest = {
     index: [traces, serviceMetrics, serviceLogs],
     size: 0,
@@ -33,13 +43,7 @@ export async function collectServices({
             },
           },
         ],
-        must: [
+        must: musts,
-          {
-            exists: {
-              field: 'service.name',
-            },
-          },
-        ],
       },
     },
     aggs: {
@@ -58,6 +62,7 @@ export async function collectServices({
               serviceEnvironment: {
                 terms: {
                   field: 'service.environment',
+                  missing_bucket: true,
                 },
               },
             },
@@ -112,14 +117,14 @@ export async function collectServices({
     }
 
     containerHosts.buckets?.forEach((containerBucket: any) => {
-      const [containerId, hostname] = containerBucket.key;
+      const [hostname, containerId] = containerBucket.key;
-      if (containerId) {
-        (service['asset.parents'] as string[]).push(`container:${containerId}`);
-      }
-
       if (hostname) {
         (service['asset.references'] as string[]).push(`host:${hostname}`);
       }
+
+      if (containerId) {
+        (service['asset.parents'] as string[]).push(`container:${containerId}`);
+      }
     });
 
     acc.push(service);

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

@@ -50,8 +50,8 @@ export function hostsRoutes<T extends RequestHandlerContext>({
 
       try {
         const response = await assetAccessor.getHosts({
-          from: datemath.parse(from)!.valueOf(),
+          from: datemath.parse(from)!.toISOString(),
-          to: datemath.parse(to)!.valueOf(),
+          to: datemath.parse(to)!.toISOString(),
           esClient,
         });
 

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

@@ -0,0 +1,70 @@
+/*
+ * 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 * as rt from 'io-ts';
+import datemath from '@kbn/datemath';
+import {
+  dateRt,
+  inRangeFromStringRt,
+  datemathStringRt,
+  createRouteValidationFunction,
+  createLiteralValueFromUndefinedRT,
+} from '@kbn/io-ts-utils';
+import { RequestHandlerContext } from '@kbn/core-http-request-handler-context-server';
+import { debug } from '../../../common/debug_log';
+import { SetupRouteOptions } from '../types';
+import { ASSET_MANAGER_API_BASE } from '../../constants';
+import { getEsClientFromContext } from '../utils';
+
+const sizeRT = rt.union([inRangeFromStringRt(1, 100), createLiteralValueFromUndefinedRT(10)]);
+const assetDateRT = rt.union([dateRt, datemathStringRt]);
+const getServiceAssetsQueryOptionsRT = rt.exact(
+  rt.partial({
+    from: assetDateRT,
+    to: assetDateRT,
+    size: sizeRT,
+    parent: rt.string,
+  })
+);
+
+export type GetServiceAssetsQueryOptions = rt.TypeOf<typeof getServiceAssetsQueryOptionsRT>;
+
+export function servicesRoutes<T extends RequestHandlerContext>({
+  router,
+  assetAccessor,
+}: SetupRouteOptions<T>) {
+  // GET /assets/services
+  router.get<unknown, GetServiceAssetsQueryOptions, unknown>(
+    {
+      path: `${ASSET_MANAGER_API_BASE}/assets/services`,
+      validate: {
+        query: createRouteValidationFunction(getServiceAssetsQueryOptionsRT),
+      },
+    },
+    async (context, req, res) => {
+      const { from = 'now-24h', to = 'now', parent } = req.query || {};
+      const esClient = await getEsClientFromContext(context);
+
+      try {
+        const response = await assetAccessor.getServices({
+          from: datemath.parse(from)!.toISOString(),
+          to: datemath.parse(to)!.toISOString(),
+          parent,
+          esClient,
+        });
+
+        return res.ok({ body: response });
+      } catch (error: unknown) {
+        debug('Error while looking up SERVICE asset records', error);
+        return res.customError({
+          statusCode: 500,
+          body: { message: 'Error while looking up service asset records - ' + `${error}` },
+        });
+      }
+    }
+  );
+}

x-pack/plugins/asset_manager/server/routes/index.ts

@@ -11,6 +11,7 @@ import { pingRoute } from './ping';
 import { assetsRoutes } from './assets';
 import { sampleAssetsRoutes } from './sample_assets';
 import { hostsRoutes } from './assets/hosts';
+import { servicesRoutes } from './assets/services';
 
 export function setupRoutes<T extends RequestHandlerContext>({
   router,
@@ -20,4 +21,5 @@ export function setupRoutes<T extends RequestHandlerContext>({
   assetsRoutes<T>({ router, assetAccessor });
   sampleAssetsRoutes<T>({ router, assetAccessor });
   hostsRoutes<T>({ router, assetAccessor });
+  servicesRoutes<T>({ router, assetAccessor });
 }

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

@@ -0,0 +1,107 @@
+/*
+ * 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 { omit } from 'lodash';
+import { apm, timerange } from '@kbn/apm-synthtrace-client';
+import expect from '@kbn/expect';
+import { ASSETS_ENDPOINT } from '../constants';
+import { FtrProviderContext } from '../../types';
+
+const SERVICES_ASSETS_ENDPOINT = `${ASSETS_ENDPOINT}/services`;
+
+export default function ({ getService }: FtrProviderContext) {
+  const supertest = getService('supertest');
+  const synthtrace = getService('apmSynthtraceEsClient');
+
+  describe('asset management', () => {
+    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 }));
+
+        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);
+      });
+
+      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);
+
+        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',
+        });
+      });
+    });
+  });
+}
+
+function generateServicesData({
+  from,
+  to,
+  count = 1,
+}: {
+  from: string;
+  to: string;
+  count: number;
+}) {
+  const range = timerange(from, to);
+
+  const services = Array(count)
+    .fill(0)
+    .map((_, idx) =>
+      apm
+        .service({
+          name: `service-${idx}`,
+          environment: 'production',
+          agentName: 'nodejs',
+        })
+        .instance(`my-host-${idx}`)
+    );
+
+  return range
+    .interval('1m')
+    .rate(1)
+    .generator((timestamp, index) =>
+      services.map((service) =>
+        service
+          .transaction({ transactionName: 'GET /foo' })
+          .timestamp(timestamp)
+          .duration(500)
+          .success()
+      )
+    );
+}