Merge pull request #1763 in SNORT/snort3 from ~MIALTIZE/snort3:daq_msg_event to master

Squashed commit of the following:

commit 2d87ba3bb1e5352e3a7a3f48692cb9d2f93e5c1f
Author: Michael Altizer <mialtize@cisco.com>
Date:   Tue Sep 24 08:22:43 2019 -0400

    pub_sub: Replace DaqMetaEvent and OtherMessageEvent with DaqMessageEvent

    Three events use this new shared event structure: DAQ_SOF_MSG_EVENT,
    DAQ_EOF_MSG_EVENT, and DAQ_OTHER_MSG_EVENT.

    Additionally, DAQ peg counts were added for SoF and EoF messages
    received.

diff --git a/src/flow/test/flow_stash_test.cc b/src/flow/test/flow_stash_test.cc
index 87543e2584207c559daf28549b8a9d55b4b3aa85..144a3489e5a8fde27a0de0dd9ad8ae80e156e076 100644 (file)
--- a/src/flow/test/flow_stash_test.cc
+++ b/src/flow/test/flow_stash_test.cc
@@ -109,7 +109,6 @@ void DataBus::publish(const char* key, DataEvent& e, Flow* f)
 
 void DataBus::publish(const char*, const uint8_t*, unsigned, Flow*) {}
 void DataBus::publish(const char*, Packet*, Flow*) {}
-void DataBus::publish(const char*, void*, int, const uint8_t*) {}
 
 void DataBus::_subscribe(const char* key, DataHandler* h)
 {

diff --git a/src/framework/data_bus.cc b/src/framework/data_bus.cc
index d948cb662e034fe7f7511c0843585b04f828b3aa..8885d3b5a5b15f0bff8c4169c1277eb1c76469f7 100644 (file)
--- a/src/framework/data_bus.cc
+++ b/src/framework/data_bus.cc
@@ -155,12 +155,6 @@ void DataBus::publish(const char* key, Packet* p, Flow* f)
     publish(key, e, f);
 }
 
-void DataBus::publish(const char* key, void* user, int type, const uint8_t* data)
-{
-    DaqMetaEvent e(user, type, data);
-    publish(key, e, nullptr);
-}
-
 //--------------------------------------------------------------------------
 // private methods
 //--------------------------------------------------------------------------

diff --git a/src/framework/data_bus.h b/src/framework/data_bus.h
index 47f66b67a59beb406f4a1daa2fdb8ec9164cea78..dbf5607c2b78f67687d17bf71e30968ccee1bef2 100644 (file)
--- a/src/framework/data_bus.h
+++ b/src/framework/data_bus.h
@@ -105,7 +105,6 @@ public:
     // convenience methods
     static void publish(const char* key, const uint8_t*, unsigned, Flow* = nullptr);
     static void publish(const char* key, Packet*, Flow* = nullptr);
-    static void publish(const char* key, void* user, int type, const uint8_t* data);
 
 private:
     void _subscribe(const char* key, DataHandler*);
@@ -116,28 +115,6 @@ private:
     DataMap map;
     DataModule mapped_module;
 };
-
-class SO_PUBLIC DaqMetaEvent : public DataEvent
-{
-public:
-    DaqMetaEvent(void* user, int type, const uint8_t *data) :
-        user(user), type(type), data(data)
-    { }
-
-    void* get_user_data()
-    { return user; }
-
-    int get_type()
-    { return type; }
-
-    const uint8_t* get_data() override
-    { return data; }
-
-private:
-    void* user;
-    int type;
-    const uint8_t* data;
-};
 }
 
 //
@@ -145,7 +122,6 @@ private:
 //
 
 #define PACKET_EVENT "detection.packet"
-#define DAQ_META_EVENT "daq.metapacket"
 #define FLOW_STATE_EVENT "flow.state_change"
 #define THREAD_IDLE_EVENT "thread.idle"
 #define THREAD_ROTATE_EVENT "thread.rotate"

diff --git a/src/loggers/log_hext.cc b/src/loggers/log_hext.cc
index b3c46c70d32c9320b4bc59b290c498dd793da183..cdd2394e8331ba5fabfd87393dc3ffe770eccac7 100644 (file)
--- a/src/loggers/log_hext.cc
+++ b/src/loggers/log_hext.cc
@@ -27,6 +27,7 @@
 #include "framework/module.h"
 #include "log/text_log.h"
 #include "protocols/packet.h"
+#include "pub_sub/daq_message_event.h"
 
 using namespace snort;
 using namespace std;
@@ -40,27 +41,33 @@ static THREAD_LOCAL TextLog* hext_log = nullptr;
 static THREAD_LOCAL unsigned s_pkt_num = 0;
 
 
-class DaqMetaEventHandler : public DataHandler
+class DaqMessageEventHandler : public DataHandler
 {
 public:
-    DaqMetaEventHandler() : DataHandler(S_NAME) { }
+    DaqMessageEventHandler() : DataHandler(S_NAME) { }
     void handle(DataEvent&, Flow*) override;
 };
 
