Share Kubernetes Version Through Relation Data

charms/worker/k8s/src/charm.py

@@ -601,6 +601,43 @@ def _get_scrape_jobs(self):
             log.exception("Failed to get COS token.")
         return []
 
+    @on_error(
+        ops.WaitingStatus("Sharing Cluster Version"),
+        AssertionError,
+    )
+    def _update_kubernetes_version(self):
+        """Update the unit Kubernetes version in the cluster relation."""
+        if not (relation := self.model.get_relation("cluster")):
+            assert False, "Missing cluster integration"  # nosec
+        if version := snap_version("k8s"):
+            relation.data[self.unit]["version"] = version
+
+    @on_error(
+        ops.WaitingStatus("Announcing Kubernetes version"),
+        AssertionError,
+    )
+    def _announce_kubernetes_version(self):
+        """Announce the Kubernetes version to the cluster.
+
+        This method ensures that the Kubernetes version is consistent across the cluster.
+        """
+        peer = self.model.get_relation("cluster")
+        worker = self.model.get_relation("k8s-cluster")
+
+        version = snap_version("k8s")
+        assert version, "k8s-snap is not installed"  # nosec
+
+        for relation in (peer, worker):
+            assert relation, "Missing cluster integration"  # nosec
+            units = (unit for unit in relation.units if unit.name != self.unit.name)
+            for unit in units:
+                unit_version = relation.data[unit].get("version")
+                assert unit_version, f"Waiting for version from {unit.name}"  # nosec
+                if unit_version != version:
+                    status.add(ops.BlockedStatus(f"Version mismatch with {unit.name}"))
+                    assert False, "Version mismatch with cluster nodes"  # nosec
+            relation.data[self.app]["version"] = version
+
     def _get_proxy_env(self) -> Dict[str, str]:
         """Retrieve the Juju model config proxy values.
 
@@ -690,6 +727,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()
@@ -700,6 +738,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, Protocol, Union
 
@@ -299,6 +300,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)
 
@@ -308,6 +313,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 re.match(r"k8s(-worker)?/\d+", 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]