diff --git a/sunbeam-microcluster/sunbeam/terraform.go b/sunbeam-microcluster/sunbeam/terraform.go index be33815d5..5c18a6437 100644 --- a/sunbeam-microcluster/sunbeam/terraform.go +++ b/sunbeam-microcluster/sunbeam/terraform.go @@ -98,6 +98,23 @@ func GetTerraformLock(ctx context.Context, s state.State, name string) (string, // UpdateTerraformLock updates the terraform lock record in the database func UpdateTerraformLock(ctx context.Context, s state.State, name string, lock string) (apitypes.Lock, error) { + tflockKey := tflockPrefix + name + return updateTerraformLock( + lock, + func() (string, error) { + return GetConfig(ctx, s, tflockKey) + }, + func(lock string) error { + return CreateConfig(ctx, s, tflockKey, lock) + }, + ) +} + +func updateTerraformLock( + lock string, + getLock func() (string, error), + createLock func(string) error, +) (apitypes.Lock, error) { var reqLock apitypes.Lock var dbLock apitypes.Lock @@ -106,22 +123,31 @@ func UpdateTerraformLock(ctx context.Context, s state.State, name string, lock s return dbLock, err } - tflockKey := tflockPrefix + name - lockInDb, err := GetConfig(ctx, s, tflockKey) + lockInDb, err := getLock() if err != nil { - if err, ok := err.(api.StatusError); ok { - // No Lock exists, add lock details in DB - if err.Status() == http.StatusNotFound { - j, err := json.Marshal(reqLock) - if err != nil { - return dbLock, err - } + if !api.StatusErrorCheck(err, http.StatusNotFound) { + return dbLock, err + } - err = UpdateConfig(ctx, s, tflockKey, string(j)) - return dbLock, err - } + j, err := json.Marshal(reqLock) + if err != nil { + return dbLock, err + } + + createErr := createLock(string(j)) + if createErr == nil { + return dbLock, nil + } + if !api.StatusErrorCheck(createErr, http.StatusConflict) { + return dbLock, createErr + } + + // Another request may have created the lock after GetConfig. + // Read that lock and use the identity checks below. + lockInDb, err = getLock() + if err != nil { + return dbLock, err } - return dbLock, err } err = json.Unmarshal([]byte(lockInDb), &dbLock) diff --git a/sunbeam-microcluster/sunbeam/terraform_test.go b/sunbeam-microcluster/sunbeam/terraform_test.go new file mode 100644 index 000000000..a6cb8e162 --- /dev/null +++ b/sunbeam-microcluster/sunbeam/terraform_test.go @@ -0,0 +1,84 @@ +package sunbeam + +import ( + "errors" + "fmt" + "net/http" + "testing" + + "github.com/canonical/lxd/shared/api" +) + +func TestUpdateTerraformLockAfterConcurrentCreate(t *testing.T) { + tests := []struct { + name string + requestLock string + existingLock string + status int + }{ + { + name: "same lock", + requestLock: `{"ID":"lock-id","Operation":"join","Who":"node-1"}`, + existingLock: `{"ID":"lock-id","Operation":"join","Who":"node-1"}`, + status: http.StatusLocked, + }, + { + name: "different lock", + requestLock: `{"ID":"lock-id","Operation":"join","Who":"node-1"}`, + existingLock: `{"ID":"other-id","Operation":"join","Who":"node-2"}`, + status: http.StatusConflict, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + getCalls := 0 + getLock := func() (string, error) { + getCalls++ + if getCalls == 1 { + return "", api.StatusErrorf(http.StatusNotFound, "Lock not found") + } + + return test.existingLock, nil + } + createLock := func(string) error { + return fmt.Errorf( + "Failed to record config item: %w", + api.StatusErrorf(http.StatusConflict, "Lock already exists"), + ) + } + + _, err := updateTerraformLock(test.requestLock, getLock, createLock) + if !api.StatusErrorCheck(err, test.status) { + t.Fatalf("Expected status %d, got %v", test.status, err) + } + if getCalls != 2 { + t.Fatalf("Expected the stored lock to be read twice, got %d", getCalls) + } + }) + } +} + +func TestUpdateTerraformLockReturnsCreateError(t *testing.T) { + createErr := errors.New("database unavailable") + getCalls := 0 + getLock := func() (string, error) { + getCalls++ + return "", api.StatusErrorf(http.StatusNotFound, "Lock not found") + } + createLock := func(string) error { + return createErr + } + + _, err := updateTerraformLock( + `{"ID":"lock-id","Operation":"join","Who":"node-1"}`, + getLock, + createLock, + ) + if !errors.Is(err, createErr) { + t.Fatalf("Expected create error, got %v", err) + } + if getCalls != 1 { + t.Fatalf("Expected no retry for create error, got %d reads", getCalls) + } +} diff --git a/sunbeam-python/sunbeam/clusterd/cluster.py b/sunbeam-python/sunbeam/clusterd/cluster.py index 6934ea1bd..9bc62fa7e 100644 --- a/sunbeam-python/sunbeam/clusterd/cluster.py +++ b/sunbeam-python/sunbeam/clusterd/cluster.py @@ -220,6 +220,23 @@ def get_terraform_lock(self, plan: str) -> dict: lock = self._get(f"/1.0/terraformlock/{plan}") return json.loads(lock) + def lock_terraform_plan(self, plan: str, lock: dict) -> None: + """Lock a Terraform plan.""" + try: + self._put(f"/1.0/terraformlock/{plan}", data=json.dumps(lock)) + except HTTPError as e: + if e.response is None: + raise + if e.response.status_code == codes.locked: + # Clusterd returns 423 when the stored lock has the same + # ID, operation, and owner as this request. + return + if e.response.status_code == codes.conflict: + raise service.TerraformPlanLockConflictException( + f"Terraform plan {plan} is locked" + ) from e + raise + def unlock_terraform_plan(self, plan: str, lock: dict) -> None: """Unlock plan.""" self._put(f"/1.0/terraformunlock/{plan}", data=json.dumps(lock)) diff --git a/sunbeam-python/sunbeam/clusterd/service.py b/sunbeam-python/sunbeam/clusterd/service.py index cf360a89d..285d1314b 100644 --- a/sunbeam-python/sunbeam/clusterd/service.py +++ b/sunbeam-python/sunbeam/clusterd/service.py @@ -28,6 +28,12 @@ class ClusterServiceUnavailableException(RemoteException): pass +class TerraformPlanLockConflictException(RemoteException): + """Raised when another operation holds a Terraform plan lock.""" + + pass + + class ConfigItemNotFoundException(RemoteException): """Raise when ConfigItem cannot be found on the remote.""" @@ -167,6 +173,8 @@ def _request(self, method, path, **kwargs): # noqa: C901 too complex except HTTPError as e: # Do some nice translating to sunbeamdexceptions error = response.json().get("error") + if not isinstance(error, str): + raise if "remote with name" in error: raise NodeAlreadyExistsException( "Already node exists in the sunbeam cluster" diff --git a/sunbeam-python/sunbeam/provider/local/commands.py b/sunbeam-python/sunbeam/provider/local/commands.py index 3655fea7f..f80a77fa4 100644 --- a/sunbeam-python/sunbeam/provider/local/commands.py +++ b/sunbeam-python/sunbeam/provider/local/commands.py @@ -1561,6 +1561,7 @@ def join( # noqa: C901 jhelper, deployment.openstack_machines_model, name, + reconcile_existing=True, ), ) plan4.append( diff --git a/sunbeam-python/sunbeam/steps/k8s.py b/sunbeam-python/sunbeam/steps/k8s.py index 88e92b6f9..6573fb7ab 100644 --- a/sunbeam-python/sunbeam/steps/k8s.py +++ b/sunbeam-python/sunbeam/steps/k8s.py @@ -6,16 +6,20 @@ import subprocess import time import typing +import uuid import jubilant import tenacity import yaml +from requests.models import HTTPError from rich.console import Console from sunbeam.clusterd.client import Client from sunbeam.clusterd.service import ( + ClusterServiceUnavailableException, ConfigItemNotFoundException, NodeNotExistInClusterException, + TerraformPlanLockConflictException, ) from sunbeam.core.common import ( BaseStep, @@ -1269,6 +1273,7 @@ def __init__( network: Networks, kube_namespace: str | None = None, fqdn: str | None = None, + reconcile_existing: bool = False, ): super().__init__(name, description) self.deployment = deployment @@ -1278,6 +1283,7 @@ def __init__( self.network = network self.kube_namespace = kube_namespace self.fqdn = fqdn + self.reconcile_existing = reconcile_existing self.to_update: list[dict] = [] self.to_delete: list[dict] = [] self._ifnames: dict[str, str] = {} @@ -1308,40 +1314,81 @@ def _get_outdated_resources( """ raise NotImplementedError + def _get_reconciliation_nodes(self, target_node: dict) -> list[dict]: + """Return the target node and tagged existing control nodes.""" + k8s_nodes = list_nodes( + self.kube, + labels={DEPLOYMENT_LABEL: self.deployment.name}, + ) + tagged_node_names = set() + for k8s_node in k8s_nodes: + if k8s_node.metadata and k8s_node.metadata.labels: + hostname = k8s_node.metadata.labels.get(HOSTNAME_LABEL) + if hostname: + tagged_node_names.add(hostname) + + eligible_nodes = [target_node] + for control_node in self.client.cluster.list_nodes_by_role( + Role.CONTROL.name.lower() + ): + name = control_node["name"] + if name == target_node["name"]: + continue + if name in tagged_node_names: + eligible_nodes.append(control_node) + else: + LOG.debug( + "Skipping control node %s without %s label", + name, + HOSTNAME_LABEL, + ) + return eligible_nodes + def is_skip(self, context: StepContext) -> Result: """Determines if the step should be skipped or not.""" self.to_update = [] self.to_delete = [] control = Role.CONTROL.name.lower() region_controller = Role.REGION_CONTROLLER.name.lower() + target_node = None if self.fqdn: node = self.client.cluster.get_node_info(self.fqdn) node_roles = node.get("role", []) if control not in node_roles and region_controller not in node_roles: return Result(ResultType.FAILED, f"{self.fqdn} is not a control node") - self.control_nodes = [node] + target_node = node + if not self.reconcile_existing: + self.control_nodes = [node] else: self.control_nodes = self.client.cluster.list_nodes_by_role(control) + juju_machines = set(self.jhelper.get_machines(self.model).keys()) + + try: + self.kube = get_kube_client(self.client, self.kube_namespace) + except KubeClientError as e: + LOG.debug("Failed to create k8s client", exc_info=True) + return Result(ResultType.FAILED, str(e)) + + if target_node and self.reconcile_existing: + try: + self.control_nodes = self._get_reconciliation_nodes(target_node) + except K8SError as e: + LOG.debug("Failed to list K8S nodes", exc_info=True) + return Result(ResultType.FAILED, str(e)) + # Skip control nodes whose machines are not yet in juju. This can # happen during node removal (machine removed before clusterd cleanup) # or during concurrent joins (machine not yet provisioned). Skipping # is safe: stale resources are cleaned up later when the node is # removed from clusterd, at which point it won't appear in # control_nodes and _get_outdated_resources will report it as deleted. - juju_machines = set(self.jhelper.get_machines(self.model).keys()) self.control_nodes = [ node for node in self.control_nodes if str(node.get("machineid", "")) in juju_machines ] - try: - self.kube = get_kube_client(self.client, self.kube_namespace) - except KubeClientError as e: - LOG.debug("Failed to create k8s client", exc_info=True) - return Result(ResultType.FAILED, str(e)) - try: outdated, deleted = self._get_outdated_resources( self.control_nodes, self.kube @@ -1351,8 +1398,8 @@ def is_skip(self, context: StepContext) -> Result: return Result(ResultType.FAILED, str(e)) if self.fqdn: - # Single-node mode (join/bootstrap): only create/update for the - # target node. Defer deletion of stale resources to full + # FQDN-scoped mode (join/bootstrap) only manages selected nodes. + # Defer deletion of stale resources to full # reconciliation (refresh, where fqdn is None). deleted = [] @@ -1379,6 +1426,7 @@ class EnsureCiliumDeviceByHostStep(_PerHostK8SResourceStep): _RESTART_TIMEOUT = 300 _RESTART_POLL_INTERVAL = 5 _RESTART_PENDING_ANNOTATION = "sunbeam/restart-pending" + _TERRAFORM_PLAN = "k8s-plan" def __init__( self, @@ -1387,6 +1435,7 @@ def __init__( jhelper: JujuHelper, model: str, fqdn: str | None = None, + reconcile_existing: bool = False, ): super().__init__( "Ensure Cilium device config", @@ -1398,10 +1447,66 @@ def __init__( Networks.INTERNAL, kube_namespace=self._CILIUM_NAMESPACE, fqdn=fqdn, + reconcile_existing=reconcile_existing, ) self.cilium_node_config_resource = ( K8SHelper.get_lightkube_cilium_node_config_resource() ) + self._reconciliation_lock = { + "ID": str(uuid.uuid4()), + "Operation": "Cilium reconciliation", + "Who": fqdn or deployment.name, + } + + @tenacity.retry( + wait=tenacity.wait_fixed(_RESTART_POLL_INTERVAL), + stop=tenacity.stop_after_delay(_RESTART_TIMEOUT), + retry=tenacity.retry_if_exception_type(TerraformPlanLockConflictException), + reraise=True, + ) + def _acquire_reconciliation_lock(self) -> None: + self.client.cluster.lock_terraform_plan( + self._TERRAFORM_PLAN, + self._reconciliation_lock, + ) + + def _release_reconciliation_lock(self) -> bool: + try: + self.client.cluster.unlock_terraform_plan( + self._TERRAFORM_PLAN, + self._reconciliation_lock, + ) + except (ClusterServiceUnavailableException, HTTPError): + LOG.debug("Failed to release Cilium reconciliation lock", exc_info=True) + return False + return True + + def _with_reconciliation_lock(self, action: typing.Callable[[], Result]) -> Result: + try: + self._acquire_reconciliation_lock() + except ( + ClusterServiceUnavailableException, + HTTPError, + TerraformPlanLockConflictException, + ) as e: + LOG.debug("Failed to acquire Cilium reconciliation lock", exc_info=True) + return Result(ResultType.FAILED, str(e)) + + try: + result = action() + finally: + lock_released = self._release_reconciliation_lock() + + if not lock_released and result.result_type != ResultType.FAILED: + return Result( + ResultType.FAILED, + "Failed to release Cilium reconciliation lock", + ) + return result + + def is_skip(self, context: StepContext) -> Result: + """Defer change detection until the k8s plan lock is held in run.""" + return Result(ResultType.COMPLETED) def _cilium_node_config_name(self, hostname: str) -> str: return f"cilium-devices-{hostname}" @@ -1594,7 +1699,17 @@ def _restart_cilium_on_node(self, sunbeam_node_name: str) -> None: self.kube.delete(core_v1.Pod, pod_name, namespace=self._CILIUM_NAMESPACE) self._wait_for_cilium_ready(k8s_node_name, deleted_pod_name=pod_name) + def _reconcile(self, context: StepContext) -> Result: + result = super().is_skip(context) + if result.result_type != ResultType.COMPLETED: + return result + return self._run(context) + def run(self, context: StepContext) -> Result: + """Recalculate and apply Cilium changes under the k8s plan lock.""" + return self._with_reconciliation_lock(lambda: self._reconcile(context)) + + def _run(self, context: StepContext) -> Result: """Apply or delete CiliumNodeConfig resources and restart cilium pods.""" for node in self.to_update: name = node["name"] diff --git a/sunbeam-python/tests/unit/sunbeam/steps/test_k8s.py b/sunbeam-python/tests/unit/sunbeam/steps/test_k8s.py index 22c0c8364..8b426d110 100644 --- a/sunbeam-python/tests/unit/sunbeam/steps/test_k8s.py +++ b/sunbeam-python/tests/unit/sunbeam/steps/test_k8s.py @@ -13,8 +13,11 @@ from lightkube import ApiError from lightkube.types import PatchType -from sunbeam.clusterd.service import ConfigItemNotFoundException -from sunbeam.core.common import ResultType +from sunbeam.clusterd.service import ( + ConfigItemNotFoundException, + TerraformPlanLockConflictException, +) +from sunbeam.core.common import Result, ResultType from sunbeam.core.deployment import Networks from sunbeam.core.juju import ( ActionFailedException, @@ -1461,33 +1464,38 @@ def setup_patches(self, step): kubeconfig_mocker.stop() kube_mocker.stop() - def test_is_skip_no_changes(self, step, step_context): + def test_run_no_changes(self, step, step_context): step._get_outdated_resources = Mock(return_value=([], [])) - result = step.is_skip(step_context) + result = step.run(step_context) assert result.result_type == ResultType.SKIPPED + step.client.cluster.lock_terraform_plan.assert_called_once() + step.client.cluster.unlock_terraform_plan.assert_called_once() - def test_is_skip_outdated_device(self, step, step_context): + def test_run_outdated_device(self, step, step_context): step._get_outdated_resources = Mock(return_value=(["node1"], [])) - result = step.is_skip(step_context) + step._run = Mock(return_value=Result(ResultType.COMPLETED)) + result = step.run(step_context) assert result.result_type == ResultType.COMPLETED assert len(step.to_update) == 1 assert step.to_update[0]["name"] == "node1" - def test_is_skip_missing_config(self, step, step_context): + def test_run_missing_config(self, step, step_context): step._get_outdated_resources = Mock(return_value=(["node1", "node2"], [])) - result = step.is_skip(step_context) + step._run = Mock(return_value=Result(ResultType.COMPLETED)) + result = step.run(step_context) assert result.result_type == ResultType.COMPLETED assert len(step.to_update) == 2 - def test_is_skip_deleted_node(self, step, step_context): + def test_run_deleted_node(self, step, step_context): """Deleted nodes (not in control_nodes) are scheduled for cleanup.""" step._get_outdated_resources = Mock(return_value=([], ["departed-node"])) - result = step.is_skip(step_context) + step._run = Mock(return_value=Result(ResultType.COMPLETED)) + result = step.run(step_context) assert result.result_type == ResultType.COMPLETED assert len(step.to_delete) == 1 assert step.to_delete[0]["name"] == "departed-node" - def test_is_skip_stale_machine_skipped(self, step, jhelper, step_context): + def test_run_stale_machine_skipped(self, step, jhelper, step_context): """Nodes whose machines are gone from juju are skipped, not deleted. This avoids accidentally deleting cilium configs for nodes that are @@ -1497,10 +1505,10 @@ def test_is_skip_stale_machine_skipped(self, step, jhelper, step_context): # node2's machine (id=2) is no longer in juju jhelper.get_machines.return_value = {"1": Mock()} step._get_outdated_resources = Mock(return_value=([], [])) - result = step.is_skip(step_context) + result = step.run(step_context) assert result.result_type == ResultType.SKIPPED - def test_is_skip_single_node_fqdn(self, deployment, client, jhelper, step_context): + def test_run_single_node_fqdn(self, deployment, client, jhelper, step_context): node_info = {"name": "node1", "machineid": "1", "role": ["control"]} client.cluster.get_node_info = Mock(return_value=node_info) step = EnsureCiliumDeviceByHostStep( @@ -1508,12 +1516,13 @@ def test_is_skip_single_node_fqdn(self, deployment, client, jhelper, step_contex ) step.kube = Mock() step._get_outdated_resources = Mock(return_value=(["node1"], [])) + step._run = Mock(return_value=Result(ResultType.COMPLETED)) with patch("sunbeam.steps.k8s.get_kube_client", return_value=step.kube): - result = step.is_skip(step_context) + result = step.run(step_context) assert result.result_type == ResultType.COMPLETED assert step.control_nodes == [node_info] - def test_is_skip_single_node_fqdn_preserves_other_configs( + def test_run_single_node_fqdn_preserves_other_configs( self, deployment, client, jhelper, step_context ): """When fqdn is set, configs for other nodes must not be deleted. @@ -1531,13 +1540,97 @@ def test_is_skip_single_node_fqdn_preserves_other_configs( step.kube = Mock() # node2 needs an update; node1 reported as deleted (not in working set) step._get_outdated_resources = Mock(return_value=(["node2"], ["node1"])) + step._run = Mock(return_value=Result(ResultType.COMPLETED)) with patch("sunbeam.steps.k8s.get_kube_client", return_value=step.kube): - result = step.is_skip(step_context) + result = step.run(step_context) assert result.result_type == ResultType.COMPLETED assert len(step.to_update) == 1 assert step.to_update[0]["name"] == "node2" assert step.to_delete == [] + def test_run_join_reconciles_existing_control_nodes( + self, deployment, client, jhelper, control_nodes, step_context + ): + concurrent_node = {"name": "node3", "machineid": "3"} + client.cluster.list_nodes_by_role.return_value = [ + *control_nodes, + concurrent_node, + ] + joining_node = {**control_nodes[1], "role": ["control"]} + client.cluster.get_node_info.return_value = joining_node + jhelper.get_machines.return_value["3"] = Mock() + step = EnsureCiliumDeviceByHostStep( + deployment, + client, + jhelper, + "test-model", + fqdn="node2.maas", + reconcile_existing=True, + ) + step.kube = Mock() + step._get_outdated_resources = Mock( + return_value=(["node1", "node2", "node3"], ["departed-node"]) + ) + step._run = Mock(return_value=Result(ResultType.COMPLETED)) + tagged_k8s_nodes = [ + _to_kube_object({"labels": {"sunbeam/hostname": node["name"]}}) + for node in control_nodes + ] + + with ( + patch("sunbeam.steps.k8s.get_kube_client", return_value=step.kube), + patch("sunbeam.steps.k8s.list_nodes", return_value=tagged_k8s_nodes), + ): + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + assert len(step.control_nodes) == 2 + assert {node["name"] for node in step.control_nodes} == {"node1", "node2"} + assert {node["name"] for node in step.to_update} == {"node1", "node2"} + assert step.to_delete == [] + + def test_is_skip_defers_reconciliation_to_run(self, step, client, step_context): + step._get_outdated_resources = Mock(return_value=(["node1"], [])) + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.COMPLETED + step._get_outdated_resources.assert_not_called() + client.cluster.lock_terraform_plan.assert_not_called() + client.cluster.unlock_terraform_plan.assert_not_called() + + def test_reconciliation_lock_retries_conflict(self, step, client): + client.cluster.lock_terraform_plan.side_effect = [ + TerraformPlanLockConflictException("plan is locked"), + None, + ] + step._acquire_reconciliation_lock.retry.wait = tenacity.wait_none() + + step._acquire_reconciliation_lock() + + assert client.cluster.lock_terraform_plan.call_count == 2 + + def test_reconciliation_lock_released_when_run_raises( + self, step, client, step_context + ): + step._get_outdated_resources = Mock(return_value=(["node1"], [])) + + with ( + patch("sunbeam.steps.k8s.get_kube_client", return_value=step.kube), + patch.object( + EnsureCiliumDeviceByHostStep, + "_run", + side_effect=KeyboardInterrupt, + ), + pytest.raises(KeyboardInterrupt), + ): + step.run(step_context) + + client.cluster.unlock_terraform_plan.assert_called_once_with( + "k8s-plan", + step._reconciliation_lock, + ) + def test_is_skip_wrong_node_selector(self, step, control_nodes, jhelper): jhelper.get_machine_interfaces.return_value = { "eth0": Mock(space="internal"), @@ -1587,7 +1680,7 @@ def list_side_effect(*args, **kwargs): step.kube.list = Mock(side_effect=list_side_effect) step.kube.delete = Mock() - result = step.run(None) + result = step._run(None) step.kube.apply.assert_called_once() step.kube.patch.assert_called_once_with( @@ -1646,7 +1739,7 @@ def list_side_effect(*args, **kwargs): step.kube.list = Mock(side_effect=list_side_effect) step.kube.delete = Mock() - result = step.run(None) + result = step._run(None) assert step.kube.apply.call_count == 2 assert step.kube.patch.call_count == 2 # clears restart-pending on both @@ -1676,7 +1769,7 @@ def list_side_effect(*args, **kwargs): step.kube.list = Mock(side_effect=list_side_effect) - result = step.run(None) + result = step._run(None) assert step.kube.delete.call_count == 2 # config + pod assert result.result_type == ResultType.COMPLETED @@ -1689,7 +1782,7 @@ def test_run_api_error(self, step): api_error.status = Mock(code=500) step.kube.apply = Mock(side_effect=api_error) - result = step.run(None) + result = step._run(None) assert result.result_type == ResultType.FAILED assert "Failed to apply CiliumNodeConfig for node1" in result.message @@ -1699,7 +1792,7 @@ def test_run_no_interface_found(self, step): step.to_delete = [] step._get_interface = Mock(side_effect=MachineNotFoundException("not found")) - result = step.run(None) + result = step._run(None) assert result.result_type == ResultType.FAILED @@ -1711,7 +1804,7 @@ def test_run_cilium_pod_not_found(self, step): step.kube.apply = Mock() step.kube.list = Mock(return_value=[]) - result = step.run(None) + result = step._run(None) assert result.result_type == ResultType.FAILED assert "No cilium pod found on node node1" in result.message @@ -1747,7 +1840,7 @@ def list_side_effect(*args, **kwargs): patch("sunbeam.steps.k8s.time.monotonic", side_effect=[0.0, 301.0]), patch("sunbeam.steps.k8s.time.sleep"), ): - result = step.run(None) + result = step._run(None) assert result.result_type == ResultType.FAILED assert "did not become Ready" in result.message @@ -1787,7 +1880,7 @@ def list_side_effect(*args, **kwargs): with patch("sunbeam.steps.k8s.time.monotonic", side_effect=[0.0, 1.0, 2.0]): with patch("sunbeam.steps.k8s.time.sleep"): - result = step.run(None) + result = step._run(None) assert result.result_type == ResultType.COMPLETED @@ -1866,7 +1959,7 @@ def list_side_effect(*args, **kwargs): step.kube.delete = Mock() with patch("sunbeam.steps.k8s.list_nodes", return_value=[k8s_node]): - result = step.run(None) + result = step._run(None) step.kube.apply.assert_called_once() step.kube.patch.assert_called_once() @@ -1909,7 +2002,7 @@ def test_run_delete_node_already_gone(self, step): step.kube.delete = Mock() with patch("sunbeam.steps.k8s.list_nodes", return_value=[]): - result = step.run(None) + result = step._run(None) # Config deletion should still happen step.kube.delete.assert_called_once() diff --git a/sunbeam-python/tests/unit/sunbeam/test_clusterd.py b/sunbeam-python/tests/unit/sunbeam/test_clusterd.py index 752f3ec7a..867f26b7c 100644 --- a/sunbeam-python/tests/unit/sunbeam/test_clusterd.py +++ b/sunbeam-python/tests/unit/sunbeam/test_clusterd.py @@ -327,6 +327,90 @@ def test_bootstrap(self): cs = ClusterService(mock_session, "http+unix://mock") cs.bootstrap_cluster("node-1", "10.10.1.10:7000") + def test_lock_terraform_plan(self): + json_data = { + "type": "sync", + "status": "Success", + "status_code": 200, + "operation": "", + "error_code": 0, + "error": "", + "metadata": {}, + } + mock_response = self._mock_response(status=200, json_data=json_data) + mock_session = MagicMock() + mock_session.request.return_value = mock_response + lock = { + "ID": "test-id", + "Operation": "Cilium reconciliation", + "Who": "node-1", + } + + cs = ClusterService(mock_session, "http+unix://mock") + cs.lock_terraform_plan("k8s-plan", lock) + + mock_session.request.assert_called_once_with( + method="put", + url="http+unix://mock/1.0/terraformlock/k8s-plan", + cert=None, + timeout=None, + data=json.dumps(lock), + ) + + def test_lock_terraform_plan_accepts_same_lock(self): + lock = { + "ID": "test-id", + "Operation": "Cilium reconciliation", + "Who": "node-1", + } + mock_response = self._mock_response(status=423, json_data=lock) + http_error = HTTPError("Locked", response=mock_response) + mock_response.raise_for_status.side_effect = http_error + mock_session = MagicMock() + mock_session.request.return_value = mock_response + + cs = ClusterService(mock_session, "http+unix://mock") + cs.lock_terraform_plan("k8s-plan", lock) + + mock_session.request.assert_called_once() + + def test_lock_terraform_plan_reports_conflict(self): + lock = { + "ID": "test-id", + "Operation": "Cilium reconciliation", + "Who": "node-1", + } + existing_lock = {**lock, "ID": "other-id"} + mock_response = self._mock_response(status=409, json_data=existing_lock) + http_error = HTTPError("Conflict", response=mock_response) + mock_response.raise_for_status.side_effect = http_error + mock_session = MagicMock() + mock_session.request.return_value = mock_response + + cs = ClusterService(mock_session, "http+unix://mock") + + with pytest.raises(service.TerraformPlanLockConflictException): + cs.lock_terraform_plan("k8s-plan", lock) + + def test_lock_terraform_plan_preserves_unexpected_error(self): + lock = { + "ID": "test-id", + "Operation": "Cilium reconciliation", + "Who": "node-1", + } + mock_response = self._mock_response(status=500, json_data=lock) + http_error = HTTPError("Internal Error", response=mock_response) + mock_response.raise_for_status.side_effect = http_error + mock_session = MagicMock() + mock_session.request.return_value = mock_response + + cs = ClusterService(mock_session, "http+unix://mock") + + with pytest.raises(HTTPError) as exc_info: + cs.lock_terraform_plan("k8s-plan", lock) + + assert exc_info.value is http_error + def test_bootstrap_when_node_already_exists(self): json_data = { "type": "error",