[Monitoring] Fix Node Advanced page (#19740) (#19749)

* [Monitoring] Fix Node Advanced page

* fix metrics snapshot

x-pack/plugins/monitoring/server/lib/details/get_series.js

@@ -216,6 +216,9 @@ export async function getSeries(req, indexPattern, metricName, filters, { min, m
   checkParam(indexPattern, 'indexPattern in details/getSeries');
 
   const metric = metrics[metricName];
+  if (!metric) {
+    throw new Error(`Not a valid metric: ${metricName}`);
+  }
   const response = await fetchSeries(req, indexPattern, metric, min, max, bucketSize, filters);
 
   return handleSeries(metric, min, max, bucketSize, response);

x-pack/plugins/monitoring/server/lib/metrics/__test__/__snapshots__/metrics.test.js.snap

@@ -2863,36 +2863,6 @@ Object {
     "units": "B",
     "uuidField": "source_node.uuid",
   },
-  "node_index_threads_bulk_queue": ElasticsearchMetric {
-    "app": "elasticsearch",
-    "derivative": true,
-    "description": "Number of bulk operations in the queue.",
-    "field": "node_stats.thread_pool.bulk.queue",
-    "format": "0,0.[00]",
-    "label": "Bulk Queue",
-    "metricAgg": "max",
-    "min": 0,
-    "timestampField": "timestamp",
-    "title": "Indexing Threads",
-    "type": "node",
-    "units": "",
-    "uuidField": "source_node.uuid",
-  },
-  "node_index_threads_bulk_rejected": ElasticsearchMetric {
-    "app": "elasticsearch",
-    "derivative": true,
-    "description": "Number of bulk operations that have been rejected, which occurs when the queue is full.",
-    "field": "node_stats.thread_pool.bulk.rejected",
-    "format": "0,0.[00]",
-    "label": "Bulk Rejections",
-    "metricAgg": "max",
-    "min": 0,
-    "timestampField": "timestamp",
-    "title": "Indexing Threads",
-    "type": "node",
-    "units": "",
-    "uuidField": "source_node.uuid",
-  },
   "node_index_threads_get_queue": ElasticsearchMetric {
     "app": "elasticsearch",
     "derivative": true,
@@ -2923,36 +2893,6 @@ Object {
     "units": "",
     "uuidField": "source_node.uuid",
   },
-  "node_index_threads_index_queue": ElasticsearchMetric {
-    "app": "elasticsearch",
-    "derivative": true,
-    "description": "Number of non-bulk, index operations in the queue.",
-    "field": "node_stats.thread_pool.index.queue",
-    "format": "0,0.[00]",
-    "label": "Index Queue",
-    "metricAgg": "max",
-    "min": 0,
-    "timestampField": "timestamp",
-    "title": "Indexing Threads",
-    "type": "node",
-    "units": "",
-    "uuidField": "source_node.uuid",
-  },
-  "node_index_threads_index_rejected": ElasticsearchMetric {
-    "app": "elasticsearch",
-    "derivative": true,
-    "description": "Number of non-bulk, index operations that have been rejected, which occurs when the queue is full. Generally indicates that bulk should be used.",
-    "field": "node_stats.thread_pool.index.rejected",
-    "format": "0,0.[00]",
-    "label": "Index Rejections",
-    "metricAgg": "max",
-    "min": 0,
-    "timestampField": "timestamp",
-    "title": "Indexing Threads",
-    "type": "node",
-    "units": "",
-    "uuidField": "source_node.uuid",
-  },
   "node_index_threads_search_queue": ElasticsearchMetric {
     "app": "elasticsearch",
     "derivative": true,
@@ -2983,6 +2923,91 @@ Object {
     "units": "",
     "uuidField": "source_node.uuid",
   },
+  "node_index_threads_write_queue": WriteThreadPoolQueueMetric {
+    "app": "elasticsearch",
+    "calculation": [Function],
+    "dateHistogramSubAggs": Object {
+      "bulk": Object {
+        "max": Object {
+          "field": "node_stats.thread_pool.bulk.queue",
+        },
+      },
+      "index": Object {
+        "max": Object {
+          "field": "node_stats.thread_pool.index.queue",
+        },
+      },
+      "write": Object {
+        "max": Object {
+          "field": "node_stats.thread_pool.write.queue",
+        },
+      },
+    },
+    "derivative": false,
+    "description": "Number of index, bulk, and write operations in the queue. The bulk threadpool was renamed to write in 6.3, and the index threadpool is deprecated.",
+    "field": "node_stats.thread_pool.write.queue",
+    "format": "0.[00]",
+    "label": "Write Queue",
+    "metricAgg": "max",
+    "timestampField": "timestamp",
+    "title": "Indexing Threads",
+    "type": "node",
+    "units": "",
+    "uuidField": "source_node.uuid",
+  },
+  "node_index_threads_write_rejected": WriteThreadPoolRejectedMetric {
+    "app": "elasticsearch",
+    "calculation": [Function],
+    "dateHistogramSubAggs": Object {
+      "bulk_deriv": Object {
+        "derivative": Object {
+          "buckets_path": "bulk_rejections",
+          "gap_policy": "skip",
+          "unit": "1s",
+        },
+      },
+      "bulk_rejections": Object {
+        "max": Object {
+          "field": "node_stats.thread_pool.bulk.rejected",
+        },
+      },
+      "index_deriv": Object {
+        "derivative": Object {
+          "buckets_path": "index_rejections",
+          "gap_policy": "skip",
+          "unit": "1s",
+        },
+      },
+      "index_rejections": Object {
+        "max": Object {
+          "field": "node_stats.thread_pool.index.rejected",
+        },
+      },
+      "write_deriv": Object {
+        "derivative": Object {
+          "buckets_path": "write_rejections",
+          "gap_policy": "skip",
+          "unit": "1s",
+        },
+      },
+      "write_rejections": Object {
+        "max": Object {
+          "field": "node_stats.thread_pool.write.rejected",
+        },
+      },
+    },
+    "derivative": false,
+    "description": "Number of index, bulk, and write operations that have been rejected, which occurs when the queue is full. The bulk threadpool was renamed to write in 6.3, and the index threadpool is deprecated.",
+    "field": "node_stats.thread_pool.write.rejected",
+    "format": "0.[00]",
+    "label": "Write Rejections",
+    "metricAgg": "max",
+    "timestampField": "timestamp",
+    "title": "Indexing Threads",
+    "type": "node",
+    "units": "",
+    "uuidField": "source_node.uuid",
+  },
   "node_index_time": ElasticsearchMetric {
     "app": "elasticsearch",
     "derivative": true,

x-pack/plugins/monitoring/server/lib/metrics/elasticsearch/classes.js

@@ -194,3 +194,102 @@ export class SingleIndexMemoryMetric extends IndexMemoryMetric {
     this.field = 'index_stats.total.segments.' + opts.field;
   }
 }
+
+export class WriteThreadPoolQueueMetric extends ElasticsearchMetric {
+  constructor(opts) {
+    super({
+      ...opts,
+      field: 'node_stats.thread_pool.write.queue', // in 7.0, we can only check for this threadpool
+      type: 'node',
+      format: SMALL_FLOAT,
+      metricAgg: 'max',
+      units: '',
+    });
+
+    this.dateHistogramSubAggs = {
+      index: {
+        max: { field: 'node_stats.thread_pool.index.queue' },
+      },
+      bulk: {
+        max: { field: 'node_stats.thread_pool.bulk.queue' },
+      },
+      write: {
+        max: { field: 'node_stats.thread_pool.write.queue' },
+      },
+    };
+
+    this.calculation = bucket => {
+      const index = _.get(bucket, 'index.value', null);
+      const bulk = _.get(bucket, 'bulk.value', null);
+      const write = _.get(bucket, 'write.value', null);
+
+      if (index !== null || bulk !== null || write !== null) {
+        return (index || 0) + (bulk || 0) + (write || 0);
+      }
+
+      // ignore the data if none of them exist
+      return null;
+    };
+  }
+}
+
+export class WriteThreadPoolRejectedMetric extends ElasticsearchMetric {
+  constructor(opts) {
+    super({
+      ...opts,
+      field: 'node_stats.thread_pool.write.rejected', // in 7.0, we can only check for this threadpool
+      type: 'node',
+      format: SMALL_FLOAT,
+      metricAgg: 'max',
+      units: '',
+    });
+
+    this.dateHistogramSubAggs = {
+      index_rejections: {
+        max: { field: 'node_stats.thread_pool.index.rejected' },
+      },
+      bulk_rejections: {
+        max: { field: 'node_stats.thread_pool.bulk.rejected' },
+      },
+      write_rejections: {
+        max: { field: 'node_stats.thread_pool.write.rejected' },
+      },
+      index_deriv: {
+        derivative: {
+          buckets_path: 'index_rejections',
+          gap_policy: 'skip',
+          unit: NORMALIZED_DERIVATIVE_UNIT,
+        },
+      },
+      bulk_deriv: {
+        derivative: {
+          buckets_path: 'bulk_rejections',
+          gap_policy: 'skip',
+          unit: NORMALIZED_DERIVATIVE_UNIT,
+        },
+      },
+      write_deriv: {
+        derivative: {
+          buckets_path: 'write_rejections',
+          gap_policy: 'skip',
+          unit: NORMALIZED_DERIVATIVE_UNIT,
+        },
+      },
+    };
+
+    this.calculation = bucket => {
+      const index = _.get(bucket, 'index_deriv.normalized_value', null);
+      const bulk = _.get(bucket, 'bulk_deriv.normalized_value', null);
+      const write = _.get(bucket, 'write_deriv.normalized_value', null);
+
+      if (index !== null || bulk !== null || write !== null) {
+        const valueOrZero = value => (value < 0 ? 0 : value || 0);
+
+        return valueOrZero(index) + valueOrZero(bulk) + valueOrZero(write);
+      }
+
+      // ignore the data if none of them exist
+      return null;
+    };
+  }
+}

x-pack/plugins/monitoring/server/lib/metrics/elasticsearch/metrics.js

@@ -13,7 +13,9 @@ import {
   IndexMemoryMetric,
   NodeIndexMemoryMetric,
   ThreadPoolQueueMetric,
-  ThreadPoolRejectedMetric
+  ThreadPoolRejectedMetric,
+  WriteThreadPoolQueueMetric,
+  WriteThreadPoolRejectedMetric
 } from './classes';
 import {
   LARGE_FLOAT,
@@ -691,31 +693,6 @@ export const metrics = {
     description:
       'Heap memory used by the Index Writer. This is NOT a part of Lucene Total.'
   }),
-  node_index_threads_bulk_queue: new ElasticsearchMetric({
-    field: 'node_stats.thread_pool.bulk.queue',
-    title: 'Indexing Threads',
-    label: 'Bulk Queue',
-    description: 'Number of bulk operations in the queue.',
-    type: 'node',
-    derivative: true,
-    format: LARGE_FLOAT,
-    metricAgg: 'max',
-    units: '',
-    min: 0
-  }),
-  node_index_threads_bulk_rejected: new ElasticsearchMetric({
-    field: 'node_stats.thread_pool.bulk.rejected',
-    title: 'Indexing Threads',
-    label: 'Bulk Rejections',
-    description:
-      'Number of bulk operations that have been rejected, which occurs when the queue is full.',
-    type: 'node',
-    derivative: true,
-    format: LARGE_FLOAT,
-    metricAgg: 'max',
-    units: '',
-    min: 0
-  }),
   node_index_threads_get_queue: new ElasticsearchMetric({
     field: 'node_stats.thread_pool.get.queue',
     title: 'Read Threads',
@@ -741,31 +718,22 @@ export const metrics = {
     units: '',
     min: 0
   }),
-  node_index_threads_index_queue: new ElasticsearchMetric({
-    field: 'node_stats.thread_pool.index.queue',
+  node_index_threads_write_queue: new WriteThreadPoolQueueMetric({
     title: 'Indexing Threads',
-    label: 'Index Queue',
-    description: 'Number of non-bulk, index operations in the queue.',
-    type: 'node',
-    derivative: true,
-    format: LARGE_FLOAT,
-    metricAgg: 'max',
-    units: '',
-    min: 0
+    label: 'Write Queue',
+    description: (
+      'Number of index, bulk, and write operations in the queue. ' +
+      'The bulk threadpool was renamed to write in 6.3, and the index threadpool is deprecated.'
+    ),
   }),
-  node_index_threads_index_rejected: new ElasticsearchMetric({
-    field: 'node_stats.thread_pool.index.rejected',
+  node_index_threads_write_rejected: new WriteThreadPoolRejectedMetric({
+    field: 'node_stats.thread_pool.bulk.rejected',
     title: 'Indexing Threads',
-    label: 'Index Rejections',
-    description:
-      'Number of non-bulk, index operations that have been rejected, which occurs when the queue is full. ' +
-      'Generally indicates that bulk should be used.',
-    type: 'node',
-    derivative: true,
-    format: LARGE_FLOAT,
-    metricAgg: 'max',
-    units: '',
-    min: 0
+    label: 'Write Rejections',
+    description: (
+      'Number of index, bulk, and write operations that have been rejected, which occurs when the queue is full. ' +
+      'The bulk threadpool was renamed to write in 6.3, and the index threadpool is deprecated.'
+    ),
   }),
   node_index_threads_search_queue: new ElasticsearchMetric({
     field: 'node_stats.thread_pool.search.queue',

x-pack/test/api_integration/apis/monitoring/elasticsearch/index.js

@@ -9,6 +9,7 @@ export default function ({ loadTestFile }) {
     loadTestFile(require.resolve('./overview'));
     loadTestFile(require.resolve('./nodes'));
     loadTestFile(require.resolve('./node_detail'));
+    loadTestFile(require.resolve('./node_detail_advanced'));
     loadTestFile(require.resolve('./indices'));
     loadTestFile(require.resolve('./index_detail'));
   });

x-pack/test/api_integration/apis/monitoring/elasticsearch/node_detail_advanced.js

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import expect from 'expect.js';
+import nodeDetailFixture from './fixtures/node_detail_advanced';
+
+export default function ({ getService }) {
+  const supertest = getService('supertest');
+  const esArchiver = getService('esArchiver');
+
+  describe('node detail advanced', () => {
+    const archive = 'monitoring/singlecluster-three-nodes-shard-relocation';
+    const timeRange = {
+      min: '2017-10-05T20:31:48.000Z',
+      max: '2017-10-05T20:35:12.000Z'
+    };
+
+    before('load archive', () => {
+      return esArchiver.load(archive);
+    });
+
+    after('unload archive', () => {
+      return esArchiver.unload(archive);
+    });
+
+    it('should summarize node with metrics', async () => {
+      const { body } = await supertest
+        .post('/api/monitoring/v1/clusters/YCxj-RAgSZCP6GuOQ8M1EQ/elasticsearch/nodes/jUT5KdxfRbORSCWkb5zjmA')
+        .set('kbn-xsrf', 'xxx')
+        .send({
+          timeRange,
+          is_advanced: true
+        })
+        .expect(200);
+
+      console.log(JSON.stringify(body));
+      expect(body).to.eql(nodeDetailFixture);
+    });
+  });
+}