Add support to packet size and throughput (#261)

* Add support to packet size and throughput

* rename size varible to payload_size

Author: Leonardo Parente

src/Metrics.h

@@ -414,6 +414,11 @@ class Rate final : public Metric
         return *this;
     }
 
+    void operator+=(uint64_t i)
+    {
+        _counter.fetch_add(i, std::memory_order_relaxed);
+    }
+
     uint64_t rate() const
     {
         return _rate.load(std::memory_order_relaxed);

src/handlers/dns/DnsStreamHandler.cpp

@@ -137,7 +137,7 @@ void DnsStreamHandler::stop()
 }
 
 // callback from input module
-void DnsStreamHandler::process_dnstap_cb(const dnstap::Dnstap &d)
+void DnsStreamHandler::process_dnstap_cb(const dnstap::Dnstap &d, [[maybe_unused]] size_t size)
 {
     if (_f_enabled[Filters::DnstapMsgType] && !_f_dnstap_types[d.message().type()]) {
         _metrics->process_dnstap(d, true);

src/handlers/dns/DnsStreamHandler.h

@@ -268,7 +268,7 @@ class DnsStreamHandler final : public visor::StreamMetricsHandler<DnsMetricsMana
     sigslot::connection _tcp_message_connection;
 
     void process_udp_packet_cb(pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, uint32_t flowkey, timespec stamp);
-    void process_dnstap_cb(const dnstap::Dnstap &);
+    void process_dnstap_cb(const dnstap::Dnstap &, size_t);
     void tcp_message_ready_cb(int8_t side, const pcpp::TcpStreamData &tcpData);
     void tcp_connection_start_cb(const pcpp::ConnectionData &connectionData);
     void tcp_connection_end_cb(const pcpp::ConnectionData &connectionData, pcpp::TcpReassembly::ConnectionEndReason reason);

src/handlers/input_resources/InputResourcesStreamHandler.cpp

@@ -98,7 +98,7 @@ void InputResourcesStreamHandler::process_sflow_cb([[maybe_unused]] const SFSamp
     }
 }
 
-void InputResourcesStreamHandler::process_dnstap_cb([[maybe_unused]] const dnstap::Dnstap &)
+void InputResourcesStreamHandler::process_dnstap_cb([[maybe_unused]] const dnstap::Dnstap &, [[maybe_unused]] size_t)
 {
     if (difftime(time(NULL), _timer) >= MEASURE_INTERVAL) {
         _timer = time(NULL);

src/handlers/input_resources/InputResourcesStreamHandler.h

@@ -95,7 +95,7 @@ class InputResourcesStreamHandler final : public visor::StreamMetricsHandler<Inp
     sigslot::connection _policies_connection;
 
     void process_sflow_cb(const SFSample &);
-    void process_dnstap_cb(const dnstap::Dnstap &);
+    void process_dnstap_cb(const dnstap::Dnstap &, size_t);
     void process_policies_cb(const Policy *policy, InputStream::Action action);
     void process_packet_cb(pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, timespec stamp);
 

src/handlers/net/NetStreamHandler.cpp

@@ -121,9 +121,9 @@ void NetStreamHandler::process_sflow_cb(const SFSample &payload)
     _metrics->process_sflow(payload);
 }
 
-void NetStreamHandler::process_dnstap_cb(const dnstap::Dnstap &payload)
+void NetStreamHandler::process_dnstap_cb(const dnstap::Dnstap &payload, size_t size)
 {
-    _metrics->process_dnstap(payload);
+    _metrics->process_dnstap(payload, size);
 }
 
 void NetStreamHandler::process_udp_packet_cb(pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, [[maybe_unused]] uint32_t flowkey, timespec stamp)
@@ -139,6 +139,8 @@ void NetworkMetricsBucket::specialized_merge(const AbstractMetricsBucket &o)
     // rates maintain their own thread safety
     _rate_in.merge(other._rate_in);
     _rate_out.merge(other._rate_out);
+    _throughput_in.merge(other._throughput_in);
+    _throughput_out.merge(other._throughput_out);
 
     std::shared_lock r_lock(other._mutex);
     std::unique_lock w_lock(_mutex);
@@ -166,13 +168,17 @@ void NetworkMetricsBucket::specialized_merge(const AbstractMetricsBucket &o)
         _topGeoLoc.merge(other._topGeoLoc);
         _topASN.merge(other._topASN);
     }
+
+    _payload_size.merge(other._payload_size);
 }
 
 void NetworkMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const
 {
 
     _rate_in.to_prometheus(out, add_labels);
     _rate_out.to_prometheus(out, add_labels);
+    _throughput_in.to_prometheus(out, add_labels);
+    _throughput_out.to_prometheus(out, add_labels);
 
     {
         auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe
@@ -208,6 +214,8 @@ void NetworkMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMa
         _topGeoLoc.to_prometheus(out, add_labels);
         _topASN.to_prometheus(out, add_labels);
     }
+
+    _payload_size.to_prometheus(out, add_labels);
 }
 
 void NetworkMetricsBucket::to_json(json &j) const