-void DaqMetaEventHandler::handle(DataEvent& event, Flow*)
+void DaqMessageEventHandler::handle(DataEvent& event, Flow*)
 {
-    if (!hext_log) return;
+    if (!hext_log)
+        return;
 
-    DaqMetaEvent* ev = (DaqMetaEvent*)&event;
+    DaqMessageEvent* dme = (DaqMessageEvent*) &event;
 
     const char* cmd;
-    switch (ev->get_type()) {
-        case DAQ_MSG_TYPE_SOF: cmd = "sof"; break;
-        case DAQ_MSG_TYPE_EOF: cmd = "eof"; break;
-        default: return;
+    switch (dme->get_type()) {
+        case DAQ_MSG_TYPE_SOF:
+            cmd = "sof";
+            break;
+        case DAQ_MSG_TYPE_EOF:
+            cmd = "eof";
+            break;
+        default:
+            return;
     }
 
-    const Flow_Stats_t* fs = (const Flow_Stats_t*)ev->get_data();
+    const Flow_Stats_t* fs = (const Flow_Stats_t*) dme->get_header();
 
     SfIp src, dst;
     char shost[INET6_ADDRSTRLEN];
@@ -251,7 +258,8 @@ HextLogger::HextLogger(HextModule* m)
     limit = m->limit;
     width = m->width;
     raw = m->raw;
-    DataBus::subscribe(DAQ_META_EVENT, new DaqMetaEventHandler());
+    DataBus::subscribe(DAQ_SOF_MSG_EVENT, new DaqMessageEventHandler());
+    DataBus::subscribe(DAQ_EOF_MSG_EVENT, new DaqMessageEventHandler());
 }
 
 void HextLogger::open()

diff --git a/src/main/analyzer.cc b/src/main/analyzer.cc
index 32acc5c7ac35b7e7b155bd570d870aa09e8f34c5..9995c35027bd7cd99d3a8f89e33915e68dea44bc 100644 (file)
--- a/src/main/analyzer.cc
+++ b/src/main/analyzer.cc
@@ -58,8 +58,8 @@
 #include "packet_io/sfdaq_module.h"
 #include "packet_tracer/packet_tracer.h"
 #include "profiler/profiler.h"
+#include "pub_sub/daq_message_event.h"
 #include "pub_sub/finalize_packet_event.h"
-#include "pub_sub/other_message_event.h"
 #include "side_channel/side_channel.h"
 #include "stream/stream.h"
 #include "time/packet_time.h"
@@ -161,16 +161,26 @@ void Analyzer::set_main_hook(MainHook_f f)
 // message processing
 //-------------------------------------------------------------------------
 
-static void process_daq_sof_eof_msg(DAQ_Msg_h msg)
+static void process_daq_sof_eof_msg(DAQ_Msg_h msg, DAQ_Verdict& verdict)
 {
     const Flow_Stats_t *stats = (const Flow_Stats_t *) daq_msg_get_hdr(msg);
+    const char* key;
 
-    if (msg->type == DAQ_MSG_TYPE_EOF)
+    if (daq_msg_get_type(msg) == DAQ_MSG_TYPE_EOF)
+    {
         packet_time_update(&stats->eof_timestamp);
+        daq_stats.eof_messages++;
+        key = DAQ_EOF_MSG_EVENT;
+    }
     else
+    {
         packet_time_update(&stats->sof_timestamp);
+        daq_stats.sof_messages++;
+        key = DAQ_SOF_MSG_EVENT;
+    }
 
-    DataBus::publish(DAQ_META_EVENT, nullptr, daq_msg_get_type(msg), (const uint8_t*) stats);
+    DaqMessageEvent event(msg, verdict);
+    DataBus::publish(key, event);
 }
 
 static bool process_packet(Packet* p)
@@ -364,14 +374,13 @@ void Analyzer::process_daq_msg(DAQ_Msg_h msg, bool retry)
             return;
         case DAQ_MSG_TYPE_SOF:
         case DAQ_MSG_TYPE_EOF:
-            process_daq_sof_eof_msg(msg);
+            process_daq_sof_eof_msg(msg, verdict);
             break;
         default:
             {
-                OtherMessageEvent event(msg, verdict);
                 daq_stats.other_messages++;
-                // the verdict can be updated by event handler
-                DataBus::publish(OTHER_MESSAGE_EVENT, event);
+                DaqMessageEvent event(msg, verdict);
+                DataBus::publish(DAQ_OTHER_MSG_EVENT, event);
             }
             break;
     }

