Labels on Metadata are able to identify the workload a series belongs to

.github/workflows/core-build-checks.yml

@@ -51,6 +51,21 @@ jobs:
       working-directory: core
       run: cargo test --locked -p cortexbrain-common --test public_contracts --no-default-features --features network-structs,monitoring-structs
 
+  test-cli:
+    name: Test CLI
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Install prerequisites
+      run: sudo apt-get install -y protobuf-compiler
+
+    - name: Test CLI
+      working-directory: cli
+      run: cargo test --locked
+
   build-core-components:
     runs-on: ubuntu-latest
 

Examples/run-with-docker/docker-compose.yaml

@@ -49,7 +49,7 @@ services:
   #    - cortexflow
 
   otel-collector:
-    image: otel/opentelemetry-collector:0.95.0
+    image: otel/opentelemetry-collector-contrib:0.95.0
     container_name: otel-collector
     command:
       - "--config=/conf/otel-collector-config.yaml"

Examples/run-with-docker/otel-collector-config.yaml

@@ -11,6 +11,13 @@ processors:
     limit_mib: 1500
     spike_limit_mib: 512
     check_interval: 5s
+  transform/identity:
+    error_mode: ignore
+    metric_statements:
+      - context: datapoint
+        statements:
+          - set(attributes["container.name"], attributes["k8s.pod.name"])
+            where attributes["container.name"] == nil and attributes["k8s.pod.name"] != nil
 
 exporters:
   logging: {}
@@ -30,5 +37,5 @@ service:
       exporters: [logging]
     metrics:
       receivers: [otlp]
-      processors: [memory_limiter]
+      processors: [memory_limiter, transform/identity]
       exporters: [logging, prometheus]

cli/Cargo.toml

@@ -26,6 +26,13 @@ cortexflow_agent_api = {version="0.1.2",features = ["client"]}
 kube = "2.0.1"
 k8s-openapi = {version = "0.26.0", features = ["v1_34"]}
 
+[dev-dependencies]
+serde_json = "1.0"
+tower = { version = "0.5", features = ["util"] }
+http = "1"
+http-body-util = "0.1"
+bytes = "1"
+
 [[bin]]
 name = "cfcli"
 path = "src/main.rs"

cli/src/command_runner.rs

@@ -0,0 +1,76 @@
+use std::process::{Command, Output};
+
+// docs:
+//
+// Abstraction over external process execution (kubectl, cargo, ...) so callers
+// can inject a stub in tests instead of shelling out to a real binary/cluster.
+
+pub trait CommandRunner {
+    fn run(&self, program: &str, args: &[String]) -> std::io::Result<Output>;
+}
+
+pub struct RealCommandRunner;
+
+impl CommandRunner for RealCommandRunner {
+    fn run(&self, program: &str, args: &[String]) -> std::io::Result<Output> {
+        Command::new(program).args(args).output()
+    }
+}
+
+#[cfg(test)]
+pub mod test_support {
+    use super::CommandRunner;
+    use std::process::{ExitStatus, Output};
+
+    #[cfg(unix)]
+    fn exit_status(success: bool) -> ExitStatus {
+        use std::os::unix::process::ExitStatusExt;
+        ExitStatus::from_raw(if success { 0 } else { 1 << 8 })
+    }
+
+    // Stubs process output so tests never shell out to a real binary or cluster.
+    pub struct StubCommandRunner {
+        result: std::io::Result<Output>,
+    }
+
+    impl StubCommandRunner {
+        pub fn success(stdout: &str) -> Self {
+            Self {
+                result: Ok(Output {
+                    status: exit_status(true),
+                    stdout: stdout.as_bytes().to_vec(),
+                    stderr: Vec::new(),
+                }),
+            }
+        }
+
+        pub fn failure(stderr: &str) -> Self {
+            Self {
+                result: Ok(Output {
+                    status: exit_status(false),
+                    stdout: Vec::new(),
+                    stderr: stderr.as_bytes().to_vec(),
+                }),
+            }
+        }
+
+        pub fn io_error() -> Self {
+            Self {
+                result: Err(std::io::Error::other("command not found")),
+            }
+        }
+    }
+
+    impl CommandRunner for StubCommandRunner {
+        fn run(&self, _program: &str, _args: &[String]) -> std::io::Result<Output> {
+            match &self.result {
+                Ok(output) => Ok(Output {
+                    status: output.status,
+                    stdout: output.stdout.clone(),
+                    stderr: output.stderr.clone(),
+                }),
+                Err(_) => Err(std::io::Error::other("command not found")),
+            }
+        }
+    }
+}

