Integrate runtime updates for etcd (#71)

---------

Co-authored-by: Adam Dyess <[email protected]>

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

@@ -293,6 +293,24 @@ class UserFacingClusterConfig(BaseModel):
     cloud_provider: Optional[str] = Field(None, alias="cloud-provider")
 
 
+class UserFacingDatastoreConfig(BaseModel, allow_population_by_field_name=True):  # type: ignore[call-arg]
+    """Aggregated configuration model for the user-facing datastore aspects of a cluster.
+
+    Attributes:
+        type: Type of the datastore. For runtime updates, this needs to be "external".
+        servers: Server addresses of the external datastore.
+        ca_crt: CA certificate of the external datastore cluster in PEM format.
+        client_crt: client certificate of the external datastore cluster in PEM format.
+        client_key: client key of the external datastore cluster in PEM format.
+    """
+
+    type: Optional[str] = Field(None)
+    servers: Optional[List[str]] = Field(None)
+    ca_crt: Optional[str] = Field(None, alias="ca-crt")
+    client_crt: Optional[str] = Field(None, alias="client-crt")
+    client_key: Optional[str] = Field(None, alias="client-key")
+
+
 class BootstrapConfig(BaseModel):
     """Configuration model for bootstrapping a Canonical K8s cluster.
 
@@ -346,9 +364,11 @@ class UpdateClusterConfigRequest(BaseModel):
 
     Attributes:
         config (Optional[UserFacingClusterConfig]): The cluster configuration.
+        datastore (Optional[UserFacingDatastoreConfig]): The clusters datastore configuration.
     """
 
-    config: UserFacingClusterConfig
+    config: Optional[UserFacingClusterConfig] = Field(None)
+    datastore: Optional[UserFacingDatastoreConfig] = Field(None)
 
 
 class NodeJoinConfig(BaseModel, allow_population_by_field_name=True):
@@ -418,11 +438,11 @@ class DatastoreStatus(BaseModel):
 
     Attributes:
         datastore_type (str): external or k8s-dqlite datastore
-        external_url: (str): list of external_urls
+        servers: (List(str)): list of server addresses of the external datastore cluster.
     """
 
     datastore_type: Optional[str] = Field(None, alias="type")
-    external_url: Optional[str] = Field(None, alias="external-url")
+    servers: Optional[List[str]] = Field(None, alias="servers")
 
 
 class ClusterStatus(BaseModel):
@@ -692,7 +712,7 @@ def update_cluster_config(self, config: UpdateClusterConfigRequest):
             config (UpdateClusterConfigRequest): The cluster configuration.
         """
         endpoint = "/1.0/k8sd/cluster/config"
-        body = config.dict(exclude_none=True)
+        body = config.dict(exclude_none=True, by_alias=True)
         self._send_request(endpoint, "PUT", EmptyResponse, body)
 
     def get_cluster_status(self) -> GetClusterStatusResponse:

charms/worker/k8s/src/charm.py

@@ -46,6 +46,7 @@
     UnixSocketConnectionFactory,
     UpdateClusterConfigRequest,
     UserFacingClusterConfig,
+    UserFacingDatastoreConfig,
 )
 from charms.kubernetes_libs.v0.etcd import EtcdReactiveRequires
 from charms.node_base import LabelMaker
@@ -388,6 +389,36 @@ def _enable_functionalities(self):
 
         self.api_manager.update_cluster_config(update_request)
 
