build: update k8 crates

Cargo.toml

@@ -144,10 +144,11 @@ fluvio-future = { version = "0.6.0", default-features = false }
 fluvio-helm = { version = "0.4.1" }
 flv-tls-proxy = { version = "0.8" }
 flv-util = { version = "0.5.2", default-features = false }
-k8-client = { version = "10.1.0" }
+k8-client = { version = "11.0.0" }
 k8-config = { version = "2.0.0" }
-k8-metadata-client = { version = "5.1.0" }
-k8-types = { version = "0.8.3" }
+k8-metadata-client = { version = "6.0.0" }
+k8-types =  { version = "0.8.5" }
+k8-diff = { version = "0.1.2" }
 trybuild = { branch = "check_option", git = "https://github.com/infinyon/trybuild" }
 
 # Internal fluvio dependencies
@@ -170,7 +171,7 @@ fluvio-smartmodule = { version = "0.7.0", path = "crates/fluvio-smartmodule", de
 fluvio-socket = { version = "0.14.3", path = "crates/fluvio-socket", default-features = false }
 fluvio-spu-schema = { version = "0.14.0", path = "crates/fluvio-spu-schema", default-features  = false }
 fluvio-storage = { path = "crates/fluvio-storage" }
-fluvio-stream-dispatcher = { path = "crates/fluvio-stream-dispatcher" }
+fluvio-stream-dispatcher = { version = "0.11.0", path = "crates/fluvio-stream-dispatcher" }
 fluvio-stream-model = { version = "0.9.3", path = "crates/fluvio-stream-model", default-features = false }
 fluvio-types = { version = "0.4.4", path = "crates/fluvio-types", default-features = false }
 

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

@@ -2,7 +2,7 @@ use serde::{Serialize, Deserialize};
 
 use futures_util::stream::StreamExt;
 
-use fluvio_protocol::api::{ResponseMessage};
+use fluvio_protocol::api::ResponseMessage;
 use fluvio_socket::FluvioSocket;
 
 use super::request::{AuthorizationScopes, AuthorizationApiRequest, AuthResponse};

crates/fluvio-cli/src/error.rs

@@ -1,4 +1,4 @@
-use std::{convert::Infallible};
+use std::convert::Infallible;
 
 use handlebars::TemplateError;
 use indicatif::style::TemplateError as ProgressTemplateError;
@@ -28,14 +28,6 @@ pub enum CliError {
     #[error("Fluvio client error: {0}")]
     ClientError(#[from] FluvioError),
 
-    #[cfg(feature = "k8s")]
-    #[error("Kubernetes config error: {0}")]
-    K8ConfigError(#[from] k8_config::ConfigError),
-
-    #[cfg(feature = "k8s")]
-    #[error("Kubernetes client error: {0}")]
-    K8ClientError(#[from] k8_client::ClientError),
-
     /// An error occurred while processing the connector yaml
     #[error("Fluvio connector config: {0}")]
     ConnectorConfig(#[from] serde_yaml::Error),

crates/fluvio-cli/src/profile/sync/k8.rs

@@ -1,6 +1,7 @@
 use std::convert::TryInto;
 
 use clap::Parser;
+use k8_client::http::status::StatusCode;
 use tracing::debug;
 use anyhow::{Result, anyhow};
 
@@ -10,7 +11,7 @@ use k8_config::K8Config;
 use k8_client::K8Client;
 use k8_client::meta_client::MetadataClient;
 use k8_types::core::service::ServiceSpec;
-use k8_types::InputObjectMeta;
+use k8_types::{InputObjectMeta, MetaStatus};
 
 use crate::common::tls::TlsClientOpt;
 
@@ -112,22 +113,26 @@ pub async fn set_k8_context(opt: K8Opt, external_addr: String) -> Result<Profile
 
 /// find fluvio addr
 pub async fn discover_fluvio_addr(namespace: Option<&str>) -> Result<Option<String>> {
-    use k8_client::http::status::StatusCode;
-
     let ns = namespace.unwrap_or("default");
     let svc = match K8Client::try_default()?
         .retrieve_item::<ServiceSpec, _>(&InputObjectMeta::named("fluvio-sc-public", ns))
         .await
     {
         Ok(svc) => svc,
-        Err(err) => match err {
-            k8_client::ClientError::ApiResponse(status)
-                if status.code == Some(StatusCode::NOT_FOUND.as_u16()) =>
+        Err(err) => {
+            if let Some(MetaStatus {
+                code: Some(code), ..
+            }) = err.downcast_ref()
             {
-                return Ok(None)
+                if *code == StatusCode::NOT_FOUND.as_u16() {
+                    return Ok(None);
+                } else {
+                    return Err(anyhow!("unable to look up fluvio service in k8: {}", err));
+                }
+            } else {
+                return Err(anyhow!("unable to look up fluvio service in k8: {}", err));
             }
-            _ => return Err(anyhow!("unable to look up fluvio service in k8: {}", err)),
-        },
+        }
     };
 
     debug!("fluvio svc: {:#?}", svc);

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

@@ -19,7 +19,6 @@ use sysinfo::{ProcessExt, System, SystemExt};
 
 use fluvio_helm::{HelmClient, HelmError};
 use k8_config::{ConfigError as K8ConfigError, K8Config};
-use k8_client::ClientError as K8ClientError;
 
 use crate::charts::{DEFAULT_HELM_VERSION, APP_CHART_NAME};
 use crate::progress::ProgressBarFactory;
@@ -53,10 +52,6 @@ pub enum ClusterCheckError {
     #[error("Kubernetes config error")]
     K8ConfigError(#[from] K8ConfigError),
 
-    /// Could not connect to K8 client
-    #[error("Kubernetes client error")]
-    K8ClientError(#[from] K8ClientError),
-
     /// Failed to parse kubernetes cluster server URL
     #[error("Failed to parse server url from Kubernetes context")]
     BadKubernetesServerUrl(#[from] ParseError),
@@ -111,10 +106,6 @@ pub enum ClusterAutoFixError {
     #[error("Kubernetes config error")]
     K8Config(#[from] K8ConfigError),
 
-    /// Could not connect to K8 client
-    #[error("Kubernetes client error")]
-    K8Client(#[from] K8ClientError),
-
     #[error("Chart Install error")]
     ChartInstall(#[from] ChartInstallError),
 }

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

@@ -50,20 +50,7 @@ impl ClusterCliError {
 impl ClusterError {
     /// Converts the plain error type into a CLI-formatted Report
     pub fn into_report(self) -> color_eyre::Report {
-        #[allow(unused)]
-        use color_eyre::Section;
         use color_eyre::Report;
-        use k8_client::ClientError as K8;
-
-        // In the future when we want to annotate errors, we do it here
-        match &self {
-            Self::InstallLocal(LocalInstallError::K8ClientError(K8::ApiResponse(it)))
-                if it.code == Some(409) =>
-            {
-                let report = Report::from(self);
-                report.suggestion("Run `fluvio cluster delete --local`, then retry")
-            }
-            _ => Report::from(self),
-        }
+        Report::from(self)
     }
 }

crates/fluvio-cluster/src/delete.rs

@@ -2,6 +2,7 @@ use std::process::Command;
 use std::fs::{remove_dir_all, remove_file};
 
 use derive_builder::Builder;
+use k8_metadata_client::MetadataClient;
 use tracing::{info, warn, debug, instrument};
 use sysinfo::{ProcessExt, System, SystemExt};
 
@@ -283,17 +284,13 @@ impl ClusterUninstaller {
 
     /// in order to remove partitions, finalizers need to be cleared
     #[instrument(skip(self))]
-    async fn remove_finalizers_for_partitions(
-        &self,
-        namespace: &str,
-    ) -> Result<(), UninstallError> {
+    async fn remove_finalizers_for_partitions(&self, namespace: &str) -> anyhow::Result<()> {
         use fluvio_controlplane_metadata::partition::PartitionSpec;
         use fluvio_controlplane_metadata::store::k8::K8ExtendedSpec;
         use k8_client::load_and_share;
-        use k8_metadata_client::MetadataClient;
         use k8_metadata_client::PatchMergeType::JsonMerge;
 
-        let client = load_and_share().map_err(UninstallError::K8ClientError)?;
+        let client = load_and_share()?;
 
         let partitions = client
             .retrieve_items::<<PartitionSpec as K8ExtendedSpec>::K8Spec, _>(namespace)

crates/fluvio-cluster/src/error.rs

@@ -6,7 +6,6 @@ use indicatif::style::TemplateError;
 
 use fluvio::FluvioError;
 use k8_config::ConfigError as K8ConfigError;
-use k8_client::ClientError as K8ClientError;
 use fluvio_helm::HelmError;
 use fluvio_command::CommandError;
 
@@ -39,9 +38,6 @@ pub enum K8InstallError {
     /// An error occurred with the Kubernetes config.
     #[error("Kubernetes config error: {0}")]
     K8ConfigError(#[from] K8ConfigError),
-    /// An error occurred with the Kubernetes client.
-    #[error("Kubernetes client error: {0}")]
-    K8ClientError(#[from] K8ClientError),
     /// An error occurred while running helm.
     #[error("Helm client error: {0}")]
     HelmError(#[from] HelmError),
@@ -115,9 +111,6 @@ pub enum LocalInstallError {
     /// An error occurred with the Kubernetes config.
     #[error("Kubernetes config error: {0}")]
     K8ConfigError(#[from] K8ConfigError),
-    /// An error occurred with the Kubernetes client.
-    #[error("Kubernetes client error: {0}")]
-    K8ClientError(#[from] K8ClientError),
     /// An error occurred while running helm.
     #[error("Helm client error: {0}")]
     HelmError(#[from] HelmError),
@@ -179,9 +172,6 @@ pub enum UninstallError {
     /// An error occurred with the Kubernetes config.
     #[error("Kubernetes config error: {0}")]
     K8ConfigError(#[from] K8ConfigError),
-    /// An error occurred with the Kubernetes client.
-    #[error("Kubernetes client error: {0}")]
-    K8ClientError(#[from] K8ClientError),
     /// An error occurred while running helm.
     #[error("Helm client error: {0}")]
     HelmError(#[from] HelmError),

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

@@ -4,7 +4,7 @@ use std::{
 };
 
 use fluvio_controlplane_metadata::spu::SpuSpec;
-use k8_client::{SharedK8Client, ClientError};
+use k8_client::SharedK8Client;
 use once_cell::sync::Lazy;
 use semver::Version;
 use tracing::{debug, error, instrument, warn};
@@ -88,7 +88,7 @@ pub async fn try_connect_to_sc(
 }
 
 // hack
-pub async fn check_crd(client: SharedK8Client) -> Result<(), ClientError> {
+pub async fn check_crd(client: SharedK8Client) -> anyhow::Result<()> {
     use k8_metadata_client::MetadataClient;
 
     for i in 0..100 {
@@ -103,5 +103,5 @@ pub async fn check_crd(client: SharedK8Client) -> Result<(), ClientError> {
         }
     }
 
-    Err(ClientError::Other("Fluvio CRD not ready".to_string()))
+    Err(anyhow::anyhow!("Fluvio CRD not ready"))
 }