Copy control-plane taint from bootstrap config

src/k8s/pkg/k8sd/types/cluster_config_convert.go

@@ -65,6 +65,7 @@ func ClusterConfigFromBootstrapConfig(b apiv1.BootstrapConfig) (ClusterConfig, e
 
 	// Kubelet
 	config.Kubelet.CloudProvider = b.ClusterConfig.CloudProvider
+	config.Kubelet.ControlPlaneTaints = &b.ControlPlaneTaints
 
 	return config, nil
 }

tests/integration/tests/test_control_plane_taints.py

@@ -0,0 +1,38 @@
+#
+# Copyright 2024 Canonical, Ltd.
+#
+import logging
+import time
+from typing import List
+
+import pytest
+import yaml
+from test_util import harness, util
+
+LOG = logging.getLogger(__name__)
+
+
+@pytest.mark.node_count(1)
+@pytest.mark.disable_k8s_bootstrapping()
+def test_control_plane_taints(instances: List[harness.Instance]):
+    k8s_instance = instances[0]
+
+    bootstrap_conf = yaml.safe_dump(
+        {"control-plane-taints": ["node-role.kubernetes.io/control-plane:NoSchedule"]}
+    )
+
+    k8s_instance.exec(
+        ["dd", "of=/root/config.yaml"],
+        input=str.encode(bootstrap_conf),
+    )
+
+    k8s_instance.exec(["k8s", "bootstrap", "--file", "/root/config.yaml"])
+    while not (nodes := util.get_nodes(k8s_instance)):
+        LOG.info("Waiting for Nodes")
+        time.sleep(5)
+
+    assert all([
+        t["effect"] == "NoSchedule"
+        for t in nodes[0]["spec"]["taints"]
+        if t["key"] == "node-role.kubernetes.io/control-plane"
+    ])

tests/integration/tests/test_util/util.py

@@ -229,8 +229,8 @@ def get_local_node_status(instance: harness.Instance) -> str:
     return resp.stdout.decode().strip()
 
 
-def ready_nodes(control_node: harness.Instance) -> List[Any]:
-    """Get a list of the ready nodes.
+def get_nodes(control_node: harness.Instance) -> List[Any]:
+    """Get a list of existing nodes.
 
     Args:
         control_node: instance on which to execute check
@@ -239,21 +239,32 @@ def ready_nodes(control_node: harness.Instance) -> List[Any]:
         list of nodes
     """
     result = control_node.exec(
-        "k8s kubectl get nodes -o json".split(" "), capture_output=True
+        ["k8s", "kubectl", "get", "nodes", "-o", "json"],
+        capture_output=True
     )
     assert result.returncode == 0, "Failed to get nodes with kubectl"
     node_list = json.loads(result.stdout.decode())
     assert node_list["kind"] == "List", "Should have found a list of nodes"
-    nodes = [
-        node
-        for node in node_list["items"]
+    return [node for node in node_list["items"]]
+
+
+def ready_nodes(control_node: harness.Instance) -> List[Any]:
+    """Get a list of the ready nodes.
+
+    Args:
+        control_node: instance on which to execute check
+
+    Returns:
+        list of nodes
+    """
+    return [
+        node for node in get_nodes(control_node)
         if all(
             condition["status"] == "False"
             for condition in node["status"]["conditions"]
             if condition["type"] != "Ready"
         )
     ]
-    return nodes
 
 
 # Create a token to join a node to an existing cluster