Merge pull request #222 from Zadamsa/update-process-plugins-events

Enhance process plugins: Allow skipping flows marked as uninteresting

Updated process plugins to enable skipping flows that are flagged as not interesting. This optimization improves efficiency by bypassing unnecessary processing for certain flows

Author: hynekkar

include/ipfixprobe/flowifc.hpp

@@ -49,6 +49,7 @@
 #include <arpa/inet.h>
 #include "ipaddr.hpp"
 #include <string>
+#include <bitset>
 
 namespace ipxp {
 
@@ -247,7 +248,24 @@ struct Record {
  * \brief Flow record struct constaining basic flow record data and extension headers.
  */
 struct Flow : public Record {
-   uint64_t flow_hash;
+    static inline const int MAXIMAL_PROCESS_PLUGIN_COUNT = 64;
+    /**
+     * \brief Plugins status struct describes flow information required by process plugins.
+     */
+    struct PluginsStatus {
+        // get_no_data[i] == true -> i-th process plugin requires no flow data
+        // get_no_data[i] == false && get_all_data[i] == true -> i-th process plugin requires all
+        // available flow data
+        // get_no_data[i] == false && get_all_data[i] == false -> i-th process plugin requires
+        // only metadata
+        std::bitset<MAXIMAL_PROCESS_PLUGIN_COUNT> get_all_data;
+        std::bitset<MAXIMAL_PROCESS_PLUGIN_COUNT> get_no_data;
+    };
+
+    uint64_t flow_hash;
+    PluginsStatus plugins_status; /**< Statuses of the process plugins for this flow, used to check
+                                     if the flow process plugins requires all available data, only
+                                     metadata or nothing of this. */
 
    struct timeval time_first;
    struct timeval time_last;

include/ipfixprobe/process.hpp

@@ -38,85 +38,103 @@
 #include "flowifc.hpp"
 
 namespace ipxp {
-
-/**
- * \brief Tell storage plugin to flush (immediately export) current flow.
- * Behavior when called from post_create, pre_update and post_update: flush current Flow and erase FlowRecord.
- */
-#define FLOW_FLUSH                  0x1
-
-/**
- * \brief Tell storage plugin to flush (immediately export) current flow.
- * Behavior when called from post_create: flush current Flow and erase FlowRecord.
- * Behavior when called from pre_update and post_update: flush current Flow, erase FlowRecord and call post_create on packet.
- */
-#define FLOW_FLUSH_WITH_REINSERT    0x3
-
 /**
  * \brief Class template for flow cache plugins.
  */
 class ProcessPlugin : public Plugin
 {
 public:
-   ProcessPlugin() {}
-   virtual ~ProcessPlugin() {}
-   virtual ProcessPlugin *copy() = 0;
+    enum FlowAction : int {
+        /**
+         * \brief Tell storage plugin that process plugin requires all incoming data for given flow.
+         */
+        GET_ALL_DATA = 0,
+        /**
+         * \brief Tell storage plugin that process plugin requires only metadata.
+         * Used to offload the cache when all process plugin return GET_METADATA.
+         */
+        GET_METADATA = 0x2,
+        /**
+         * \brief Tell storage plugin that process plugin has ended up its work and doesn't require
+         * any new data. Used to offload the cache when all process plugin return NO_PROCESS.
+         */
+        NO_PROCESS = 0x4,
+        /**
+         * \brief Tell storage plugin to flush (immediately export) current flow.
+         * Behavior when called from post_create, pre_update and post_update: flush current Flow and
+         * erase FlowRecord.
+         */
+        FLUSH = 0x1,
+
+        /**
+         * \brief Tell storage plugin to flush (immediately export) current flow.
+         * Behavior when called from post_create: flush current Flow and erase FlowRecord.
+         * Behavior when called from pre_update and post_update: flush current Flow, erase
+         * FlowRecord and call post_create on packet.
+         */
+        FLUSH_WITH_REINSERT = 0x7
+    };
+
+    ProcessPlugin() {}
+    virtual ~ProcessPlugin() {}
+    virtual ProcessPlugin* copy() = 0;
+
+    virtual RecordExt *get_ext() const
+    {
+        return nullptr;
+    }
 
-   virtual RecordExt *get_ext() const
-   {
-      return nullptr;
-   }
+    /**
+     * \brief Called before a new flow record is created.
+     * \param [in] pkt Parsed packet.
+     * \return 0 on success or FLOW_FLUSH option.
+     */
+    virtual FlowAction pre_create(Packet& pkt)
+    {
+        return FlowAction::GET_ALL_DATA;
+    }
 
-   /**
-    * \brief Called before a new flow record is created.
-    * \param [in] pkt Parsed packet.
-    * \return 0 on success or FLOW_FLUSH option.
-    */
-   virtual int pre_create(Packet &pkt)
-   {
-      return 0;
-   }
+    /**
+     * \brief Called after a new flow record is created.
+     * \param [in,out] rec Reference to flow record.
+     * \param [in] pkt Parsed packet.
+     * \return 0 on success or FLOW_FLUSH option.
+     */
+    virtual FlowAction post_create(Flow& rec, const Packet& pkt)
+    {
+        return FlowAction::GET_ALL_DATA;
+    }
 
-   /**
-    * \brief Called after a new flow record is created.
-    * \param [in,out] rec Reference to flow record.
-    * \param [in] pkt Parsed packet.
-    * \return 0 on success or FLOW_FLUSH option.
-    */
-   virtual int post_create(Flow &rec, const Packet &pkt)
-   {
-      return 0;
-   }
+    /**
+     * \brief Called before an existing record is update.
+     * \param [in,out] rec Reference to flow record.
+     * \param [in,out] pkt Parsed packet.
+     * \return 0 on success or FLOW_FLUSH option.
+     */
+    virtual FlowAction pre_update(Flow& rec, Packet& pkt)
+    {
+        return FlowAction::GET_ALL_DATA;
+    }
 
-   /**
-    * \brief Called before an existing record is update.
-    * \param [in,out] rec Reference to flow record.
-    * \param [in,out] pkt Parsed packet.
-    * \return 0 on success or FLOW_FLUSH option.
-    */
-   virtual int pre_update(Flow &rec, Packet &pkt)
-   {
-      return 0;
-   }
+    /**
+     * \brief Called after an existing record is updated.
+     * \param [in,out] rec Reference to flow record.
+     * \param [in,out] pkt Parsed packet.
+     * \return 0 on success or FLOW_FLUSH option.
+     */
+    virtual FlowAction post_update(Flow& rec, const Packet& pkt)
+    {
+        return FlowAction::GET_ALL_DATA;
+    }
 
-   /**
-    * \brief Called after an existing record is updated.
-    * \param [in,out] rec Reference to flow record.
-    * \param [in,out] pkt Parsed packet.
-    * \return 0 on success or FLOW_FLUSH option.
-    */
-   virtual int post_update(Flow &rec, const Packet &pkt)
-   {
-      return 0;
-   }
+    /**
+     * \brief Called before a flow record is exported from the cache.
+     * \param [in,out] rec Reference to flow record.
+     */
+    virtual void pre_export(Flow& rec)
+    {
 
-   /**
-    * \brief Called before a flow record is exported from the cache.
-    * \param [in,out] rec Reference to flow record.
-    */
-   virtual void pre_export(Flow &rec)
-   {
-   }
+    }
 };
 
 }