@@ -217,6 +225,8 @@ void NetworkMetricsBucket::to_json(json &j) const
     bool live_rates = !read_only() && !recorded_stream();
     _rate_in.to_json(j, live_rates);
     _rate_out.to_json(j, live_rates);
+    _throughput_in.to_json(j, live_rates);
+    _throughput_out.to_json(j, live_rates);
 
     {
         auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe
@@ -252,13 +262,15 @@ void NetworkMetricsBucket::to_json(json &j) const
         _topGeoLoc.to_json(j);
         _topASN.to_json(j);
     }
+
+    _payload_size.to_json(j);
 }
 
 // the main bucket analysis
 void NetworkMetricsBucket::process_packet(bool deep, pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4)
 {
     if (!deep) {
-        process_net_layer(dir, l3, l4);
+        process_net_layer(dir, l3, l4, payload.getRawPacket()->getRawDataLen());
         return;
     }
 
@@ -285,9 +297,9 @@ void NetworkMetricsBucket::process_packet(bool deep, pcpp::Packet &payload, Pack
         }
     }
 
-    process_net_layer(dir, l3, l4, is_ipv6, ipv4_in, ipv4_out, ipv6_in, ipv6_out);
+    process_net_layer(dir, l3, l4, payload.getRawPacket()->getRawDataLen(), is_ipv6, ipv4_in, ipv4_out, ipv6_in, ipv6_out);
 }
-void NetworkMetricsBucket::process_dnstap(bool deep, const dnstap::Dnstap &payload)
+void NetworkMetricsBucket::process_dnstap(bool deep, const dnstap::Dnstap &payload, size_t size)
 {
     pcpp::ProtocolType l3;
     bool is_ipv6{false};
@@ -335,7 +347,7 @@ void NetworkMetricsBucket::process_dnstap(bool deep, const dnstap::Dnstap &paylo
     }
 
     if (!deep) {
-        process_net_layer(dir, l3, l4);
+        process_net_layer(dir, l3, l4, size);
         return;
     }
 
@@ -354,7 +366,7 @@ void NetworkMetricsBucket::process_dnstap(bool deep, const dnstap::Dnstap &paylo
         ipv6_out = pcpp::IPv6Address(reinterpret_cast<const uint8_t *>(payload.message().response_address().data()));
     }
 
-    process_net_layer(dir, l3, l4, is_ipv6, ipv4_in, ipv4_out, ipv6_in, ipv6_out);
+    process_net_layer(dir, l3, l4, size, is_ipv6, ipv4_in, ipv4_out, ipv6_in, ipv6_out);
 }
 
 void NetworkMetricsBucket::process_sflow(bool deep, const SFSample &payload)
@@ -385,7 +397,7 @@ void NetworkMetricsBucket::process_sflow(bool deep, const SFSample &payload)
         }
 
         if (!deep) {
-            process_net_layer(dir, l3, l4);
+            process_net_layer(dir, l3, l4, payload.rawSampleLen);
             return;
         }
 
@@ -410,20 +422,22 @@ void NetworkMetricsBucket::process_sflow(bool deep, const SFSample &payload)
             ipv6_out = pcpp::IPv6Address(sample.ipdst.address.ip_v6.addr);
         }
 
-        process_net_layer(dir, l3, l4, is_ipv6, ipv4_in, ipv4_out, ipv6_in, ipv6_out);
+        process_net_layer(dir, l3, l4, payload.rawSampleLen, is_ipv6, ipv4_in, ipv4_out, ipv6_in, ipv6_out);
     }
 }
 
-void NetworkMetricsBucket::process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4)
+void NetworkMetricsBucket::process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, size_t payload_size)
 {
     std::unique_lock lock(_mutex);
 
     switch (dir) {
     case PacketDirection::fromHost:
         ++_rate_out;
+        _throughput_out += payload_size;
         break;
     case PacketDirection::toHost:
         ++_rate_in;
+        _throughput_in += payload_size;
         break;
     case PacketDirection::unknown:
         break;
@@ -464,18 +478,22 @@ void NetworkMetricsBucket::process_net_layer(PacketDirection dir, pcpp::Protocol
             break;
         }
     }
+
+    _payload_size.update(payload_size);
 }
 
