Add option to ignore outgoing messages from the server.

If Ignore_outgoing is set to true, then the agent is not publishing
the messages that are originating from the server on which the agent
is installed. This can help with removing duplicates in cases where
there is communication between two servers running agents.

packetbeat.conf

@@ -52,6 +52,11 @@ device = "any"
 #
 #name=
 
+# Uncomment the following if you want to ignore transactions created
+# by the server on which the agent is installed. This option is useful
+# to remove duplicates if agents are installed on multiple servers.
+#ignore_outgoing = true
+
 [runoptions]
 # The Packetbeat agent can drop privileges after creating the sniffing
 # socket. Root access is required for opening the socket but everything

publish.go

@@ -29,6 +29,7 @@ var Publisher PublisherType
 type tomlAgent struct {
     Name                  string
     Refresh_topology_freq int
+    Ignore_outgoing       bool
 }
 type tomlMothership struct {
     Host string
@@ -102,12 +103,14 @@ func (publisher *PublisherType) PublishHttpTransaction(t *HttpTransaction) error
     src_server := publisher.GetServerName(t.Src.Ip)
     dst_server := publisher.GetServerName(t.Dst.Ip)
 
-    if dst_server != publisher.name {
+    if _Config.Agent.Ignore_outgoing && dst_server != "" &&
+            dst_server != publisher.name {
         // duplicated transaction -> ignore it
-        DEBUG("publish", "Ignore duplicated Http transaction on %s: %s -> %s", publisher.name, src_server, dst_server)
+        DEBUG("publish", "Ignore duplicated HTTP transaction on %s: %s -> %s", publisher.name, src_server, dst_server)
         return nil
     }
 
+
     var src_country = ""
     if _GeoLite != nil {
         if len(src_server) == 0 { // only for external IP addresses
@@ -150,6 +153,13 @@ func (publisher *PublisherType) PublishMysqlTransaction(t *MysqlTransaction) err
     src_server := publisher.GetServerName(t.Src.Ip)
     dst_server := publisher.GetServerName(t.Dst.Ip)
 
+    if _Config.Agent.Ignore_outgoing && dst_server != "" &&
+            dst_server != publisher.name {
+        // duplicated transaction -> ignore it
+        DEBUG("publish", "Ignore duplicated MySQL transaction on %s: %s -> %s", publisher.name, src_server, dst_server)
+        return nil
+    }
+
     event := Event{
         t.ts, "mysql", t.Src.Ip, t.Src.Port, t.Src.Proc, "", src_server,
         t.Dst.Ip, t.Dst.Port, t.Dst.Proc, dst_server,
@@ -179,6 +189,13 @@ func (publisher *PublisherType) PublishRedisTransaction(t *RedisTransaction) err
     src_server := publisher.GetServerName(t.Src.Ip)
     dst_server := publisher.GetServerName(t.Dst.Ip)
 
+    if _Config.Agent.Ignore_outgoing && dst_server != "" &&
+            dst_server != publisher.name {
+        // duplicated transaction -> ignore it
+        DEBUG("publish", "Ignore duplicated REDIS transaction on %s: %s -> %s", publisher.name, src_server, dst_server)
+        return nil
+    }
+
     event := Event{
         t.ts, "redis", t.Src.Ip, t.Src.Port, t.Src.Proc, "", src_server,
         t.Dst.Ip, t.Dst.Port, t.Dst.Proc, dst_server,