Added timestamp, flow count and app. level protocol to (-b) diagnose statistics

1 files changed, 91 insertions, 56 deletions

diff --git a/example/ndpiReader.c b/example/ndpiReader.c
index c00d630f0..92bebc0d2 100644
--- a/example/ndpiReader.c
+++ b/example/ndpiReader.c
@@ -63,9 +63,11 @@ static char *_protoFilePath   = NULL; /**< Protocol file path  */
 static char *_statsFilePath   = NULL; /**< Top stats file path */
 #ifdef HAVE_JSON_C
 static char *_jsonFilePath    = NULL; /**< JSON file path  */
+static FILE *stats_fp = NULL;         /**< for Top Stats JSON file */
 #endif
 #ifdef HAVE_JSON_C
 static json_object *jArray_known_flows, *jArray_unknown_flows;
+static json_object *jArray_topStats;
 #endif
 static u_int8_t live_capture = 0;
 static u_int8_t undetected_flows_deleted = 0;
@@ -89,20 +91,21 @@ static u_int32_t num_flows;
 
 struct info_pair{
     char addr[48];
-    char protocol[64]; /*l4 protocol*/
+    char proto[48]; /*l4 protocol*/
     int count;
 };
 
 typedef struct node_a{
     char addr[48];
     int count;
-    char protocol[64]; /*l4 protocol*/
+    char proto[48]; /*l4 protocol*/
     struct node_a *left, *right;
 }addr_node;
 
 struct port_stats {
   u_int32_t port; /* we'll use this field as the key */
   u_int32_t num_pkts, num_bytes;
+  u_int32_t num_flows; 
   u_int32_t num_addr; /*to hold number of distinct IP addresses */
   u_int32_t cumulative_addr; /*to hold cumulative some of IP addresses */
   addr_node *addr_tree; /* to hold distinct IP addresses */
@@ -116,10 +119,11 @@ struct port_stats *srcStats = NULL, *dstStats = NULL;
 struct top_stats {  
   u_int32_t port; /* we'll use this field as the key */
   char top_ip[48]; /*ip address that is contributed to > 95% of traffic*/
-  char protocol[64]; /*application level protocol of top_ip */
+  char proto[64]; /*application level protocol of top_ip */
   u_int32_t num_pkts;
   float prcnt_pkt; /*percent of packets respect to total packets */
   u_int32_t num_addr; /*to hold number of distinct IP addresses */
+  u_int32_t num_flows; 
   UT_hash_handle hh;  /* makes this structure hashable */
 };
 
@@ -807,7 +811,7 @@ static void node_proto_guess_walker(const void *node, ndpi_VISIT which, int dept
 
 /* *********************************************** */
 
-int updateIpTree(const char *key, addr_node **vrootp) {
+int updateIpTree(const char *key, addr_node **vrootp, const char *proto) {
   addr_node *q;
   addr_node **rootp = vrootp;
   int r;
@@ -829,6 +833,7 @@ int updateIpTree(const char *key, addr_node **vrootp) {
   if(q != (addr_node *)0) {	                /* make new node */
     *rootp = q;			                /* link new node to old */
     strncpy(q->addr, key, sizeof(q->addr));     /* initialize new node */
+    strncpy(q->proto, proto, sizeof(q->proto));    
     q->count = UPDATED_TREE;
     q->left = q->right = (addr_node *)0;
     return q->count;
@@ -858,7 +863,7 @@ void freeIpTree(addr_node *root) {
 
 /* *********************************************** */
 
-void updateTopIpAddress(const char *addr, int count, struct info_pair top[], int size){
+void updateTopIpAddress(const char *addr, const char *proto, int count, struct info_pair top[], int size){
     int update = 0;
     int r;
     int i;
@@ -869,6 +874,7 @@ void updateTopIpAddress(const char *addr, int count, struct info_pair top[], int
     if(count == 0) return;
 
     strncpy(pair.addr, addr, sizeof(pair.addr)); 
+    strncpy(pair.proto, proto, sizeof(pair.proto)); 
     pair.count = count;
 
     
@@ -905,7 +911,7 @@ void updateTopIpAddress(const char *addr, int count, struct info_pair top[], int
 
 static void updatePortStats(struct port_stats **stats, u_int32_t port,
 			    const char *addr, u_int32_t num_pkts,
-			    u_int32_t num_bytes) {
+			    u_int32_t num_bytes, const char *proto) {
   struct port_stats *s;
   int count = 0;
   
@@ -915,15 +921,16 @@ static void updatePortStats(struct port_stats **stats, u_int32_t port,
     if(!s) return;
 
     s->port = port, s->num_pkts = num_pkts, s->num_bytes = num_bytes,
-      s->num_addr = 1, s->cumulative_addr = 1;
+    s->num_addr = 1, s->cumulative_addr = 1; s->num_flows = 1; 
 
     memset(s->top_ip_addrs, 0, MAX_NUM_IP_ADDRESS*sizeof(struct info_pair));
-    updateTopIpAddress(addr, 1, s->top_ip_addrs, MAX_NUM_IP_ADDRESS);
+    updateTopIpAddress(addr, proto, 1, s->top_ip_addrs, MAX_NUM_IP_ADDRESS);
     
     s->addr_tree = (addr_node *) malloc(sizeof(addr_node));
     if(!s->addr_tree) return;
 
     strncpy(s->addr_tree->addr, addr, sizeof(s->addr_tree->addr));
+    strncpy(s->addr_tree->proto, proto, sizeof(s->addr_tree->proto));
     s->addr_tree->count = 1;
     s->addr_tree->left = NULL;
     s->addr_tree->right = NULL;
@@ -931,16 +938,16 @@ static void updatePortStats(struct port_stats **stats, u_int32_t port,
     HASH_ADD_INT(*stats, port, s);
   }
   else{
-    count = updateIpTree(addr, &(*s).addr_tree);
+    count = updateIpTree(addr, &(*s).addr_tree, proto);
 
     if(count == UPDATED_TREE) s->num_addr++;
     
     if(count) {
         s->cumulative_addr++;
-        updateTopIpAddress(addr, count, s->top_ip_addrs, MAX_NUM_IP_ADDRESS);
+        updateTopIpAddress(addr, proto, count, s->top_ip_addrs, MAX_NUM_IP_ADDRESS);
     }
 
-    s->num_pkts += num_pkts, s->num_bytes += num_bytes;
+    s->num_pkts += num_pkts, s->num_bytes += num_bytes, s->num_flows++;
   }
 }
 
@@ -967,13 +974,23 @@ static void port_stats_walker(const void *node, ndpi_VISIT which, int depth, voi
 	  struct ndpi_flow_info *flow = *(struct ndpi_flow_info **) node;
 	  u_int16_t sport, dport;
 	  char saddr[48], daddr[48];
+          char proto[48];
+          u_int16_t thread_id = *(int *)user_data;
 	  
 	  sport = ntohs(flow->src_port), dport = ntohs(flow->dst_port);
 	  strncpy(saddr, flow->src_name, sizeof(saddr));
 	  strncpy(daddr, flow->dst_name, sizeof(daddr));
+
+          /* get app level protocol */
+          if(flow->detected_protocol.master_protocol)
+            ndpi_protocol2name(ndpi_thread_info[thread_id].workflow->ndpi_struct,
+                                flow->detected_protocol, proto, sizeof(proto));
+          else
+            strncpy(proto, ndpi_get_proto_name(ndpi_thread_info[thread_id].workflow->ndpi_struct, 
+                    flow->detected_protocol.app_protocol),sizeof(proto));
 	  
-	  updatePortStats(&srcStats, sport, saddr, flow->src2dst_packets, flow->src2dst_bytes);
-	  if(flow->dst2src_packets > 0) updatePortStats(&dstStats, dport, daddr, flow->dst2src_packets, flow->dst2src_bytes);
+	  updatePortStats(&srcStats, sport, saddr, flow->src2dst_packets, flow->src2dst_bytes, proto);
+	  if(flow->dst2src_packets > 0) updatePortStats(&dstStats, dport, daddr, flow->dst2src_packets, flow->dst2src_bytes, proto);
   }
 }
 
@@ -1178,6 +1195,22 @@ char* formatPackets(float numPkts, char *buf) {
 static void json_init() {
   jArray_known_flows = json_object_new_array();
   jArray_unknown_flows = json_object_new_array();
+  jArray_topStats = json_object_new_array();
+}
+
+static void json_open_stats_file() {
+     if((file_first_time && ((stats_fp = fopen(_statsFilePath,"w")) == NULL))
+       ||
+       (!file_first_time && (stats_fp = fopen(_statsFilePath,"a")) == NULL)) {
+      printf("Error creating/opening file %s\n", _statsFilePath);
+      stats_flag = 0;
+    } 
+    else file_first_time = 0;
+}
+
+static void json_close_stats_file() {
+  fprintf(stats_fp,"%s\n",json_object_to_json_string(jArray_topStats));
+  fclose(stats_fp);
 }
 #endif
 
@@ -1267,13 +1300,14 @@ static int getTopStats(struct top_stats **topStats, struct port_stats *stats, u_
     s->num_pkts = sp->num_pkts; 
     s->prcnt_pkt = (sp->num_pkts*100.0)/total_packet_count;
     s->num_addr = sp->num_addr;
+    s->num_flows = sp->num_flows;
 
     qsort(&sp->top_ip_addrs[0], MAX_NUM_IP_ADDRESS, sizeof(struct info_pair), info_pair_cmp);
     inf = sp->top_ip_addrs[0];
 
     if(((inf.count * 100.0)/sp->cumulative_addr) > AGGRESSIVE_PERCENT){
       strncpy(s->top_ip, inf.addr, sizeof(s->top_ip));
-      strncpy(s->protocol, inf.protocol, sizeof(s->protocol));
+      strncpy(s->proto, inf.proto, sizeof(s->proto));
     }
 
     HASH_ADD_INT(*topStats, port, s);
@@ -1290,55 +1324,56 @@ static int getTopStats(struct top_stats **topStats, struct port_stats *stats, u_
 /*
  * @brief Save Top Stats in json format
  */
-static void saveTopStats(FILE *fp, struct top_stats *stats, int direction, u_int64_t total_ip_addr){
+static void saveTopStats(json_object **jObj_group, struct top_stats *stats, int direction, u_int64_t total_ip_addr){
 #ifdef HAVE_JSON_C
   struct top_stats *s, *tmp;
-  json_object *jsMain = json_object_new_object();
-  json_object *jArray_filters  = json_object_new_array();
+  json_object *jArray_stats  = json_object_new_array();
   int i = 0;
 
   /* stats for packet burst diagnose */
   HASH_ITER(hh, stats, s, tmp) {
 
     if(s->top_ip[0] != '\0'){
-      json_object *jObj_topStats = json_object_new_object();
-      json_object_object_add(jObj_topStats,"port",json_object_new_int(s->port));
-      json_object_object_add(jObj_topStats,"packets.number",json_object_new_int64(s->num_pkts));
-      json_object_object_add(jObj_topStats,"packets.percent",json_object_new_double(s->prcnt_pkt));
-      json_object_object_add(jObj_topStats,"aggressive.ip",json_object_new_string(s->top_ip));
+      json_object *jObj_stat = json_object_new_object();
+      json_object_object_add(jObj_stat,"port",json_object_new_int(s->port));
+      json_object_object_add(jObj_stat,"packets.number",json_object_new_int64(s->num_pkts));
+      json_object_object_add(jObj_stat,"packets.percent",json_object_new_double(s->prcnt_pkt));
+      json_object_object_add(jObj_stat,"flows.number",json_object_new_double(s->num_flows));
+      json_object_object_add(jObj_stat,"aggressive.ip",json_object_new_string(s->top_ip));
+      json_object_object_add(jObj_stat,"protocol",json_object_new_string(s->proto));
       
-      json_object_array_add(jArray_filters,jObj_topStats);
+      json_object_array_add(jArray_stats, jObj_stat);
       i++;
 
       if(i >= 10) break;
     }
   }  
 
-  json_object_object_add(jsMain, (direction == DIR_SRC) ? "top.src.pkts.stats" : "top.dst.pkts.stats", jArray_filters);
+  json_object_object_add(*jObj_group, (direction == DIR_SRC) ? 
+      "top.src.pkts.stats" : "top.dst.pkts.stats", jArray_stats);
+
+  jArray_stats  = json_object_new_array();
+  i=0;
 
   /*sort top stats by ip addr count*/
   HASH_SORT(stats, top_stats_sort);
 
-  jArray_filters  = json_object_new_array();
-  i=0;
-
   /* stats for ip burst diagnose */
   HASH_ITER(hh, stats, s, tmp) {
 
-    json_object *jObj_topStats = json_object_new_object();
-    json_object_object_add(jObj_topStats,"port",json_object_new_int(s->port));
-    json_object_object_add(jObj_topStats,"ip.total",json_object_new_int64(s->num_addr));
-    json_object_object_add(jObj_topStats,"ip.percent",json_object_new_double((s->num_addr*100.0)/total_ip_addr));
+    json_object *jObj_stat = json_object_new_object();
+    json_object_object_add(jObj_stat,"port",json_object_new_int(s->port));
+    json_object_object_add(jObj_stat,"ip.total",json_object_new_int64(s->num_addr));
+    json_object_object_add(jObj_stat,"ip.percent",json_object_new_double((s->num_addr*100.0)/total_ip_addr));
 
-    json_object_array_add(jArray_filters,jObj_topStats);
+    json_object_array_add(jArray_stats,jObj_stat);
     i++;
 
     if(i >= 10) break;
   }
 
-  json_object_object_add(jsMain, (direction == DIR_SRC) ? "top.src.ip.stats" : "top.dst.ip.stats", jArray_filters);
-
-  fprintf(fp,"%s\n",json_object_to_json_string(jsMain));
+  json_object_object_add(*jObj_group, (direction == DIR_SRC) ? 
+          "top.src.ip.stats" : "top.dst.ip.stats", jArray_stats);
 #endif
 }
 
@@ -1350,8 +1385,8 @@ void printPortStats(struct port_stats *stats) {
 
   HASH_ITER(hh, stats, s, tmp) {
     i++;
-    printf("\t%2d\tPort %5u\t[%u IP address(es)/%u pkts/%u bytes]\n\t\tTop IP Stats:\n",
-	   i, s->port, s->num_addr, s->num_pkts, s->num_bytes);
+    printf("\t%2d\tPort %5u\t[%u IP address(es)/%u flows/%u pkts/%u bytes]\n\t\tTop IP Stats:\n",
+	   i, s->port, s->num_addr, s->num_flows, s->num_pkts, s->num_bytes);
 
     qsort(&s->top_ip_addrs[0], MAX_NUM_IP_ADDRESS, sizeof(struct info_pair), info_pair_cmp);
 
@@ -1381,7 +1416,6 @@ static void printResults(u_int64_t tot_usec) {
   char buf[32];
 #ifdef HAVE_JSON_C
   FILE *json_fp = NULL;
-  FILE *stats_fp = NULL;
   json_object *jObj_main = NULL, *jObj_trafficStats, *jArray_detProto = NULL, *jObj;
 #endif
   long long unsigned int breed_stats[NUM_BREEDS] = { 0 };
@@ -1637,26 +1671,19 @@ static void printResults(u_int64_t tot_usec) {
     u_int64_t total_src_addr = getTopStats(&topSrcStats, srcStats, cumulative_stats.ip_packet_count);
     u_int64_t total_dst_addr = getTopStats(&topDstStats, dstStats, cumulative_stats.ip_packet_count);
 
-    if((file_first_time && ((stats_fp = fopen(_statsFilePath,"w")) == NULL))
-       ||
-       (!file_first_time && (stats_fp = fopen(_statsFilePath,"a")) == NULL)) {
-      printf("Error creating file %s\n", _statsFilePath);
-      stats_flag = 0;
-    } 
-    else {
-      file_first_time = 0;
+    json_object *jObj_stats = json_object_new_object();
+    char timestamp[64];
 
-      fprintf(stats_fp,"[\n");
-      saveTopStats(stats_fp, topSrcStats, DIR_SRC, total_src_addr);
-      fprintf(stats_fp,",\n");
-      saveTopStats(stats_fp, topDstStats, DIR_DST, total_dst_addr);
-      fprintf(stats_fp,"]\n");
-      
-      fclose(stats_fp);
+    strftime(timestamp, sizeof(timestamp), "%FT%TZ", localtime(&pcap_start.tv_sec));
+    json_object_object_add(jObj_stats, "time", json_object_new_string(timestamp));
 
-      deleteTopStats(topSrcStats), deleteTopStats(topDstStats);
-      topSrcStats = NULL, topDstStats = NULL;
-    }
+    saveTopStats(&jObj_stats, topSrcStats, DIR_SRC, total_src_addr);
+    saveTopStats(&jObj_stats, topDstStats, DIR_DST, total_dst_addr);
+
+    json_object_array_add(jArray_topStats, jObj_stats);
+
+    deleteTopStats(topSrcStats), deleteTopStats(topDstStats);
+    topSrcStats = NULL, topDstStats = NULL;
 #endif
   }
 
@@ -1901,6 +1928,7 @@ static void pcap_process_packet(u_char *args,
     tot_usec = end.tv_sec*1000000 + end.tv_usec - (begin.tv_sec*1000000 + begin.tv_usec);
 
     printResults(tot_usec);
+    
 
     for(i=0; i<ndpi_thread_info[thread_id].workflow->prefs.num_roots; i++) {
       ndpi_tdestroy(ndpi_thread_info[thread_id].workflow->ndpi_flows_root[i], ndpi_flow_info_freer);
@@ -1975,6 +2003,7 @@ void test_lib() {
 
 #ifdef HAVE_JSON_C
   json_init();
+  if(stats_flag)  json_open_stats_file();
 #endif
 
 #ifdef DEBUG_TRACE
@@ -2026,6 +2055,12 @@ void test_lib() {
   /* Printing cumulative results */
   printResults(tot_usec);
 
+if(stats_flag){
+#ifdef HAVE_JSON_C
+  json_close_stats_file();
+#endif
+}
+
   for(thread_id = 0; thread_id < num_threads; thread_id++) {
     if(ndpi_thread_info[thread_id].workflow->pcap_handle != NULL)
       pcap_close(ndpi_thread_info[thread_id].workflow->pcap_handle);