-void NetworkMetricsBucket::process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, bool is_ipv6, pcpp::IPv4Address &ipv4_in, pcpp::IPv4Address &ipv4_out, pcpp::IPv6Address &ipv6_in, pcpp::IPv6Address &ipv6_out)
+void NetworkMetricsBucket::process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, size_t payload_size, bool is_ipv6, pcpp::IPv4Address &ipv4_in, pcpp::IPv4Address &ipv4_out, pcpp::IPv6Address &ipv6_in, pcpp::IPv6Address &ipv6_out)
 {
     std::unique_lock lock(_mutex);
 
     switch (dir) {
     case PacketDirection::fromHost:
         ++_rate_out;
+        _throughput_out += payload_size;
         break;
     case PacketDirection::toHost:
         ++_rate_in;
+        _throughput_in += payload_size;
         break;
     case PacketDirection::unknown:
         break;
@@ -517,6 +535,8 @@ void NetworkMetricsBucket::process_net_layer(PacketDirection dir, pcpp::Protocol
         }
     }
 
+    _payload_size.update(payload_size);
+
     struct sockaddr_in sa4;
     struct sockaddr_in6 sa6;
 
@@ -586,7 +606,7 @@ void NetworkMetricsManager::process_packet(pcpp::Packet &payload, PacketDirectio
     live_bucket()->process_packet(_deep_sampling_now, payload, dir, l3, l4);
 }
 
-void NetworkMetricsManager::process_dnstap(const dnstap::Dnstap &payload)
+void NetworkMetricsManager::process_dnstap(const dnstap::Dnstap &payload, size_t size)
 {
     // dnstap message type
     auto mtype = payload.message().type();
@@ -616,7 +636,7 @@ void NetworkMetricsManager::process_dnstap(const dnstap::Dnstap &payload)
     // base event
     new_event(stamp);
     // process in the "live" bucket. this will parse the resources if we are deep sampling
-    live_bucket()->process_dnstap(_deep_sampling_now, payload);
+    live_bucket()->process_dnstap(_deep_sampling_now, payload, size);
 }
 
 void NetworkMetricsManager::process_sflow(const SFSample &payload)

src/handlers/net/NetStreamHandler.h

@@ -67,8 +67,12 @@ class NetworkMetricsBucket final : public visor::AbstractMetricsBucket
     };
     counters _counters;
 
+    Quantile<std::size_t> _payload_size;
+
     Rate _rate_in;
     Rate _rate_out;
+    Rate _throughput_in;
+    Rate _throughput_out;
 
 public:
     NetworkMetricsBucket()
@@ -78,8 +82,11 @@ class NetworkMetricsBucket final : public visor::AbstractMetricsBucket
         , _topASN("packets", "asn", {"top_ASN"}, "Top ASNs by IP")
         , _topIPv4("packets", "ipv4", {"top_ipv4"}, "Top IPv4 IP addresses")
         , _topIPv6("packets", "ipv6", {"top_ipv6"}, "Top IPv6 IP addresses")
+        , _payload_size("packets", {"payload_size"}, "Quantiles of payload sizes, in bytes")
         , _rate_in("packets", {"rates", "pps_in"}, "Rate of ingress in packets per second")
         , _rate_out("packets", {"rates", "pps_out"}, "Rate of egress in packets per second")
+        , _throughput_in("payload", {"rates", "bps_in"}, "Rate of ingress packets size in bytes per second")
+        , _throughput_out("payload", {"rates", "bps_out"}, "Rate of egress packets size in bytes per second")
     {
         set_event_rate_info("packets", {"rates", "pps_total"}, "Rate of all packets (combined ingress and egress) in packets per second");
         set_num_events_info("packets", {"total"}, "Total packets processed");
@@ -104,13 +111,15 @@ class NetworkMetricsBucket final : public visor::AbstractMetricsBucket
         // stop rate collection
         _rate_in.cancel();
         _rate_out.cancel();
+        _throughput_in.cancel();
+        _throughput_out.cancel();
     }
 
     void process_packet(bool deep, pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4);
-    void process_dnstap(bool deep, const dnstap::Dnstap &payload);
+    void process_dnstap(bool deep, const dnstap::Dnstap &payload, size_t size);
     void process_sflow(bool deep, const SFSample &payload);
-    void process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4);
-    void process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, bool is_ipv6, pcpp::IPv4Address &ipv4_in, pcpp::IPv4Address &ipv4_out, pcpp::IPv6Address &ipv6_in, pcpp::IPv6Address &ipv6_out);
+    void process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, size_t payload_size);
+    void process_net_layer(PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, size_t payload_size, bool is_ipv6, pcpp::IPv4Address &ipv4_in, pcpp::IPv4Address &ipv4_out, pcpp::IPv6Address &ipv6_in, pcpp::IPv6Address &ipv6_out);
 };
 
 class NetworkMetricsManager final : public visor::AbstractMetricsManager<NetworkMetricsBucket>
@@ -122,7 +131,7 @@ class NetworkMetricsManager final : public visor::AbstractMetricsManager<Network
     }
 
     void process_packet(pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, timespec stamp);
