fix(k8s): reconcile Cilium devices on join

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)

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)
+	}
+}

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))

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"

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(