diff --git a/src/packet_io/sfdaq_module.cc b/src/packet_io/sfdaq_module.cc
index 09675ee7c34bde79e6435d125951d6ce85262776..ee3bbeabf21f3fedc34c4f52afc661bb8afdd33f 100644 (file)
--- a/src/packet_io/sfdaq_module.cc
+++ b/src/packet_io/sfdaq_module.cc
@@ -197,6 +197,8 @@ const PegInfo daq_names[] =
     { CountType::SUM, "retries_dropped", "messages dropped when overrunning the retry queue" },
     { CountType::SUM, "retries_processed", "messages processed from the retry queue" },
     { CountType::SUM, "retries_discarded", "messages discarded when purging the retry queue" },
+    { CountType::SUM, "sof_messages", "start of flow messages received from DAQ" },
+    { CountType::SUM, "eof_messages", "end of flow messages received from DAQ" },
     { CountType::SUM, "other_messages", "messages received from DAQ with unrecognized message type" },
     { CountType::END, nullptr, nullptr }
 };

diff --git a/src/packet_io/sfdaq_module.h b/src/packet_io/sfdaq_module.h
index 0c9ae1a8cfa1a82a7c15508499c138256c3c73f5..8f7b4c942a66d6624bac797e9dc94a66223982f0 100644 (file)
--- a/src/packet_io/sfdaq_module.h
+++ b/src/packet_io/sfdaq_module.h
@@ -74,6 +74,8 @@ struct DAQStats
     PegCount retries_dropped;
     PegCount retries_processed;
     PegCount retries_discarded;
+    PegCount sof_messages;
+    PegCount eof_messages;
     PegCount other_messages;
 };
 

diff --git a/src/pub_sub/CMakeLists.txt b/src/pub_sub/CMakeLists.txt
index 73a44aa42cfa8f92cc5fa1876391e3d4cb5aa4ef..4f0cb9c5307161bad430003e0b90013c190f822a 100644 (file)
--- a/src/pub_sub/CMakeLists.txt
+++ b/src/pub_sub/CMakeLists.txt
@@ -1,9 +1,9 @@
 set (PUB_SUB_INCLUDES
     appid_events.h
+    daq_message_event.h
     expect_events.h
     finalize_packet_event.h
     http_events.h
-    other_message_event.h
     sip_events.h
 )
 

diff --git a/src/pub_sub/other_message_event.h b/src/pub_sub/daq_message_event.h
similarity index 54%
rename from src/pub_sub/other_message_event.h
rename to src/pub_sub/daq_message_event.h
index 350fcfb94d8c760ca84758cea855f761915d79d9..3a1438c9178d805cd9e0e9669ee661e839aa999c 100644 (file)
--- a/src/pub_sub/other_message_event.h
+++ b/src/pub_sub/daq_message_event.h
@@ -15,39 +15,54 @@
 // with this program; if not, write to the Free Software Foundation, Inc.,
 // 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 //--------------------------------------------------------------------------
-// other_message_event.h author Steven Baigal <sbaigal@cisco.com>
+// daq_message_event.h author Michael Altizer <mialtize@cisco.com>
 
-#ifndef OTHER_MESSAGE_EVENT_H
-#define OTHER_MESSAGE_EVENT_H
+#ifndef DAQ_MESSAGE_EVENT_H
+#define DAQ_MESSAGE_EVENT_H
 
-#include <daq_common.h>
+#include <daq.h>
 
 #include "framework/data_bus.h"
 
-#define OTHER_MESSAGE_EVENT "daq.other.message"
+#define DAQ_SOF_MSG_EVENT   "daq.message.sof"
+#define DAQ_EOF_MSG_EVENT   "daq.message.eof"
+#define DAQ_OTHER_MSG_EVENT "daq.message.other"
 
 namespace snort
 {
-
-class SO_PUBLIC OtherMessageEvent : public snort::DataEvent
+class SO_PUBLIC DaqMessageEvent : public snort::DataEvent
 {
 public:
-    OtherMessageEvent(DAQ_Msg_h msg, DAQ_Verdict& v) :
-        daq_msg(msg), verdict(v)
-    {
-    }
+    DaqMessageEvent(DAQ_Msg_h msg, DAQ_Verdict& v) : msg(msg), verdict(v) { }
+
+    DAQ_Msg_h get_message()
+    { return msg; }
+
+    DAQ_MsgType get_type() const
+    { return daq_msg_get_type(msg); }
+
+    size_t get_header_length() const
+    { return daq_msg_get_hdr_len(msg); }
 
-    DAQ_Msg_h get_daq_msg()
-    { return daq_msg; }
+    const void* get_header() const
+    { return daq_msg_get_hdr(msg); }
 
-    DAQ_Verdict& get_verdict()
+    uint32_t get_data_length() const
+    { return daq_msg_get_data_len(msg); }
+
+    const uint8_t* get_data() override
+    { return daq_msg_get_data(msg); }
+
+    DAQ_Verdict get_verdict()
     { return verdict; }
 
+    void set_verdict(DAQ_Verdict v)
+    { verdict = v; }
+
 private:
-    DAQ_Msg_h daq_msg;
+    DAQ_Msg_h msg;
     DAQ_Verdict& verdict;
 };
-
 }
 
 #endif