[Stack Monitoring] Add support for beats datastream patterns (#146184)

## Summary
Closes https://github.com/elastic/kibana/issues/146686

Update apm and beats queries to read from the Beat package data streams
created in https://github.com/elastic/integrations/pull/4708. Also
updates health API to fetch from the data streams.

API tests follow up in https://github.com/elastic/kibana/pull/147755

### Testing
That's a fairly heavy workflow. I'm investigating ways to make that
easier

- build `beat` package with `elastic-package build`. The package is not
published yet so you'll need to pull
https://github.com/elastic/integrations/pull/4708 locally
- start a stack `elastic-package stack up -v -d --version
8.7.0-SNAPSHOT`. Make sure you do this within `integrations` repo so it
picks up the previously built package
- start a beat service with `elastic-package service up -v --variant
metricbeat_8.7.0`
- install both elasticsearch and beat packages from the kibana started
by the stack command (`https://localhost:5601`). Nothing shows up in SM
if we don't have elasticsearch data.
- both packages are prerelease versions and we need to explicitly tell
Integrations plugin to show them up[1]
- elasticsearch hosts is `https://elasticsearch:9200` and beat _should
be_ `http://elastic-package-service_beat_1:5066` but it may differ
depending on your docker version
- start a Kibana with this branch connected to the elasticsearch
instance from the `stack up` command. [see
howto](https://github.com/elastic/observability-dev/blob/main/docs/infra-obs-ui/stack-monitoring_integration-packages.md#connecting-a-local-kibana)
- navigate to Stack Monitoring and verify the metricbeat is properly
monitored
- start an apm server with `elastic-package service up -v --variant
apm_8.7.0`
- navigate to Stack Monitoring and verify apm-server is properly
monitored


[1]
<img width="853" alt="Screenshot 2022-12-01 at 00 20 20"
src="https://user-images.githubusercontent.com/5239883/204929352-c7656679-f88c-4013-b1f8-cf7c67d4c830.png">

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

x-pack/plugins/monitoring/server/lib/apm/create_apm_query.ts

@@ -7,6 +7,7 @@
 
 import { ApmMetric, ApmMetricFields } from '../metrics';
 import { createQuery } from '../create_query';
+import { getBeatDataset } from '../cluster/get_index_patterns';
 
 /**
  * {@code createQuery} for all APM instances.
@@ -27,7 +28,7 @@ export function createApmQuery(options: {
     metric: ApmMetric.getMetricFields(),
     type: 'beats_stats',
     metricset: 'stats',
-    dsDataset: 'beats.stats',
+    dsDataset: getBeatDataset('stats'),
     ...(options ?? {}),
   };
 

x-pack/plugins/monitoring/server/lib/apm/get_apms_for_clusters.ts

@@ -11,7 +11,7 @@ import { ApmMetric } from '../metrics';
 import { apmAggResponseHandler, apmUuidsAgg, apmAggFilterPath } from './_apm_stats';
 import { getTimeOfLastEvent } from './_get_time_of_last_event';
 import { ElasticsearchResponse } from '../../../common/types/es';
-import { getLegacyIndexPattern } from '../cluster/get_index_patterns';
+import { getIndexPatterns } from '../cluster/get_index_patterns';
 import { Globals } from '../../static_globals';
 
 export function handleResponse(clusterUuid: string, response: ElasticsearchResponse) {
@@ -40,8 +40,9 @@ export function getApmsForClusters(req: LegacyRequest, clusters: Cluster[], ccs?
   const maxBucketSize = config.ui.max_bucket_size;
   const cgroup = config.ui.container.apm.enabled;
 
-  const indexPatterns = getLegacyIndexPattern({
+  const indexPatterns = getIndexPatterns({
     moduleType: 'beats',
+    dataset: 'stats',
     ccs: ccs || req.payload.ccs,
     config: Globals.app.config,
   });

x-pack/plugins/monitoring/server/lib/beats/create_beats_query.ts

@@ -7,6 +7,7 @@
 
 import { BeatsMetric, BeatsMetricFields } from '../metrics';
 import { createQuery } from '../create_query';
+import { getBeatDataset } from '../cluster/get_index_patterns';
 
 /**
  * {@code createQuery} for all Beats instances.
@@ -31,7 +32,7 @@ export function createBeatsQuery(options: {
     metric: BeatsMetric.getMetricFields(),
     type: 'beats_stats',
     metricset: 'stats',
-    dsDataset: 'beats.stats',
+    dsDataset: getBeatDataset('stats'),
     ...(options ?? {}),
   };
 

x-pack/plugins/monitoring/server/lib/beats/get_beats_for_clusters.ts

@@ -10,7 +10,7 @@ import { createBeatsQuery } from './create_beats_query';
 import { beatsAggFilterPath, beatsUuidsAgg, beatsAggResponseHandler } from './_beats_stats';
 import type { ElasticsearchResponse } from '../../../common/types/es';
 import { LegacyRequest, Cluster } from '../../types';
-import { getLegacyIndexPattern } from '../cluster/get_index_patterns';
+import { getIndexPatterns } from '../cluster/get_index_patterns';
 import { Globals } from '../../static_globals';
 
 export function handleResponse(clusterUuid: string, response: ElasticsearchResponse) {
@@ -37,8 +37,9 @@ export function getBeatsForClusters(req: LegacyRequest, clusters: Cluster[], ccs
   const end = req.payload.timeRange.max;
   const config = req.server.config;
   const maxBucketSize = config.ui.max_bucket_size;
-  const indexPatterns = getLegacyIndexPattern({
+  const indexPatterns = getIndexPatterns({
     moduleType: 'beats',
+    dataset: 'stats',
     ccs,
     config: Globals.app.config,
   });

x-pack/plugins/monitoring/server/lib/cluster/get_index_patterns.ts

@@ -126,6 +126,7 @@ const getDataset = (moduleType: INDEX_PATTERN_TYPES) => (dataset: string) =>
 export const getElasticsearchDataset = getDataset('elasticsearch');
 export const getKibanaDataset = getDataset('kibana');
 export const getLogstashDataset = getDataset('logstash');
+export const getBeatDataset = getDataset('beats');
 
 function buildDatasetPattern(
   moduleType?: INDEX_PATTERN_TYPES,

x-pack/plugins/monitoring/server/lib/standalone_clusters/has_standalone_clusters.ts

@@ -11,9 +11,9 @@ import { LegacyRequest } from '../../types';
 import { standaloneClusterFilter } from '.';
 import { Globals } from '../../static_globals';
 import {
-  getLegacyIndexPattern,
   getIndexPatterns,
   getLogstashDataset,
+  getBeatDataset,
 } from '../cluster/get_index_patterns';
 
 export async function hasStandaloneClusters(req: LegacyRequest, ccs: string) {
@@ -22,8 +22,7 @@ export async function hasStandaloneClusters(req: LegacyRequest, ccs: string) {
     moduleType: 'logstash',
     ccs,
   });
-  // use legacy because no integration exists for beats
-  const beatsIndexPatterns = getLegacyIndexPattern({
+  const beatsIndexPatterns = getIndexPatterns({
     moduleType: 'beats',
     config: Globals.app.config,
     ccs,
@@ -51,7 +50,12 @@ export async function hasStandaloneClusters(req: LegacyRequest, ccs: string) {
           },
           {
             terms: {
-              'data_stream.dataset': [getLogstashDataset('node'), getLogstashDataset('node_stats')],
+              'data_stream.dataset': [
+                getLogstashDataset('node'),
+                getLogstashDataset('node_stats'),
+                getBeatDataset('state'),
+                getBeatDataset('stats'),
+              ],
             },
           },
         ],

x-pack/plugins/monitoring/server/routes/api/v1/_health/monitored_clusters/build_monitored_clusters.ts

@@ -43,7 +43,7 @@ const metricbeatMonitoring7Pattern =
   /(.*:)?\.monitoring-(es|kibana|beats|logstash|ent-search)-7.*-mb.*/;
 const metricbeatMonitoring8Pattern =
   /(.*:)?\.ds-\.monitoring-(es|kibana|beats|logstash|ent-search)-8-mb.*/;
-const packagePattern = /(.*:)?\.ds-metrics-(elasticsearch|kibana|logstash)\..*/;
+const packagePattern = /(.*:)?\.ds-metrics-(elasticsearch|kibana|beats|logstash)\..*/;
 
 const getCollectionMode = (index: string): CollectionMode => {
   if (internalMonitoringPattern.test(index)) return CollectionMode.Internal;

x-pack/plugins/monitoring/server/routes/api/v1/_health/monitored_clusters/monitored_clusters_query.ts

@@ -6,15 +6,15 @@
  */
 
 import { TimeRange } from '../../../../../../common/http_api/shared';
+import {
+  getBeatDataset,
+  getElasticsearchDataset,
+  getKibanaDataset,
+  getLogstashDataset,
+} from '../../../../../lib/cluster/get_index_patterns';
 
 const MAX_BUCKET_SIZE = 100;
 
-const getDataset = (product: string) => (metricset: string) =>
-  `${product}.stack_monitoring.${metricset}`;
-const getElasticsearchDataset = getDataset('elasticsearch');
-const getKibanaDataset = getDataset('kibana');
-const getLogstashDataset = getDataset('logstash');
-
 interface QueryOptions {
   timeRange?: TimeRange;
   timeout: number; // in seconds
@@ -509,6 +509,11 @@ const beatsAggregations = {
                   'metricset.name': 'stats',
                 },
               },
+              {
+                term: {
+                  'data_stream.dataset': getBeatDataset('stats'),
+                },
+              },
             ],
           },
         },
