Slim down extra args module to its core elements

charms/worker/k8s/src/charm.py

@@ -330,7 +330,8 @@ def _assemble_bootstrap_config(self):
         bootstrap_config.pod_cidr = str(self.config["bootstrap-pod-cidr"])
         bootstrap_config.control_plane_taints = str(self.config["bootstrap-node-taints"]).split()
         bootstrap_config.extra_sans = [_get_public_address()]
-        config.extra_args.craft(self, bootstrap_config)
+        cluster_name = self.get_cluster_name()
+        config.extra_args.craft(self.config, bootstrap_config, cluster_name)
         return bootstrap_config
 
     @on_error(
@@ -742,28 +743,29 @@ def _join_cluster(self, event: ops.EventBase):
         with self.collector.recover_token(relation) as token:
             remote_cluster = self.collector.cluster_name(relation, False) if relation else ""
             self.cloud_integration.integrate(remote_cluster, event)
-            self._join_with_token(relation, token)
+            self._join_with_token(relation, token, remote_cluster)
 
-    def _join_with_token(self, relation: ops.Relation, token: str):
+    def _join_with_token(self, relation: ops.Relation, token: str, cluster_name: str):
         """Join the cluster with the given token.
 
         Args:
             relation (ops.Relation): The relation to use for the token.
             token (str): The token to use for joining the cluster.
+            cluster_name (str): The name of the cluster to join.
         """
         binding = self.model.get_binding(relation.name)
         address = binding and binding.network.ingress_address
         node_name = self.get_node_name()
         cluster_addr = f"{address}:{K8SD_PORT}"
-        log.info("Joining %s(%s) to %s...", self.unit, node_name, cluster_addr)
+        log.info("Joining %s(%s) to %s...", self.unit, node_name, cluster_name)
         request = JoinClusterRequest(name=node_name, address=cluster_addr, token=token)
         if self.is_control_plane:
             request.config = ControlPlaneNodeJoinConfig()
             request.config.extra_sans = [_get_public_address()]
-            config.extra_args.craft(self, request.config)
+            config.extra_args.craft(self.config, request.config, cluster_name)
         else:
             request.config = NodeJoinConfig()
-            config.extra_args.craft(self, request.config)
+            config.extra_args.craft(self.config, request.config, cluster_name)
 
         self.api_manager.join_cluster(request)
         log.info("Joined %s(%s)", self.unit, node_name)

charms/worker/k8s/src/config/extra_args.py

@@ -6,47 +6,35 @@
 """Parse extra arguments for Kubernetes components."""
 from typing import Dict, Union
 
+import ops
 from charms.k8s.v0.k8sd_api_manager import (
     BootstrapConfig,
     ControlPlaneNodeJoinConfig,
     NodeJoinConfig,
 )
-from protocols import K8sCharmProtocol
 
 
-class ArgMapper:
-    """A class to map string key,val pairs to be used as cmd args.
+def _parse(config_data) -> Dict[str, str]:
+    """Parse user config data into a dictionary.
 
-    Attributes:
-        args: the args to be used made into command args
+    Args:
+        config_data: the charm config data for an extra-args
     """
-
-    def __init__(self, config_data):
-        """Initialise the ArgMapper class.
-
-        Args:
-            config_data: the charm config data for an extra-args
-        """
-        self.args: Dict[str, str] = {}
-        for element in str(config_data).split():
-            if "=" in element:
-                key, _, value = element.partition("=")
-            else:
-                key, value = element, "true"
-            self.args[key.lstrip("-")] = value
-
-    def dict(self) -> Dict[str, str]:
-        """Return an args based representative view.
-
-        Returns:
-            Dict[str, str]: the args as a dictionary
-        """
-        return {f"--{k}": v for k, v in self.args.items() if v is not None}
+    args: Dict[str, str] = {}
+    for element in str(config_data).split():
+        if "=" in element:
+            key, _, value = element.partition("=")
+        else:
+            key, value = element, "true"
+        if value is not None:
+            args["--" + key.lstrip("-")] = value
+    return args
 
 
 def craft(
-    charm: K8sCharmProtocol,
-    config: Union[BootstrapConfig, ControlPlaneNodeJoinConfig, NodeJoinConfig],
+    src: ops.ConfigData,
+    dest: Union[BootstrapConfig, ControlPlaneNodeJoinConfig, NodeJoinConfig],
+    cluster_name: str,
 ):
     """Set extra arguments for Kubernetes components based on the provided configuration.
 
@@ -58,26 +46,27 @@ def craft(
         - extra_node_kubelet_args: arguments for kubelet.
 
     Args:
-        charm (ops.CharmBase): the charm instance to get the configuration from.
-        config (Union[BootstrapConfig, ControlPlaneNodeJoinConfig, NodeJoinConfig]):
+        src (ops.ConfigData): the charm instance to get the configuration from.
+        dest (Union[BootstrapConfig, ControlPlaneNodeJoinConfig, NodeJoinConfig]):
             The configuration object to be updated with extra arguments.
+        cluster_name (str): the name of the cluster to override in the extra arguments.
     """
-    if isinstance(config, (BootstrapConfig, ControlPlaneNodeJoinConfig)):
-        cmd = ArgMapper(charm.config["kube-apiserver-extra-args"])
-        config.extra_node_kube_apiserver_args = cmd.dict()
+    if isinstance(dest, (BootstrapConfig, ControlPlaneNodeJoinConfig)):
+        cmd = _parse(src["kube-apiserver-extra-args"])
+        dest.extra_node_kube_apiserver_args = cmd
 
-        cmd = ArgMapper(charm.config["kube-controller-manager-extra-args"])
-        if cluster_name := charm.get_cluster_name():
-            cmd.args.update(**{"cluster-name": cluster_name})
+        cmd = _parse(src["kube-controller-manager-extra-args"])
+        if cluster_name:
+            cmd.update(**{"--cluster-name": cluster_name})
         else:
-            cmd.args.pop("cluster-name", None)
-        config.extra_node_kube_controller_manager_args = cmd.dict()
+            cmd.pop("--cluster-name", None)
+        dest.extra_node_kube_controller_manager_args = cmd
 
-        cmd = ArgMapper(charm.config["kube-scheduler-extra-args"])
-        config.extra_node_kube_scheduler_args = cmd.dict()
+        cmd = _parse(src["kube-scheduler-extra-args"])
+        dest.extra_node_kube_scheduler_args = cmd
 
-    cmd = ArgMapper(charm.config["kube-proxy-extra-args"])
-    config.extra_node_kube_proxy_args = cmd.dict()
+    cmd = _parse(src["kube-proxy-extra-args"])
+    dest.extra_node_kube_proxy_args = cmd
 
-    cmd = ArgMapper(charm.config["kubelet-extra-args"])
-    config.extra_node_kubelet_args = cmd.dict()
+    cmd = _parse(src["kubelet-extra-args"])
+    dest.extra_node_kubelet_args = cmd