Action to get kubeconfig from k8s control-plane units (#42)

* Action to get kubeconfig from k8s control-plane units

* Addresses tests and linting

* Apply suggestions from code review

Co-authored-by: Mateo Florido <[email protected]>

---------

Co-authored-by: Mateo Florido <[email protected]>

charms/worker/k8s/charmcraft.yaml

@@ -61,6 +61,15 @@ config:
       description: |
         Labels can be used to organize and to select subsets of nodes in the
         cluster. Declare node labels in key=value format, separated by spaces.      
+
+actions:
+  get-kubeconfig:
+    description: Retrieve Public Kubernetes cluster config, including credentials
+    params:
+      server:
+        description: Override the server endpoint with this field
+        type: string
+
 parts:
   charm:
     build-packages: [git]

charms/worker/k8s/lib/charms/k8s/v0/k8sd_api_manager.py

@@ -335,6 +335,26 @@ class GetClusterStatusResponse(BaseRequestModel):
     metadata: Optional[ClusterMetadata] = None
 
 
+class KubeConfigMetadata(BaseModel):
+    """Metadata containing kubeconfig.
+
+    Attributes:
+        kubeconfig (KubeConfigMetadata): The status of the k8sd cluster.
+    """
+
+    kubeconfig: str
+
+
+class GetKubeConfigResponse(BaseRequestModel):
+    """Response model for getting the kubeconfig from the cluster.
+
+    Attributes:
+        metadata (KubeconfigMetadata): Metadata containing the kubeconfig.
+    """
+
+    metadata: KubeConfigMetadata
+
+
 T = TypeVar("T", bound=BaseRequestModel)
 
 
@@ -610,3 +630,17 @@ def request_auth_token(self, username: str, groups: List[str]) -> str:
         body = {"username": username, "groups": groups}
         auth_response = self._send_request(endpoint, "POST", AuthTokenResponse, body)
         return auth_response.metadata.token
+
+    def get_kubeconfig(self, server: Optional[str]) -> str:
+        """Request a Kubernetes admin config.
+
+        Args:
+            server (str): Optional server to replace in the kubeconfig endpoint
+
+        Returns:
+            str: The authentication token.
+        """
+        endpoint = "/1.0/k8sd/kubeconfig"
+        body = {"server": server or ""}
+        response = self._send_request(endpoint, "GET", GetKubeConfigResponse, body)
+        return response.metadata.kubeconfig

charms/worker/k8s/src/charm.py

@@ -85,7 +85,7 @@ def __init__(self, *args):
         self.distributor = TokenDistributor(self, self.get_node_name(), self.api_manager)
         self.collector = TokenCollector(self, self.get_node_name())
         self.labeler = LabelMaker(
-            self, kubeconfig_path=self._source_kubeconfig, kubectl=KUBECTL_PATH
+            self, kubeconfig_path=self._internal_kubeconfig, kubectl=KUBECTL_PATH
         )
         self._stored.set_default(removing=False)
 
@@ -103,6 +103,8 @@ def __init__(self, *args):
         )
 
         self.framework.observe(self.on.update_status, self._on_update_status)
