Merge branch 'main' into KU-2000/use-v2-kubecontrol

charms/worker/k8s/src/charm.py

@@ -34,7 +34,7 @@
 import ops
 import reschedule
 import yaml
-from charms.contextual_status import WaitingStatus, on_error
+from charms.contextual_status import ReconcilerError, WaitingStatus, on_error
 from charms.grafana_agent.v0.cos_agent import COSAgentProvider
 from charms.interface_external_cloud_provider import ExternalCloudProvider
 from charms.k8s.v0.k8sd_api_manager import (
@@ -677,6 +677,48 @@ def _get_scrape_jobs(self):
             log.exception("Failed to get COS token.")
         return []
 
+    @on_error(ops.WaitingStatus("Sharing Cluster Version"))
+    def _update_kubernetes_version(self):
+        """Update the unit Kubernetes version in the cluster relation.
+
+        Raises:
+            ReconcilerError: If the cluster integration is missing.
+        """
+        relation = self.model.get_relation("cluster")
+        if not relation:
+            raise ReconcilerError("Missing cluster integration")
+        if version := snap_version("k8s"):
+            relation.data[self.unit]["version"] = version
+
+    @on_error(ops.WaitingStatus("Announcing Kubernetes version"))
+    def _announce_kubernetes_version(self):
+        """Announce the Kubernetes version to the cluster.
+
+        This method ensures that the Kubernetes version is consistent across the cluster.
+
+        Raises:
+            ReconcilerError: If the k8s snap is not installed, the version is missing,
+                or the version does not match the local version.
+        """
+        if not (local_version := snap_version("k8s")):
+            raise ReconcilerError("k8s-snap is not installed")
+
+        peer = self.model.get_relation("cluster")
+        worker = self.model.get_relation("k8s-cluster")
+
+        for relation in (peer, worker):
+            if not relation:
+                continue
+            units = (unit for unit in relation.units if unit.name != self.unit.name)
+            for unit in units:
+                unit_version = relation.data[unit].get("version")
+                if not unit_version:
+                    raise ReconcilerError(f"Waiting for version from {unit.name}")
+                if unit_version != local_version:
+                    status.add(ops.BlockedStatus(f"Version mismatch with {unit.name}"))
+                    raise ReconcilerError(f"Version mismatch with {unit.name}")
+            relation.data[self.app]["version"] = local_version
+
     def _get_proxy_env(self) -> Dict[str, str]:
         """Retrieve the Juju model config proxy values.
 
@@ -766,6 +808,7 @@ def _reconcile(self, event: ops.EventBase):
         self._install_snaps()
         self._apply_snap_requirements()
         self._check_k8sd_ready()
+        self._update_kubernetes_version()
         if self.lead_control_plane:
             self._k8s_info(event)
             self._bootstrap_k8s_snap()
@@ -776,6 +819,7 @@ def _reconcile(self, event: ops.EventBase):
             self._apply_cos_requirements()
             self._revoke_cluster_tokens(event)
             self._ensure_cluster_config()
+            self._announce_kubernetes_version()
         self._join_cluster()
         self._config_containerd_registries()
         self._configure_cos_integration()

charms/worker/k8s/src/token_distributor.py

@@ -5,6 +5,7 @@
 
 import contextlib
 import logging
+import re
 from enum import Enum, auto
 from typing import Dict, Optional, Union
 
@@ -23,6 +24,8 @@
 
 SECRET_ID = "{0}-secret-id"  # nosec
 
+UNIT_RE = re.compile(r"k8s(-worker)?/\d+")
+
 
 class TokenStrategy(Enum):
     """Enumeration defining strategy for token creation.
@@ -282,6 +285,10 @@ def _revoke_juju_secret(self, relation: ops.Relation, unit: ops.Unit) -> None:
     def active_nodes(self, relation: ops.Relation):
         """Get nodes from application databag for given relation.
 
+        This method filters out entries in the application databag that are not
+        to the cluster units. It uses the regex pattern, which matches patterns
+        like k8s/0, k8s-worker/0, etc.
+
         Args:
             relation (ops.Relation): Which relation (cluster or k8s-cluster)
 
@@ -291,6 +298,7 @@ def active_nodes(self, relation: ops.Relation):
         return {
             self.charm.model.get_unit(str(u)): data
             for u, data in relation.data[self.charm.app].items()
+            if UNIT_RE.match(u)
         }
 
     def drop_node(self, relation: ops.Relation, unit: ops.Unit):

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

@@ -54,6 +54,7 @@ def mock_reconciler_handlers(harness):
         "_configure_cos_integration",
         "_update_status",
         "_apply_node_labels",
+        "_update_kubernetes_version",
     }
     if harness.charm.is_control_plane:
         handler_names |= {
@@ -66,6 +67,7 @@ def mock_reconciler_handlers(harness):
             "_revoke_cluster_tokens",
             "_ensure_cluster_config",
             "_expose_ports",
+            "_announce_kubernetes_version",
         }
 
     handlers = [mock.patch(f"charm.K8sCharm.{name}") for name in handler_names]