Use OpenSearch for locking with fallback to peer databag

lib/charms/opensearch/v0/helper_cluster.py

@@ -46,14 +46,8 @@ def suggest_roles(nodes: List[Node], planned_units: int) -> List[str]:
             — odd: "all" the nodes are cm_eligible nodes.
             — even: "all - 1" are cm_eligible and 1 data node.
         """
-        max_cms = ClusterTopology.max_cluster_manager_nodes(planned_units)
-
-        base_roles = ["data", "ingest", "ml", "coordinating_only"]
-        full_roles = base_roles + ["cluster_manager"]
-        nodes_by_roles = ClusterTopology.nodes_count_by_role(nodes)
-        if nodes_by_roles.get("cluster_manager", 0) == max_cms:
-            return base_roles
-        return full_roles
+        # TODO: remove in https://github.com/canonical/opensearch-operator/issues/230
+        return ["data", "ingest", "ml", "coordinating_only", "cluster_manager"]
 
     @staticmethod
     def get_cluster_settings(
@@ -74,6 +68,7 @@ def get_cluster_settings(
 
     @staticmethod
     def recompute_nodes_conf(app_name: str, nodes: List[Node]) -> Dict[str, Node]:
+        # TODO: remove in https://github.com/canonical/opensearch-operator/issues/230
         """Recompute the configuration of all the nodes (cluster set to auto-generate roles)."""
         if not nodes:
             return {}
@@ -86,19 +81,11 @@ def recompute_nodes_conf(app_name: str, nodes: List[Node]) -> Dict[str, Node]:
             else:
                 # Leave node unchanged
                 nodes_by_name[node.name] = node
-        base_roles = ["data", "ingest", "ml", "coordinating_only"]
-        full_roles = base_roles + ["cluster_manager"]
-        highest_unit_number = max(node.unit_number for node in current_cluster_nodes)
         for node in current_cluster_nodes:
-            # we do this in order to remove any non-default role / add any missing default role
-            if len(current_cluster_nodes) % 2 == 0 and node.unit_number == highest_unit_number:
-                roles = base_roles
-            else:
-                roles = full_roles
-
             nodes_by_name[node.name] = Node(
                 name=node.name,
-                roles=roles,
+                # we do this in order to remove any non-default role / add any missing default role
+                roles=["data", "ingest", "ml", "coordinating_only", "cluster_manager"],
                 ip=node.ip,
                 app_name=node.app_name,
                 unit_number=node.unit_number,

lib/charms/opensearch/v0/opensearch_base_charm.py

@@ -66,7 +66,7 @@
 from charms.opensearch.v0.opensearch_fixes import OpenSearchFixes
 from charms.opensearch.v0.opensearch_health import HealthColors, OpenSearchHealth
 from charms.opensearch.v0.opensearch_internal_data import RelationDataStore, Scope
-from charms.opensearch.v0.opensearch_locking import OpenSearchOpsLock
+from charms.opensearch.v0.opensearch_locking import OpenSearchNodeLock
 from charms.opensearch.v0.opensearch_nodes_exclusions import (
     ALLOCS_TO_DELETE,
     VOTING_TO_DELETE,
@@ -87,7 +87,6 @@
 from charms.opensearch.v0.opensearch_secrets import OpenSearchSecrets
 from charms.opensearch.v0.opensearch_tls import OpenSearchTLS
 from charms.opensearch.v0.opensearch_users import OpenSearchUserManager
-from charms.rolling_ops.v0.rollingops import RollingOpsManager
 from charms.tls_certificates_interface.v3.tls_certificates import (
     CertificateAvailableEvent,
 )
@@ -105,7 +104,7 @@
     StorageDetachingEvent,
     UpdateStatusEvent,
 )
-from ops.framework import EventBase
+from ops.framework import EventBase, EventSource
 from ops.model import BlockedStatus, MaintenanceStatus, WaitingStatus
 from tenacity import RetryError, Retrying, stop_after_attempt, wait_fixed
 
@@ -127,9 +126,26 @@
 logger = logging.getLogger(__name__)
 
 
+class _StartOpenSearch(EventBase):
+    """Attempt to acquire lock & start OpenSearch.
+
+    This event will be deferred until OpenSearch starts.
+    """
+
+
+class _RestartOpenSearch(EventBase):
+    """Attempt to acquire lock & restart OpenSearch.
+
+    This event will be deferred until OpenSearch stops. Then, `_StartOpenSearch` will be emitted.
+    """
+
+
 class OpenSearchBaseCharm(CharmBase):
     """Base class for OpenSearch charms."""
 
+    _start_opensearch_event = EventSource(_StartOpenSearch)
+    _restart_opensearch_event = EventSource(_RestartOpenSearch)
+
     def __init__(self, *args, distro: Type[OpenSearchDistribution] = None):
         super().__init__(*args)
 
@@ -147,7 +163,7 @@ def __init__(self, *args, distro: Type[OpenSearchDistribution] = None):
         self.tls = OpenSearchTLS(self, TLS_RELATION)
         self.status = Status(self)
         self.health = OpenSearchHealth(self)
-        self.ops_lock = OpenSearchOpsLock(self)
+        self.node_lock = OpenSearchNodeLock(self)
         self.cos_integration = COSAgentProvider(
             self,
             relation_name=COSRelationName,
@@ -161,14 +177,14 @@ def __init__(self, *args, distro: Type[OpenSearchDistribution] = None):
         self.plugin_manager = OpenSearchPluginManager(self)
         self.backup = OpenSearchBackup(self)
 
-        self.service_manager = RollingOpsManager(
-            self, relation=SERVICE_MANAGER, callback=self._start_opensearch
-        )
         self.user_manager = OpenSearchUserManager(self)
         self.opensearch_provider = OpenSearchProvider(self)
         self.peer_cluster_provider = OpenSearchPeerClusterProvider(self)
         self.peer_cluster_requirer = OpenSearchPeerClusterRequirer(self)
 
+        self.framework.observe(self._start_opensearch_event, self._start_opensearch)
+        self.framework.observe(self._restart_opensearch_event, self._restart_opensearch)
+
         self.framework.observe(self.on.leader_elected, self._on_leader_elected)
         self.framework.observe(self.on.start, self._on_start)
         self.framework.observe(self.on.update_status, self._on_update_status)
@@ -271,7 +287,7 @@ def _on_start(self, event: StartEvent):
 
         # request the start of OpenSearch
         self.status.set(WaitingStatus(RequestUnitServiceOps.format("start")))
-        self.on[self.service_manager.name].acquire_lock.emit(callback_override="_start_opensearch")
+        self._start_opensearch_event.emit()
 
     def _apply_peer_cm_directives_and_check_if_can_start(self) -> bool:
         """Apply the directives computed by the opensearch peer cluster manager."""
@@ -404,7 +420,9 @@ def _on_peer_relation_departed(self, event: RelationDepartedEvent):
     def _on_opensearch_data_storage_detaching(self, _: StorageDetachingEvent):  # noqa: C901
         """Triggered when removing unit, Prior to the storage being detached."""
         # acquire lock to ensure only 1 unit removed at a time
-        self.ops_lock.acquire()
+        if not self.node_lock.acquired:
+            # Raise uncaught exception to prevent Juju from removing unit
+            raise Exception("Unable to acquire lock: Another unit is starting or stopping.")
 
         # if the leader is departing, and this hook fails "leader elected" won"t trigger,
         # so we want to re-balance the node roles from here
@@ -446,7 +464,7 @@ def _on_opensearch_data_storage_detaching(self, _: StorageDetachingEvent):  # no
                     raise OpenSearchHAError(ClusterHealthUnknown)
         finally:
             # release lock
-            self.ops_lock.release()
+            self.node_lock.release()
 
     def _on_update_status(self, event: UpdateStatusEvent):
         """On update status event.
