CortexFlow[#143]: CLI changes & packet loss metrics

Author: siddh34

cli/src/main.rs

@@ -16,7 +16,7 @@ use tracing::debug;
 use crate::essential::{ CliError, info, update_cli };
 use crate::install::{ InstallArgs, InstallCommands, install_cortexflow, install_simple_example };
 use crate::logs::{ LogsArgs, logs_command };
-use crate::monitoring::{ MonitorArgs, MonitorCommands, list_features, monitor_identity_events };
+use crate::monitoring::{ MonitorArgs, MonitorCommands, list_features, monitor_dropped_packets, monitor_identity_events, monitor_latency_metrics };
 use crate::policies::{
     PoliciesArgs,
     PoliciesCommands,
@@ -109,6 +109,12 @@ async fn args_parser() -> Result<(), CliError> {
                 MonitorCommands::Connections => {
                     let _ = monitor_identity_events().await.map_err(|e| eprintln!("{}",e) )?;
                 }
+                MonitorCommands::Latencymetrics => {
+                    let _ = monitor_latency_metrics().await.map_err(|e| eprintln!("{}",e) )?;
+                }
+                MonitorCommands::Droppedpackets => {
+                    let _ = monitor_dropped_packets().await.map_err(|e| eprintln!("{}",e) )?;
+                }
             }
         Some(Commands::Policies(policies_args)) => {
             match policies_args.policy_cmd {

cli/src/monitoring.rs

@@ -23,6 +23,14 @@ pub enum MonitorCommands {
         name = "connections",
         about = "Monitor the recent connections detected by the identity service"
     )] Connections,
+    #[command(
+        name = "latencymetrics",
+        about = "Monitor the latency metrics detected by the metrics service"
+    )] Latencymetrics,
+    #[command(
+        name = "droppedpackets",
+        about = "Monitor the dropped packets metrics detected by the metrics service"
+    )] Droppedpackets,
 }
 
 // cfcli monitor <args>
@@ -140,3 +148,120 @@ pub async fn monitor_identity_events() -> Result<(), Error> {
 
     Ok(())
 }
+
+pub async fn monitor_latency_metrics() -> Result<(), Error> {
+    //function to monitor latency metrics
+    println!("{} {}", "=====>".blue().bold(), "Connecting to cortexflow Client".white());
+
+    match connect_to_client().await {
+        Ok(client) => {
+            println!("{} {}", "=====>".blue().bold(), "Connected to CortexFlow Client".green());
+            //send request to get latency metrics
+            match agent_api::requests::send_latency_metrics_request(client).await {
+                Ok(response) => {
+                    let resp = response.into_inner();
+                    if resp.metrics.is_empty() {
+                        println!("{} No latency metrics found", "=====>".blue().bold());
+                    } else {
+                        println!("{} Found {} latency metrics", "=====>".blue().bold(), resp.metrics.len());
+                        for (i, metric) in resp.metrics.iter().enumerate() {
+                            println!(
+                                "index {} Latency[{}], tgid {} process_name {} address_family {} delta_us {} src_address_v4 {} dst_address_v4 {} src_address_v6 {} dst_address_v6 {} local_port {} remote_port {} timestamp_us {}",
+                                "=====>".blue().bold(),
+                                i,
+                                metric.tgid,
+                                metric.process_name,
+                                metric.address_family,
+                                metric.delta_us,
+                                metric.src_address_v4,
+                                metric.dst_address_v4,
+                                format!("{:?}", metric.src_address_v6),
+                                format!("{:?}", metric.dst_address_v6),
+                                metric.local_port,
+                                metric.remote_port,
+                                metric.timestamp_us
+                            );
+                        }
+                    }
+                }
+                Err(e) => {
+                    println!(
+                        "{} {} {} {}",
+                        "=====>".blue().bold(),
+                        "An error occured".red(),
+                        "Error:",
+                        e
+                    );
+                    return Err(e);
+                }
+            }
+        }
+        Err(e) =>{
+            println!(
+                "{} {}",
+                "=====>".blue().bold(),
+                "Failed to connect to CortexFlow Client".red()
+            );
+            return Err(e);
+        }
+    }
+    Ok(())
+}
+
+pub async fn monitor_dropped_packets() -> Result<(), Error> {
+    //function to monitor dropped packets metrics
+    println!("{} {}", "=====>".blue().bold(), "Connecting to cortexflow Client".white());
+
+    match connect_to_client().await {
+        Ok(client) => {
+            println!("{} {}", "=====>".blue().bold(), "Connected to CortexFlow Client".green());
+            //send request to get dropped packets metrics
+            match agent_api::requests::send_dropped_packets_request(client).await {
+                Ok(response) => {
+                    let resp = response.into_inner();
+                    if resp.metrics.is_empty() {
+                        println!("{} No dropped packets metrics found", "=====>".blue().bold());
+                    } else {
+                        println!("{} Found {} dropped packets metrics", "=====>".blue().bold(), resp.metrics.len());
+                        for (i, metric) in resp.metrics.iter().enumerate() {
+                            println!(
+                                "{} DroppedPackets[{}]\n  TGID: {}\n  Process: {}\n  SK Drops: {}\n  Socket Errors: {}\n  Soft Errors: {}\n  Backlog Length: {}\n  Write Memory Queued: {}\n  Receive Buffer Size: {}\n  ACK Backlog: {}\n  Timestamp: {} µs",
+                                "=====>".blue().bold(),
+                                i,
+                                metric.tgid,
+                                metric.process_name,
+                                metric.sk_drops,
+                                metric.sk_err,
+                                metric.sk_err_soft,
+                                metric.sk_backlog_len,
+                                metric.sk_wmem_queued,
+                                metric.sk_rcvbuf,
+                                metric.sk_ack_backlog,
+                                metric.timestamp_us
+                            );
+                        }
+                    }
+                }
+                Err(e) => {
+                    println!(
+                        "{} {} {} {}",
+                        "=====>".blue().bold(),
+                        "An error occured".red(),
+                        "Error:",
+                        e
+                    );
+                    return Err(e);
+                }
+            }
+        }
+        Err(e) =>{
+            println!(
+                "{} {}",
+                "=====>".blue().bold(),
+                "Failed to connect to CortexFlow Client".red()
+            );
+            return Err(e);
+        }
+    }
+    Ok(())
+}

