feat(k8s-plugin): add support for kubeconfig context switching

This feature will allow user to choose between different kubeconfig contexts.
Useful in case there are multiple clusters in a single kubeconfig file.

Signed-off-by: Hristo Voyvodov <hristo.voyvodov@redsift.io>

Author: voyvodov

k8s/proxy/src/lib.rs

@@ -16,16 +16,19 @@ pub use proxy::{ConfigBuilder, ForwardingProxy, LokiClient, Scheme};
 /// Get the `kube::Config` from the given kubeconfig file, or the default.
 pub async fn config_from_kubeconfig(
     kube_config_path: Option<PathBuf>,
+    options: KubeConfigOptions,
 ) -> anyhow::Result<kube::Config> {
     let mut config = match kube_config_path {
         Some(config_path) => {
             // NOTE: Kubeconfig file may hold multiple contexts to communicate
-            //       with different kubernetes clusters. We have to pick master
-            //       address of current-context config only
+            //       with different kubernetes clusters. We'll pick either the one
+            //       specified in the options, or the current-context from the
+            //       kubeconfig file.
             let kube_config = kube::config::Kubeconfig::read_from(&config_path)?;
-            kube::Config::from_custom_kubeconfig(kube_config, &Default::default()).await?
+            kube::Config::from_custom_kubeconfig(kube_config, &options).await?
         }
-        None => kube::Config::from_kubeconfig(&KubeConfigOptions::default()).await?,
+
+        None => kube::Config::from_kubeconfig(&options).await?,
     };
     config.apply_debug_overrides();
     Ok(config)
@@ -34,8 +37,21 @@ pub async fn config_from_kubeconfig(
 /// Get the `kube::Client` from the given kubeconfig file, or the default.
 pub async fn client_from_kubeconfig(
     kube_config_path: Option<PathBuf>,
+    context: Option<String>,
 ) -> anyhow::Result<kube::Client> {
+    let options = kubeconfig_options_from_context(context);
     Ok(kube::Client::try_from(
-        config_from_kubeconfig(kube_config_path).await?,
+        config_from_kubeconfig(kube_config_path, options).await?,
     )?)
 }
+
+/// Create a `KubeConfigOptions` from the given context.
+pub fn kubeconfig_options_from_context(context: Option<String>) -> KubeConfigOptions {
+    match context {
+        Some(ctx) => KubeConfigOptions {
+            context: Some(ctx),
+            ..Default::default()
+        },
+        None => KubeConfigOptions::default(),
+    }
+}

k8s/proxy/src/proxy.rs

@@ -22,6 +22,7 @@ use tower::{util::BoxService, ServiceExt};
 /// ```
 pub struct ConfigBuilder<T> {
     kube_config: Option<PathBuf>,
+    context: Option<String>,
     target: kube_forward::Target,
     timeout: Option<std::time::Duration>,
     jwt: Option<String>,
@@ -77,6 +78,7 @@ impl Default for ConfigBuilder<ApiRest> {
     fn default() -> Self {
         Self {
             kube_config: None,
+            context: None,
             target: kube_forward::Target::new(
                 kube_forward::TargetSelector::ServiceLabel(utils::API_REST_LABEL.to_string()),
                 utils::API_REST_HTTP_PORT,
@@ -94,6 +96,7 @@ impl Default for ConfigBuilder<Etcd> {
     fn default() -> Self {
         Self {
             kube_config: None,
+            context: None,
             target: kube_forward::Target::new(
                 kube_forward::TargetSelector::PodLabel(utils::ETCD_LABEL.to_string()),
                 utils::ETCD_PORT,
@@ -111,6 +114,7 @@ impl Default for ConfigBuilder<Loki> {
     fn default() -> Self {
         Self {
             kube_config: None,
+            context: None,
             target: kube_forward::Target::new(
                 kube_forward::TargetSelector::ServiceLabel(utils::LOKI_LABEL.to_string()),
                 utils::LOKI_PORT,
@@ -155,6 +159,11 @@ impl<T> ConfigBuilder<T> {
         self.target = target;
         self
     }
+    /// Move self with the following context.
+    pub fn with_context(mut self, context: Option<String>) -> Self {
+        self.context = context;
+        self
+    }
     /// Move self with the following target closure.
     pub fn with_target_mod(
         mut self,
@@ -191,7 +200,7 @@ impl ConfigBuilder<ApiRest> {
     }
     /// Tries to build an HTTP `Configuration` from the current self.
     async fn build_http(self) -> Result<Configuration, Error> {
-        let client = super::client_from_kubeconfig(self.kube_config).await?;
+        let client = super::client_from_kubeconfig(self.kube_config, self.context).await?;
         let uri =
             kube_forward::HttpForward::new(self.target, Some(self.scheme.into()), client.clone())
                 .uri()
@@ -209,7 +218,7 @@ impl ConfigBuilder<ApiRest> {
     }
     /// Tries to build a TCP `Configuration` from the current self.
     async fn build_tcp(self) -> Result<Configuration, Error> {
-        let client = super::client_from_kubeconfig(self.kube_config).await?;
+        let client = super::client_from_kubeconfig(self.kube_config, self.context).await?;
 
         let pf = kube_forward::PortForward::new(self.target, None, client);
 
@@ -238,7 +247,7 @@ impl ConfigBuilder<ApiRest> {
 impl ConfigBuilder<Etcd> {
     /// Tries to build a TCP `Configuration` from the current self.
     pub async fn build(self) -> Result<Uri, Error> {
-        let client = super::client_from_kubeconfig(self.kube_config).await?;
+        let client = super::client_from_kubeconfig(self.kube_config, self.context).await?;
 
         let pf = kube_forward::PortForward::new(self.target, None, client);
 
@@ -284,7 +293,7 @@ impl ConfigBuilder<Loki> {
     }
     /// Tries to build an HTTP `Configuration` from the current self.
     async fn build_http(self) -> Result<(Uri, LokiClient), Error> {
-        let client = super::client_from_kubeconfig(self.kube_config).await?;
+        let client = super::client_from_kubeconfig(self.kube_config, self.context).await?;
 
         let uri =
             kube_forward::HttpForward::new(self.target, Some(self.scheme.into()), client.clone())
@@ -300,7 +309,7 @@ impl ConfigBuilder<Loki> {
     }
     /// Tries to build a TCP `Configuration` from the current self.
     async fn build_tcp(self) -> Result<(Uri, LokiClient), Error> {
-        let client = super::client_from_kubeconfig(self.kube_config).await?;
+        let client = super::client_from_kubeconfig(self.kube_config, self.context).await?;
 
         let pf = kube_forward::PortForward::new(self.target, None, client);