List ingress and egress separately on JSON and CSV exports. Include BW and also raw values per second (#246)

filipecosta90 · web-flow · commit 20c013d370b0 · 2023-12-24T00:36:17.000Z
diff --git a/client.cpp b/client.cpp
@@ -445,14 +445,16 @@ void client::handle_response(unsigned int conn_id, struct timeval timestamp,
     switch (request->m_type) {
         case rt_get:
             m_stats.update_get_op(&timestamp,
-                                  request->m_size + response->get_total_len(),
+                                  response->get_total_len(),
+                                  request->m_size,
                                   ts_diff(request->m_sent_time, timestamp),
                                   response->get_hits(),
                                   request->m_keys - response->get_hits());
             break;
         case rt_set:
             m_stats.update_set_op(&timestamp,
-                                  request->m_size + response->get_total_len(),
+                                  response->get_total_len(),
+                                  request->m_size,
                                   ts_diff(request->m_sent_time, timestamp));
             break;
         case rt_wait:
@@ -462,7 +464,8 @@ void client::handle_response(unsigned int conn_id, struct timeval timestamp,
         case rt_arbitrary: {
             arbitrary_request *ar = static_cast<arbitrary_request *>(request);
             m_stats.update_arbitrary_op(&timestamp,
-                                        request->m_size + response->get_total_len(),
+                                        response->get_total_len(),
+                                        request->m_size,
                                         ts_diff(request->m_sent_time, timestamp),
                                         ar->index);
             break;
diff --git a/cluster_client.cpp b/cluster_client.cpp
@@ -418,16 +418,19 @@ void cluster_client::handle_moved(unsigned int conn_id, struct timeval timestamp
     // update stats
     if (request->m_type == rt_get) {
         m_stats.update_moved_get_op(&timestamp,
-                                    request->m_size + response->get_total_len(),
+                                    response->get_total_len(),
+                                    request->m_size,
                                     ts_diff(request->m_sent_time, timestamp));
     } else if (request->m_type == rt_set) {
         m_stats.update_moved_set_op(&timestamp,
-                                    request->m_size + response->get_total_len(),
+                                    response->get_total_len(),
+                                    request->m_size,
                                     ts_diff(request->m_sent_time, timestamp));
      } else if (request->m_type == rt_arbitrary) {
         arbitrary_request *ar = static_cast<arbitrary_request *>(request);
         m_stats.update_moved_arbitrary_op(&timestamp,
-                                    request->m_size + response->get_total_len(),
+                                    response->get_total_len(),
+                                    request->m_size,
                                     ts_diff(request->m_sent_time, timestamp),
                                     ar->index);
     } else {
@@ -452,16 +455,19 @@ void cluster_client::handle_ask(unsigned int conn_id, struct timeval timestamp,
     // update stats
     if (request->m_type == rt_get) {
         m_stats.update_ask_get_op(&timestamp,
-                                    request->m_size + response->get_total_len(),
+                                    response->get_total_len(),
+                                    request->m_size,
                                     ts_diff(request->m_sent_time, timestamp));
     } else if (request->m_type == rt_set) {
         m_stats.update_ask_set_op(&timestamp,
-                                    request->m_size + response->get_total_len(),
+                                    response->get_total_len(),
+                                    request->m_size,
                                     ts_diff(request->m_sent_time, timestamp));
     } else if (request->m_type == rt_arbitrary) {
         arbitrary_request *ar = static_cast<arbitrary_request *>(request);
         m_stats.update_ask_arbitrary_op(&timestamp,
-                                    request->m_size + response->get_total_len(),
+                                    response->get_total_len(),
+                                    request->m_size,
                                     ts_diff(request->m_sent_time, timestamp),
                                     ar->index);
     } else {
diff --git a/run_stats.cpp b/run_stats.cpp
diff --git a/run_stats.h b/run_stats.h
@@ -121,21 +121,21 @@ class run_stats {
     void set_start_time(struct timeval* start_time);
     void set_end_time(struct timeval* end_time);
 
-    void update_get_op(struct timeval* ts, unsigned int bytes, unsigned int latency, unsigned int hits, unsigned int misses);
-    void update_set_op(struct timeval* ts, unsigned int bytes, unsigned int latency);
+    void update_get_op(struct timeval* ts, unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency, unsigned int hits, unsigned int misses);
+    void update_set_op(struct timeval* ts, unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency);
 
-    void update_moved_get_op(struct timeval* ts, unsigned int bytes, unsigned int latency);
-    void update_moved_set_op(struct timeval* ts, unsigned int bytes, unsigned int latency);
-    void update_moved_arbitrary_op(struct timeval *ts, unsigned int bytes,
+    void update_moved_get_op(struct timeval* ts, unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency);
+    void update_moved_set_op(struct timeval* ts, unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency);
+    void update_moved_arbitrary_op(struct timeval *ts, unsigned int bytes_rx, unsigned int bytes_tx,
                              unsigned int latency, size_t arbitrary_index);
 
-    void update_ask_get_op(struct timeval* ts, unsigned int bytes, unsigned int latency);
-    void update_ask_set_op(struct timeval* ts, unsigned int bytes, unsigned int latency);
-    void update_ask_arbitrary_op(struct timeval *ts, unsigned int bytes,
+    void update_ask_get_op(struct timeval* ts, unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency);
+    void update_ask_set_op(struct timeval* ts, unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency);
+    void update_ask_arbitrary_op(struct timeval *ts, unsigned int bytes_rx, unsigned int bytes_tx,
                                    unsigned int latency, size_t arbitrary_index);
 
     void update_wait_op(struct timeval* ts, unsigned int latency);
-    void update_arbitrary_op(struct timeval *ts, unsigned int bytes,
+    void update_arbitrary_op(struct timeval *ts, unsigned int bytes_rx, unsigned int bytes_tx,
                              unsigned int latency, size_t arbitrary_index);
 
     void aggregate_average(const std::vector<run_stats>& all_stats);
diff --git a/run_stats_types.cpp b/run_stats_types.cpp
@@ -27,7 +27,8 @@
 
 
 one_sec_cmd_stats::one_sec_cmd_stats() :
-    m_bytes(0),
+    m_bytes_rx(0),
+    m_bytes_tx(0),
     m_ops(0),
     m_hits(0),
     m_misses(0),
@@ -41,7 +42,8 @@ one_sec_cmd_stats::one_sec_cmd_stats() :
 
 
 void one_sec_cmd_stats::reset() {
-    m_bytes = 0;
+    m_bytes_rx = 0;
+    m_bytes_tx = 0;
     m_ops = 0;
     m_hits = 0;
     m_misses = 0;
@@ -55,7 +57,8 @@ void one_sec_cmd_stats::reset() {
 }
 
 void one_sec_cmd_stats::merge(const one_sec_cmd_stats& other) {
-    m_bytes += other.m_bytes;
+    m_bytes_rx += other.m_bytes_rx;
+    m_bytes_tx += other.m_bytes_tx;
     m_ops += other.m_ops;
     m_hits += other.m_hits;
     m_misses += other.m_misses;
@@ -79,26 +82,27 @@ void one_sec_cmd_stats::summarize_quantiles(safe_hdr_histogram histogram, std::v
     m_min_latency = has_samples ? hdr_min(histogram)/ (double) LATENCY_HDR_RESULTS_MULTIPLIER : 0.0;
 }
 
-void one_sec_cmd_stats::update_op(unsigned int bytes, unsigned int latency) {
-    m_bytes += bytes;
+void one_sec_cmd_stats::update_op(unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency) {
+    m_bytes_rx += bytes_rx;
+    m_bytes_tx += bytes_tx;
     m_ops++;
     m_total_latency += latency;
 }
 
-void one_sec_cmd_stats::update_op(unsigned int bytes, unsigned int latency,
+void one_sec_cmd_stats::update_op(unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency,
                                   unsigned int hits, unsigned int misses) {
-    update_op(bytes, latency);
+    update_op(bytes_rx, bytes_tx, latency);
     m_hits += hits;
     m_misses += misses;
 }
 
-void one_sec_cmd_stats::update_moved_op(unsigned int bytes, unsigned int latency) {
-    update_op(bytes, latency);
+void one_sec_cmd_stats::update_moved_op(unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency) {
+    update_op(bytes_rx, bytes_tx, latency);
     m_moved++;
 }
 
-void one_sec_cmd_stats::update_ask_op(unsigned int bytes, unsigned int latency) {
-    update_op(bytes, latency);
+void one_sec_cmd_stats::update_ask_op(unsigned int bytes_rx, unsigned int bytes_tx, unsigned int latency) {
+    update_op(bytes_rx, bytes_tx, latency);
     m_ask++;
 }
 
@@ -132,7 +136,8 @@ unsigned long int ar_one_sec_cmd_stats::ops() {
 unsigned long int ar_one_sec_cmd_stats::bytes() {
     unsigned long int total_bytes = 0;
     for (size_t i = 0; i<m_commands.size(); i++) {
-        total_bytes += m_commands[i].m_bytes;
+        total_bytes += m_commands[i].m_bytes_rx;
+        total_bytes += m_commands[i].m_bytes_tx;
     }
 
     return total_bytes;
@@ -187,6 +192,8 @@ void one_second_stats::merge(const one_second_stats& other) {
 totals_cmd::totals_cmd() :
         m_ops_sec(0),
         m_bytes_sec(0),
+        m_bytes_sec_rx(0),
+        m_bytes_sec_tx(0),
         m_moved_sec(0),
         m_ask_sec(0),
         m_latency(0),
@@ -198,6 +205,8 @@ void totals_cmd::add(const totals_cmd& other) {
     m_moved_sec += other.m_moved_sec;
     m_ask_sec += other.m_ask_sec;
     m_bytes_sec += other.m_bytes_sec;
+    m_bytes_sec_rx += other.m_bytes_sec_rx;
+    m_bytes_sec_tx += other.m_bytes_sec_tx;
     m_latency += other.m_latency;
     m_ops += other.m_ops;
 }
@@ -207,6 +216,8 @@ void totals_cmd::aggregate_average(size_t stats_size) {
     m_moved_sec /= stats_size;
     m_ask_sec /= stats_size;
     m_bytes_sec /= stats_size;
+    m_bytes_sec_rx /= stats_size;
+    m_bytes_sec_tx /= stats_size;
     m_latency /= stats_size;
 }
 
@@ -219,8 +230,9 @@ void totals_cmd::summarize(const one_sec_cmd_stats& other, unsigned long test_du
     } else {
         m_latency = 0;
     }
-
-    m_bytes_sec = (other.m_bytes / 1024.0) / test_duration_usec * 1000000;
+    m_bytes_sec = ((other.m_bytes_rx + other.m_bytes_tx) / 1024.0) / test_duration_usec * 1000000;
+    m_bytes_sec_rx = (other.m_bytes_rx / 1024.0) / test_duration_usec * 1000000;
+    m_bytes_sec_tx = (other.m_bytes_tx / 1024.0) / test_duration_usec * 1000000;
     m_moved_sec = (double) other.m_moved / test_duration_usec * 1000000;
     m_ask_sec = (double) other.m_ask / test_duration_usec * 1000000;
 }
@@ -265,7 +277,8 @@ totals::totals() :
         m_moved_sec(0),
         m_ask_sec(0),
         m_latency(0),
-        m_bytes(0),
+        m_bytes_rx(0),
+        m_bytes_tx(0),
         m_ops(0) {
 }
 
@@ -285,16 +298,18 @@ void totals::add(const totals& other) {
     m_moved_sec += other.m_moved_sec;
     m_ask_sec += other.m_ask_sec;
     m_bytes_sec += other.m_bytes_sec;
+    m_bytes_rx += other.m_bytes_rx;
+    m_bytes_tx += other.m_bytes_tx;
     m_latency += other.m_latency;
-    m_bytes += other.m_bytes;
     m_ops += other.m_ops;
 
     // aggregate latency data
     hdr_add(latency_histogram,other.latency_histogram);
 }
 
-void totals::update_op(unsigned long int bytes, unsigned int latency) {
-    m_bytes += bytes;
+void totals::update_op(unsigned long int bytes_rx, unsigned long int bytes_tx, unsigned int latency) {
+    m_bytes_rx += bytes_rx;
+    m_bytes_tx += bytes_tx;
     m_ops++;
     m_latency += latency;
     hdr_record_value(latency_histogram,latency);
diff --git a/run_stats_types.h b/run_stats_types.h
@@ -75,7 +75,8 @@ class safe_hdr_histogram {
 
 class one_sec_cmd_stats {
 public:
-    unsigned long int m_bytes;
+    unsigned long int m_bytes_rx;
+    unsigned long int m_bytes_tx;
     unsigned long int m_ops;
     unsigned int m_hits;
     unsigned int m_misses;
@@ -90,10 +91,10 @@ class one_sec_cmd_stats {
     void reset();
     void merge(const one_sec_cmd_stats& other);
     void summarize_quantiles(safe_hdr_histogram histogram, std::vector<float> quantiles);
-    void update_op(unsigned int bytes, unsigned int latency);
-    void update_op(unsigned int bytes, unsigned int latency, unsigned int hits, unsigned int misses);
-    void update_moved_op(unsigned int bytes, unsigned int latency);
-    void update_ask_op(unsigned int bytes, unsigned int latency);
+    void update_op(unsigned int bytes_rx, unsigned int bytes_tx,  unsigned int latency);
+    void update_op(unsigned int bytes_rx, unsigned int bytes_tx,  unsigned int latency, unsigned int hits, unsigned int misses);
+    void update_moved_op(unsigned int bytes_rx, unsigned int bytes_tx,  unsigned int latency);
+    void update_ask_op(unsigned int bytes_rx, unsigned int bytes_tx,  unsigned int latency);
 };
 
 class one_second_stats; // forward declaration
@@ -135,6 +136,8 @@ class totals_cmd {
 public:
     double m_ops_sec;
     double m_bytes_sec;
+    double m_bytes_sec_rx;
+    double m_bytes_sec_tx;
     double m_moved_sec;
     double m_ask_sec;
     double m_latency;
@@ -173,17 +176,22 @@ class totals {
     safe_hdr_histogram latency_histogram;
     double m_ops_sec;
     double m_bytes_sec;
+    double m_bytes_sec_rx;
+    double m_bytes_sec_tx;
     double m_hits_sec;
     double m_misses_sec;
     double m_moved_sec;
     double m_ask_sec;
     unsigned long long int m_latency;
-    unsigned long int m_bytes;
+    // number of bytes received
+    unsigned long int m_bytes_rx;
+    // number of bytes sent
+    unsigned long int m_bytes_tx;
     unsigned long int m_ops;
     totals();
     void setup_arbitrary_commands(size_t n_arbitrary_commands);
     void add(const totals& other);
-    void update_op(unsigned long int bytes, unsigned int latency);
+    void update_op(unsigned long int bytes_rx, unsigned long int bytes_tx, unsigned int latency);
 };
 
 
diff --git a/tests/include.py b/tests/include.py
@@ -158,3 +158,31 @@ def agg_keyspace_range(master_nodes_connections):
                 shard_count = int(shard_reply['db0']['keys'])
         overall_keyspace_range = overall_keyspace_range + shard_count
     return overall_keyspace_range
+
+
+def get_column_csv(filename,column_name):
+    found = False
+    with open(filename,"r") as fd:
+        stop_line = 0
+        lines = fd.readlines()
+        for line in lines:
+            # CSV is the first part of file
+            if "Full-Test GET Latency" in line or len(line) == 0:
+                break
+            stop_line = stop_line + 1
+        print(stop_line)
+        csv_lines = lines[1:stop_line-1]
+        header_line = csv_lines[0].strip().split(",")
+        col_pos = -1
+        for col_index,col in enumerate(header_line):
+            if column_name == col:
+                col_pos = col_index
+                found = True
+        data_lines = []
+        for line in csv_lines[1:]:
+            data_lines.append(line.strip().split(","))
+        column_data = []
+        if found is True:
+            for line in data_lines:
+                column_data.append(line[col_pos])
+    return found, column_data
diff --git a/tests/tests_oss_simple_flow.py b/tests/tests_oss_simple_flow.py
