[Metricbeat] Remove strict parsing on RabbitMQ module (#30090)

elastic · Jan 31, 2022 · 16fded1 · 16fded1
1 parent ca0d305
commit 16fded1
Show file tree

Hide file tree

Showing 9 changed files with 30 additions and 73 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -198,6 +198,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add `elasticsearch.cluster.id` field to Logstash module. {pull}29625[29625]
 - Add `xpack.enabled` support for Enterprise Search module. {pull}29871[29871]
 - Add gcp firestore metricset. {pull}29918[29918] 
+- Remove strict parsing on RabbitMQ module {pull}30090[30090]
 
 *Packetbeat*
 

diff --git a/metricbeat/module/rabbitmq/connection/connection.go b/metricbeat/module/rabbitmq/connection/connection.go
@@ -53,5 +53,5 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 		return errors.Wrap(err, "error in fetch")
 	}
 
-	return eventsMapping(content, r, m)
+	return eventsMapping(content, r)
 }
diff --git a/metricbeat/module/rabbitmq/connection/data.go b/metricbeat/module/rabbitmq/connection/data.go
@@ -60,33 +60,22 @@ var (
 	}
 )
 
-func eventsMapping(content []byte, r mb.ReporterV2, m *MetricSet) error {
+func eventsMapping(content []byte, r mb.ReporterV2) error {
 	var connections []map[string]interface{}
 	err := json.Unmarshal(content, &connections)
 	if err != nil {
 		return errors.Wrap(err, "error in unmarshal")
 	}
 
 	for _, node := range connections {
-		evt, err := eventMapping(node)
-		if err != nil {
-			m.Logger().Errorf("error in mapping: %s", err)
-			r.Error(err)
-			continue
-		}
-
-		if !r.Event(evt) {
-			return nil
-		}
+		evt := eventMapping(node)
+		r.Event(evt)
 	}
 	return nil
 }
 
-func eventMapping(connection map[string]interface{}) (mb.Event, error) {
-	fields, err := schema.Apply(connection, s.FailOnRequired)
-	if err != nil {
-		return mb.Event{}, errors.Wrap(err, "error applying schema")
-	}
+func eventMapping(connection map[string]interface{}) mb.Event {
+	fields, _ := schema.Apply(connection, s.FailOnRequired)
 
 	rootFields := common.MapStr{}
 	if v, err := fields.GetValue("user"); err == nil {
@@ -110,5 +99,5 @@ func eventMapping(connection map[string]interface{}) (mb.Event, error) {
 		RootFields:      rootFields,
 		ModuleFields:    moduleFields,
 	}
-	return event, nil
+	return event
 }
diff --git a/metricbeat/module/rabbitmq/exchange/data.go b/metricbeat/module/rabbitmq/exchange/data.go
@@ -55,32 +55,22 @@ var (
 	}
 )
 
-func eventsMapping(content []byte, r mb.ReporterV2, m *MetricSet) error {
+func eventsMapping(content []byte, r mb.ReporterV2) error {
 	var exchanges []map[string]interface{}
 	err := json.Unmarshal(content, &exchanges)
 	if err != nil {
 		return errors.Wrap(err, "error in unmarshal")
 	}
 
 	for _, exchange := range exchanges {
-		evt, err := eventMapping(exchange)
-		if err != nil {
-			m.Logger().Errorf("error in mapping: %s", err)
-			r.Error(err)
-			continue
-		}
-		if !r.Event(evt) {
-			return nil
-		}
+		evt := eventMapping(exchange)
+		r.Event(evt)
 	}
 	return nil
 }
 
-func eventMapping(exchange map[string]interface{}) (mb.Event, error) {
-	fields, err := schema.Apply(exchange)
-	if err != nil {
-		return mb.Event{}, err
-	}
+func eventMapping(exchange map[string]interface{}) mb.Event {
+	fields, _ := schema.Apply(exchange)
 
 	rootFields := common.MapStr{}
 	if v, err := fields.GetValue("user"); err == nil {
@@ -99,6 +89,6 @@ func eventMapping(exchange map[string]interface{}) (mb.Event, error) {
 		RootFields:      rootFields,
 		ModuleFields:    moduleFields,
 	}
-	return event, nil
+	return event
 
 }
diff --git a/metricbeat/module/rabbitmq/exchange/exchange.go b/metricbeat/module/rabbitmq/exchange/exchange.go
@@ -55,5 +55,5 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 		return errors.Wrap(err, "error in fetch")
 	}
 
-	return eventsMapping(content, r, m)
+	return eventsMapping(content, r)
 }
diff --git a/metricbeat/module/rabbitmq/node/data.go b/metricbeat/module/rabbitmq/node/data.go
@@ -147,34 +147,24 @@ var (
 	}
 )
 
-func eventsMapping(r mb.ReporterV2, content []byte, m *ClusterMetricSet) error {
+func eventsMapping(r mb.ReporterV2, content []byte) error {
 	var nodes []map[string]interface{}
 	err := json.Unmarshal(content, &nodes)
 	if err != nil {
 		return errors.Wrap(err, "error in Unmarshal")
 	}
 
 	for _, node := range nodes {
-		evt, err := eventMapping(node)
-		if err != nil {
-			m.Logger().Errorf("error in mapping: %s", err)
-			r.Error(err)
-			continue
-		}
-		if !r.Event(evt) {
-			return nil
-		}
+		evt := eventMapping(node)
+		r.Event(evt)
 	}
 	return nil
 }
 
-func eventMapping(node map[string]interface{}) (mb.Event, error) {
-	event, err := schema.Apply(node)
-	if err != nil {
-		return mb.Event{}, errors.Wrap(err, "error applying schema")
-	}
+func eventMapping(node map[string]interface{}) mb.Event {
+	event, _ := schema.Apply(node)
 	return mb.Event{
 		MetricSetFields: event,
-	}, nil
+	}
 
 }
diff --git a/metricbeat/module/rabbitmq/node/node.go b/metricbeat/module/rabbitmq/node/node.go
@@ -105,10 +105,7 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 		return errors.Wrap(err, "error in fetch")
 	}
 
-	evt, err := eventMapping(content)
-	if err != nil {
-		return errors.Wrap(err, "error in mapping")
-	}
+	evt := eventMapping(content)
 	r.Event(evt)
 	return nil
 }
@@ -120,5 +117,5 @@ func (m *ClusterMetricSet) Fetch(r mb.ReporterV2) error {
 		return errors.Wrap(err, "error in fetch")
 	}
 
-	return eventsMapping(r, content, m)
+	return eventsMapping(r, content)
 }