@@ -524,9 +542,7 @@ def _on_config_changed(self, event: ConfigChangedEvent):  # noqa C901
             if self.unit.is_leader():
                 self.status.set(MaintenanceStatus(PluginConfigCheck), app=True)
             if self.plugin_manager.run():
-                self.on[self.service_manager.name].acquire_lock.emit(
-                    callback_override="_restart_opensearch"
-                )
+                self._restart_opensearch_event.emit()
         except (OpenSearchNotFullyReadyError, OpenSearchPluginError) as e:
             if isinstance(e, OpenSearchNotFullyReadyError):
                 logger.warning("Plugin management: cluster not ready yet at config changed")
@@ -618,9 +634,7 @@ def on_tls_conf_set(
 
         # In case of renewal of the unit transport layer cert - restart opensearch
         if renewal and self.is_admin_user_configured() and self.is_tls_fully_configured():
-            self.on[self.service_manager.name].acquire_lock.emit(
-                callback_override="_restart_opensearch"
-            )
+            self._restart_opensearch_event.emit()
 
     def on_tls_relation_broken(self, _: RelationBrokenEvent):
         """As long as all certificates are produced, we don't do anything."""
@@ -699,8 +713,12 @@ def _handle_change_to_main_orchestrator_if_needed(
 
             self.tls.request_new_admin_certificate()
 
-    def _start_opensearch(self, event: EventBase) -> None:  # noqa: C901
+    def _start_opensearch(self, event: _StartOpenSearch) -> None:  # noqa: C901
         """Start OpenSearch, with a generated or passed conf, if all resources configured."""
+        if not self.node_lock.acquired:
+            logger.debug("Lock to start opensearch not acquired. Will retry next event")
+            event.defer()
+            return
         self.peers_data.delete(Scope.UNIT, "started")
         if self.opensearch.is_started():
             try:
@@ -710,25 +728,18 @@ def _start_opensearch(self, event: EventBase) -> None:  # noqa: C901
             return
 
         if not self._can_service_start():
-            self.peers_data.delete(Scope.UNIT, "starting")
+            self.node_lock.release()
             event.defer()
             return
 
-        if self.peers_data.get(Scope.UNIT, "starting", False) and self.opensearch.is_failed():
-            self.peers_data.delete(Scope.UNIT, "starting")
+        if self.opensearch.is_failed():
+            self.node_lock.release()
+            self.status.set(BlockedStatus(ServiceStartError))
             event.defer()
             return
 
         self.unit.status = WaitingStatus(WaitingToStart)
 
-        rel = self.model.get_relation(PeerRelationName)
-        for unit in rel.units.union({self.unit}):
-            if rel.data[unit].get("starting") == "True":
-                event.defer()
-                return
-
-        self.peers_data.put(Scope.UNIT, "starting", True)
-
         try:
             # Retrieve the nodes of the cluster, needed to configure this node
             nodes = self._get_nodes(False)
@@ -739,12 +750,12 @@ def _start_opensearch(self, event: EventBase) -> None:  # noqa: C901
             # Set the configuration of the node
             self._set_node_conf(nodes)
         except OpenSearchHttpError:
-            self.peers_data.delete(Scope.UNIT, "starting")
+            self.node_lock.release()
             event.defer()
             return
         except OpenSearchProvidedRolesException as e:
             logger.exception(e)
-            self.peers_data.delete(Scope.UNIT, "starting")
+            self.node_lock.release()
             event.defer()
             self.unit.status = BlockedStatus(str(e))
             return
@@ -761,7 +772,7 @@ def _start_opensearch(self, event: EventBase) -> None:  # noqa: C901
             event.defer()
         except OpenSearchStartError as e:
             logger.exception(e)
-            self.peers_data.delete(Scope.UNIT, "starting")
+            self.node_lock.release()
             self.status.set(BlockedStatus(ServiceStartError))
             event.defer()
 
@@ -788,8 +799,8 @@ def _post_start_init(self, event: EventBase):
         # Remove the exclusions that could not be removed when no units were online
         self.opensearch_exclusions.delete_current()
 
-        # Remove the 'starting' flag on the unit
-        self.peers_data.delete(Scope.UNIT, "starting")
+        self.node_lock.release()
+
         self.peers_data.put(Scope.UNIT, "started", True)
 
         # apply post_start fixes to resolve start related upstream bugs
@@ -827,18 +838,23 @@ def _stop_opensearch(self) -> None:
         # 3. Remove the exclusions
         self.opensearch_exclusions.delete_current()
 
-    def _restart_opensearch(self, event: EventBase) -> None:
+    def _restart_opensearch(self, event: _RestartOpenSearch) -> None:
         """Restart OpenSearch if possible."""
-        if not self.peers_data.get(Scope.UNIT, "starting", False):
-            try:
-                self._stop_opensearch()
-            except OpenSearchStopError as e:
-                logger.exception(e)
-                event.defer()
-                self.status.set(WaitingStatus(ServiceIsStopping))
-                return
+        if not self.node_lock.acquired:
+            logger.debug("Lock to restart opensearch not acquired. Will retry next event")
+            event.defer()
+            return
 
-        self._start_opensearch(event)
+        try:
+            self._stop_opensearch()
+        except OpenSearchStopError as e:
+            logger.exception(e)
+            self.node_lock.release()
+            event.defer()
+            self.status.set(WaitingStatus(ServiceIsStopping))
+            return
+
+        self._start_opensearch_event.emit()
 
     def _can_service_start(self) -> bool:
         """Return if the opensearch service can start."""
@@ -917,9 +933,7 @@ def _remove_data_role_from_dedicated_cm_if_needed(  # noqa: C901
             return False
 
         self.status.set(WaitingStatus(WaitingToStart))
-        self.on[self.service_manager.name].acquire_lock.emit(
-            callback_override="_restart_opensearch"
-        )
+        self._restart_opensearch_event.emit()
         return True
 
     def _purge_users(self):
@@ -1148,9 +1162,7 @@ def _reconfigure_and_restart_unit_if_needed(self):
             return
 
         self.status.set(WaitingStatus(WaitingToStart))
-        self.on[self.service_manager.name].acquire_lock.emit(
-            callback_override="_restart_opensearch"
-        )
+        self._restart_opensearch_event.emit()
 
     def _recompute_roles_if_needed(self, event: RelationChangedEvent):
         """Recompute node roles:self-healing that didn't trigger leader related event occurred."""

lib/charms/opensearch/v0/opensearch_exceptions.py

@@ -63,10 +63,6 @@ class OpenSearchNotFullyReadyError(OpenSearchError):
     """Exception thrown when a node is started but not full ready to take on requests."""
 
 
-class OpenSearchOpsLockAlreadyAcquiredError(OpenSearchError):
-    """Exception thrown when a node is started but not full ready to take on requests."""
-
-
 class OpenSearchCmdError(OpenSearchError):
     """Exception thrown when an OpenSearch bin command fails."""