core/api/protos/agent.proto

@@ -22,17 +22,8 @@ message ActiveConnectionResponse{
 
 // Network metrics
 
-
-
 // Latency metrics request and response messages
 
-message LatencyMetricsRequest {
-    optional uint32 tgid = 1;         // Filter by thread group ID
-    optional string process_name = 2; // Filter by process name
-    optional uint64 start_time = 3;   // Start timestamp (microseconds)
-    optional uint64 end_time = 4;     // End timestamp (microseconds)
-}
-
 message LatencyMetric {
     uint64 delta_us = 1;          // Latency in microseconds
     uint64 timestamp_us = 2;      // Event timestamp
@@ -56,40 +47,8 @@ message LatencyMetricsResponse {
     double max_latency_us = 6;        // Maximum latency
 }
 
-// Packet Loss Metrics
-
-message PacketLossMetricsRequest {
-    optional uint32 tgid = 1;         // Filter by thread group ID
-    optional uint64 start_time = 2;   // Start timestamp
-    optional uint64 end_time = 3;     // End timestamp
-}
-
-message PacketLossMetric {
-    uint32 tgid = 1;                          // Thread group ID
-    string process_name = 2;                  // Process name
-    uint64 timestamp_us = 3;                  // Event timestamp
-    uint32 total_packets_lost = 4;            // Total packets lost
-    uint32 total_packets_transmitted = 5;     // Total packets transmitted
-    double packet_loss_percentage = 6;        // % of total packet loss
-    uint64 total_data_loss_bytes = 7;         // Total size of data loss
-    uint64 total_data_transmitted_bytes = 8;  // Total transmitted data
-    double data_loss_ratio = 9;               // Ratio between loss and transmitted
-}
-
-message PacketLossMetricsResponse {
-    string status = 1;
-    repeated PacketLossMetric metrics = 2;
-    uint32 total_connections = 3;
-}
-
 // Dropped TCP Packets
 
-message DroppedPacketsRequest {
-    optional uint32 tgid = 1;         // Filter by thread group ID
-    optional uint64 start_time = 2;   // Start timestamp
-    optional uint64 end_time = 3;     // End timestamp
-}
-
 message DroppedPacketMetric {
     uint32 tgid = 1;                  // Thread group ID
     string process_name = 2;          // Process name
@@ -112,7 +71,6 @@ message DroppedPacketsResponse {
 
 //declare agent api 
 service Agent{
-    
     // active connections endpoint 
     rpc ActiveConnections(RequestActiveConnections) returns (ActiveConnectionResponse);
 
@@ -124,11 +82,10 @@ service Agent{
     rpc RmIpFromBlocklist(RmIpFromBlocklistRequest) returns (RmIpFromBlocklistResponse);
 
     // metrics data
-    rpc GetLatencyMetrics(LatencyMetricsRequest) returns (LatencyMetricsResponse);
-
-    rpc GetPacketLossMetrics(PacketLossMetricsRequest) returns (PacketLossMetricsResponse);
+    rpc GetLatencyMetrics(google.protobuf.Empty) returns (LatencyMetricsResponse);
 
-    rpc GetDroppedPacketsMetrics(DroppedPacketsRequest) returns (DroppedPacketsResponse);
+    // dropped packets
+    rpc GetDroppedPacketsMetrics(google.protobuf.Empty) returns (DroppedPacketsResponse);
 }
 
 message AddIpToBlocklistRequest{