[asset manager] merge obsasset signals collection (#162222)

## Summary
Closes #161887

Merges most of the functionality from `feat/obs-asset-manager-demo`
branch. We remove implicit collection code while including:
- asset collectors from signals (also include pods and containers but we
don't use then directly)
- source configuration code (assets or signals)
- `assetAccessor` logic that determines which indices to query

The change also enables ftr test suite. We'll also merge the services
endpoint #160294 when approved.

---------

Co-authored-by: kibanamachine <[email protected]>
Co-authored-by: Jason Rhodes <[email protected]>

.buildkite/ftr_configs.yml

@@ -68,13 +68,11 @@ disabled:
   - x-pack/test/plugin_api_perf/config.js
   - x-pack/test/screenshot_creation/config.ts
   - x-pack/test/fleet_packages/config.ts
+  - x-pack/test/api_integration/apis/asset_manager/config_with_assets_source.ts
 
   # Scalability testing config that we run in its own pipeline
   - x-pack/test/scalability/config.ts
 
-  # Asset Manager configs, in tech preview, will move to enabled after more stability introduced
-  - x-pack/test/api_integration/apis/asset_manager/config.ts
-
   # Serverless base config files
   - x-pack/test_serverless/api_integration/config.base.ts
   - x-pack/test_serverless/functional/config.base.ts
@@ -176,6 +174,7 @@ enabled:
   - x-pack/test/api_integration/config_security_trial.ts
   - x-pack/test/api_integration/apis/aiops/config.ts
   - x-pack/test/api_integration/apis/asset_manager/config_when_disabled.ts
+  - x-pack/test/api_integration/apis/asset_manager/config_with_signals_source.ts
   - x-pack/test/api_integration/apis/cases/config.ts
   - x-pack/test/api_integration/apis/cloud_security_posture/config.ts
   - x-pack/test/api_integration/apis/console/config.ts

x-pack/plugins/asset_manager/common/types_api.ts

@@ -15,7 +15,17 @@ export const assetTypeRT = rt.union([
 
 export type AssetType = rt.TypeOf<typeof assetTypeRT>;
 
-export type AssetKind = 'unknown' | 'node';
+export const assetKindRT = rt.union([
+  rt.literal('cluster'),
+  rt.literal('host'),
+  rt.literal('pod'),
+  rt.literal('container'),
+  rt.literal('service'),
+  rt.literal('alert'),
+]);
+
+export type AssetKind = rt.TypeOf<typeof assetKindRT>;
+
 export type AssetStatus =
   | 'CREATING'
   | 'ACTIVE'
@@ -47,17 +57,20 @@ export interface ECSDocument extends WithTimestamp {
   'orchestrator.cluster.version'?: string;
 
   'cloud.provider'?: CloudProviderName;
+  'cloud.instance.id'?: string;
   'cloud.region'?: string;
   'cloud.service.name'?: string;
+
+  'service.environment'?: string;
 }
 
 export interface Asset extends ECSDocument {
   'asset.collection_version'?: string;
   'asset.ean': string;
   'asset.id': string;
-  'asset.kind'?: AssetKind;
+  'asset.kind': AssetKind;
   'asset.name'?: string;
-  'asset.type': AssetType;
+  'asset.type'?: AssetType;
   'asset.status'?: AssetStatus;
   'asset.parents'?: string | string[];
   'asset.children'?: string | string[];
@@ -121,14 +134,15 @@ export interface K8sCluster extends WithTimestamp {
 
 export interface AssetFilters {
   type?: AssetType | AssetType[];
-  kind?: AssetKind;
+  kind?: AssetKind | AssetKind[];
   ean?: string | string[];
   id?: string;
   typeLike?: string;
+  kindLike?: string;
   eanLike?: string;
   collectionVersion?: number | 'latest' | 'all';
-  from?: string;
-  to?: string;
+  from?: string | number;
+  to?: string | number;
 }
 
 export const relationRT = rt.union([

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

@@ -6,8 +6,9 @@
  */
 
 import { PluginInitializerContext } from '@kbn/core-plugins-server';
-import { AssetManagerServerPlugin, config, AssetManagerConfig } from './plugin';
+import { AssetManagerServerPlugin, config } from './plugin';
 import type { WriteSamplesPostBody } from './routes/sample_assets';
+import { AssetManagerConfig } from './types';
 
 export type { AssetManagerConfig, WriteSamplesPostBody };
 export { config };

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

@@ -0,0 +1,27 @@
+/*
+ * 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 { GetHostsOptionsInjected } from '.';
+import { getAssets } from '../../get_assets';
+
+export async function getHostsByAssets(
+  options: GetHostsOptionsInjected
+): Promise<{ hosts: Asset[] }> {
+  const hosts = await getAssets({
+    esClient: options.esClient,
+    filters: {
+      kind: 'host',
+      from: options.from,
+      to: options.to,
+    },
+  });
+
+  return {
+    hosts,
+  };
+}

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

@@ -0,0 +1,24 @@
+/*
+ * 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 { GetHostsOptionsInjected } from '.';
+import { collectHosts } from '../../collectors/hosts';
+
+export async function getHostsBySignals(
+  options: GetHostsOptionsInjected
+): Promise<{ hosts: Asset[] }> {
+  const { assets } = await collectHosts({
+    client: options.esClient,
+    from: options.from,
+    to: options.to,
+    sourceIndices: options.sourceIndices,
+  });
+  return {
+    hosts: assets,
+  };
+}

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

@@ -0,0 +1,20 @@
+/*
+ * 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 GetHostsOptions extends AccessorOptions {
+  from: number;
+  to: number;
+}
+export type GetHostsOptionsInjected = OptionsWithInjectedValues<GetHostsOptions>;
+
+export interface HostIdentifier {
+  'asset.ean': string;
+  'asset.id': string;
+  'asset.name'?: string;
+}

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

@@ -0,0 +1,19 @@
+/*
+ * 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 { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+import { AssetManagerConfig } from '../../types';
+
+export interface InjectedValues {
+  sourceIndices: AssetManagerConfig['sourceIndices'];
+}
+
+export type OptionsWithInjectedValues<T extends object> = T & InjectedValues;
+
+export interface AccessorOptions {
+  esClient: ElasticsearchClient;
+}

x-pack/plugins/asset_manager/server/lib/asset_accessor.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 { AssetManagerConfig } from '../types';
+import { OptionsWithInjectedValues } from './accessors';
+import { GetHostsOptions } from './accessors/hosts';
+import { getHostsByAssets } from './accessors/hosts/get_hosts_by_assets';
+import { getHostsBySignals } from './accessors/hosts/get_hosts_by_signals';
+
+interface AssetAccessorClassOptions {
+  sourceIndices: AssetManagerConfig['sourceIndices'];
+  source: AssetManagerConfig['lockedSource'];
+}
+
+export class AssetAccessor {
+  constructor(private options: AssetAccessorClassOptions) {}
+
+  injectOptions<T extends object = {}>(options: T): OptionsWithInjectedValues<T> {
+    return {
+      ...options,
+      sourceIndices: this.options.sourceIndices,
+    };
+  }
+
+  async getHosts(options: GetHostsOptions): Promise<{ hosts: Asset[] }> {
+    const withInjected = this.injectOptions(options);
+    if (this.options.source === 'assets') {
+      return await getHostsByAssets(withInjected);
+    } else {
+      return await getHostsBySignals(withInjected);
+    }
+  }
+}

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

@@ -0,0 +1,91 @@
+/*
+ * 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 { estypes } from '@elastic/elasticsearch';
+import { Asset } from '../../../common/types_api';
+import { CollectorOptions, QUERY_MAX_SIZE } from '.';
+
+export async function collectContainers({
+  client,
+  from,
+  to,
+  sourceIndices,
+  afterKey,
+}: CollectorOptions) {
+  const { metrics, logs, traces } = sourceIndices;
+  const dsl: estypes.SearchRequest = {
+    index: [traces, logs, metrics],
+    size: QUERY_MAX_SIZE,
+    collapse: {
+      field: 'container.id',
+    },
+    sort: [{ 'container.id': 'asc' }],
+    _source: false,
+    fields: [
+      'kubernetes.*',
+      'cloud.provider',
+      'orchestrator.cluster.name',
+      'host.name',
+      'host.hostname',
+    ],
+    query: {
+      bool: {
+        filter: [
+          {
+            range: {
+              '@timestamp': {
+                gte: from,
+                lte: to,
+              },
+            },
+          },
+        ],
+        should: [
+          { exists: { field: 'kubernetes.container.id' } },
+          { exists: { field: 'kubernetes.pod.uid' } },
+          { exists: { field: 'kubernetes.node.name' } },
+          { exists: { field: 'host.hostname' } },
+        ],
+      },
+    },
+  };
+
+  if (afterKey) {
+    dsl.search_after = afterKey;
+  }
+
+  const esResponse = await client.search(dsl);
+
+  const assets = esResponse.hits.hits.reduce<Asset[]>((acc: Asset[], hit: any) => {
+    const { fields = {} } = hit;
+    const containerId = fields['container.id'];
+    const podUid = fields['kubernetes.pod.uid'];
+    const nodeName = fields['kubernetes.node.name'];
+
+    const parentEan = podUid ? `pod:${podUid}` : `host:${fields['host.hostname']}`;
+
+    const container: Asset = {
+      '@timestamp': new Date().toISOString(),
+      'asset.kind': 'container',
+      'asset.id': containerId,
+      'asset.ean': `container:${containerId}`,
+      'asset.parents': [parentEan],
+    };
+
+    if (nodeName) {
+      container['asset.references'] = [`host:${nodeName}`];
+    }
+
+    acc.push(container);
+
+    return acc;
+  }, []);
+
+  const hitsLen = esResponse.hits.hits.length;
+  const next = hitsLen === QUERY_MAX_SIZE ? esResponse.hits.hits[hitsLen - 1].sort : undefined;
+  return { assets, afterKey: next };
+}