@@ -543,6 +548,11 @@ const beatsAggregations = {
                   'metricset.name': 'state',
                 },
               },
+              {
+                term: {
+                  'data_stream.dataset': getBeatDataset('state'),
+                },
+              },
             ],
           },
         },

x-pack/plugins/monitoring/server/routes/api/v1/apm/instance.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { prefixIndexPatternWithCcs } from '../../../../../common/ccs_utils';
-import { INDEX_PATTERN_BEATS } from '../../../../../common/constants';
 import {
   postApmInstanceRequestParamsRT,
   postApmInstanceRequestPayloadRT,
@@ -16,6 +14,7 @@ import { getApmInfo } from '../../../../lib/apm';
 import { createValidationFunction } from '../../../../lib/create_route_validation_function';
 import { getMetrics } from '../../../../lib/details/get_metrics';
 import { handleError } from '../../../../lib/errors';
+import { getIndexPatterns } from '../../../../lib/cluster/get_index_patterns';
 import { MonitoringCore } from '../../../../types';
 import { metricSet } from './metric_set_instance';
 
@@ -35,7 +34,12 @@ export function apmInstanceRoute(server: MonitoringCore) {
       const config = server.config;
       const clusterUuid = req.params.clusterUuid;
       const ccs = req.payload.ccs;
-      const apmIndexPattern = prefixIndexPatternWithCcs(config, INDEX_PATTERN_BEATS, ccs);
+      const apmIndexPattern = getIndexPatterns({
+        ccs,
+        config,
+        moduleType: 'beats',
+        dataset: 'stats',
+      });
 
       const showCgroupMetrics = config.ui.container.apm.enabled;
       if (showCgroupMetrics) {

x-pack/plugins/monitoring/server/routes/api/v1/apm/instances.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { prefixIndexPatternWithCcs } from '../../../../../common/ccs_utils';
-import { INDEX_PATTERN_BEATS } from '../../../../../common/constants';
 import {
   postApmInstancesRequestParamsRT,
   postApmInstancesRequestPayloadRT,
@@ -15,6 +13,7 @@ import {
 import { getApms, getStats } from '../../../../lib/apm';
 import { createValidationFunction } from '../../../../lib/create_route_validation_function';
 import { handleError } from '../../../../lib/errors';
+import { getIndexPatterns } from '../../../../lib/cluster/get_index_patterns';
 import { MonitoringCore } from '../../../../types';
 
 export function apmInstancesRoute(server: MonitoringCore) {
@@ -32,7 +31,12 @@ export function apmInstancesRoute(server: MonitoringCore) {
       const config = server.config;
       const ccs = req.payload.ccs;
       const clusterUuid = req.params.clusterUuid;
-      const apmIndexPattern = prefixIndexPatternWithCcs(config, INDEX_PATTERN_BEATS, ccs);
+      const apmIndexPattern = getIndexPatterns({
+        ccs,
+        config,
+        moduleType: 'beats',
+        dataset: 'stats',
+      });
 
       try {
         const [stats, apms] = await Promise.all([

x-pack/plugins/monitoring/server/routes/api/v1/apm/overview.ts

@@ -43,7 +43,9 @@ export function apmOverviewRoute(server: MonitoringCore) {
       try {
         const [stats, metrics] = await Promise.all([
           getApmClusterStatus(req, { clusterUuid }),
-          getMetrics(req, 'beats', metricSet),
+          getMetrics(req, 'beats', metricSet, [
+            { term: { 'beats_stats.beat.type': 'apm-server' } },
+          ]),
         ]);
 
         return postApmOverviewResponsePayloadRT.encode({

x-pack/plugins/monitoring/server/routes/api/v1/beats/beat_detail.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { prefixIndexPatternWithCcs } from '../../../../../common/ccs_utils';
-import { INDEX_PATTERN_BEATS } from '../../../../../common/constants';
 import {
   postBeatDetailRequestParamsRT,
   postBeatDetailRequestPayloadRT,
@@ -16,6 +14,7 @@ import { getBeatSummary } from '../../../../lib/beats';
 import { createValidationFunction } from '../../../../lib/create_route_validation_function';
 import { getMetrics } from '../../../../lib/details/get_metrics';
 import { handleError } from '../../../../lib/errors';
+import { getIndexPatterns } from '../../../../lib/cluster/get_index_patterns';
 import { MonitoringCore } from '../../../../types';
 import { metricSet } from './metric_set_detail';
 
@@ -35,7 +34,11 @@ export function beatsDetailRoute(server: MonitoringCore) {
       const beatUuid = req.params.beatUuid;
       const config = server.config;
       const ccs = req.payload.ccs;
-      const beatsIndexPattern = prefixIndexPatternWithCcs(config, INDEX_PATTERN_BEATS, ccs);
+      const beatsIndexPattern = getIndexPatterns({
+        ccs,
+        config,
+        moduleType: 'beats',
+      });
 
       const summaryOptions = {
         clusterUuid,

x-pack/plugins/monitoring/server/routes/api/v1/beats/beats.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { prefixIndexPatternWithCcs } from '../../../../../common/ccs_utils';
-import { INDEX_PATTERN_BEATS } from '../../../../../common/constants';
 import {
   postBeatsListingRequestParamsRT,
   postBeatsListingRequestPayloadRT,
@@ -15,6 +13,7 @@ import {
 import { getBeats, getStats } from '../../../../lib/beats';
 import { createValidationFunction } from '../../../../lib/create_route_validation_function';
 import { handleError } from '../../../../lib/errors';
+import { getIndexPatterns } from '../../../../lib/cluster/get_index_patterns';
 import { MonitoringCore } from '../../../../types';
 
 export function beatsListingRoute(server: MonitoringCore) {
@@ -32,7 +31,11 @@ export function beatsListingRoute(server: MonitoringCore) {
       const config = server.config;
       const ccs = req.payload.ccs;
       const clusterUuid = req.params.clusterUuid;
-      const beatsIndexPattern = prefixIndexPatternWithCcs(config, INDEX_PATTERN_BEATS, ccs);
+      const beatsIndexPattern = getIndexPatterns({
+        ccs,
+        config,
+        moduleType: 'beats',
+      });
 
       try {
         const [stats, listing] = await Promise.all([

x-pack/plugins/monitoring/server/routes/api/v1/beats/overview.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { prefixIndexPatternWithCcs } from '../../../../../common/ccs_utils';
-import { INDEX_PATTERN_BEATS } from '../../../../../common/constants';
 import {
   postBeatsOverviewRequestParamsRT,
   postBeatsOverviewRequestPayloadRT,
@@ -16,6 +14,7 @@ import { getLatestStats, getStats } from '../../../../lib/beats';
 import { createValidationFunction } from '../../../../lib/create_route_validation_function';
 import { getMetrics } from '../../../../lib/details/get_metrics';
 import { handleError } from '../../../../lib/errors';
+import { getIndexPatterns } from '../../../../lib/cluster/get_index_patterns';
 import { MonitoringCore } from '../../../../types';
 import { metricSet } from './metric_set_overview';
 
@@ -34,7 +33,11 @@ export function beatsOverviewRoute(server: MonitoringCore) {
       const config = server.config;
       const ccs = req.payload.ccs;
       const clusterUuid = req.params.clusterUuid;
-      const beatsIndexPattern = prefixIndexPatternWithCcs(config, INDEX_PATTERN_BEATS, ccs);
+      const beatsIndexPattern = getIndexPatterns({
+        ccs,
+        config,
+        moduleType: 'beats',
+      });
 
       try {
         const [latest, stats, metrics] = await Promise.all([