[Infra] infra services api (#173875)

## Summary
Creation of a new endpoint within Infra to get services from APM indices
that are related to a give host through `host.name`. These services will
be listed in the Host Detail view in another PR. This endpoint queries
apm transaction metrics and apm logs to get services.

Closes #171661

### Test
The easiest way to test this api is to visit it directly using a host
that has some services attached to it using our test cluster

URL: http://localhost:5601/api/infra/services
eg usage:
`http://localhost:5601/api/infra/services?from=now-15m&to=now&filters={"host.name":"gke-edge-oblt-edge-oblt-pool-5fbec7a6-nfy0"}&size=5`

response:

```
{
    "services": [
        {
            "service.name": "productcatalogservice",
            "agent.name": "opentelemetry/go"
        },
        {
            "service.name": "frontend",
            "agent.name": "opentelemetry/nodejs"
        }
    ]
}
```



### Follow up 
- Have APM server collect host.name as part of service_summary metrics
and query that instead. Service summary aggregates transaction, error,
log, and metric events into service-summary metrics. This would simplify
the query.

- `added apm-synthtrace` to `metrics_ui` api tests and created follow up
PR for removing the code i needed to duplicate
#175064

---------

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

x-pack/plugins/infra/common/constants.ts

@@ -21,6 +21,8 @@ export const TIEBREAKER_FIELD = '_doc';
 export const HOST_FIELD = 'host.name';
 export const CONTAINER_FIELD = 'container.id';
 export const POD_FIELD = 'kubernetes.pod.uid';
+export const CMDLINE_FIELD = 'system.process.cmdline';
+export const HOST_NAME_FIELD = 'host.name';
 
 export const O11Y_AAD_FIELDS = [
   'cloud.*',

x-pack/plugins/infra/common/http_api/host_details/get_infra_services.ts

@@ -0,0 +1,78 @@
+/*
+ * 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 {
+  createLiteralValueFromUndefinedRT,
+  inRangeFromStringRt,
+  dateRt,
+  datemathStringRt,
+} from '@kbn/io-ts-utils';
+import * as rt from 'io-ts';
+
+export const sizeRT = rt.union([
+  inRangeFromStringRt(1, 100),
+  createLiteralValueFromUndefinedRT(10),
+]);
+export const assetDateRT = rt.union([dateRt, datemathStringRt]);
+
+export const servicesFiltersRT = rt.strict({
+  ['host.name']: rt.string,
+});
+
+export type ServicesFilter = rt.TypeOf<typeof servicesFiltersRT>;
+
+export const GetServicesRequestQueryRT = rt.intersection([
+  rt.strict({ from: assetDateRT, to: assetDateRT, filters: rt.string }),
+  rt.partial({
+    size: sizeRT,
+    validatedFilters: servicesFiltersRT,
+  }),
+]);
+
+export type GetServicesRequestQuery = rt.TypeOf<typeof GetServicesRequestQueryRT>;
+
+export interface ServicesAPIRequest {
+  filters: ServicesFilter;
+  from: string;
+  to: string;
+  size?: number;
+}
+
+const AgentNameRT = rt.union([rt.string, rt.null]);
+
+export const ServicesAPIQueryAggregationRT = rt.type({
+  services: rt.type({
+    buckets: rt.array(
+      rt.type({
+        key: rt.string,
+        latestAgent: rt.type({
+          top: rt.array(
+            rt.type({
+              sort: rt.array(rt.string),
+              metrics: rt.type({
+                'agent.name': AgentNameRT,
+              }),
+            })
+          ),
+        }),
+      })
+    ),
+  }),
+});
+
+export type ServicesAPIQueryAggregation = rt.TypeOf<typeof ServicesAPIQueryAggregationRT>;
+
+export const ServiceRT = rt.type({
+  'service.name': rt.string,
+  'agent.name': AgentNameRT,
+});
+
+export type Service = rt.TypeOf<typeof ServiceRT>;
+
+export const ServicesAPIResponseRT = rt.type({
+  services: rt.array(ServiceRT),
+});

x-pack/plugins/infra/common/http_api/host_details/index.ts

@@ -6,3 +6,4 @@
  */
 
 export * from './process_list';
+export * from './get_infra_services';

x-pack/plugins/infra/kibana.jsonc

@@ -33,7 +33,8 @@
       "uiActions",
       "unifiedSearch",
       "usageCollection",
-      "visTypeTimeseries"
+      "visTypeTimeseries",
+      "apmDataAccess"
     ],
     "optionalPlugins": ["spaces", "ml", "home", "embeddable", "osquery", "cloud", "profilingDataAccess"],
     "requiredBundles": [

x-pack/plugins/infra/server/infra_server.ts

@@ -33,6 +33,7 @@ import { initSnapshotRoute } from './routes/snapshot';
 import { initInfraMetricsRoute } from './routes/infra';
 import { initMetricsExplorerViewRoutes } from './routes/metrics_explorer_views';
 import { initProfilingRoutes } from './routes/profiling';
+import { initServicesRoute } from './routes/services';
 
 export const initInfraServer = (libs: InfraBackendLibs) => {
   initIpToHostName(libs);
@@ -61,4 +62,5 @@ export const initInfraServer = (libs: InfraBackendLibs) => {
   initOverviewRoute(libs);
   initInfraMetricsRoute(libs);
   initProfilingRoutes(libs);
+  initServicesRoute(libs);
 };

x-pack/plugins/infra/server/lib/adapters/framework/adapter_types.ts

@@ -31,6 +31,10 @@ import {
   ProfilingDataAccessPluginSetup,
   ProfilingDataAccessPluginStart,
 } from '@kbn/profiling-data-access-plugin/server';
+import {
+  ApmDataAccessPluginSetup,
+  ApmDataAccessPluginStart,
+} from '@kbn/apm-data-access-plugin/server';
 
 export interface InfraServerPluginSetupDeps {
   alerting: AlertingPluginContract;
@@ -47,13 +51,15 @@ export interface InfraServerPluginSetupDeps {
   logsShared: LogsSharedPluginSetup;
   metricsDataAccess: MetricsDataPluginSetup;
   profilingDataAccess?: ProfilingDataAccessPluginSetup;
+  apmDataAccess: ApmDataAccessPluginSetup;
 }
 
 export interface InfraServerPluginStartDeps {
   data: DataPluginStart;
   dataViews: DataViewsPluginStart;
   logsShared: LogsSharedPluginStart;
   profilingDataAccess?: ProfilingDataAccessPluginStart;
+  apmDataAccess: ApmDataAccessPluginStart;
 }
 
 export interface CallWithRequestParams extends estypes.RequestBase {

x-pack/plugins/infra/server/lib/host_details/get_services.ts

@@ -0,0 +1,143 @@
+/*
+ * 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 { QueryDslQueryContainer } from '@elastic/elasticsearch/lib/api/types';
+import { APMDataAccessConfig } from '@kbn/apm-data-access-plugin/server';
+import { termQuery } from '@kbn/observability-plugin/server';
+import { ESSearchClient } from '../metrics/types';
+import {
+  ServicesAPIRequest,
+  ServicesAPIQueryAggregation,
+} from '../../../common/http_api/host_details';
+import { HOST_NAME_FIELD } from '../../../common/constants';
+
+export const getServices = async (
+  client: ESSearchClient,
+  apmIndices: APMDataAccessConfig['indices'],
+  options: ServicesAPIRequest
+) => {
+  const { error, metric } = apmIndices;
+  const { filters, size = 10, from, to } = options;
+  const commonFiltersList: QueryDslQueryContainer[] = [
+    {
+      range: {
+        '@timestamp': {
+          gte: from,
+          lte: to,
+        },
+      },
+    },
+    {
+      exists: {
+        field: 'service.name',
+      },
+    },
+  ];
+
+  if (filters['host.name']) {
+    // also query for host.hostname field along with host.name, as some services may use this field
+    const HOST_HOSTNAME_FIELD = 'host.hostname';
+    commonFiltersList.push({
+      bool: {
+        should: [
+          ...termQuery(HOST_NAME_FIELD, filters[HOST_NAME_FIELD]),
+          ...termQuery(HOST_HOSTNAME_FIELD, filters[HOST_NAME_FIELD]),
+        ],
+        minimum_should_match: 1,
+      },
+    });
+  }
+  const aggs = {
+    services: {
+      terms: {
+        field: 'service.name',
+        size,
+      },
+      aggs: {
+        latestAgent: {
+          top_metrics: {
+            metrics: [{ field: 'agent.name' }],
+            sort: {
+              '@timestamp': 'desc',
+            },
+            size: 1,
+          },
+        },
+      },
+    },
+  };
+  // get services from transaction metrics
+  const metricsQuery = {
+    size: 0,
+    _source: false,
+    query: {
+      bool: {
+        filter: [
+          {
+            term: {
+              'metricset.name': 'transaction',
+            },
+          },
+          {
+            term: {
+              'metricset.interval': '1m', // make this dynamic if we start returning time series data
+            },
+          },
+          ...commonFiltersList,
+        ],
+      },
+    },
+    aggs,
+  };
+  // get services from logs
+  const logsQuery = {
+    size: 0,
+    _source: false,
+    query: {
+      bool: {
+        filter: commonFiltersList,
+      },
+    },
+    aggs,
+  };
+
+  const resultMetrics = await client<{}, ServicesAPIQueryAggregation>({
+    body: metricsQuery,
+    index: [metric],
+  });
+  const resultLogs = await client<{}, ServicesAPIQueryAggregation>({
+    body: logsQuery,
+    index: [error],
+  });
+
+  const servicesListBucketsFromMetrics = resultMetrics.aggregations?.services?.buckets || [];
+  const servicesListBucketsFromLogs = resultLogs.aggregations?.services?.buckets || [];
+  const serviceMap = [...servicesListBucketsFromMetrics, ...servicesListBucketsFromLogs].reduce(
+    (acc, bucket) => {
+      const serviceName = bucket.key;
+      const latestAgentEntry = bucket.latestAgent.top[0];
+      const latestTimestamp = latestAgentEntry.sort[0];
+      const agentName = latestAgentEntry.metrics['agent.name'];
+      // dedup and get the latest timestamp
+      const existingService = acc.get(serviceName);
+      if (!existingService || existingService.latestTimestamp < latestTimestamp) {
+        acc.set(serviceName, { latestTimestamp, agentName });
+      }
+
+      return acc;
+    },
+    new Map<string, { latestTimestamp: string; agentName: string | null }>()
+  );
+
+  const services = Array.from(serviceMap)
+    .slice(0, size)
+    .map(([serviceName, { agentName }]) => ({
+      'service.name': serviceName,
+      'agent.name': agentName,
+    }));
+  return { services };
+};

x-pack/plugins/infra/server/lib/host_details/process_list.ts

@@ -5,10 +5,9 @@
  * 2.0.
  */
 
-import { TIMESTAMP_FIELD } from '../../../common/constants';
+import { TIMESTAMP_FIELD, CMDLINE_FIELD } from '../../../common/constants';
 import { ProcessListAPIRequest, ProcessListAPIQueryAggregation } from '../../../common/http_api';
 import { ESSearchClient } from '../metrics/types';
-import { CMDLINE_FIELD } from './common';
 
 const TOP_N = 10;
 

x-pack/plugins/infra/server/lib/host_details/process_list_chart.ts

@@ -6,15 +6,14 @@
  */
 
 import { first } from 'lodash';
-import { TIMESTAMP_FIELD } from '../../../common/constants';
+import { TIMESTAMP_FIELD, CMDLINE_FIELD } from '../../../common/constants';
 import {
   ProcessListAPIChartRequest,
   ProcessListAPIChartQueryAggregation,
   ProcessListAPIRow,
   ProcessListAPIChartResponse,
 } from '../../../common/http_api';
 import { ESSearchClient } from '../metrics/types';
-import { CMDLINE_FIELD } from './common';
 
 export const getProcessListChart = async (
   search: ESSearchClient,

x-pack/plugins/infra/server/lib/infra_types.ts

@@ -13,6 +13,8 @@ import { ObservabilityConfig } from '@kbn/observability-plugin/server';
 import type { LocatorPublic } from '@kbn/share-plugin/common';
 import type { ILogsSharedLogEntriesDomain } from '@kbn/logs-shared-plugin/server';
 import type { MetricsDataClient } from '@kbn/metrics-data-access-plugin/server';
+import { APMDataAccessConfig } from '@kbn/apm-data-access-plugin/server';
+import { SavedObjectsClientContract } from '@kbn/core-saved-objects-api-server';
 import { RulesServiceSetup } from '../services/rules';
 import { InfraConfig, InfraPluginStartServicesAccessor } from '../types';
 import { KibanaFramework } from './adapters/framework/kibana_framework_adapter';
@@ -41,4 +43,5 @@ export interface InfraBackendLibs extends InfraDomainLibs {
   logger: Logger;
   alertsLocator?: LocatorPublic<AlertsLocatorParams>;
   metricsClient: MetricsDataClient;
+  getApmIndices: (soClient: SavedObjectsClientContract) => Promise<APMDataAccessConfig['indices']>;
 }

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

@@ -175,6 +175,7 @@ export class InfraServerPlugin
   setup(core: InfraPluginCoreSetup, plugins: InfraServerPluginSetupDeps) {
     const framework = new KibanaFramework(core, this.config, plugins);
     const metricsClient = plugins.metricsDataAccess.client;
+    const getApmIndices = plugins.apmDataAccess.getApmIndices;
     metricsClient.setDefaultMetricIndicesHandler(async (options: GetMetricIndicesOptions) => {
       const sourceConfiguration = await sources.getInfraSourceConfiguration(
         options.savedObjectsClient,
@@ -219,6 +220,7 @@ export class InfraServerPlugin
       sources,
       sourceStatus,
       metricsClient,
+      getApmIndices,
       ...domainLibs,
       handleEsError,
       logsRules: this.logsRules.setup(core, plugins),