diff --git a/metricbeat/module/rabbitmq/queue/data.go b/metricbeat/module/rabbitmq/queue/data.go
@@ -84,33 +84,23 @@ var (
 	}
 )
 
-func eventsMapping(content []byte, r mb.ReporterV2, m *MetricSet) error {
+func eventsMapping(content []byte, r mb.ReporterV2) error {
 	var queues []map[string]interface{}
 	err := json.Unmarshal(content, &queues)
 	if err != nil {
 		return errors.Wrap(err, "error in mapping")
 	}
 
 	for _, queue := range queues {
-		evt, err := eventMapping(queue)
-		if err != nil {
-			m.Logger().Errorf("error in mapping: %s", err)
-			r.Error(err)
-			continue
-		}
-		if !r.Event(evt) {
-			return nil
-		}
+		evt := eventMapping(queue)
+		r.Event(evt)
 	}
 
 	return nil
 }
 
-func eventMapping(queue map[string]interface{}) (mb.Event, error) {
-	fields, err := schema.Apply(queue)
-	if err != nil {
-		return mb.Event{}, errors.Wrap(err, "error applying schema")
-	}
+func eventMapping(queue map[string]interface{}) mb.Event {
+	fields, _ := schema.Apply(queue)
 
 	moduleFields := common.MapStr{}
 	if v, err := fields.GetValue("vhost"); err == nil {
@@ -127,5 +117,5 @@ func eventMapping(queue map[string]interface{}) (mb.Event, error) {
 		MetricSetFields: fields,
 		ModuleFields:    moduleFields,
 	}
-	return event, nil
+	return event
 }
diff --git a/metricbeat/module/rabbitmq/queue/queue.go b/metricbeat/module/rabbitmq/queue/queue.go
@@ -53,5 +53,5 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 		return errors.Wrap(err, "error in fetch")
 	}
 
-	return eventsMapping(content, r, m)
+	return eventsMapping(content, r)
 }