[Stack Monitoring] compatibility for agent data streams

x-pack/plugins/monitoring/common/ccs_utils.ts

@@ -13,32 +13,17 @@ type Config = Partial<MonitoringConfig> & {
   get?: (key: string) => any;
 };
 
-export function appendMetricbeatIndex(
-  config: Config,
-  indexPattern: string,
-  ccs?: string,
-  bypass: boolean = false
-) {
-  if (bypass) {
-    return indexPattern;
-  }
-  // Leverage this function to also append the dynamic metricbeat index too
-  let mbIndex = null;
+export function getConfigCcs(config: Config): boolean | string {
+  let ccsEnabled = false;
   // TODO: NP
   // This function is called with both NP config and LP config
   if (isFunction(config.get)) {
-    mbIndex = config.get('monitoring.ui.metricbeat.index');
+    ccsEnabled = config.get('monitoring.ui.ccs.enabled');
   } else {
-    mbIndex = get(config, 'ui.metricbeat.index');
-  }
-
-  if (ccs) {
-    mbIndex = `${mbIndex},${ccs}:${mbIndex}`;
+    ccsEnabled = get(config, 'ui.ccs.enabled');
   }
-
-  return `${indexPattern},${mbIndex}`;
+  return ccsEnabled;
 }
-
 /**
  * Prefix all comma separated index patterns within the original {@code indexPattern}.
  *
@@ -50,44 +35,20 @@ export function appendMetricbeatIndex(
  * @param  {String} ccs The optional cluster-prefix to prepend.
  * @return {String} The index pattern with the {@code cluster} prefix appropriately prepended.
  */
-export function prefixIndexPattern(
-  config: Config,
-  indexPattern: string,
-  ccs?: string,
-  monitoringIndicesOnly: boolean = false
-) {
-  let ccsEnabled = false;
-  // TODO: NP
-  // This function is called with both NP config and LP config
-  if (isFunction(config.get)) {
-    ccsEnabled = config.get('monitoring.ui.ccs.enabled');
-  } else {
-    ccsEnabled = get(config, 'ui.ccs.enabled');
-  }
-
+export function prefixIndexPattern(config: Config, indexPattern: string, ccs?: string) {
+  const ccsEnabled = getConfigCcs(config);
   if (!ccsEnabled || !ccs) {
-    return appendMetricbeatIndex(
-      config,
-      indexPattern,
-      ccsEnabled ? ccs : undefined,
-      monitoringIndicesOnly
-    );
+    return indexPattern;
   }
 
   const patterns = indexPattern.split(',');
   const prefixedPattern = patterns.map((pattern) => `${ccs}:${pattern}`).join(',');
 
   // if a wildcard is used, then we also want to search the local indices
   if (ccs === '*') {
-    return appendMetricbeatIndex(
-      config,
-      `${prefixedPattern},${indexPattern}`,
-      ccs,
-      monitoringIndicesOnly
-    );
+    return `${prefixedPattern},${indexPattern}`;
   }
-
-  return appendMetricbeatIndex(config, prefixedPattern, ccs, monitoringIndicesOnly);
+  return prefixedPattern;
 }
 
 /**

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

@@ -129,6 +129,9 @@ export const INDEX_PATTERN_LOGSTASH = '.monitoring-logstash-6-*,.monitoring-logs
 export const INDEX_PATTERN_BEATS = '.monitoring-beats-6-*,.monitoring-beats-7-*';
 export const INDEX_ALERTS = '.monitoring-alerts-6*,.monitoring-alerts-7*';
 export const INDEX_PATTERN_ELASTICSEARCH = '.monitoring-es-6-*,.monitoring-es-7-*';
+export const DS_INDEX_PATTERN_METRICS = 'metrics';
+export const DS_INDEX_PATTERN_LOGS = 'logs';
+export const DS_INDEX_PATTERN_ES = 'elasticsearch';
 export const INDEX_PATTERN_ENTERPRISE_SEARCH = '.monitoring-ent-search-*';
 
 // This is the unique token that exists in monitoring indices collected by metricbeat
@@ -584,3 +587,12 @@ export const ALERT_EMAIL_SERVICES = ['gmail', 'hotmail', 'icloud', 'outlook365',
 export const SAVED_OBJECT_TELEMETRY = 'monitoring-telemetry';
 
 export const TELEMETRY_METRIC_BUTTON_CLICK = 'btnclick__';
+
+export type INDEX_PATTERN_TYPES =
+  | 'elasticsearch'
+  | 'kibana'
+  | 'logstash'
+  | 'beats'
+  | 'enterprisesearch';
+
+export type DS_INDEX_PATTERN_TYPES = typeof DS_INDEX_PATTERN_METRICS | typeof DS_INDEX_PATTERN_LOGS;

x-pack/plugins/monitoring/server/alerts/base_rule.ts

@@ -28,12 +28,10 @@ import {
   CommonAlertParams,
 } from '../../common/types/alerts';
 import { fetchClusters } from '../lib/alerts/fetch_clusters';
-import { getCcsIndexPattern } from '../lib/alerts/get_ccs_index_pattern';
-import { INDEX_PATTERN_ELASTICSEARCH } from '../../common/constants';
 import { AlertSeverity } from '../../common/enums';
-import { appendMetricbeatIndex } from '../lib/alerts/append_mb_index';
 import { parseDuration } from '../../../alerting/common';
 import { Globals } from '../static_globals';
+import { getNewIndexPatterns } from '../lib/cluster/get_index_patterns';
 
 type ExecutedState =
   | {
@@ -226,23 +224,18 @@ export class BaseRule {
     );
 
     const esClient = services.scopedClusterClient.asCurrentUser;
-    const availableCcs = Globals.app.config.ui.ccs.enabled;
-    const clusters = await this.fetchClusters(esClient, params as CommonAlertParams, availableCcs);
-    const data = await this.fetchData(params, esClient, clusters, availableCcs);
+    const clusters = await this.fetchClusters(esClient, params as CommonAlertParams);
+    const data = await this.fetchData(params, esClient, clusters);
     return await this.processData(data, clusters, services, state);
   }
 
-  protected async fetchClusters(
-    esClient: ElasticsearchClient,
-    params: CommonAlertParams,
-    ccs?: boolean
-  ) {
-    let esIndexPattern = appendMetricbeatIndex(Globals.app.config, INDEX_PATTERN_ELASTICSEARCH);
-    if (ccs) {
-      esIndexPattern = getCcsIndexPattern(esIndexPattern, ccs);
-    }
+  protected async fetchClusters(esClient: ElasticsearchClient, params: CommonAlertParams) {
+    const indexPatterns = getNewIndexPatterns({
+      config: Globals.app.config,
+      moduleType: 'elasticsearch',
+    });
     if (!params.limit) {
-      return await fetchClusters(esClient, esIndexPattern);
+      return await fetchClusters(esClient, indexPatterns);
     }
     const limit = parseDuration(params.limit);
     const rangeFilter = this.ruleOptions.fetchClustersRange
@@ -253,14 +246,13 @@ export class BaseRule {
           },
         }
       : undefined;
-    return await fetchClusters(esClient, esIndexPattern, rangeFilter);
+    return await fetchClusters(esClient, indexPatterns, rangeFilter);
   }
 
   protected async fetchData(
     params: CommonAlertParams | unknown,
     esClient: ElasticsearchClient,
-    clusters: AlertCluster[],
-    availableCcs: boolean
+    clusters: AlertCluster[]
   ): Promise<Array<AlertData & unknown>> {
     throw new Error('Child classes must implement `fetchData`');
   }

x-pack/plugins/monitoring/server/alerts/ccr_read_exceptions_rule.ts

@@ -22,19 +22,14 @@ import {
   CCRReadExceptionsStats,
 } from '../../common/types/alerts';
 import { AlertInstance } from '../../../alerting/server';
-import {
-  INDEX_PATTERN_ELASTICSEARCH,
-  RULE_CCR_READ_EXCEPTIONS,
-  RULE_DETAILS,
-} from '../../common/constants';
+import { RULE_CCR_READ_EXCEPTIONS, RULE_DETAILS } from '../../common/constants';
 import { fetchCCRReadExceptions } from '../lib/alerts/fetch_ccr_read_exceptions';
-import { getCcsIndexPattern } from '../lib/alerts/get_ccs_index_pattern';
 import { AlertMessageTokenType, AlertSeverity } from '../../common/enums';
 import { parseDuration } from '../../../alerting/common/parse_duration';
 import { SanitizedAlert, RawAlertInstance } from '../../../alerting/common';
 import { AlertingDefaults, createLink } from './alert_helpers';
-import { appendMetricbeatIndex } from '../lib/alerts/append_mb_index';
 import { Globals } from '../static_globals';
+import { getNewIndexPatterns } from '../lib/cluster/get_index_patterns';
 
 export class CCRReadExceptionsRule extends BaseRule {
   constructor(public sanitizedRule?: SanitizedAlert) {
@@ -72,20 +67,20 @@ export class CCRReadExceptionsRule extends BaseRule {
   protected async fetchData(
     params: CommonAlertParams,
     esClient: ElasticsearchClient,
-    clusters: AlertCluster[],
-    availableCcs: boolean
+    clusters: AlertCluster[]
   ): Promise<AlertData[]> {
-    let esIndexPattern = appendMetricbeatIndex(Globals.app.config, INDEX_PATTERN_ELASTICSEARCH);
-    if (availableCcs) {
-      esIndexPattern = getCcsIndexPattern(esIndexPattern, availableCcs);
-    }
+    const indexPatterns = getNewIndexPatterns({
+      config: Globals.app.config,
+      moduleType: 'elasticsearch',
+      dataset: 'ccr',
+    });
     const { duration: durationString } = params;
     const duration = parseDuration(durationString);
     const endMs = +new Date();
     const startMs = endMs - duration;
     const stats = await fetchCCRReadExceptions(
       esClient,
-      esIndexPattern,
+      indexPatterns,
       startMs,
       endMs,
       Globals.app.config.ui.max_bucket_size,

x-pack/plugins/monitoring/server/alerts/cluster_health_rule.ts

@@ -19,18 +19,13 @@ import {
   AlertInstanceState,
 } from '../../common/types/alerts';
 import { AlertInstance } from '../../../alerting/server';
-import {
-  RULE_CLUSTER_HEALTH,
-  LEGACY_RULE_DETAILS,
-  INDEX_PATTERN_ELASTICSEARCH,
-} from '../../common/constants';
+import { RULE_CLUSTER_HEALTH, LEGACY_RULE_DETAILS } from '../../common/constants';
 import { AlertMessageTokenType, AlertClusterHealthType, AlertSeverity } from '../../common/enums';
 import { AlertingDefaults } from './alert_helpers';
 import { SanitizedAlert } from '../../../alerting/common';
 import { Globals } from '../static_globals';
-import { getCcsIndexPattern } from '../lib/alerts/get_ccs_index_pattern';
-import { appendMetricbeatIndex } from '../lib/alerts/append_mb_index';
 import { fetchClusterHealth } from '../lib/alerts/fetch_cluster_health';
+import { getNewIndexPatterns } from '../lib/cluster/get_index_patterns';
 
 const RED_STATUS_MESSAGE = i18n.translate('xpack.monitoring.alerts.clusterHealth.redMessage', {
   defaultMessage: 'Allocate missing primary and replica shards',
@@ -66,19 +61,14 @@ export class ClusterHealthRule extends BaseRule {
   protected async fetchData(
     params: CommonAlertParams,
     esClient: ElasticsearchClient,
-    clusters: AlertCluster[],
-    availableCcs: boolean
+    clusters: AlertCluster[]
   ): Promise<AlertData[]> {
-    let esIndexPattern = appendMetricbeatIndex(Globals.app.config, INDEX_PATTERN_ELASTICSEARCH);
-    if (availableCcs) {
-      esIndexPattern = getCcsIndexPattern(esIndexPattern, availableCcs);
-    }
-    const healths = await fetchClusterHealth(
-      esClient,
-      clusters,
-      esIndexPattern,
-      params.filterQuery
-    );
+    const indexPatterns = getNewIndexPatterns({
+      config: Globals.app.config,
+      moduleType: 'elasticsearch',
+      dataset: 'cluster_stats',
+    });
+    const healths = await fetchClusterHealth(esClient, clusters, indexPatterns, params.filterQuery);
     return healths.map((clusterHealth) => {
       const shouldFire = clusterHealth.health !== AlertClusterHealthType.Green;
       const severity =

x-pack/plugins/monitoring/server/alerts/cpu_usage_rule.ts

@@ -23,17 +23,16 @@ import {
   CommonAlertFilter,
 } from '../../common/types/alerts';
 import { AlertInstance } from '../../../alerting/server';
-import { INDEX_PATTERN_ELASTICSEARCH, RULE_CPU_USAGE, RULE_DETAILS } from '../../common/constants';
+import { RULE_CPU_USAGE, RULE_DETAILS } from '../../common/constants';
 // @ts-ignore
 import { ROUNDED_FLOAT } from '../../common/formatting';
 import { fetchCpuUsageNodeStats } from '../lib/alerts/fetch_cpu_usage_node_stats';
-import { getCcsIndexPattern } from '../lib/alerts/get_ccs_index_pattern';
 import { AlertMessageTokenType, AlertSeverity } from '../../common/enums';
 import { RawAlertInstance, SanitizedAlert } from '../../../alerting/common';
 import { parseDuration } from '../../../alerting/common/parse_duration';
 import { AlertingDefaults, createLink } from './alert_helpers';
-import { appendMetricbeatIndex } from '../lib/alerts/append_mb_index';
 import { Globals } from '../static_globals';
+import { getNewIndexPatterns } from '../lib/cluster/get_index_patterns';
 
 export class CpuUsageRule extends BaseRule {
   constructor(public sanitizedRule?: SanitizedAlert) {
@@ -60,20 +59,20 @@ export class CpuUsageRule extends BaseRule {
   protected async fetchData(
     params: CommonAlertParams,
     esClient: ElasticsearchClient,
-    clusters: AlertCluster[],
-    availableCcs: boolean
+    clusters: AlertCluster[]
   ): Promise<AlertData[]> {
-    let esIndexPattern = appendMetricbeatIndex(Globals.app.config, INDEX_PATTERN_ELASTICSEARCH);
-    if (availableCcs) {
-      esIndexPattern = getCcsIndexPattern(esIndexPattern, availableCcs);
-    }
+    const indexPatterns = getNewIndexPatterns({
+      config: Globals.app.config,
+      moduleType: 'elasticsearch',
+      dataset: 'node_stats',
+    });
     const duration = parseDuration(params.duration);
     const endMs = +new Date();
     const startMs = endMs - duration;
     const stats = await fetchCpuUsageNodeStats(
       esClient,
       clusters,
-      esIndexPattern,
+      indexPatterns,
       startMs,
       endMs,
       Globals.app.config.ui.max_bucket_size,

x-pack/plugins/monitoring/server/alerts/disk_usage_rule.ts

@@ -23,16 +23,15 @@ import {
   CommonAlertFilter,
 } from '../../common/types/alerts';
 import { AlertInstance } from '../../../alerting/server';
-import { INDEX_PATTERN_ELASTICSEARCH, RULE_DISK_USAGE, RULE_DETAILS } from '../../common/constants';
+import { RULE_DISK_USAGE, RULE_DETAILS } from '../../common/constants';
 // @ts-ignore
 import { ROUNDED_FLOAT } from '../../common/formatting';
 import { fetchDiskUsageNodeStats } from '../lib/alerts/fetch_disk_usage_node_stats';
-import { getCcsIndexPattern } from '../lib/alerts/get_ccs_index_pattern';
 import { AlertMessageTokenType, AlertSeverity } from '../../common/enums';
 import { RawAlertInstance, SanitizedAlert } from '../../../alerting/common';
 import { AlertingDefaults, createLink } from './alert_helpers';
-import { appendMetricbeatIndex } from '../lib/alerts/append_mb_index';
 import { Globals } from '../static_globals';
+import { getNewIndexPatterns } from '../lib/cluster/get_index_patterns';
 
 export class DiskUsageRule extends BaseRule {
   constructor(public sanitizedRule?: SanitizedAlert) {
@@ -59,18 +58,18 @@ export class DiskUsageRule extends BaseRule {
   protected async fetchData(
     params: CommonAlertParams,
     esClient: ElasticsearchClient,
-    clusters: AlertCluster[],
-    availableCcs: boolean
+    clusters: AlertCluster[]
   ): Promise<AlertData[]> {
-    let esIndexPattern = appendMetricbeatIndex(Globals.app.config, INDEX_PATTERN_ELASTICSEARCH);
-    if (availableCcs) {
-      esIndexPattern = getCcsIndexPattern(esIndexPattern, availableCcs);
-    }
+    const indexPatterns = getNewIndexPatterns({
+      config: Globals.app.config,
+      moduleType: 'elasticsearch',
+      dataset: 'node_stats',
+    });
     const { duration, threshold } = params;
     const stats = await fetchDiskUsageNodeStats(
       esClient,
       clusters,
-      esIndexPattern,
+      indexPatterns,
       duration as string,
       Globals.app.config.ui.max_bucket_size,
       params.filterQuery