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.

(cherry picked from commit 0f92583702)

etc/schema.json

@@ -2079,6 +2079,10 @@
                 "elephant": {
                     "type": "boolean"
                 },
+                "elephant_direction": {
+                    "type": "array",
+                    "description": "Direction(s) in which flow was found to be elephant"
+                },
                 "emergency": {
                     "type": "boolean"
                 },
@@ -7493,6 +7497,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

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

src/decode.h

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

src/flow.c

@@ -357,7 +357,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) {
@@ -367,9 +369,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) {
-                    StatsIncr(tv, dtv->counter_flow_elephant);
+                if (dir == TOSERVER) {
+                    f->flags |= FLOW_IS_ELEPHANT_TOSERVER;
+                    if (tv != NULL) {
+                        if ((f->flags & FLOW_IS_ELEPHANT_TOCLIENT) == 0) {
+                            StatsIncr(tv, dtv->counter_flow_elephant);
+                        }
+                        StatsIncr(tv, dtv->counter_flow_elephant_toserver);
+                    }
+                } else {
+                    f->flags |= FLOW_IS_ELEPHANT_TOCLIENT;
+                    if (tv != NULL) {
+                        if ((f->flags & FLOW_IS_ELEPHANT_TOSERVER) == 0) {
+                            StatsIncr(tv, dtv->counter_flow_elephant);
+                        }
+                        StatsIncr(tv, 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");