flow: counters for total number of flows

flow.tcp
flow.udp
flow.icmpv4
flow.icmpv6

src/decode.c

@@ -430,6 +430,11 @@ void DecodeRegisterPerfCounters(DecodeThreadVars *dtv, ThreadVars *tv)
     dtv->counter_erspan = StatsRegisterMaxCounter("decoder.erspan", tv);
     dtv->counter_flow_memcap = StatsRegisterCounter("flow.memcap", tv);
 
+    dtv->counter_flow_tcp = StatsRegisterCounter("flow.tcp", tv);
+    dtv->counter_flow_udp = StatsRegisterCounter("flow.udp", tv);
+    dtv->counter_flow_icmp4 = StatsRegisterCounter("flow.icmpv4", tv);
+    dtv->counter_flow_icmp6 = StatsRegisterCounter("flow.icmpv6", tv);
+
     dtv->counter_defrag_ipv4_fragments =
         StatsRegisterCounter("defrag.ipv4.fragments", tv);
     dtv->counter_defrag_ipv4_reassembled =

src/decode.h

@@ -665,6 +665,11 @@ typedef struct DecodeThreadVars_
 
     uint16_t counter_flow_memcap;
 
+    uint16_t counter_flow_tcp;
+    uint16_t counter_flow_udp;
+    uint16_t counter_flow_icmp4;
+    uint16_t counter_flow_icmp6;
+
      uint16_t counter_invalid_events[DECODE_EVENT_PACKET_MAX];
     /* thread data for flow logging api: only used at forced
      * flow recycle during lookups */

src/flow-hash.c

@@ -339,6 +339,31 @@ static inline int FlowCreateCheck(const Packet *p)
     return 1;
 }
 
+static inline void FlowUpdateCounter(ThreadVars *tv, DecodeThreadVars *dtv,
+        uint8_t proto)
+{
+#ifdef UNITTESTS
+    if (tv && dtv) {
+#endif
+        switch (proto){
+            case IPPROTO_UDP:
+                StatsIncr(tv, dtv->counter_flow_udp);
+                break;
+            case IPPROTO_TCP:
+                StatsIncr(tv, dtv->counter_flow_tcp);
+                break;
+            case IPPROTO_ICMP:
+                StatsIncr(tv, dtv->counter_flow_icmp4);
+                break;
+            case IPPROTO_ICMPV6:
+                StatsIncr(tv, dtv->counter_flow_icmp6);
+                break;
+        }
+#ifdef UNITTESTS
+    }
+#endif
+}
+
 /**
  *  \brief Get a new flow
  *
@@ -406,6 +431,7 @@ static Flow *FlowGetNew(ThreadVars *tv, DecodeThreadVars *dtv, const Packet *p)
     }
 
     FLOWLOCK_WRLOCK(f);
+    FlowUpdateCounter(tv, dtv, p->proto);
     return f;
 }
 

src/flow-worker.c

@@ -119,6 +119,7 @@ static TmEcode FlowWorkerThreadInit(ThreadVars *tv, const void *initdata, void *
         return TM_ECODE_FAILED;
     }
 
+    DecodeRegisterPerfCounters(fw->dtv, tv);
     AppLayerRegisterThreadCounters(tv);
 
     /* setup pq for stream end pkts */