+        if self.is_control_plane:
+            self.framework.observe(self.on.get_kubeconfig_action, self._get_external_kubeconfig)
 
     @status.on_error(
         ops.WaitingStatus("Installing COS requirements"),
@@ -119,6 +121,7 @@ def _apply_cos_requirements(self):
         if not self.model.get_relation("cos-agent"):
             return
 
+        log.info("Apply COS Integrations")
         status.add(ops.MaintenanceStatus("Configuring COS Integration"))
         subprocess.check_call(shlex.split("k8s kubectl apply -f templates/cos_roles.yaml"))
         subprocess.check_call(shlex.split("k8s kubectl apply -f templates/ksm.yaml"))
@@ -147,6 +150,7 @@ def _apply_proxy_environment(self):
         """Apply the proxy settings from environment variables."""
         proxy_settings = self._get_proxy_env()
         if proxy_settings:
+            log.info("Applying Proxied Environment Settings")
             with open("/etc/environment", mode="r", encoding="utf-8") as file:
                 current_env = dict(line.strip().split("=", 1) for line in file if "=" in line)
 
@@ -171,15 +175,11 @@ def get_cloud_name(self) -> str:
         # TODO: adjust to detect the correct cloud
         return ""
 
-    @property
-    def _source_kubeconfig(self) -> Path:
-        """Return the highest authority kube config for this unit."""
-        return ETC_KUBERNETES / ("admin.conf" if self.is_control_plane else "kubelet.conf")
-
     @on_error(ops.BlockedStatus("Failed to install k8s snap."), SnapError)
     def _install_k8s_snap(self):
         """Install the k8s snap package."""
         status.add(ops.MaintenanceStatus("Installing k8s snap"))
+        log.info("Ensuring k8s snap version")
         snap_ensure("k8s", SnapState.Latest.value, self.config["channel"])
 
     @on_error(WaitingStatus("Failed to apply snap requirements"), subprocess.CalledProcessError)
@@ -190,12 +190,14 @@ def _apply_snap_requirements(self):
         meets the network and interface requirements.
         """
         status.add(ops.MaintenanceStatus("Applying K8s requirements"))
+        log.info("Applying K8s requirements")
         init_sh = "/snap/k8s/current/k8s/hack/init.sh"
         subprocess.check_call(shlex.split(init_sh))
 
     @on_error(WaitingStatus("Waiting for k8sd"), InvalidResponseError, K8sdConnectionError)
     def _check_k8sd_ready(self):
         """Check if k8sd is ready to accept requests."""
+        log.info("Check if k8ds is ready")
         status.add(ops.MaintenanceStatus("Check k8sd ready"))
         self.api_manager.check_k8sd_ready()
 
@@ -208,11 +210,12 @@ def _bootstrap_k8s_snap(self):
         """Bootstrap k8s if it's not already bootstrapped."""
         if not self.api_manager.is_cluster_bootstrapped():
             status.add(ops.MaintenanceStatus("Bootstrapping Cluster"))
+            log.info("Bootstrapping Cluster")
             binding = self.model.get_binding("juju-info")
             address = binding and binding.network.ingress_address
             node_name = self.get_node_name()
             # TODO: Make port (and address) configurable.
-            self.api_manager.bootstrap_k8s_snap(node_name, f"{str(address)}:{K8SD_PORT}")
+            self.api_manager.bootstrap_k8s_snap(node_name, f"{address}:{K8SD_PORT}")
 
     @status.on_error(
         ops.WaitingStatus("Configuring COS Integration"),
@@ -225,7 +228,7 @@ def _configure_cos_integration(self):
             return
 
         status.add(ops.MaintenanceStatus("Configuring COS integration"))
-
+        log.info("Configuring COS integration")
         if relation := self.model.get_relation("cos-tokens"):
             self.collector.request(relation)
 
@@ -234,6 +237,7 @@ def _revoke_cluster_tokens(self):
 
         if self is dying, only try to remove itself from the cluster
         """
+        log.info("Garbage collect cluster tokens")
         to_remove = {self.unit} if self.is_dying else None
 
         if peer := self.model.get_relation("cluster"):
@@ -254,6 +258,7 @@ def _revoke_cluster_tokens(self):
 
     def _create_cluster_tokens(self):
         """Create tokens for the units in the cluster and k8s-cluster relations."""
+        log.info("Prepare clustering")
         if peer := self.model.get_relation("cluster"):
             node_name = self.get_node_name()
             peer.data[self.unit]["node-name"] = node_name
@@ -281,18 +286,22 @@ def _create_cos_tokens(self):
         if not self.model.get_relation("cos-agent"):
             return
 
+        log.info("Prepare cos tokens")
         if rel := self.model.get_relation("cos-tokens"):
             self.distributor.allocate_tokens(relation=rel, token_strategy=TokenStrategy.COS)
 
         if rel := self.model.get_relation("cos-worker-tokens"):
             self.distributor.allocate_tokens(relation=rel, token_strategy=TokenStrategy.COS)
 
     @on_error(
-        WaitingStatus("Waiting for enable components"), InvalidResponseError, K8sdConnectionError
+        WaitingStatus("Waiting for enable functionalities"),
+        InvalidResponseError,
+        K8sdConnectionError,
     )
     def _enable_functionalities(self):
         """Enable necessary components for the Kubernetes cluster."""
-        status.add(ops.MaintenanceStatus("Enabling DNS and Network"))
+        status.add(ops.MaintenanceStatus("Enabling Functionalities"))
+        log.info("Enabling Functionalities")
         dns_config = DNSConfig(enabled=True)
         network_config = NetworkConfig(enabled=True)
         user_cluster_config = UserFacingClusterConfig(dns=dns_config, network=network_config)
@@ -418,7 +427,7 @@ def _reconcile(self, event):
         self._update_status()
         self._apply_node_labels()
         if self.is_control_plane:
-            self._generate_kubeconfig()
+            self._copy_internal_kubeconfig()
 
     @on_error(
         ops.WaitingStatus("Cluster not yet ready"),
@@ -470,13 +479,6 @@ def _last_gasp(self, event):
             sleep(1)
             busy_wait -= 1
 
-    @on_error(ops.WaitingStatus(""))
-    def _generate_kubeconfig(self):
-        """Generate kubeconfig."""
-        status.add(ops.MaintenanceStatus("Generating KubeConfig"))
-        KUBECONFIG.parent.mkdir(parents=True, exist_ok=True)
-        KUBECONFIG.write_bytes(self._source_kubeconfig.read_bytes())
-
     @status.on_error(ops.BlockedStatus("Cannot apply node-labels"), LabelMaker.NodeLabelError)
     def _apply_node_labels(self):
         """Apply labels to the node."""
@@ -498,6 +500,37 @@ def _on_update_status(self, _event: ops.UpdateStatusEvent):
         except status.ReconcilerError:
             log.exception("Can't to update_status")
 
+    @property
+    def _internal_kubeconfig(self) -> Path:
+        """Return the highest authority kube config for this unit."""
+        return ETC_KUBERNETES / ("admin.conf" if self.is_control_plane else "kubelet.conf")
+
+    @on_error(ops.WaitingStatus(""))
+    def _copy_internal_kubeconfig(self):
+        """Write internal kubeconfig to /root/.kube/config."""
+        status.add(ops.MaintenanceStatus("Generating KubeConfig"))
+        KUBECONFIG.parent.mkdir(parents=True, exist_ok=True)
+        KUBECONFIG.write_bytes(self._internal_kubeconfig.read_bytes())
+
+    def _get_external_kubeconfig(self, event: ops.ActionEvent):
+        """Retrieve a public kubeconfig via a charm action.
+
+        Args:
+            event: ops.ActionEvent - event that triggered the action
+        """
+        try:
+            server = event.params.get("server")
+            if not server:
+                log.info("No server requested, use public-address")
+                cmd = ["unit-get", "public-address"]
+                addr = subprocess.check_output(cmd).decode("UTF-8").strip()
+                server = f"{addr}:6443"
+            log.info("Requesting kubeconfig for server=%s", server)
+            resp = self.api_manager.get_kubeconfig(server)
+            event.set_results({"kubeconfig": resp})
+        except (InvalidResponseError, K8sdConnectionError) as e:
+            event.fail(f"Failed to retrieve kubeconfig: {e}")
+
 
 if __name__ == "__main__":  # pragma: nocover
     ops.main.main(K8sCharm)

charms/worker/k8s/tests/unit/test_base.py

@@ -53,7 +53,7 @@ def mock_reconciler_handlers(harness):
             "_create_cluster_tokens",
             "_create_cos_tokens",
             "_apply_cos_requirements",
-            "_generate_kubeconfig",
+            "_copy_internal_kubeconfig",
             "_revoke_cluster_tokens",
         }