[Monitoring] De-duplicate pipeline ids based on the ephemeral_id changing (#49978)

* De-duplicate pipeline ids based on the ephemeral_id changing

* Add tests

Author: chrisronline

x-pack/legacy/plugins/monitoring/server/lib/logstash/get_pipeline_ids.js

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 import moment from 'moment';
-import { get } from 'lodash';
+import { get, uniq } from 'lodash';
 import { createQuery } from '../create_query';
 import { LogstashMetric } from '../metrics';
 
@@ -74,5 +74,6 @@ export async function getLogstashPipelineIds(req, logstashIndexPattern, { cluste
 
   const { callWithRequest } = req.server.plugins.elasticsearch.getCluster('monitoring');
   const response = await callWithRequest(req, 'search', params);
-  return get(response, 'aggregations.nested_context.composite_data.buckets', []).map(bucket => bucket.key);
+  const data = get(response, 'aggregations.nested_context.composite_data.buckets', []).map(bucket => bucket.key);
+  return uniq(data, item => item.id);
 }

x-pack/test/api_integration/apis/monitoring/logstash/fixtures/pipelines.json

@@ -0,0 +1 @@
+{"pipelines":[{"id":"eight","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"eighteen","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"eleven","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"fifteen","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"five","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"four","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"fourteen","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"nine","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"nineteen","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1},{"id":"one","metrics":{"throughput":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.pipelines.events.out","label":"Pipeline Throughput","description":"Number of events emitted per second by the Logstash pipeline at the outputs stage.","units":"e/s","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,0],[1572882260000,0],[1572882270000,0]]},"nodesCount":{"bucket_size":"10 seconds","timeRange":{"min":1572882044855,"max":1572882638667},"metric":{"app":"logstash","field":"logstash_stats.logstash.uuid","label":"Pipeline Node Count","description":"Number of nodes on which the Logstash pipeline is running.","units":"","format":"0,0.[00]","hasCalculation":true,"isDerivative":false},"data":[[1572882220000,1],[1572882260000,1],[1572882270000,1]]}},"latestThroughput":0,"latestNodesCount":1}],"clusterStatus":{"node_count":1,"events_in_total":312,"events_out_total":234,"avg_memory":1037959168,"avg_memory_used":205063840,"max_uptime":40598,"pipeline_count":26,"queue_types":{"memory":26,"persisted":0},"versions":["8.0.0"]},"totalPipelineCount":26}

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

@@ -10,6 +10,7 @@ export default function ({ loadTestFile }) {
     loadTestFile(require.resolve('./nodes'));
     loadTestFile(require.resolve('./node_detail'));
     loadTestFile(require.resolve('./multicluster_pipelines'));
+    loadTestFile(require.resolve('./pipelines'));
   });
 }
 

x-pack/test/api_integration/apis/monitoring/logstash/pipelines.js

@@ -0,0 +1,66 @@
+/*
+ * 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 '@kbn/expect';
+import pipelinesFixture from './fixtures/pipelines';
+
+export default function ({ getService }) {
+  const supertest = getService('supertest');
+  const esArchiver = getService('esArchiver');
+
+  describe('pipelines', () => {
+    const archive = 'monitoring/logstash/changing_pipelines';
+    const timeRange = {
+      min: '2019-11-04T15:40:44.855Z',
+      max: '2019-11-04T15:50:38.667Z'
+    };
+    const pagination = {
+      size: 10,
+      index: 0
+    };
+    const sort = {
+      field: 'id',
+      direction: 'asc'
+    };
+
+    before('load archive', () => {
+      return esArchiver.load(archive);
+    });
+
+    after('unload archive', () => {
+      return esArchiver.unload(archive);
+    });
+
+    it('should return paginated pipelines', async () => {
+      const { body } = await supertest
+        .post('/api/monitoring/v1/clusters/TUjQLdHNTh2SB9Wy0gOtWg/logstash/pipelines')
+        .set('kbn-xsrf', 'xxx')
+        .send({ timeRange, pagination, sort })
+        .expect(200);
+
+      expect(body).to.eql(pipelinesFixture);
+    });
+
+    it('should get one of each after enough pagination', async () => {
+      async function getIds(page) {
+        const { body } = await supertest
+          .post('/api/monitoring/v1/clusters/TUjQLdHNTh2SB9Wy0gOtWg/logstash/pipelines')
+          .set('kbn-xsrf', 'xxx')
+          .send({ timeRange, pagination: { ...pagination, index: page }, sort })
+          .expect(200);
+
+        return body.pipelines.map(pipeline => pipeline.id);
+      }
+
+      const ids = [
+        ...await getIds(0),
+        ...await getIds(1),
+        ...await getIds(2),
+      ];
+      expect(ids.length).to.be(26);
+    });
+  });
+}