Add host-service GetNetworkInfo GRPC call including test and cli

Signed-off-by: Guvenc Gulce <guevenc.guelce@sap.com>

Author: guvenc

cli/src/host_commands.rs

@@ -2,9 +2,9 @@ use anyhow::{Context, Result};
 use clap::{Args, Subcommand};
 use digest::Digest;
 use feos_proto::host_service::{
-    host_service_client::HostServiceClient, upgrade_request, GetCpuInfoRequest, HostnameRequest,
-    MemoryRequest, RebootRequest, ShutdownRequest, StreamKernelLogsRequest, UpgradeMetadata,
-    UpgradeRequest,
+    host_service_client::HostServiceClient, upgrade_request, GetCpuInfoRequest,
+    GetNetworkInfoRequest, HostnameRequest, MemoryRequest, RebootRequest, ShutdownRequest,
+    StreamKernelLogsRequest, UpgradeMetadata, UpgradeRequest,
 };
 use sha2::Sha256;
 use std::path::PathBuf;
@@ -34,6 +34,7 @@ pub enum HostCommand {
     Hostname,
     Memory,
     CpuInfo,
+    NetworkInfo,
     Upgrade {
         #[arg(required = true)]
         binary_path: PathBuf,
@@ -55,6 +56,7 @@ pub async fn handle_host_command(args: HostArgs) -> Result<()> {
         HostCommand::Hostname => get_hostname(&mut client).await?,
         HostCommand::Memory => get_memory(&mut client).await?,
         HostCommand::CpuInfo => get_cpu_info(&mut client).await?,
+        HostCommand::NetworkInfo => get_network_info(&mut client).await?,
         HostCommand::Upgrade { binary_path } => upgrade_feos(&mut client, binary_path).await?,
         HostCommand::Klogs => stream_klogs(&mut client).await?,
         HostCommand::Shutdown => shutdown_host(&mut client).await?,
@@ -185,6 +187,33 @@ async fn get_cpu_info(client: &mut HostServiceClient<Channel>) -> Result<()> {
     Ok(())
 }
 
+async fn get_network_info(client: &mut HostServiceClient<Channel>) -> Result<()> {
+    let request = GetNetworkInfoRequest {};
+    let response = client.get_network_info(request).await?.into_inner();
+
+    if response.devices.is_empty() {
+        println!("No network devices found on the host.");
+        return Ok(());
+    }
+
+    for dev in response.devices {
+        println!("Interface: {}", dev.name);
+        println!("  RX");
+        println!("    Bytes:    {:>15}", dev.rx_bytes);
+        println!("    Packets:  {:>15}", dev.rx_packets);
+        println!("    Errors:   {:>15}", dev.rx_errors);
+        println!("    Dropped:  {:>15}", dev.rx_dropped);
+        println!("  TX");
+        println!("    Bytes:    {:>15}", dev.tx_bytes);
+        println!("    Packets:  {:>15}", dev.tx_packets);
+        println!("    Errors:   {:>15}", dev.tx_errors);
+        println!("    Dropped:  {:>15}", dev.tx_dropped);
+        println!();
+    }
+
+    Ok(())
+}
+
 async fn stream_klogs(client: &mut HostServiceClient<Channel>) -> Result<()> {
     println!("Streaming kernel logs... Press Ctrl+C to stop.");
     let request = StreamKernelLogsRequest {};

feos/services/host-service/src/api.rs

@@ -1,8 +1,9 @@
 use crate::Command;
 use feos_proto::host_service::{
-    host_service_server::HostService, GetCpuInfoRequest, GetCpuInfoResponse, HostnameRequest,
-    HostnameResponse, KernelLogEntry, MemoryRequest, MemoryResponse, RebootRequest, RebootResponse,
-    ShutdownRequest, ShutdownResponse, StreamKernelLogsRequest, UpgradeRequest, UpgradeResponse,
+    host_service_server::HostService, GetCpuInfoRequest, GetCpuInfoResponse, GetNetworkInfoRequest,
+    GetNetworkInfoResponse, HostnameRequest, HostnameResponse, KernelLogEntry, MemoryRequest,
+    MemoryResponse, RebootRequest, RebootResponse, ShutdownRequest, ShutdownResponse,
+    StreamKernelLogsRequest, UpgradeRequest, UpgradeResponse,
 };
 use log::info;
 use std::pin::Pin;
@@ -88,6 +89,27 @@ impl HostService for HostApiHandler {
         }
     }
 
+    async fn get_network_info(
+        &self,
+        _request: Request<GetNetworkInfoRequest>,
+    ) -> Result<Response<GetNetworkInfoResponse>, Status> {
+        info!("HOST_API_HANDLER: Received GetNetworkInfo request.");
+        let (resp_tx, resp_rx) = oneshot::channel();
+        let cmd = Command::GetNetworkInfo(resp_tx);
+        self.dispatcher_tx
+            .send(cmd)
+            .await
+            .map_err(|e| Status::internal(format!("Failed to send command to dispatcher: {e}")))?;
+
+        match resp_rx.await {
+            Ok(Ok(result)) => Ok(Response::new(result)),
+            Ok(Err(status)) => Err(status),
+            Err(_) => Err(Status::internal(
+                "Dispatcher task dropped response channel.",
+            )),
+        }
+    }
+
     async fn shutdown(
         &self,
         request: Request<ShutdownRequest>,

feos/services/host-service/src/dispatcher.rs

@@ -25,6 +25,9 @@ impl HostServiceDispatcher {
                 Command::GetCPUInfo(responder) => {
                     tokio::spawn(worker::handle_get_cpu_info(responder));
                 }
+                Command::GetNetworkInfo(responder) => {
+                    tokio::spawn(worker::handle_get_network_info(responder));
+                }
                 Command::UpgradeFeosBinary(stream, responder) => {
                     let restart_tx = self.restart_tx.clone();
                     tokio::spawn(worker::handle_upgrade(restart_tx, *stream, responder));

feos/services/host-service/src/lib.rs

@@ -1,6 +1,7 @@
 use feos_proto::host_service::{
-    GetCpuInfoResponse, HostnameResponse, KernelLogEntry, MemoryResponse, RebootRequest,
-    RebootResponse, ShutdownRequest, ShutdownResponse, UpgradeRequest, UpgradeResponse,
+    GetCpuInfoResponse, GetNetworkInfoResponse, HostnameResponse, KernelLogEntry, MemoryResponse,
+    RebootRequest, RebootResponse, ShutdownRequest, ShutdownResponse, UpgradeRequest,
+    UpgradeResponse,
 };
 use std::path::PathBuf;
 use tokio::sync::{mpsc, oneshot};
@@ -15,6 +16,7 @@ pub enum Command {
     GetHostname(oneshot::Sender<Result<HostnameResponse, Status>>),
     GetMemory(oneshot::Sender<Result<MemoryResponse, Status>>),
     GetCPUInfo(oneshot::Sender<Result<GetCpuInfoResponse, Status>>),
+    GetNetworkInfo(oneshot::Sender<Result<GetNetworkInfoResponse, Status>>),
     UpgradeFeosBinary(
         Box<Streaming<UpgradeRequest>>,
         oneshot::Sender<Result<UpgradeResponse, Status>>,

feos/services/host-service/src/worker.rs

@@ -1,9 +1,9 @@
 use crate::RestartSignal;
 use digest::Digest;
 use feos_proto::host_service::{
-    upgrade_request, CpuInfo, GetCpuInfoResponse, HostnameResponse, KernelLogEntry, MemInfo,
-    MemoryResponse, RebootRequest, RebootResponse, ShutdownRequest, ShutdownResponse,
-    UpgradeRequest, UpgradeResponse,
+    upgrade_request, CpuInfo, GetCpuInfoResponse, GetNetworkInfoResponse, HostnameResponse,
+    KernelLogEntry, MemInfo, MemoryResponse, NetDev, RebootRequest, RebootResponse,
+    ShutdownRequest, ShutdownResponse, UpgradeRequest, UpgradeResponse,
 };
 use log::{error, info, warn};
 use nix::sys::reboot::{reboot, RebootMode};
@@ -13,9 +13,9 @@ use std::collections::HashMap;
 use std::fs::Permissions;
 use std::io::Write;
 use std::os::unix::fs::PermissionsExt;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use tempfile::NamedTempFile;
-use tokio::fs::File;
+use tokio::fs::{self, File};
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
@@ -235,6 +235,81 @@ pub async fn handle_get_cpu_info(responder: oneshot::Sender<Result<GetCpuInfoRes
     }
 }
 
+async fn read_net_stat(base_path: &Path, stat_name: &str) -> u64 {
+    let stat_path = base_path.join(stat_name);
+    fs::read_to_string(stat_path)
+        .await
+        .ok()
+        .and_then(|s| s.trim().parse::<u64>().ok())
+        .unwrap_or(0)
+}
+
+async fn read_all_net_stats() -> Result<Vec<NetDev>, std::io::Error> {
+    let mut devices = Vec::new();
+    let mut entries = fs::read_dir("/sys/class/net").await?;
+
+    while let Some(entry) = entries.next_entry().await? {
+        let path = entry.path();
+        if !path.is_dir() {
+            continue;
+        }
+
+        let name = entry
+            .file_name()
+            .into_string()
+            .unwrap_or_else(|_| "invalid_utf8".to_string());
+        let stats_path = path.join("statistics");
+
+        if !stats_path.is_dir() {
+            continue;
+        }
+
+        let device = NetDev {
+            name,
+            rx_bytes: read_net_stat(&stats_path, "rx_bytes").await,
+            rx_packets: read_net_stat(&stats_path, "rx_packets").await,
+            rx_errors: read_net_stat(&stats_path, "rx_errors").await,
+            rx_dropped: read_net_stat(&stats_path, "rx_dropped").await,
+            rx_fifo: read_net_stat(&stats_path, "rx_fifo_errors").await,
+            rx_frame: read_net_stat(&stats_path, "rx_frame_errors").await,
+            rx_compressed: read_net_stat(&stats_path, "rx_compressed").await,
+            rx_multicast: read_net_stat(&stats_path, "multicast").await,
+            tx_bytes: read_net_stat(&stats_path, "tx_bytes").await,
+            tx_packets: read_net_stat(&stats_path, "tx_packets").await,
+            tx_errors: read_net_stat(&stats_path, "tx_errors").await,
+            tx_dropped: read_net_stat(&stats_path, "tx_dropped").await,
+            tx_fifo: read_net_stat(&stats_path, "tx_fifo_errors").await,
+            tx_collisions: read_net_stat(&stats_path, "collisions").await,
+            tx_carrier: read_net_stat(&stats_path, "tx_carrier_errors").await,
+            tx_compressed: read_net_stat(&stats_path, "tx_compressed").await,
+        };
+        devices.push(device);
+    }
+
+    Ok(devices)
+}
+
+pub async fn handle_get_network_info(
+    responder: oneshot::Sender<Result<GetNetworkInfoResponse, Status>>,
+) {
+    info!("HOST_WORKER: Processing GetNetworkInfo request.");
+    let result = match read_all_net_stats().await {
+        Ok(devices) => Ok(GetNetworkInfoResponse { devices }),
+        Err(e) => {
+            error!("HOST_WORKER: ERROR - Failed to get network info: {e}");
+            Err(Status::internal(format!(
+                "Failed to get network info from sysfs: {e}"
+            )))
+        }
+    };
+
+    if responder.send(result).is_err() {
+        error!(
+            "HOST_WORKER: Failed to send response for GetNetworkInfo. API handler may have timed out."
+        );
+    }
+}
+
 pub async fn handle_stream_kernel_logs(grpc_tx: mpsc::Sender<Result<KernelLogEntry, Status>>) {
     info!("HOST_WORKER: Opening {KMSG_PATH} for streaming kernel logs.");
 

feos/services/vm-service/src/dispatcher.rs

@@ -12,7 +12,7 @@ use feos_proto::{
         VmStateChangedEvent,
     },
 };
-use log::{error, info, warn, debug};
+use log::{debug, error, info, warn};
 use prost::Message;
 use prost_types::Any;
 use std::sync::Arc;

feos/services/vm-service/src/worker.rs

@@ -211,7 +211,9 @@ pub async fn handle_stream_vm_events(
     loop {
         match broadcast_rx.recv().await {
             Ok(VmEventWrapper { event, .. }) => {
-                if vm_id_to_watch.as_ref().is_none_or(|id| event.vm_id == *id) && stream_tx.send(Ok(event)).await.is_err() {
+                if vm_id_to_watch.as_ref().is_none_or(|id| event.vm_id == *id)
+                    && stream_tx.send(Ok(event)).await.is_err()
+                {
                     info!("VM_WORKER (Stream): Client for '{watcher_desc}' disconnected.");
                     break;
                 }

feos/tests/integration_tests.rs

@@ -1,7 +1,8 @@
 use anyhow::{Context, Result};
 use feos_proto::{
     host_service::{
-        host_service_client::HostServiceClient, GetCpuInfoRequest, HostnameRequest, MemoryRequest,
+        host_service_client::HostServiceClient, GetCpuInfoRequest, GetNetworkInfoRequest,
+        HostnameRequest, MemoryRequest,
     },
     image_service::{
         image_service_client::ImageServiceClient, DeleteImageRequest, ImageState,
@@ -434,6 +435,39 @@ async fn test_get_cpu_info() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_get_network_info() -> Result<()> {
+    ensure_server().await;
+    let (_, mut host_client) = get_public_clients().await?;
+
+    info!("Sending GetNetworkInfo request");
+    let response = host_client
+        .get_network_info(GetNetworkInfoRequest {})
+        .await?
+        .into_inner();
+
+    assert!(
+        !response.devices.is_empty(),
+        "The list of network devices should not be empty"
+    );
+    info!("Received {} network devices", response.devices.len());
+
+    let lo = response
+        .devices
+        .iter()
+        .find(|d| d.name == "lo")
+        .context("Could not find the loopback interface 'lo'")?;
+
+    info!("Found loopback interface 'lo'");
+    assert_eq!(lo.name, "lo");
+    assert!(
+        lo.rx_packets > 0 || lo.tx_packets > 0,
+        "Loopback interface should have some packets transferred"
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_image_lifecycle() -> Result<()> {
     if skip_if_ch_binary_missing() {

proto/v1/host.proto

@@ -8,6 +8,8 @@ service HostService {
   rpc Hostname(HostnameRequest) returns (HostnameResponse);
   rpc GetMemory(MemoryRequest) returns (MemoryResponse);
   rpc GetCPUInfo(GetCPUInfoRequest) returns (GetCPUInfoResponse);
+  // Retrieves statistics for all network interfaces.
+  rpc GetNetworkInfo(GetNetworkInfoRequest) returns (GetNetworkInfoResponse);
   rpc Shutdown(ShutdownRequest) returns (ShutdownResponse);
   rpc Reboot(RebootRequest) returns (RebootResponse);
 
@@ -148,4 +150,30 @@ message CPUInfo {
   uint32 cache_alignment = 24;
   string address_sizes = 25;
   string power_management = 26;
+}
+
+message GetNetworkInfoRequest {}
+
+message GetNetworkInfoResponse {
+  repeated NetDev devices = 1;
+}
+
+message NetDev {
+  string name = 1;
+  uint64 rx_bytes = 2;
+  uint64 rx_packets = 3;
+  uint64 rx_errors = 4;
+  uint64 rx_dropped = 5;
+  uint64 rx_fifo = 6;
+  uint64 rx_frame = 7;
+  uint64 rx_compressed = 8;
+  uint64 rx_multicast = 9;
+  uint64 tx_bytes = 10;
+  uint64 tx_packets = 11;
+  uint64 tx_errors = 12;
+  uint64 tx_dropped = 13;
+  uint64 tx_fifo = 14;
+  uint64 tx_collisions = 15;
+  uint64 tx_carrier = 16;
+  uint64 tx_compressed = 17;
 }