cli/src/errors.rs

@@ -122,3 +122,18 @@ impl fmt::Display for CliError {
         }
     }
 }
+
+# [cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_display_base_error() {
+        let err = CliError::BaseError {
+            reason: "some reason".to_string(),
+        };
+        let err_str = format!("{}", err);
+        assert!(err_str.contains("An error occured. Reason:"));
+        assert!(err_str.contains("some reason"));
+    }
+}

cli/src/essential.rs

@@ -1,3 +1,4 @@
+use crate::command_runner::{CommandRunner, RealCommandRunner};
 use crate::errors::CliError;
 use std::borrow::Cow;
 use std::thread;
@@ -130,10 +131,17 @@ pub fn update_cli() -> Result<(), CliError> {
 //
 // This function returns the latest version of the CLI from the crates.io registry
 pub fn get_latest_cfcli_version() -> Result<String, CliError> {
-    let output = Command::new("cargo")
-        .args(["search", "cortexflow-cli", "--limit", "1"])
-        .output()
-        .expect("Error");
+    get_latest_cfcli_version_with(&RealCommandRunner)
+}
+
+fn get_latest_cfcli_version_with(runner: &dyn CommandRunner) -> Result<String, CliError> {
+    let args = [
+        "search".to_string(),
+        "cortexflow-cli".to_string(),
+        "--limit".to_string(),
+        "1".to_string(),
+    ];
+    let output = runner.run("cargo", &args).expect("Error");
 
     if !output.status.success() {
         return Err(CliError::InstallerError {
@@ -410,10 +418,18 @@ pub async fn update_configmap(config_struct: MetadataConfigFile) -> Result<(), C
 
 #[cfg(test)]
 mod tests {
-    use crate::essential::extract_version_from_output;
+    use crate::command_runner::test_support::StubCommandRunner;
+    use crate::essential::{create_configs, extract_version_from_output, get_latest_cfcli_version_with};
+
+    #[test]
+    fn creates_an_empty_blocklist_configuration() {
+        let configs = create_configs();
+
+        assert_eq!(configs.blocklist, vec![String::new()]);
+    }
 
     #[test]
-    fn test_version_extraction() {
+    fn extracts_the_version_from_cargo_search_output() {
         let command_stdout = String::from(
             r#"cortexflow-cli = "0.1.4-test_123"    
             # CortexFlow command line interface made to interact with the CortexBrain core components...
@@ -423,4 +439,19 @@ mod tests {
         let extracted_command = extract_version_from_output(command_stdout.into());
         assert_eq!(extracted_command, "0.1.4-test_123");
     }
+
+    #[test]
+    fn get_latest_cfcli_version_returns_parsed_version_on_success() {
+        let runner = StubCommandRunner::success(
+            r#"cortexflow-cli = "0.1.5"    # CortexFlow command line interface"#,
+        );
+        let version = get_latest_cfcli_version_with(&runner);
+        assert_eq!(version.unwrap(), "0.1.5");
+    }
+
+    #[test]
+    fn get_latest_cfcli_version_errors_on_command_failure() {
+        let runner = StubCommandRunner::failure("network unreachable");
+        assert!(get_latest_cfcli_version_with(&runner).is_err());
+    }
 }