From 2831eb8aa2e0149e7246605b7d8e45cdfc535f92 Mon Sep 17 00:00:00 2001 From: Shaunak Kashyap Date: Tue, 1 Dec 2020 21:12:21 +0530 Subject: [PATCH] Cherry-pick to 7.10: Fixing logic to keep list of unique cluster UUIDs (#22808) (#22817) * Fixing logic to keep list of unique cluster UUIDs (#22808) * Fixing logic to keep list of unique cluster UUIDs * Adding CHANGELOG entry * Use common.StringSet * Adding more test cases * Adding back logic to broadly override cluster UUID for all pipelines, if set * Removing ToSlice() * Fixing loop * Fixing up CHANGELOG --- CHANGELOG.next.asciidoc | 1 + metricbeat/module/logstash/logstash.go | 10 +- metricbeat/module/logstash/node/data_xpack.go | 13 +- .../module/logstash/node/data_xpack_test.go | 328 ++++++++++++++++++ .../module/logstash/node_stats/data_xpack.go | 8 +- .../logstash/node_stats/data_xpack_test.go | 273 +++++++++++++++ 6 files changed, 620 insertions(+), 13 deletions(-) create mode 100644 metricbeat/module/logstash/node/data_xpack_test.go create mode 100644 metricbeat/module/logstash/node_stats/data_xpack_test.go diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index cdd5a87b34a..c9920929468 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -283,6 +283,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d - Revert change to report `process.memory.rss` as `process.memory.wss` on Windows. {pull}22055[22055] - Remove io.time from windows {pull}22237[22237] - Stop generating NaN values from Cloud Foundry module to avoid errors in outputs. {pull}22634[22634] +- Fix `logstash` module when `xpack.enabled: true` is set from emitting redundant events. {pull}22808[22808] *Packetbeat* diff --git a/metricbeat/module/logstash/logstash.go b/metricbeat/module/logstash/logstash.go index fbc030fa310..abd737f3ed4 100644 --- a/metricbeat/module/logstash/logstash.go +++ b/metricbeat/module/logstash/logstash.go @@ -57,13 +57,13 @@ type MetricSet struct { XPack bool } -type graph struct { +type Graph struct { Vertices []map[string]interface{} `json:"vertices"` Edges []map[string]interface{} `json:"edges"` } -type graphContainer struct { - Graph *graph `json:"graph,omitempty"` +type GraphContainer struct { + Graph *Graph `json:"graph,omitempty"` Type string `json:"type"` Version string `json:"version"` Hash string `json:"hash"` @@ -74,8 +74,8 @@ type PipelineState struct { ID string `json:"id"` Hash string `json:"hash"` EphemeralID string `json:"ephemeral_id"` - Graph *graphContainer `json:"graph,omitempty"` - Representation *graphContainer `json:"representation"` + Graph *GraphContainer `json:"graph,omitempty"` + Representation *GraphContainer `json:"representation"` BatchSize int `json:"batch_size"` Workers int `json:"workers"` } diff --git a/metricbeat/module/logstash/node/data_xpack.go b/metricbeat/module/logstash/node/data_xpack.go index 96fc252158c..66d3623c7de 100644 --- a/metricbeat/module/logstash/node/data_xpack.go +++ b/metricbeat/module/logstash/node/data_xpack.go @@ -66,21 +66,26 @@ func makeClusterToPipelinesMap(pipelines []logstash.PipelineState, overrideClust var clusterToPipelinesMap map[string][]logstash.PipelineState clusterToPipelinesMap = make(map[string][]logstash.PipelineState) + if overrideClusterUUID != "" { + clusterToPipelinesMap[overrideClusterUUID] = pipelines + return clusterToPipelinesMap + } + for _, pipeline := range pipelines { - var clusterUUIDs []string + clusterUUIDs := common.StringSet{} for _, vertex := range pipeline.Graph.Graph.Vertices { clusterUUID := logstash.GetVertexClusterUUID(vertex, overrideClusterUUID) if clusterUUID != "" { - clusterUUIDs = append(clusterUUIDs, clusterUUID) + clusterUUIDs.Add(clusterUUID) } } // If no cluster UUID was found in this pipeline, assign it a blank one if len(clusterUUIDs) == 0 { - clusterUUIDs = []string{""} + clusterUUIDs.Add("") } - for _, clusterUUID := range clusterUUIDs { + for clusterUUID := range clusterUUIDs { clusterPipelines := clusterToPipelinesMap[clusterUUID] if clusterPipelines == nil { clusterToPipelinesMap[clusterUUID] = []logstash.PipelineState{} diff --git a/metricbeat/module/logstash/node/data_xpack_test.go b/metricbeat/module/logstash/node/data_xpack_test.go new file mode 100644 index 00000000000..17ae0aaaf91 --- /dev/null +++ b/metricbeat/module/logstash/node/data_xpack_test.go @@ -0,0 +1,328 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// +build !integration + +package node + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/elastic/beats/v7/metricbeat/module/logstash" +) + +func TestMakeClusterToPipelinesMap(t *testing.T) { + tests := map[string]struct { + pipelines []logstash.PipelineState + overrideClusterUUID string + expectedMap map[string][]logstash.PipelineState + }{ + "no_vertex_cluster_id": { + pipelines: []logstash.PipelineState{ + { + ID: "test_pipeline", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + }, + }, + overrideClusterUUID: "prod_cluster_id", + expectedMap: map[string][]logstash.PipelineState{ + "prod_cluster_id": { + { + ID: "test_pipeline", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + }, + }, + }, + }, + "one_vertex_cluster_id": { + pipelines: []logstash.PipelineState{ + { + ID: "test_pipeline", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + }, + }, + overrideClusterUUID: "prod_cluster_id", + expectedMap: map[string][]logstash.PipelineState{ + "prod_cluster_id": { + { + ID: "test_pipeline", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + }, + }, + }, + }, + "two_pipelines": { + pipelines: []logstash.PipelineState{ + { + ID: "test_pipeline_1", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + }, + }, + { + ID: "test_pipeline_2", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + }, + }, + overrideClusterUUID: "prod_cluster_id", + expectedMap: map[string][]logstash.PipelineState{ + "prod_cluster_id": { + { + ID: "test_pipeline_1", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + }, + }, + { + ID: "test_pipeline_2", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + }, + }, + }, + }, + "no_override_cluster_id": { + pipelines: []logstash.PipelineState{ + { + ID: "test_pipeline_1", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + "cluster_uuid": "es_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + }, + }, + { + ID: "test_pipeline_2", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + }, + }, + overrideClusterUUID: "", + expectedMap: map[string][]logstash.PipelineState{ + "es_1": { + { + ID: "test_pipeline_1", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + "cluster_uuid": "es_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + }, + }, + }, + "es_2": { + { + ID: "test_pipeline_1", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + "cluster_uuid": "es_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + }, + }, + }, + "": { + { + ID: "test_pipeline_2", + Graph: &logstash.GraphContainer{ + Graph: &logstash.Graph{ + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + }, + }, + }, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + actualMap := makeClusterToPipelinesMap(test.pipelines, test.overrideClusterUUID) + require.Equal(t, test.expectedMap, actualMap) + }) + } +} diff --git a/metricbeat/module/logstash/node_stats/data_xpack.go b/metricbeat/module/logstash/node_stats/data_xpack.go index a6a9867b7cd..e5d82365b53 100644 --- a/metricbeat/module/logstash/node_stats/data_xpack.go +++ b/metricbeat/module/logstash/node_stats/data_xpack.go @@ -219,20 +219,20 @@ func makeClusterToPipelinesMap(pipelines []PipelineStats, overrideClusterUUID st } for _, pipeline := range pipelines { - var clusterUUIDs []string + clusterUUIDs := common.StringSet{} for _, vertex := range pipeline.Vertices { clusterUUID := logstash.GetVertexClusterUUID(vertex, overrideClusterUUID) if clusterUUID != "" { - clusterUUIDs = append(clusterUUIDs, clusterUUID) + clusterUUIDs.Add(clusterUUID) } } // If no cluster UUID was found in this pipeline, assign it a blank one if len(clusterUUIDs) == 0 { - clusterUUIDs = []string{""} + clusterUUIDs.Add("") } - for _, clusterUUID := range clusterUUIDs { + for clusterUUID := range clusterUUIDs { clusterPipelines := clusterToPipelinesMap[clusterUUID] if clusterPipelines == nil { clusterToPipelinesMap[clusterUUID] = []PipelineStats{} diff --git a/metricbeat/module/logstash/node_stats/data_xpack_test.go b/metricbeat/module/logstash/node_stats/data_xpack_test.go new file mode 100644 index 00000000000..6593be72534 --- /dev/null +++ b/metricbeat/module/logstash/node_stats/data_xpack_test.go @@ -0,0 +1,273 @@ +// Licensed to Elasticsearch B.V. under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Elasticsearch B.V. licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// +build !integration + +package node_stats + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestMakeClusterToPipelinesMap(t *testing.T) { + tests := map[string]struct { + pipelines []PipelineStats + overrideClusterUUID string + expectedMap map[string][]PipelineStats + }{ + "no_vertex_cluster_id": { + pipelines: []PipelineStats{ + { + ID: "test_pipeline", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + overrideClusterUUID: "prod_cluster_id", + expectedMap: map[string][]PipelineStats{ + "prod_cluster_id": { + { + ID: "test_pipeline", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + }, + }, + "one_vertex_cluster_id": { + pipelines: []PipelineStats{ + { + ID: "test_pipeline", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + overrideClusterUUID: "prod_cluster_id", + expectedMap: map[string][]PipelineStats{ + "prod_cluster_id": { + { + ID: "test_pipeline", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_2", + }, + { + "id": "vertex_3", + }, + }, + }, + }, + }, + }, + "two_pipelines": { + pipelines: []PipelineStats{ + { + ID: "test_pipeline_1", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + { + ID: "test_pipeline_2", + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + overrideClusterUUID: "prod_cluster_id", + expectedMap: map[string][]PipelineStats{ + "prod_cluster_id": { + { + ID: "test_pipeline_1", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + { + ID: "test_pipeline_2", + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + }, + }, + "no_override_cluster_id": { + pipelines: []PipelineStats{ + { + ID: "test_pipeline_1", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + "cluster_uuid": "es_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + { + ID: "test_pipeline_2", + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + expectedMap: map[string][]PipelineStats{ + "es_1": { + { + ID: "test_pipeline_1", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + "cluster_uuid": "es_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + }, + "es_2": { + { + ID: "test_pipeline_1", + Vertices: []map[string]interface{}{ + { + "id": "vertex_1_1", + "cluster_uuid": "es_1", + }, + { + "id": "vertex_1_2", + "cluster_uuid": "es_2", + }, + { + "id": "vertex_1_3", + }, + }, + }, + }, + "": { + { + ID: "test_pipeline_2", + Vertices: []map[string]interface{}{ + { + "id": "vertex_2_1", + }, + { + "id": "vertex_2_2", + }, + { + "id": "vertex_2_3", + }, + }, + }, + }, + }, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + actualMap := makeClusterToPipelinesMap(test.pipelines, test.overrideClusterUUID) + require.Equal(t, test.expectedMap, actualMap) + }) + } +}