chore: cleanup format in some crates and fix fluvio cluster status wh… (

infinyon#3500)

…en there is not access to the underlying k8 context

crates/fluvio-auth/src/x509/authenticator.rs

@@ -10,7 +10,7 @@ use fluvio_future::{net::TcpStream, openssl::DefaultServerTlsStream};
 use fluvio_protocol::api::{RequestMessage, ResponseMessage};
 use flv_tls_proxy::authenticator::Authenticator;
 
-use super::request::{AuthRequest};
+use super::request::AuthRequest;
 
 #[derive(Debug)]
 struct ScopeBindings(HashMap<String, Vec<String>>);

crates/fluvio-cluster/src/charts/chart.rs

@@ -1,4 +1,4 @@
-use std::path::{PathBuf};
+use std::path::PathBuf;
 
 use tracing::{info, debug, instrument};
 use derive_builder::Builder;

crates/fluvio-cluster/src/check/mod.rs

@@ -2,7 +2,7 @@ use std::cmp::Ordering;
 use std::collections::HashSet;
 use std::io::Error as IoError;
 use std::fmt::Debug;
-use std::process::{Command};
+use std::process::Command;
 use std::time::Duration;
 
 pub mod render;
@@ -753,6 +753,12 @@ impl ClusterChecker {
         self
     }
 
+    pub fn with_no_k8_checks(mut self) -> Self {
+        let checks: Vec<Box<(dyn ClusterCheck)>> = vec![Box::new(LocalClusterCheck)];
+        self.checks.extend(checks);
+        self
+    }
+
     /// Adds all checks required for starting a cluster on minikube.
     ///
     /// Note that no checks are run until the [`run`] method is invoked.

crates/fluvio-cluster/src/cli/error.rs

@@ -4,7 +4,7 @@ use fluvio::FluvioError;
 use fluvio_extension_common::output::OutputError;
 use fluvio_extension_common::target::TargetError;
 use crate::check::ClusterCheckError;
-use crate::{LocalInstallError};
+use crate::LocalInstallError;
 use crate::ClusterError;
 
 /// Cluster Command Error

crates/fluvio-cluster/src/cli/shutdown.rs

@@ -1,21 +1,26 @@
-use std::fs::{remove_file};
+use std::fs::remove_file;
+use std::process::Command;
 
 use clap::Parser;
 use tracing::debug;
 use sysinfo::{ProcessExt, System, SystemExt};
 
 use fluvio_types::defaults::SPU_MONITORING_UNIX_SOCKET;
+use fluvio_command::CommandExt;
 
 use crate::render::ProgressRenderer;
-use crate::{cli::ClusterCliError};
+use crate::cli::ClusterCliError;
 use crate::progress::ProgressBarFactory;
-use crate::ClusterError;
+use crate::{ClusterError, UninstallError};
 
 #[derive(Debug, Parser)]
 pub struct ShutdownOpt {
     /// shutdown local spu/sc
     #[arg(long)]
     local: bool,
+
+    #[arg(long)]
+    no_k8: bool,
 }
 
 impl ShutdownOpt {
@@ -74,6 +79,10 @@ impl ShutdownOpt {
         kill_proc("fluvio", Some(&["run".into()]));
         kill_proc("fluvio-run", None);
 
+        if !self.no_k8 {
+            let _ = self.remove_custom_objects("spus", true);
+        }
+
         // remove monitoring socket
         match remove_file(SPU_MONITORING_UNIX_SOCKET) {
             Ok(_) => {
@@ -93,4 +102,18 @@ impl ShutdownOpt {
 
         Ok(())
     }
+
+    /// Remove objects of specified type, namespace
+    fn remove_custom_objects(&self, object_type: &str, force: bool) -> Result<(), UninstallError> {
+        let mut cmd = Command::new("kubectl");
+        cmd.arg("delete");
+        cmd.arg(object_type);
+        cmd.arg("--all");
+        if force {
+            cmd.arg("--force");
+        }
+        cmd.result()?;
+
+        Ok(())
+    }
 }

crates/fluvio-cluster/src/cli/status.rs

@@ -16,7 +16,11 @@ use crate::{cli::ClusterCliError, cli::ClusterTarget};
 use crate::progress::ProgressBarFactory;
 
 #[derive(Debug, Parser)]
-pub struct StatusOpt {}
+pub struct StatusOpt {
+    /// Skip Kubernetes cluster checks
+    #[clap(long)]
+    no_k8: bool,
+}
 
 macro_rules! pad_format {
     ( $e:expr ) => {
@@ -41,7 +45,9 @@ impl StatusOpt {
             Self::profile_name(&config_file).italic()
         ));
 
-        Self::check_k8s_cluster(&pb).await?;
+        if !self.no_k8 {
+            let _ = Self::check_k8s_cluster(&pb).await;
+        }
         Self::check_sc(&pb, &fluvio_config, &config_file).await?;
         Self::check_spus(&pb, &fluvio_config).await?;
         Self::check_topics(&pb, &fluvio_config).await?;

crates/fluvio-cluster/src/delete.rs

@@ -12,7 +12,7 @@ use crate::helm::HelmClient;
 use crate::charts::{APP_CHART_NAME, SYS_CHART_NAME};
 use crate::progress::ProgressBarFactory;
 use crate::render::ProgressRenderer;
-use crate::{DEFAULT_NAMESPACE};
+use crate::DEFAULT_NAMESPACE;
 use crate::error::UninstallError;
 use crate::ClusterError;
 use crate::start::local::DEFAULT_DATA_DIR;
@@ -60,14 +60,20 @@ pub struct ClusterUninstaller {
     /// Configuration options for this process
     config: ClusterUninstallConfig,
     /// Helm client for performing uninstalls
-    helm_client: HelmClient,
+    helm_client: Option<HelmClient>,
     pb_factory: ProgressBarFactory,
 }
 
 impl ClusterUninstaller {
     fn from_config(config: ClusterUninstallConfig) -> Result<Self, ClusterError> {
+        let helm_client = if config.uninstall_k8 || config.uninstall_sys {
+            Some(HelmClient::new().map_err(UninstallError::HelmError)?)
+        } else {
+            None
+        };
+
         Ok(ClusterUninstaller {
-            helm_client: HelmClient::new().map_err(UninstallError::HelmError)?,
+            helm_client,
             pb_factory: ProgressBarFactory::new(config.hide_spinner),
             config,
         })
@@ -100,7 +106,13 @@ impl ClusterUninstaller {
         let uninstall = UninstallArg::new(self.config.app_chart_name.to_owned())
             .namespace(self.config.namespace.to_owned())
             .ignore_not_found();
-        self.helm_client
+
+        let Some(ref helm_client) = self.helm_client else {
+            return Err(ClusterError::Uninstall(UninstallError::Other(
+                "Helm client not found".to_string(),
+            )));
+        };
+        helm_client
             .uninstall(uninstall)
             .map_err(UninstallError::HelmError)?;
 
@@ -116,7 +128,12 @@ impl ClusterUninstaller {
 
         let pb = self.pb_factory.create()?;
         pb.set_message("Uninstalling Fluvio sys chart");
-        self.helm_client
+        let Some(ref helm_client) = self.helm_client else {
+            return Err(ClusterError::Uninstall(UninstallError::Other(
+                "Helm client not found".to_string(),
+            )));
+        };
+        helm_client
             .uninstall(
                 UninstallArg::new(self.config.sys_chart_name.to_owned())
                     .namespace(self.config.namespace.to_owned())

crates/fluvio-cluster/src/error.rs

@@ -5,8 +5,8 @@ use anyhow::Error as AnyError;
 use indicatif::style::TemplateError;
 
 use fluvio::FluvioError;
-use k8_config::{ConfigError as K8ConfigError};
-use k8_client::{ClientError as K8ClientError};
+use k8_config::ConfigError as K8ConfigError;
+use k8_client::ClientError as K8ClientError;
 use fluvio_helm::HelmError;
 use fluvio_command::CommandError;
 

crates/fluvio-cluster/src/progress.rs

@@ -2,9 +2,7 @@ use std::{borrow::Cow, time::Duration};
 
 use indicatif::{ProgressBar, ProgressStyle, style::TemplateError};
 
-use crate::{
-    render::{ProgressRenderedText, ProgressRenderer},
-};
+use crate::render::{ProgressRenderedText, ProgressRenderer};
 
 #[derive(Debug)]
 pub(crate) enum InstallProgressMessage {

crates/fluvio-cluster/src/runtime/local/mod.rs

@@ -11,7 +11,7 @@ mod error {
 
     use std::io::Error as IoError;
 
-    use fluvio_command::{CommandError};
+    use fluvio_command::CommandError;
 
     #[derive(thiserror::Error, Debug)]
     pub enum LocalRuntimeError {

crates/fluvio-cluster/src/runtime/local/spu.rs

@@ -6,8 +6,8 @@ use tracing::{debug, info, instrument};
 
 use fluvio_controlplane_metadata::spu::{Endpoint, IngressAddr, IngressPort, SpuSpec, SpuType};
 
-use fluvio_command::{CommandExt};
-use fluvio::config::{TlsPolicy};
+use fluvio_command::CommandExt;
+use fluvio::config::TlsPolicy;
 use fluvio_types::SpuId;
 
 use crate::runtime::spu::{SpuClusterManager, SpuTarget};

crates/fluvio-cluster/src/start/k8.rs

@@ -50,7 +50,7 @@ use crate::tls_config_to_cert_paths;
 use crate::{ClusterError, StartStatus, DEFAULT_NAMESPACE, ClusterChecker};
 use crate::charts::{ChartConfig, ChartInstaller};
 use crate::UserChartLocation;
-use crate::progress::{InstallProgressMessage};
+use crate::progress::InstallProgressMessage;
 
 use super::constants::*;
 use super::common::try_connect_to_sc;

crates/fluvio-sc-schema/src/objects/classic.rs

@@ -7,7 +7,7 @@ mod object_macro {
 
     /// Macro to objectify generic Request/Response for Admin Objects
     /// AdminSpec is difficult to turn into TraitObject due to associated types and use of other derived
-    /// properties such as `PartialEq`.  This generates all possible variation of given API.  
+    /// properties such as `PartialEq`.  This generates all possible variation of given API.
     /// Not all variation will be constructed or used
     macro_rules! ClassicObjectApiEnum {
         ($api:ident) => {

crates/fluvio-sc-schema/src/tableformat/mod.rs

@@ -4,7 +4,7 @@ mod convert {
 
     use crate::{DeletableAdminSpec, CreatableAdminSpec};
 
-    use crate::{AdminSpec};
+    use crate::AdminSpec;
     use super::TableFormatSpec;
 
     impl AdminSpec for TableFormatSpec {}

crates/fluvio-sc-schema/src/topic/mod.rs

@@ -15,7 +15,7 @@ mod convert {
 
     use crate::CreatableAdminSpec;
     use crate::DeletableAdminSpec;
-    use crate::{AdminSpec};
+    use crate::AdminSpec;
 
     use super::TopicSpec;
 

crates/fluvio-service/src/server.rs

@@ -14,7 +14,7 @@ use fluvio_future::net::{TcpListener, TcpStream};
 use fluvio_future::task::spawn;
 use fluvio_protocol::api::ApiMessage;
 use fluvio_protocol::Decoder as FluvioDecoder;
-use fluvio_socket::{FluvioSocket};
+use fluvio_socket::FluvioSocket;
 use fluvio_types::event::StickyEvent;
 
 pub struct ConnectInfo {

crates/fluvio-types/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-types"
-version = "0.4.3"
+version = "0.4.4"
 authors = ["Fluvio Contributors <[email protected]>"]
 edition = "2021"
 description = "Fluvio common types and objects"

crates/fluvio-types/src/event.rs

@@ -55,6 +55,7 @@ impl StickyEvent {
 }
 
 pub mod offsets {
+    use std::fmt;
     use std::sync::atomic::{AtomicI64, Ordering};
     use std::sync::Arc;
 
@@ -117,6 +118,12 @@ pub mod offsets {
         last_value: i64,
     }
 
+    impl fmt::Debug for OffsetChangeListener {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            write!(f, "OffsetCL{}", self.last_value)
+        }
+    }
+
     impl OffsetChangeListener {
         fn new(publisher: Arc<OffsetPublisher>) -> Self {
             Self {

crates/fluvio/src/config/config.rs

@@ -96,6 +96,7 @@ impl ConfigFile {
     /// read from file
     fn from_file<T: AsRef<Path>>(path: T) -> Result<Self, FluvioError> {
         let path_ref = path.as_ref();
+        debug!(?path_ref, "loading from");
         let file_str: String = read_to_string(path_ref)
             .map_err(|e| config_file_error(&format!("{:?}", path_ref.as_os_str()), e))?;
         let config = toml::from_str(&file_str).map_err(|e| ConfigError::TomlError {

crates/fluvio/src/consumer.rs

@@ -19,7 +19,7 @@ use fluvio_protocol::record::ReplicaKey;
 use fluvio_protocol::link::ErrorCode;
 use fluvio_protocol::record::Batch;
 
-use crate::{FluvioError};
+use crate::FluvioError;
 use crate::metrics::ClientMetrics;
 use crate::offset::{Offset, fetch_offsets};
 use crate::spu::{SpuDirectory, SpuPool};

crates/fluvio/src/fluvio.rs

@@ -302,11 +302,8 @@ mod wasm_tests {
     use async_trait::async_trait;
     use fluvio_ws_stream_wasm::WsMeta;
     use std::io::Error as IoError;
-    use fluvio_future::{
-        net::{
-            BoxReadConnection, BoxWriteConnection, DomainConnector, TcpDomainConnector,
-            ConnectionFd,
-        },
+    use fluvio_future::net::{
+        BoxReadConnection, BoxWriteConnection, DomainConnector, TcpDomainConnector, ConnectionFd,
     };
     #[derive(Clone, Default)]
     pub struct FluvioWebsocketConnector {}