+    @on_error(
+        WaitingStatus("Ensure that the cluster configuration is up-to-date"),
+        InvalidResponseError,
+        K8sdConnectionError,
+    )
+    def _ensure_cluster_config(self):
+        """Ensure that the cluster configuration is up-to-date.
+
+        The snap will detect any changes and only perform necessary steps.
+        There is no need to track changes in the charm.
+        """
+        status.add(ops.MaintenanceStatus("Ensure cluster config"))
+        log.info("Ensure cluster-config")
+
+        update_request = UpdateClusterConfigRequest()
+
+        # TODO: Ensure other configs here as well.
+
+        if self.config.get("datastore") == "etcd":
+            etcd_config = self.etcd.get_client_credentials()
+            update_request.datastore = UserFacingDatastoreConfig(
+                type="external",
+                servers=self.etcd.get_connection_string().split(","),
+                ca_crt=etcd_config.get("client_ca", ""),
+                client_crt=etcd_config.get("client_cert", ""),
+                client_key=etcd_config.get("client_key", ""),
+            )
+
+        self.api_manager.update_cluster_config(update_request)
+
     def _get_scrape_jobs(self):
         """Retrieve the Prometheus Scrape Jobs.
 
@@ -506,6 +537,7 @@ def _reconcile(self, event):
             self._create_cos_tokens()
             self._apply_cos_requirements()
             self._revoke_cluster_tokens()
+            self._ensure_cluster_config()
         self._join_cluster()
         self._configure_cos_integration()
         self._update_status()

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

@@ -59,6 +59,7 @@ def mock_reconciler_handlers(harness):
             "_apply_cos_requirements",
             "_copy_internal_kubeconfig",
             "_revoke_cluster_tokens",
+            "_ensure_cluster_config",
             "_expose_ports",
         }
 

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

@@ -27,6 +27,7 @@
     UnixSocketHTTPConnection,
     UpdateClusterConfigRequest,
     UserFacingClusterConfig,
+    UserFacingDatastoreConfig,
 )
 
 
@@ -289,13 +290,29 @@ def test_update_cluster_config(self, mock_send_request):
 
         dns_config = DNSConfig(enabled=True)
         user_config = UserFacingClusterConfig(dns=dns_config)
-        request = UpdateClusterConfigRequest(config=user_config)
+        datastore = UserFacingDatastoreConfig(
+            type="external",
+            servers=["localhost:123"],
+            ca_crt="ca-crt",
+            client_crt="client-crt",
+            client_key="client-key",
+        )
+        request = UpdateClusterConfigRequest(config=user_config, datastore=datastore)
         self.api_manager.update_cluster_config(request)
         mock_send_request.assert_called_once_with(
             "/1.0/k8sd/cluster/config",
             "PUT",
             EmptyResponse,
-            {"config": {"dns": {"enabled": True}}},
+            {
+                "config": {"dns": {"enabled": True}},
+                "datastore": {
+                    "type": "external",
+                    "servers": ["localhost:123"],
+                    "ca-crt": "ca-crt",
+                    "client-crt": "client-crt",
+                    "client-key": "client-key",
+                },
+            },
         )
 
     @patch("lib.charms.k8s.v0.k8sd_api_manager.K8sdAPIManager._send_request")

tests/integration/test_etcd.py

@@ -41,3 +41,26 @@ async def test_etcd_datastore(kubernetes_cluster: model.Model):
     assert status["ready"], "Cluster isn't ready"
     assert status["datastore"]["type"] == "external", "Not bootstrapped against etcd"
     assert status["datastore"]["servers"] == [f"https://{etcd.public_address}:{etcd_port}"]
+
+
+@pytest.mark.abort_on_fail
+async def test_update_etcd_cluster(kubernetes_cluster: model.Model):
+    """Test that adding etcd clusters are propagated to the k8s cluster."""
+    k8s: unit.Unit = kubernetes_cluster.applications["k8s"].units[0]
+    etcd = kubernetes_cluster.applications["etcd"]
+    count = 3 - len(etcd.units)
+    if count > 0:
+        await etcd.add_unit(count=count)
+    await kubernetes_cluster.wait_for_idle(status="active", timeout=20 * 60)
+
+    expected_servers = []
+    for u in etcd.units:
+        etcd_port = u.safe_data["ports"][0]["number"]
+        expected_servers.append(f"https://{u.public_address}:{etcd_port}")
+
+    event = await k8s.run("k8s status --output-format json")
+    result = await event.wait()
+    status = json.loads(result.results["stdout"])
+    assert status["ready"], "Cluster isn't ready"
+    assert status["datastore"]["type"] == "external", "Not bootstrapped against etcd"
+    assert set(status["datastore"]["servers"]) == set(expected_servers)