New flow event: 'analysis'.

 * The goal was to provide a separate event for extracted feature that are not required
   and only useful for a few (e.g. someone who wants do ML).
 * Increased network buffer size to 32kB (8192 * 4).
 * Switched timestamp precision from ms to us for *ALL* timestamps.

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>

2 files changed, 39 insertions, 18 deletions

diff --git a/dependencies/nDPIsrvd.h b/dependencies/nDPIsrvd.h
index cafe25395..b6cb3e20e 100644
--- a/dependencies/nDPIsrvd.h
+++ b/dependencies/nDPIsrvd.h
@@ -964,7 +964,7 @@ static inline struct nDPIsrvd_thread_data * nDPIsrvd_get_thread_data(
     struct nDPIsrvd_socket * const sock,
     struct nDPIsrvd_instance * const instance,
     struct nDPIsrvd_json_token const * const thread_id_token,
-    struct nDPIsrvd_json_token const * const ts_msec_token)
+    struct nDPIsrvd_json_token const * const ts_usec_token)
 {
     struct nDPIsrvd_thread_data * thread_data;
     nDPIsrvd_hashkey thread_id;
@@ -1000,14 +1000,14 @@ static inline struct nDPIsrvd_thread_data * nDPIsrvd_get_thread_data(
 #endif
     }
 
-    if (ts_msec_token != NULL)
+    if (ts_usec_token != NULL)
     {
-        nDPIsrvd_ull thread_ts_msec;
-        TOKEN_VALUE_TO_ULL(ts_msec_token, &thread_ts_msec);
+        nDPIsrvd_ull thread_ts_usec;
+        TOKEN_VALUE_TO_ULL(ts_usec_token, &thread_ts_usec);
 
-        if (thread_ts_msec > thread_data->most_recent_flow_time)
+        if (thread_ts_usec > thread_data->most_recent_flow_time)
         {
-            thread_data->most_recent_flow_time = thread_ts_msec;
+            thread_data->most_recent_flow_time = thread_ts_usec;
         }
     }
 
@@ -1023,8 +1023,9 @@ static inline struct nDPIsrvd_flow * nDPIsrvd_get_flow(struct nDPIsrvd_socket *
                                                          TOKEN_GET_SZ(sock, "source"),
                                                          TOKEN_GET_SZ(sock, "thread_id"),
                                                          TOKEN_GET_SZ(sock, "flow_id"),
-                                                         TOKEN_GET_SZ(sock, "thread_ts_msec"),
-                                                         TOKEN_GET_SZ(sock, "flow_last_seen"),
+                                                         TOKEN_GET_SZ(sock, "thread_ts_usec"),
+                                                         TOKEN_GET_SZ(sock, "flow_src_last_pkt_time"),
+                                                         TOKEN_GET_SZ(sock, "flow_dst_last_pkt_time"),
                                                          TOKEN_GET_SZ(sock, "flow_idle_time")};
     enum
     {
@@ -1033,7 +1034,8 @@ static inline struct nDPIsrvd_flow * nDPIsrvd_get_flow(struct nDPIsrvd_socket *
         TOKEN_THREAD_ID,
         TOKEN_FLOW_ID,
         TOKEN_THREAD_TS_MSEC,
-        TOKEN_FLOW_LAST_SEEN,
+        TOKEN_FLOW_SRC_LAST_PKT_TIME,
+        TOKEN_FLOW_DST_LAST_PKT_TIME,
         TOKEN_FLOW_IDLE_TIME
     };
     nDPIsrvd_hashkey flow_key;
@@ -1074,11 +1076,23 @@ static inline struct nDPIsrvd_flow * nDPIsrvd_get_flow(struct nDPIsrvd_socket *
 #endif
     }
 
-    if (tokens[TOKEN_FLOW_LAST_SEEN] != NULL)
+    if (tokens[TOKEN_FLOW_SRC_LAST_PKT_TIME] != NULL)
     {
-        nDPIsrvd_ull flow_last_seen;
-        TOKEN_VALUE_TO_ULL(tokens[TOKEN_FLOW_LAST_SEEN], &flow_last_seen);
-        flow->last_seen = flow_last_seen;
+        nDPIsrvd_ull nmb;
+        TOKEN_VALUE_TO_ULL(tokens[TOKEN_FLOW_SRC_LAST_PKT_TIME], &nmb);
+        if (nmb > flow->last_seen)
+        {
+            flow->last_seen = nmb;
+        }
+    }
+    if (tokens[TOKEN_FLOW_DST_LAST_PKT_TIME] != NULL)
+    {
+        nDPIsrvd_ull nmb;
+        TOKEN_VALUE_TO_ULL(tokens[TOKEN_FLOW_DST_LAST_PKT_TIME], &nmb);
+        if (nmb > flow->last_seen)
+        {
+            flow->last_seen = nmb;
+        }
     }
 
     if (tokens[TOKEN_FLOW_IDLE_TIME] != NULL)
diff --git a/dependencies/nDPIsrvd.py b/dependencies/nDPIsrvd.py
index 29b9458d5..4bb0f451a 100644
--- a/dependencies/nDPIsrvd.py
+++ b/dependencies/nDPIsrvd.py
@@ -21,7 +21,7 @@ DEFAULT_PORT = 7000
 DEFAULT_UNIX = '/tmp/ndpid-distributor.sock'
 
 NETWORK_BUFFER_MIN_SIZE = 6 # NETWORK_BUFFER_LENGTH_DIGITS + 1
-NETWORK_BUFFER_MAX_SIZE = 16384 # Please keep this value in sync with the one in config.h
+NETWORK_BUFFER_MAX_SIZE = 32768 # Please keep this value in sync with the one in config.h
 
 PKT_TYPE_ETH_IP4 = 0x0800
 PKT_TYPE_ETH_IP6 = 0x86DD
@@ -125,9 +125,9 @@ class Instance:
         if 'thread_id' not in json_dict:
             return
         thread_id = json_dict['thread_id']
-        if 'thread_ts_msec' in json_dict:
+        if 'thread_ts_usec' in json_dict:
             mrtf = self.getMostRecentFlowTime(thread_id) if thread_id in self.thread_data else 0
-            self.setMostRecentFlowTime(thread_id, max(json_dict['thread_ts_msec'], mrtf))
+            self.setMostRecentFlowTime(thread_id, max(json_dict['thread_ts_usec'], mrtf))
 
 class Flow:
 
@@ -176,6 +176,10 @@ class FlowManager:
 
         return self.instances[alias][source]
 
+    @staticmethod
+    def getLastPacketTime(instance, flow_id, json_dict):
+        return max(int(json_dict['flow_src_last_pkt_time']), int(json_dict['flow_dst_last_pkt_time']), instance.flows[flow_id].flow_last_seen)
+
     def getFlow(self, instance, json_dict):
         if 'flow_id' not in json_dict:
             return None
@@ -183,13 +187,13 @@ class FlowManager:
         flow_id = int(json_dict['flow_id'])
 
         if flow_id in instance.flows:
-            instance.flows[flow_id].flow_last_seen = int(json_dict['flow_last_seen'])
+            instance.flows[flow_id].flow_last_seen = FlowManager.getLastPacketTime(instance, flow_id, json_dict)
             instance.flows[flow_id].flow_idle_time = int(json_dict['flow_idle_time'])
             return instance.flows[flow_id]
 
         thread_id = int(json_dict['thread_id'])
         instance.flows[flow_id] = Flow(flow_id, thread_id)
-        instance.flows[flow_id].flow_last_seen = int(json_dict['flow_last_seen'])
+        instance.flows[flow_id].flow_last_seen = FlowManager.getLastPacketTime(instance, flow_id, json_dict)
         instance.flows[flow_id].flow_idle_time = int(json_dict['flow_idle_time'])
         instance.flows[flow_id].cleanup_reason = FlowManager.CLEANUP_REASON_INVALID
 
@@ -450,6 +454,9 @@ def defaultArgumentParser(desc='nDPIsrvd Python Interface',
     parser.add_argument('--unix', type=str, help='nDPIsrvd unix socket path')
     return parser
 
+def toSeconds(usec):
+    return usec / (1000 * 1000)
+
 def validateAddress(args):
     tcp_addr_set = False
     address = None