Cherry-pick #10878 to 7.x: [Packetbeat] [MongoDB] Report unknown opcodes once

CHANGELOG.next.asciidoc

@@ -208,6 +208,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 - Fix DHCPv4 dashboard that wouldn't load in Kibana. {issue}9850[9850]
 - Fixed a crash when using af_packet capture {pull}10477[10477]
+- Avoid reporting unknown MongoDB opcodes more than once. {pull}10878[10878]
 
 *Winlogbeat*
 

packetbeat/protos/mongodb/mongodb_parser.go

@@ -21,13 +21,19 @@ import (
 	"encoding/json"
 	"errors"
 	"strings"
+	"sync"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
 
 	"gopkg.in/mgo.v2/bson"
 )
 
+var (
+	unknownOpcodes = map[opCode]struct{}{}
+	mutex          sync.Mutex
+)
+
 func mongodbMessageParser(s *stream) (bool, bool) {
 	d := newDecoder(s.data)
 
@@ -56,7 +62,12 @@ func mongodbMessageParser(s *stream) (bool, bool) {
 	opCode := opCode(code)
 
 	if !validOpcode(opCode) {
-		logp.Err("Unknown operation code: %v", opCode)
+		mutex.Lock()
+		defer mutex.Unlock()
+		if _, reported := unknownOpcodes[opCode]; !reported {
+			logp.Err("Unknown operation code: %v", opCode)
+			unknownOpcodes[opCode] = struct{}{}
+		}
 		return false, false
 	}
 

packetbeat/tests/system/test_0025_mongodb_basic.py

@@ -219,3 +219,15 @@ def test_request_after_reply(self):
         o = objs[0]
         assert o["type"] == "mongodb"
         assert o["event.duration"] >= 0
+
+    def test_unknown_opcode_flood(self):
+        """
+        Tests that a repeated unknown opcode is reported just once.
+        """
+        self.render_config_template(
+            mongodb_ports=[9991]
+        )
+        self.run_packetbeat(pcap="mongodb_op_msg_opcode.pcap",
+                            debug_selectors=["mongodb"])
+        num_msgs = self.log_contains_count('Unknown operation code: ')
+        assert num_msgs == 1, "Unknown opcode reported more than once: {0}".format(num_msgs)