feat(host_metrics source): add a new collector for tcp stats (#22057)

* feat(host_metrics): add a collector for TCP stats

Add a new `tcp` collector  to the `host_metrics` source to expose
information about the systems's TCP stack. It exposes three metrics:
* `tcp_connections_total`: The total number of TCP connections. It
  includes the `state` of the connection as a tag.
* `tcp_tx_queued_bytes_total`: The sum of the number of bytes in the
   send queue across all connections.
* `tcp_rx_queued_bytes_total`: The sum of the number of bytes in the
  receive queue across all connections.

The collector is only enabled for Linux as it uses the netlink
subsystem.

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

* chore(host_metrics): add changelog fragment for tcp collector

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

* chore(host_metrics): add docs for tcp collector

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

* chore: update third party licenses

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

* chore(host_metrics): include error source in snafu display msg

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

* chore(host_metrics): update cue docs

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

---------

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

Author: aryan9600

Cargo.lock

@@ -384,13 +384,20 @@ heim = { git = "https://github.com/vectordotdev/heim.git", branch = "update-nix"
 # make sure to update the external docs when the Lua version changes
 mlua = { version = "0.10.2", default-features = false, features = ["lua54", "send", "vendored", "macros"], optional = true }
 sysinfo = "0.32.1"
+byteorder = "1.5.0"
 
 [target.'cfg(windows)'.dependencies]
 windows-service = "0.7.0"
 
 [target.'cfg(unix)'.dependencies]
 nix = { version = "0.26.2", default-features = false, features = ["socket", "signal"] }
 
+[target.'cfg(target_os = "linux")'.dependencies]
+netlink-packet-utils = "0.5.2"
+netlink-packet-sock-diag = "0.4.2"
+netlink-packet-core = "0.7.0"
+netlink-sys = { version = "0.8.7", features = ["tokio_socket"] }
+
 [build-dependencies]
 prost-build = { workspace = true, optional = true }
 tonic-build = { workspace = true, optional = true }

Cargo.toml

@@ -385,6 +385,10 @@ mongodb,https://github.com/mongodb/mongo-rust-driver,Apache-2.0,"Saghm Rossi <sa
 multer,https://github.com/rousan/multer-rs,MIT,Rousan Ali <hello@rousan.io>
 native-tls,https://github.com/sfackler/rust-native-tls,MIT OR Apache-2.0,Steven Fackler <sfackler@gmail.com>
 ndk-context,https://github.com/rust-windowing/android-ndk-rs,MIT OR Apache-2.0,The Rust Windowing contributors
+netlink-packet-core,https://github.com/rust-netlink/netlink-packet-core,MIT,Corentin Henry <corentinhenry@gmail.com>
+netlink-packet-sock-diag,https://github.com/rust-netlink/netlink-packet-sock-diag,MIT,"Flier Lu <flier.lu@gmail.com>, Corentin Henry <corentinhenry@gmail.com>"
+netlink-packet-utils,https://github.com/rust-netlink/netlink-packet-utils,MIT,Corentin Henry <corentinhenry@gmail.com>
+netlink-sys,https://github.com/rust-netlink/netlink-sys,MIT,Corentin Henry <corentinhenry@gmail.com>
 nibble_vec,https://github.com/michaelsproul/rust_nibble_vec,MIT,Michael Sproul <micsproul@gmail.com>
 nix,https://github.com/nix-rust/nix,MIT,The nix-rust Project Developers
 nkeys,https://github.com/wasmcloud/nkeys,Apache-2.0,wasmCloud Team

LICENSE-3rdparty.csv

@@ -0,0 +1,14 @@
+The `host_metrics` source has a new collector, `tcp`. The `tcp`
+collector exposes three metrics related to the TCP stack of the
+system:
+
+* `tcp_connections_total`: The total number of TCP connections. It
+  includes the `state` of the connection as a tag.
+* `tcp_tx_queued_bytes_total`: The sum of the number of bytes in the
+   send queue across all connections.
+* `tcp_rx_queued_bytes_total`: The sum of the number of bytes in the
+  receive queue across all connections.
+
+This collector is enabled only on Linux systems.
+
+authors: aryan9600

changelog.d/21972-add-tcp-collector-host-metrics.feature.md

@@ -11,6 +11,7 @@
 # `ring` has a custom license that is mostly "ISC-style" but parts of it also fall under OpenSSL licensing.
 "ring-0.16.20" = { license = "ISC AND Custom" }
 "ring-0.17.5" = { license = "ISC AND Custom" }
+"ring-0.17.8" = { license = "ISC AND Custom" }
 
 # `rustls-webpki` doesn't specify their license in the metadata, but the file contains the ISC terms.
 "rustls-webpki-0.100.1" = { license = "ISC" }

license-tool.toml

@@ -259,6 +259,26 @@ impl DiskMetrics {
     }
 }
 
+pub struct TCPMetrics(Vec<Metric>);
+
+#[Object]
+impl TCPMetrics {
+    /// Total TCP connections
+    async fn tcp_conns_total(&self) -> f64 {
+        filter_host_metric(&self.0, "tcp_connections_total")
+    }
+
+    /// Total bytes in the send queue across all connections.
+    async fn tcp_tx_queued_bytes_total(&self) -> f64 {
+        filter_host_metric(&self.0, "tcp_tx_queued_bytes_total")
+    }
+
+    /// Total bytes in the receive queue across all connections.
+    async fn tcp_rx_queued_bytes_total(&self) -> f64 {
+        filter_host_metric(&self.0, "tcp_rx_queued_bytes_total")
+    }
+}
+
 pub struct HostMetrics(host_metrics::HostMetrics);
 
 impl HostMetrics {
@@ -324,6 +344,14 @@ impl HostMetrics {
         self.0.disk_metrics(&mut buffer).await;
         DiskMetrics(buffer.metrics)
     }
+
+    #[cfg(target_os = "linux")]
+    /// TCP metrics
+    async fn tcp(&self) -> TCPMetrics {
+        let mut buffer = self.0.buffer();
+        self.0.tcp_metrics(&mut buffer).await;
+        TCPMetrics(buffer.metrics)
+    }
 }
 
 /// Filters a [`Vec<Metric>`] by name, returning the inner `value` or 0.00 if not found

src/api/schema/metrics/host.rs

@@ -35,6 +35,8 @@ mod filesystem;
 mod memory;
 mod network;
 mod process;
+#[cfg(target_os = "linux")]
+mod tcp;
 
 /// Collector types.
 #[serde_as]
@@ -70,6 +72,9 @@ pub enum Collector {
 
     /// Metrics related to network utilization.
     Network,
+
+    /// Metrics related to TCP connections.
+    TCP,
 }
 
 /// Filtering configuration.
@@ -178,7 +183,7 @@ pub fn default_namespace() -> Option<String> {
     Some(String::from("host"))
 }
 
-const fn example_collectors() -> [&'static str; 8] {
+const fn example_collectors() -> [&'static str; 9] {
     [
         "cgroups",
         "cpu",
@@ -188,6 +193,7 @@ const fn example_collectors() -> [&'static str; 8] {
         "host",
         "memory",
         "network",
+        "tcp",
     ]
 }
 
@@ -206,10 +212,12 @@ fn default_collectors() -> Option<Vec<Collector>> {
     #[cfg(target_os = "linux")]
     {
         collectors.push(Collector::CGroups);
+        collectors.push(Collector::TCP);
     }
     #[cfg(not(target_os = "linux"))]
     if std::env::var("VECTOR_GENERATE_SCHEMA").is_ok() {
         collectors.push(Collector::CGroups);
+        collectors.push(Collector::TCP);
     }
 
     Some(collectors)
@@ -284,6 +292,9 @@ impl SourceConfig for HostMetricsConfig {
             if self.cgroups.is_some() || self.has_collector(Collector::CGroups) {
                 return Err("CGroups collector is only available on Linux systems".into());
             }
+            if self.has_collector(Collector::TCP) {
+                return Err("TCP collector is only available on Linux systems".into());
+            }
         }
 
         let mut config = self.clone();
@@ -399,6 +410,10 @@ impl HostMetrics {
         if self.config.has_collector(Collector::Network) {
             self.network_metrics(&mut buffer).await;
         }
+        #[cfg(target_os = "linux")]
+        if self.config.has_collector(Collector::TCP) {
+            self.tcp_metrics(&mut buffer).await;
+        }
 
         let metrics = buffer.metrics;
         self.events_received.emit(CountByteSize(