Implement k8s clustering (#15)

* Implement Control Plane clustering

* Fix metadata.yaml

* Fix lint

* Remove StoredState

* Tidy Reconciler logic

charms/k8s/metadata.yaml

@@ -34,3 +34,7 @@ description: |
   This charm can optionally disable the following components:
   * A Kubernetes Backing Store
   * A Kubernetes CNI
+
+peers:
+  cluster:
+    interface: cluster

charms/k8s/src/charm.py

@@ -59,16 +59,6 @@ def __init__(self, *args):
 
         self.framework.observe(self.on.update_status, self._on_update_status)
 
-    def _reconcile(self, _):
-        """Reconcile state change events."""
-        # TODO: Implement clustering using leader units.
-        self._install_k8s_snap()
-        self._apply_snap_requirements()
-        self._check_k8sd_ready()
-        self._bootstrap_k8s_snap()
-        self._enable_components()
-        self._update_status()
-
     @on_error(WaitingStatus("Failed to apply snap requirements"), subprocess.CalledProcessError)
     def _apply_snap_requirements(self):
         """Apply necessary snap requirements for the k8s snap.
@@ -109,6 +99,29 @@ def _bootstrap_k8s_snap(self):
             # TODO: Make port (and address) configurable.
             self.api_manager.bootstrap_k8s_snap(name, f"{str(address)}:6400")
 
+    def _create_cluster_tokens(self):
+        """Create tokens for the units in the peer cluster relation."""
+        if not self.unit.is_leader():
+            return
+
+        relation = self.model.get_relation("cluster")
+        if not relation:
+            return
+
+        units = {u for u in relation.units if u.name != self.unit.name}
+        app_databag = relation.data.get(self.model.app, {})
+
+        for unit in units:
+            if app_databag.get(unit.name):
+                continue
+
+            name = unit.name.replace("/", "-")
+            token = self.api_manager.create_join_token(name)
+            content = {"token": token}
+            secret = self.app.add_secret(content)
+            secret.grant(relation, unit=unit)
+            relation.data[self.app][unit.name] = secret.id or ""
+
     @on_error(
         WaitingStatus("Waiting for enable components"), InvalidResponseError, K8sdConnectionError
     )
@@ -146,6 +159,34 @@ def _install_k8s_snap(self):
             channel = self.config["channel"]
             k8s_snap.ensure(SnapState.Latest, channel=channel)
 
+    @on_error(WaitingStatus("Waiting for Cluster token"), TypeError)
+    def _join_cluster(self):
+        """Retrieve the join token from secret databag and join the cluster."""
+        if self.api_manager.is_cluster_bootstrapped():
+            return
+
+        status.add(ops.MaintenanceStatus("Joining cluster"))
+
+        if relation := self.model.get_relation("cluster"):
+            app_databag = relation.data.get(self.model.app, {})
+            secret_id = app_databag.get(self.unit.name, "")
+            secret = self.model.get_secret(id=secret_id)
+            content = secret.get_content()
+            token = content["token"]
+            cmd = f"k8s join-cluster {shlex.quote(token)}"
+            subprocess.check_call(shlex.split(cmd))
+
+    def _reconcile(self, _):
+        """Reconcile state change events."""
+        self._install_k8s_snap()
+        self._apply_snap_requirements()
+        if self.unit.is_leader():
+            self._bootstrap_k8s_snap()
+            self._enable_components()
+            self._create_cluster_tokens()
+        self._join_cluster()
+        self._update_status()
+
     @on_error(
         ops.WaitingStatus("Cluster not yet ready"),
         subprocess.CalledProcessError,

tests/integration/conftest.py

@@ -39,12 +39,14 @@ class CharmDeploymentArgs:
         application_name:  name of juju application
         resources:         all resources for this charm
         series:            os series for the machine
+        num_units:         unit instances of the charm
     """
 
     entity_url: str
     application_name: str
     resources: dict
     series: str
+    num_units: int
 
 
 @dataclass
@@ -189,6 +191,7 @@ async def kubernetes_cluster(request: pytest.FixtureRequest, ops_test: OpsTest):
             application_name=charm.app_name,
             resources=charm.resources,
             series="jammy",
+            num_units=1 if charm.app_name == "k8s-worker" else 2,
         )
         for path, charm in zip(charm_files, charms)
     ]