add flow_statistics table to DB

code/ID2TLib/StatsDatabase.py

@@ -59,7 +59,9 @@ class StatsDatabase:
         """
         :return: True if the database was already existent, otherwise False
         """
-        return self.existing_db
+        # Aidmar - for testing: return false always
+        #return self.existing_db
+        return False
 
     @staticmethod
     def _get_selector_keywords():

code_boost/src/cxx/pcap_processor.cpp

@@ -208,23 +208,25 @@ void pcap_processor::process_packets(const Packet &pkt) {
         if (p == PDU::PDUType::TCP) {
             TCP tcpPkt = (const TCP &) *pdu_l4;
             stats.incrementProtocolCount(ipAddressSender, "TCP");
-            try {
+            
+            // Aidmar
+            // Flow statistics
+            stats.addFlowStat(ipAddressSender, tcpPkt.sport(), ipAddressReceiver, tcpPkt.dport(), pkt.timestamp());  
+            
+            // Aidmar
+            // Check window size for SYN noly
+            if(tcpPkt.get_flag(TCP::SYN)) {
+                int win = tcpPkt.window();
+                stats.incrementWinCount(ipAddressSender, win);
+            }   
+                
+            try {                                                                
                 int val = tcpPkt.mss();
                 stats.addMSS(ipAddressSender, val);
                 
                 // Aidmar
                 // MSS distribution
-                stats.incrementMSScount(ipAddressSender, val);
-                // Check window size for SYN noly
-                 if(tcpPkt.get_flag(TCP::SYN)) {
-                    int win = tcpPkt.window();
-                    stats.incrementWinCount(ipAddressSender, win);
-                }
-                    
-                // Aidmar
-                // Flow statistics
-                stats.addFlowStat(ipAddressSender, tcpPkt.sport(), ipAddressReceiver, tcpPkt.dport());
-
+                stats.incrementMSScount(ipAddressSender, val);                          
             } catch (Tins::option_not_found) {
                 // Ignore MSS if option not set
             }

code_boost/src/cxx/statistics.cpp

@@ -101,21 +101,41 @@ void statistics::addIPEntropy(){
 }
 
 // Aidmar
-void statistics::addFlowStat(std::string ipAddressSender,int sport,std::string ipAddressReceiver,int dport){
-    std::cout<<ipAddressSender<<":"<<sport<<","<<ipAddressReceiver<<":"<<dport<<"\n";
+/**
+ * Increments the packet counter for the given flow.
+ * @param ipAddressSender The sender IP address.
+ * @param sport The source port.
+ * @param ipAddressReceiver The receiver IP address.
+ * @param dport The destination port.
+ * @param timestamp The timestamp of the packet.
+ */
+void statistics::addFlowStat(std::string ipAddressSender,int sport,std::string ipAddressReceiver,int dport, std::chrono::microseconds timestamp){   
+    
+    flow f1 = {ipAddressReceiver, dport, ipAddressSender, sport};
+    flow f2 = {ipAddressSender, sport, ipAddressReceiver, dport};
     
     // if already exist A(ipAddressReceiver, dport), B(ipAddressSender, sport)
-    /*if (flow_statistics.count({ipAddressReceiver, dport, ipAddressSender, sport})>0){
-        flow_statistics[{ipAddressReceiver, dport, ipAddressSender, sport}].pkts_B_A++;
-        std::cout<<flow_statistics[{ipAddressReceiver, dport, ipAddressSender, sport}].pkts_A_B<<"\n";
-        std::cout<<flow_statistics[{ipAddressReceiver, dport, ipAddressSender, sport}].pkts_B_A<<"\n";
+    if (flow_statistics.count(f1)>0){
+        flow_statistics[f1].pkts_B_A++;
+        flow_statistics[f1].pkts_B_A_timestamp.push_back(timestamp);
+        if(flow_statistics[f1].pkts_A_B_timestamp.size()>0){
+            flow_statistics[f1].pkts_delay.push_back(std::chrono::duration_cast<std::chrono::microseconds> (timestamp - flow_statistics[f1].pkts_A_B_timestamp[flow_statistics[f1].pkts_A_B_timestamp.size()-1]));
+        }
+        
+        std::cout<<timestamp.count()<<"::"<<ipAddressReceiver<<":"<<dport<<","<<ipAddressSender<<":"<<sport<<"\n"; 
+        std::cout<<flow_statistics[f1].pkts_A_B<<"\n";
+        std::cout<<flow_statistics[f1].pkts_B_A<<"\n";
     }
-    else{*/
-    std::cout<<flow_statistics[{ipAddressSender, sport, ipAddressReceiver, dport}].pkts_A_B<<"\n";
-        flow_statistics[{ipAddressSender, sport, ipAddressReceiver, dport}].pkts_A_B++;
-        std::cout<<flow_statistics[{ipAddressSender, sport, ipAddressReceiver, dport}].pkts_A_B<<"\n";
-        std::cout<<flow_statistics[{ipAddressSender, sport, ipAddressReceiver, dport}].pkts_B_A<<"\n";
-    //}      
+    else{
+        flow_statistics[f2].pkts_A_B++;
+        flow_statistics[f2].pkts_A_B_timestamp.push_back(timestamp);
+         if(flow_statistics[f2].pkts_B_A_timestamp.size()>0){
+            flow_statistics[f2].pkts_delay.push_back(std::chrono::duration_cast<std::chrono::microseconds> (timestamp - flow_statistics[f2].pkts_B_A_timestamp[flow_statistics[f2].pkts_B_A_timestamp.size()-1]));
+        }
+        std::cout<<timestamp.count()<<"::"<<ipAddressSender<<":"<<sport<<","<<ipAddressReceiver<<":"<<dport<<"\n"; 
+        std::cout<<flow_statistics[f2].pkts_A_B<<"\n";
+        std::cout<<flow_statistics[f2].pkts_B_A<<"\n";
+    }    
     
 }
     

code_boost/src/cxx/statistics.h

@@ -4,6 +4,8 @@
 #ifndef CPP_PCAPREADER_STATISTICS_H
 #define CPP_PCAPREADER_STATISTICS_H
 
+// Aidmar
+#include <vector>
 
 #include <unordered_map>
 #include <list>
@@ -164,10 +166,18 @@ struct entry_ipStat {
 struct entry_flowStat {
     long pkts_A_B;
     long pkts_B_A;
-
+    std::vector<std::chrono::microseconds> pkts_A_B_timestamp;
+    std::vector<std::chrono::microseconds> pkts_B_A_timestamp;
+    std::vector<std::chrono::microseconds> pkts_delay;
+    //std::chrono::duration<double, std::micro> median_delay;
+    std::chrono::microseconds median_delay;
+    
     bool operator==(const entry_flowStat &other) const {
         return pkts_A_B == other.pkts_A_B
-               && pkts_B_A == other.pkts_B_A;
+               && pkts_A_B_timestamp == other.pkts_A_B_timestamp
+               && pkts_B_A_timestamp == other.pkts_B_A_timestamp
+               && pkts_delay == other.pkts_delay
+               && median_delay == other.median_delay;
     }
 };
 
@@ -237,9 +247,9 @@ namespace std {
             using std::hash;
             using std::string;
             return ((hash<string>()(k.ipAddressA)
-                    ^ (hash<int>()(k.portA) << 1)) >> 1)
-                    ^ ((hash<string>()(k.ipAddressB)
-                    ^ (hash<int>()(k.portB) << 1)) >> 1);
+                     ^ (hash<int>()(k.portA) << 1)) >> 1)
+                     ^ ((hash<string>()(k.ipAddressB)
+                     ^ (hash<int>()(k.portB) << 1)) >> 1);
         }
     };
     
@@ -288,7 +298,7 @@ public:
     void incrementMSScount(std::string ipAddress, int mssValue);
     void incrementWinCount(std::string ipAddress, int winSize);
     void addIPEntropy();
-    void addFlowStat(std::string ipAddressSender,int sport,std::string ipAddressReceiver,int dport);
+    void addFlowStat(std::string ipAddressSender,int sport,std::string ipAddressReceiver,int dport, std::chrono::microseconds timestamp);
     
 
     void incrementTTLcount(std::string ipAddress, int ttlValue);

code_boost/src/cxx/statistics_db.cpp

@@ -1,4 +1,6 @@
 #include "statistics_db.h"
+#include <iostream>
+#include <sstream>
 
 /**
  * Creates a new statistics_db object. Opens an existing database located at database_path. If not existing, creates
@@ -307,7 +309,7 @@ void statistics_db::writeStatisticsWin(std::unordered_map<ipAddress_win, int> wi
  * Writes the flow statistics into the database.
  * @param flowStatistics The flow from class statistics.
  */
-void statistics_db::writeStatisticsFlow(std::unordered_map<flow, entry_flowStat> flowStatistics){
+void statistics_db::writeStatisticsFlow(std::unordered_map<flow, entry_flowStat> flowStatistics){          
     std::cout<<"write to DB"<<"\n";
     try {
         db->exec("DROP TABLE IF EXISTS flow_statistics");
@@ -319,18 +321,26 @@ void statistics_db::writeStatisticsFlow(std::unordered_map<flow, entry_flowStat>
                 "portB INTEGER,"
                 "pkts_A_B INTEGER,"
                 "pkts_B_A INTEGER,"
+                "medianDelay INTEGER,"
+                //"medianDelay TEXT,"
                 "PRIMARY KEY(ipAddressA,portA,ipAddressB,portB));";
         db->exec(createTable);
-        SQLite::Statement query(*db, "INSERT INTO flow_statistics VALUES (?, ?, ?, ?, ?, ?)");
+        SQLite::Statement query(*db, "INSERT INTO flow_statistics VALUES (?, ?, ?, ?, ?, ?, ?)");
         for (auto it = flowStatistics.begin(); it != flowStatistics.end(); ++it) {
             flow f = it->first;
             entry_flowStat e = it->second;
+            
+            // Compute the median delay
+            e.median_delay = e.pkts_delay[e.pkts_delay.size()/2];
+            
             query.bind(1, f.ipAddressA);
             query.bind(2, f.portA);
             query.bind(3, f.ipAddressB);
             query.bind(4, f.portB);
             query.bind(5, (int) e.pkts_A_B);
             query.bind(6, (int) e.pkts_B_A);
+            query.bind(7, (int) e.median_delay.count());
+            //query.bind(7,  std::to_string(e.median_delay.count()));            
             query.exec();
             query.reset();
         }