OpenTelemetry Protcol (OTLP) identity integration (#180)

* [#158]: moved map_handlers module from identity crate to common crate

* [#158]: simplified identity logic. removed duplicated code and functions

* [#158]: added program handlers function in the common crate. Remove duplicated code in metrics module

* [#168]: added load from program-handlers in identity user space implementation. Added a small doc in the conntracker/main.rs file

* [#158]: fixed typos in the map names

* [#158]: fixed bpf error: Error: the program is already loaded.Improved map handlers code

* [#174]: added open telemetry (otel) logger for logs. Added otel daemonset with otel agent and collector

* [#158]: improved docs for the conntracker data structures. removed useless conversion from u8 to 64 with .into() for state variable

* [#174]: Added otel libraries and features in the common crate. .update identity kubernetes manifest with the otel env variables.

* [#158]: imroved documentation in the user space for the identity (VethLog) data structure

* [#158]: restored blocklist map initialization

* [#158]: added better docs. Updated while true pattern with "loop" pattern.  Code cleaning

* [#174]: added prettify to logger

* [#181]: added command to repair blocklist configmaps

* [#182]: added GetTrackedVeth grpc endpoint definition

* [#158]: added load_perf_event_array_from_mapdata function in map_handlers.rs

* [#182]: added total monitored veth_events (tot_monitored_veth)

* [#182]: added "cfcli monitoring veth" command frontend. added send_tracked_veth_requests function in api/requests.rs

* [refactoring]: separate experimental service discovery from the helpers in the identity service

* [refactoring]: created BufferType enum to centralize the event readers for PacketLog, VethLog and TcpPacketRegistry

* updated common cargo.toml

* [#158]: moved IpProtocols and network structures from the identity crate to the common crate. Added BufferType enum to list different buffers readers. Added buffer_type module in cortexbrain common (experimental)

* [update]: cli packages update

* [update]: IpProtocols update . Btter code formatting

* [#158]: fixed typos from latest commit

* [#182]: Added tracked events agent API implementation. Added VethEvent protobuffer message type

* [proposal]: added batcher module intial bones

* [158]: removed old deprecated functions to show veth logs. added packed representation for the VethLog structure. changed dev_addr type from [u32;8] to [u8;6] (see https://wiki.osdev.org/Address_Resolution_Protocol). Added unit tests to check VethLog structure bytes size. Changed "actual program" to "kernel_symbol" in the load_program function

* [#158]: added a map_manager to call BpfMaps using the map name instead of relying on the declaration order during the startup

* [#182]: added VethEvent protobuf message. Added log to see the veth event in the agent api

* [docker] pushed new development images

Author: LorenzoTettamanti

cli/Cargo.lock

@@ -22,7 +22,7 @@ tonic = "0.14.2"
 tonic-reflection = "0.14.2"
 prost-types = "0.14.3"
 prost = "0.14.3"
-cortexflow_agent_api = {version = "0.1.1",features = ["client"]}
+cortexflow_agent_api = {path = "../core/api",features = ["client"]}
 kube = "2.0.1"
 k8s-openapi = {version = "0.26.0", features = ["v1_34"]}
 

cli/Cargo.toml

@@ -2,8 +2,10 @@ use crate::errors::CliError;
 use crate::essential::{BASE_COMMAND, connect_to_client, create_config_file, create_configs};
 use clap::{Args, Subcommand};
 use colored::Colorize;
-use kube::Error;
+use k8s_openapi::api::core::v1::ConfigMap;
 use kube::core::ErrorResponse;
+use kube::{Api, Client, Error};
+use std::thread::sleep;
 use std::{process::Command, thread, time::Duration};
 
 // docs:
@@ -38,6 +40,8 @@ pub enum InstallCommands {
         about = "Deploys a simple example contained in deploy-test-pod.yaml"
     )]
     TestPods,
+    #[command(name = "blocklist", about = "Install or Repair blocklist configmap")]
+    Blocklist,
 }
 
 //install args
@@ -206,6 +210,84 @@ async fn install_simple_example_component() -> Result<(), CliError> {
     }
 }
 
+// docs:
+pub async fn install_blocklist_configmap() -> Result<(), CliError> {
+    match connect_to_client().await {
+        Ok(client) => {
+            println!(
+                "{} {}",
+                "=====>".blue().bold(),
+                "Checking if the Blocklist configmap exists"
+            );
+            sleep(Duration::from_secs(1));
+            let blocklist_exists = check_if_blocklist_exists(client).await?;
+            if !blocklist_exists {
+                println!(
+                    "{} {}",
+                    "=====>".blue().bold(),
+                    "Blocklist configmap does not exist".red().bold()
+                );
+                sleep(Duration::from_secs(1));
+                println!("{} {}", "=====>".bold().blue(), "Creating configmap");
+                let metdata_configs = create_configs();
+                sleep(Duration::from_secs(1));
+                match create_config_file(metdata_configs).await {
+                    Ok(_) => {
+                        println!(
+                            "{} {}",
+                            "=====>".bold().blue(),
+                            "Configmap created/repaired successfully".bold().green()
+                        )
+                    }
+                    Err(e) => {
+                        return Err(CliError::InstallerError {
+                            reason: e.to_string(),
+                        });
+                    }
+                }
+                return Ok(());
+            } else {
+                println!()
+            }
+
+            Ok(())
+        }
+        Err(e) => {
+            return Err(CliError::ClientError(Error::Api(ErrorResponse {
+                status: "failed".to_string(),
+                message: "Failed to connect to kubernetes client".to_string(),
+                reason: e.to_string(),
+                code: 404,
+            })));
+        }
+    }
+}
+
+// docs:
+async fn check_if_blocklist_exists(client: Client) -> Result<bool, CliError> {
+    let namespace = "cortexflow";
+    let name = "cortexbrain-client-config";
+    let api: Api<ConfigMap> = Api::namespaced(client, namespace);
+    match api.get(name).await {
+        Ok(_) => {
+            println!(
+                "{} {}",
+                "=====>".bold().blue(),
+                "Blocklist configmap exists".green().bold()
+            );
+            Ok(true)
+        }
+        Err(_) => {
+            println!(
+                "{} {}",
+                "=====>".bold().blue(),
+                "Blocklist configmap doesn not exists".red().bold(),
+            );
+            Ok(false)
+        }
+    }
+}
+
 //docs:
 //
 // This is an auxiliary function to help manage the cortexflow components during the installation

cli/src/install.rs

@@ -18,7 +18,7 @@ use crate::install::{InstallArgs, InstallCommands, install_cortexflow, install_s
 use crate::logs::{LogsArgs, logs_command};
 use crate::monitoring::{
     MonitorArgs, MonitorCommands, list_features, monitor_dropped_packets, monitor_identity_events,
-    monitor_latency_metrics,
+    monitor_latency_metrics, monitor_tracked_veth,
 };
 use crate::policies::{
     PoliciesArgs, PoliciesCommands, check_blocklist, create_blocklist, remove_ip,
@@ -68,7 +68,7 @@ enum Commands {
 struct SetArgs {
     val: String,
 }
-
+//TODO: add command for monitoring veth interfaces
 async fn args_parser() -> Result<(), CliError> {
     let args = Cli::parse();
     debug!("Arguments {:?}", args.cmd);
@@ -80,6 +80,10 @@ async fn args_parser() -> Result<(), CliError> {
             InstallCommands::TestPods => {
                 install_simple_example().await?;
             }
+            InstallCommands::Blocklist => {
+                //install or repair blocklist configmap
+                let _ = install::install_blocklist_configmap().await?;
+            }
         },
         Some(Commands::Uninstall) => {
             uninstall().await?;
@@ -120,6 +124,9 @@ async fn args_parser() -> Result<(), CliError> {
             MonitorCommands::Droppedpackets => {
                 let _ = monitor_dropped_packets().await?;
             }
+            MonitorCommands::Veth => {
+                let _ = monitor_tracked_veth().await?;
+            }
         },
         Some(Commands::Policies(policies_args)) => {
             match policies_args.policy_cmd {

cli/src/main.rs

@@ -8,7 +8,10 @@ use std::result::Result::Ok;
 use tonic_reflection::pb::v1::server_reflection_response::MessageResponse;
 
 use agent_api::client::{connect_to_client, connect_to_server_reflection};
-use agent_api::requests::{get_all_features, send_active_connection_request};
+use agent_api::requests::{
+    get_all_features, send_active_connection_request, send_dropped_packets_request,
+    send_latency_metrics_request, send_tracked_veth_request,
+};
 
 use crate::errors::CliError;
 use clap::{Args, Subcommand};
@@ -33,15 +36,18 @@ pub enum MonitorCommands {
         about = "Monitor the dropped packets metrics detected by the metrics service"
     )]
     Droppedpackets,
+    #[command(
+        name = "veth",
+        about = "Monitor tracked veth interfaces from the identity service"
+    )]
+    Veth,
 }
 
 // cfcli monitor <args>
 #[derive(Args, Debug, Clone)]
 pub struct MonitorArgs {
     #[command(subcommand)]
     pub monitor_cmd: MonitorCommands,
-    //#[arg(long, short)]
-    //pub flags: Option<String>,
 }
 
 pub async fn list_features() -> Result<(), CliError> {
@@ -168,7 +174,7 @@ pub async fn monitor_latency_metrics() -> Result<(), CliError> {
                 "Connected to CortexFlow Client".green()
             );
             //send request to get latency metrics
-            match agent_api::requests::send_latency_metrics_request(client).await {
+            match send_latency_metrics_request(client).await {
                 Ok(response) => {
                     let resp = response.into_inner();
                     if resp.metrics.is_empty() {
@@ -237,7 +243,7 @@ pub async fn monitor_dropped_packets() -> Result<(), CliError> {
                 "Connected to CortexFlow Client".green()
             );
             //send request to get dropped packets metrics
-            match agent_api::requests::send_dropped_packets_request(client).await {
+            match send_dropped_packets_request(client).await {
                 Ok(response) => {
                     let resp = response.into_inner();
                     if resp.metrics.is_empty() {
@@ -291,6 +297,50 @@ pub async fn monitor_dropped_packets() -> Result<(), CliError> {
     Ok(())
 }
 
+pub async fn monitor_tracked_veth() -> Result<(), CliError> {
+    println!(
+        "{} {}",
+        "=====>".blue().bold(),
+        "Connecting to cortexflow Client".white()
+    );
+    match connect_to_client().await {
+        Ok(client) => match send_tracked_veth_request(client).await {
+            Ok(response) => {
+                let veth_response = response.into_inner();
+                if veth_response.tot_monitored_veth == 0 {
+                    println!("{} {} ", "=====>".blue().bold(), "No tracked veth found");
+                    Ok(())
+                } else {
+                    println!(
+                        "{} {} {} {} ",
+                        "=====>".blue().bold(),
+                        "Found:",
+                        &veth_response.tot_monitored_veth,
+                        "tracked veth"
+                    );
+                    for veth in veth_response.veth_names.iter() {
+                        println!("{} {}", "=====>".blue().bold(), &veth);
+                    }
+                    Ok(())
+                }
+            }
+            Err(e) => {
+                return Err(CliError::AgentError(
+                    tonic_reflection::server::Error::InvalidFileDescriptorSet(e.to_string()),
+                ));
+            }
+        },
+        Err(e) => {
+            return Err(CliError::ClientError(kube::Error::Api(ErrorResponse {
+                status: "failed".to_string(),
+                message: "Failed to connect to kubernetes client".to_string(),
+                reason: e.to_string(),
+                code: 404,
+            })));
+        }
+    }
+}
+
 fn convert_timestamp_to_date(timestamp: u64) -> String {
     DateTime::from_timestamp_micros(timestamp as i64)
         .map(|dt| dt.to_string())