[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 {
+  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

@@ -132,6 +132,9 @@ export const INDEX_PATTERN_ELASTICSEARCH = '.monitoring-es-*';
 // ECS-compliant patterns (metricbeat >8 and agent)
 export const INDEX_PATTERN_ELASTICSEARCH_ECS = '.monitoring-es-8-*';
 export const INDEX_PATTERN_ENTERPRISE_SEARCH = '.monitoring-ent-search-*';
+export const DS_INDEX_PATTERN_METRICS = 'metrics';
+export const DS_INDEX_PATTERN_LOGS = 'logs';
+export const DS_INDEX_PATTERN_ES = 'elasticsearch';
 
 // This is the unique token that exists in monitoring indices collected by metricbeat
 export const METRICBEAT_INDEX_NAME_UNIQUE_TOKEN = '-mb-';
@@ -586,3 +589,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,10 +28,7 @@ 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';
 
@@ -226,23 +223,14 @@ 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) {
     if (!params.limit) {
-      return await fetchClusters(esClient, esIndexPattern);
+      return await fetchClusters(esClient);
     }
     const limit = parseDuration(params.limit);
     const rangeFilter = this.ruleOptions.fetchClustersRange
@@ -253,14 +241,13 @@ export class BaseRule {
           },
         }
       : undefined;
-    return await fetchClusters(esClient, esIndexPattern, rangeFilter);
+    return await fetchClusters(esClient, 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.test.ts

@@ -39,7 +39,6 @@ jest.mock('../static_globals', () => ({
       config: {
         ui: {
           ccs: { enabled: true },
-          metricbeat: { index: 'metricbeat-*' },
           container: { elasticsearch: { enabled: false } },
         },
       },

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

@@ -22,18 +22,12 @@ 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';
 
 export class CCRReadExceptionsRule extends BaseRule {
@@ -72,20 +66,14 @@ 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 { duration: durationString } = params;
     const duration = parseDuration(durationString);
     const endMs = +new Date();
     const startMs = endMs - duration;
     const stats = await fetchCCRReadExceptions(
       esClient,
-      esIndexPattern,
       startMs,
       endMs,
       Globals.app.config.ui.max_bucket_size,

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

@@ -21,7 +21,6 @@ jest.mock('../static_globals', () => ({
       config: {
         ui: {
           ccs: { enabled: true },
-          metricbeat: { index: 'metricbeat-*' },
         },
       },
     },

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

@@ -19,17 +19,10 @@ 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';
 
 const RED_STATUS_MESSAGE = i18n.translate('xpack.monitoring.alerts.clusterHealth.redMessage', {
@@ -66,19 +59,9 @@ 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 healths = await fetchClusterHealth(esClient, clusters, params.filterQuery);
     return healths.map((clusterHealth) => {
       const shouldFire = clusterHealth.health !== AlertClusterHealthType.Green;
       const severity =

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

@@ -27,7 +27,6 @@ jest.mock('../static_globals', () => ({
       config: {
         ui: {
           ccs: { enabled: true },
-          metricbeat: { index: 'metricbeat-*' },
           container: { elasticsearch: { enabled: false } },
         },
       },

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

@@ -23,16 +23,14 @@ 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';
 
 export class CpuUsageRule extends BaseRule {
@@ -60,20 +58,14 @@ 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 duration = parseDuration(params.duration);
     const endMs = +new Date();
     const startMs = endMs - duration;
     const stats = await fetchCpuUsageNodeStats(
       esClient,
       clusters,
-      esIndexPattern,
       startMs,
       endMs,
       Globals.app.config.ui.max_bucket_size,

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

@@ -40,7 +40,6 @@ jest.mock('../static_globals', () => ({
       config: {
         ui: {
           ccs: { enabled: true },
-          metricbeat: { index: 'metricbeat-*' },
           container: { elasticsearch: { enabled: false } },
         },
       },

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

@@ -23,15 +23,13 @@ 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';
 
 export class DiskUsageRule extends BaseRule {
@@ -59,18 +57,12 @@ 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 { duration, threshold } = params;
     const stats = await fetchDiskUsageNodeStats(
       esClient,
       clusters,
-      esIndexPattern,
       duration as string,
       Globals.app.config.ui.max_bucket_size,
       params.filterQuery

x-pack/plugins/monitoring/server/alerts/elasticsearch_version_mismatch_rule.test.ts

@@ -28,7 +28,6 @@ jest.mock('../static_globals', () => ({
       config: {
         ui: {
           ccs: { enabled: true },
-          metricbeat: { index: 'metricbeat-*' },
           container: { elasticsearch: { enabled: false } },
         },
       },