-    void process_dnstap(const dnstap::Dnstap &payload);
+    void process_dnstap(const dnstap::Dnstap &payload, size_t size);
     void process_sflow(const SFSample &payload);
 };
 
@@ -155,7 +164,7 @@ class NetStreamHandler final : public visor::StreamMetricsHandler<NetworkMetrics
         {"top_ips", group::NetMetrics::TopIps}};
 
     void process_sflow_cb(const SFSample &);
-    void process_dnstap_cb(const dnstap::Dnstap &);
+    void process_dnstap_cb(const dnstap::Dnstap &, size_t);
     void process_packet_cb(pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, pcpp::ProtocolType l4, timespec stamp);
     void process_udp_packet_cb(pcpp::Packet &payload, PacketDirection dir, pcpp::ProtocolType l3, uint32_t flowkey, timespec stamp);
     void set_start_tstamp(timespec stamp);

src/handlers/net/tests/test_net_layer.cpp

@@ -163,6 +163,7 @@ TEST_CASE("Parse net (dns) random UDP/TCP tests", "[pcap][net]")
     CHECK(j["cardinality"]["src_ips_in"] == 1);
     CHECK(j["top_ipv4"][0]["estimate"] == 16147);
     CHECK(j["top_ipv4"][0]["name"] == "8.8.8.8");
+    CHECK(j["payload_size"]["p50"] == 74);
 }
 
 TEST_CASE("Parse net (dns) with DNS filter only_qname_suffix", "[pcap][dns][net]")
@@ -245,6 +246,47 @@ TEST_CASE("Parse net (dns) sflow stream", "[sflow][net]")
     CHECK(j["cardinality"]["src_ips_in"] == 4);
     CHECK(j["top_ipv4"][0]["estimate"] == 27054);
     CHECK(j["top_ipv4"][0]["name"] == "10.4.2.2");
+    CHECK(j["payload_size"]["p50"] == 1372);
+    CHECK(j["payload_size"]["p99"] == 1392);
+}
+
+TEST_CASE("Parse net dnstap stream", "[dnstap][net]")
+{
+
+    DnstapInputStream stream{"dnstap-test"};
+    stream.config_set("dnstap_file", "inputs/dnstap/tests/fixtures/fixture.dnstap");
+    stream.config_set<visor::Configurable::StringList>("only_hosts", {"192.168.0.0/24", "2001:db8::/48"});
+    visor::Config c;
+    c.config_set<uint64_t>("num_periods", 1);
+    NetStreamHandler net_handler{"dns-test", &stream, &c};
+
+    net_handler.start();
+    stream.start();
+    stream.stop();
+    net_handler.stop();
+
+    auto counters = net_handler.metrics()->bucket(0)->counters();
+    auto event_data = net_handler.metrics()->bucket(0)->event_data_locked();
+
+    // confirmed with wireshark
+    CHECK(event_data.num_events->value() == 153);
+    CHECK(event_data.num_samples->value() == 153);
+    CHECK(counters.TCP.value() == 0);
+    CHECK(counters.UDP.value() == 153);
+    CHECK(counters.IPv4.value() == 153);
+    CHECK(counters.IPv6.value() == 0);
+    CHECK(counters.total_in.value() == 74);
+    CHECK(counters.total_out.value() == 79);
+
+    nlohmann::json j;
+    net_handler.metrics()->bucket(0)->to_json(j);
+
+    CHECK(j["cardinality"]["dst_ips_out"] == 1);
+    CHECK(j["cardinality"]["src_ips_in"] == 1);
+    CHECK(j["top_ipv4"][0]["estimate"] == 153);
+    CHECK(j["top_ipv4"][0]["name"] == "192.168.0.54");
+    CHECK(j["payload_size"]["p50"] == 100);
+    CHECK(j["payload_size"]["p99"] == 350);
 }
 
 TEST_CASE("Net groups", "[pcap][net]")

src/handlers/net/tests/window-schema.json

@@ -63,6 +63,12 @@
             "length": 31,
             "start_ts": 1614874231
           },
+          "payload_size": {
+            "p50": 74,
+            "p90": 176,
+            "p95": 187,
+            "p99": 202
+          },
           "tcp": 13176,
           "top_ASN": [],
           "top_geoLoc": [],
@@ -86,6 +92,7 @@
         "other_l4",
         "out",
         "period",
+        "payload_size",
         "tcp",
         "top_ASN",
         "top_geoLoc",
@@ -235,6 +242,16 @@
           },
           "additionalProperties": false
         },
+        "payload_size": {
+          "$id": "#/properties/packets/properties/payload_size",
+          "type": "object",
+          "title": "The payload_size schema",
+          "description": "An explanation about the purpose of this instance.",
+          "default": {},
+          "examples": [
+            {}
+          ]
+        },
         "tcp": {
           "$id": "#/properties/packets/properties/tcp",
           "type": "integer",