flow: split elephant flow detection per dir

The tracking for elephant flow is done per direction however, the flag
was set on the flow whenever either of the directions crossed the
rate-limit defined in the settings. Given that the tracking was already
split, it makes sense to split the detection tracking per direction as
well and allow user to have a better control via rule language.

etc/schema.json

@@ -2082,6 +2082,10 @@
                 "elephant": {
                     "type": "boolean"
                 },
+                "elephant_direction": {
+                    "type": "array",
+                    "description": "Direction(s) in which flow was found to be elephant"
+                },
                 "emergency": {
                     "type": "boolean"
                 },
@@ -7434,6 +7438,14 @@
                             "type": "integer",
                             "description": "Total number of elephant flows"
                         },
+                        "elephant_toclient": {
+                            "type": "integer",
+                            "description": "Total number of elephant flows in toclient direction"
+                        },
+                        "elephant_toserver": {
+                            "type": "integer",
+                            "description": "Total number of elephant flows in toserver direction"
+                        },
                         "emerg_mode_entered": {
                             "type": "integer",
                             "description": "Number of times emergency mode was entered"

src/decode.c

@@ -696,6 +696,10 @@ void DecodeRegisterPerfCounters(DecodeThreadVars *dtv, ThreadVars *tv)
     dtv->counter_flow_icmp6 = StatsRegisterCounter("flow.icmpv6", &tv->stats);
     dtv->counter_flow_tcp_reuse = StatsRegisterCounter("flow.tcp_reuse", &tv->stats);
     dtv->counter_flow_elephant = StatsRegisterCounter("flow.elephant", &tv->stats);
+    dtv->counter_flow_elephant_toserver =
+            StatsRegisterCounter("flow.elephant_toserver", &tv->stats);
+    dtv->counter_flow_elephant_toclient =
+            StatsRegisterCounter("flow.elephant_toclient", &tv->stats);
     dtv->counter_flow_get_used = StatsRegisterCounter("flow.get_used", &tv->stats);
     dtv->counter_flow_get_used_eval = StatsRegisterCounter("flow.get_used_eval", &tv->stats);
     dtv->counter_flow_get_used_eval_reject =

src/decode.h

@@ -1039,6 +1039,8 @@ typedef struct DecodeThreadVars_
     StatsCounterId counter_flow_icmp6;
     StatsCounterId counter_flow_tcp_reuse;
     StatsCounterId counter_flow_elephant;
+    StatsCounterId counter_flow_elephant_toserver;
+    StatsCounterId counter_flow_elephant_toclient;
     StatsCounterId counter_flow_get_used;
     StatsCounterId counter_flow_get_used_eval;
     StatsCounterId counter_flow_get_used_eval_reject;

src/flow.c

@@ -351,7 +351,9 @@ static inline void FlowUpdateFlowRate(
 {
     if (FlowRateStorageEnabled()) {
         /* No need to update the struct if flow is already marked as elephant flow */
-        if (f->flags & FLOW_IS_ELEPHANT)
+        if ((dir == TOSERVER) && (f->flags & FLOW_IS_ELEPHANT_TOSERVER))
+            return;
+        if ((dir == TOCLIENT) && (f->flags & FLOW_IS_ELEPHANT_TOCLIENT))
             return;
         FlowRateStore *frs = FlowGetStorageById(f, FlowRateGetStorageID());
         if (frs != NULL) {
@@ -361,9 +363,22 @@ static inline void FlowUpdateFlowRate(
                 SCLogDebug("Flow rate for flow %p exceeds the configured values, marking it as an "
                            "elephant flow",
                         f);
-                f->flags |= FLOW_IS_ELEPHANT;
-                if (tv != NULL) {
-                    StatsCounterIncr(&tv->stats, dtv->counter_flow_elephant);
+                if (dir == TOSERVER) {
+                    f->flags |= FLOW_IS_ELEPHANT_TOSERVER;
+                    if (tv != NULL) {
+                        if ((f->flags & FLOW_IS_ELEPHANT_TOCLIENT) == 0) {
+                            StatsCounterIncr(&tv->stats, dtv->counter_flow_elephant);
+                        }
+                        StatsCounterIncr(&tv->stats, dtv->counter_flow_elephant_toserver);
+                    }
+                } else {
+                    f->flags |= FLOW_IS_ELEPHANT_TOCLIENT;
+                    if (tv != NULL) {
+                        if ((f->flags & FLOW_IS_ELEPHANT_TOSERVER) == 0) {
+                            StatsCounterIncr(&tv->stats, dtv->counter_flow_elephant);
+                        }
+                        StatsCounterIncr(&tv->stats, dtv->counter_flow_elephant_toclient);
+                    }
                 }
             }
         }

src/flow.h

@@ -56,12 +56,11 @@ typedef struct AppLayerParserState_ AppLayerParserState;
 #define FLOW_TC_APP_UPDATE_NEXT BIT_U32(2)
 
 /** Flow is marked an elephant flow */
-#define FLOW_IS_ELEPHANT BIT_U32(3)
+#define FLOW_IS_ELEPHANT_TOSERVER BIT_U32(3)
+#define FLOW_IS_ELEPHANT_TOCLIENT BIT_U32(4)
 
 /** All packets in this flow should be accepted */
-#define FLOW_ACTION_ACCEPT BIT_U32(4)
-
-// vacancy bit 5
+#define FLOW_ACTION_ACCEPT BIT_U32(5)
 
 /** Packet payloads belonging to this flow should not be inspected */
 #define FLOW_NOPAYLOAD_INSPECTION       BIT_U32(6)

src/output-json-flow.c

@@ -342,8 +342,16 @@ static void EveFlowLogJSON(OutputJsonThreadCtx *aft, SCJsonBuilder *jb, Flow *f)
     if (f->flags & FLOW_WRONG_THREAD)
         JB_SET_TRUE(jb, "wrong_thread");
 
-    if (f->flags & FLOW_IS_ELEPHANT)
+    if (f->flags & (FLOW_IS_ELEPHANT_TOSERVER | FLOW_IS_ELEPHANT_TOCLIENT)) {
         JB_SET_TRUE(jb, "elephant");
+        SCJbOpenArray(jb, "elephant_direction");
+        if (f->flags & FLOW_IS_ELEPHANT_TOSERVER)
+            SCJbAppendString(jb, "toserver");
+        if (f->flags & FLOW_IS_ELEPHANT_TOCLIENT)
+            SCJbAppendString(jb, "toclient");
+
+        SCJbClose(jb);
+    }
 
     if (f->flags & FLOW_ACTION_DROP) {
         JB_SET_STRING(jb, "action", "drop");