CA-403379: pre-flight cluster_host state before pool-ha-enable (#7130)

When the chosen HA cluster_stack is corosync (i.e. for a gfs2 heartbeat
SR) every pool host must have an enabled, joined cluster_host on the
matching cluster stack, and this host must currently be quorate. Without
this preflight, that failure surfaces much later inside
Xha_statefile.check_sr_can_host_statefile with the misleading SR_NO_PBDS
error from pool-ha-enable.

This change adds a per-host preflight in `Xapi_ha.enable` that reuses
the existing `NO_COMPATIBLE_CLUSTER_HOST`, `CLUSTERING_DISABLED` and
`CLUSTER_HOST_NOT_JOINED` errors so the caller can pinpoint exactly
which host is the problem.

The preflight runs BEFORE the cluster_stack persisted to the pool DB and
local db, matching the pattern of the existing host_offline check, so a
failed precondition does not leak ha_cluster_stack into the pool state.

The final assert_cluster_host_quorate call queries xapi-clusterd
diagnostics directly rather than reading the Cluster_host.live DB field,
which the corosync_notifyd watcher only updates asynchronously and which
is reset to false for all hosts on any transient quorum blip.

Author: LunfanZhang

ocaml/xapi/xapi_clustering.ml

@@ -363,6 +363,52 @@ let assert_cluster_host_quorate ~__context ~self =
       warn "Cannot query cluster host quorate status" ;
       handle_error error
 
+(* Pre-flight for pool-ha-enable when the chosen [cluster_stack] is corosync:
+   every pool host must have an enabled, joined cluster_host on that stack, and
+   the local coordinator host must currently be quorate. Otherwise the gfs2
+   heartbeat SR's PBD cannot plug and pool-ha-enable would later fail with the
+   misleading SR_NO_PBDS from check_sr_can_host_statefile. Intended to run
+   before ha_cluster_stack is persisted, so a failed precondition does not leak
+   it into the pool DB. *)
+let assert_pool_ready_for_corosync_ha ~__context ~cluster_stack =
+  let localhost = Helpers.get_localhost ~__context in
+  List.iter
+    (fun host ->
+      match find_cluster_host ~__context ~host with
+      | None ->
+          raise
+            Api_errors.(
+              Server_error (no_compatible_cluster_host, [Ref.string_of host])
+            )
+      | Some self ->
+          let cluster = Db.Cluster_host.get_cluster ~__context ~self in
+          let ch_stack =
+            Db.Cluster.get_cluster_stack ~__context ~self:cluster
+          in
+          if ch_stack <> cluster_stack then
+            raise
+              Api_errors.(
+                Server_error (no_compatible_cluster_host, [Ref.string_of host])
+              ) ;
+          assert_cluster_host_enabled ~__context ~self ~expected:true ;
+          if not (Db.Cluster_host.get_joined ~__context ~self) then
+            raise
+              Api_errors.(
+                Server_error (cluster_host_not_joined, [Ref.string_of self])
+              )
+    )
+    (Db.Host.get_all ~__context) ;
+  (* Quorum only needs asserting once, for the local coordinator: it checks live
+     quorum directly via xapi-clusterd diagnostics, sidestepping the
+     Cluster_host.live DB field which the corosync_notifyd watcher only updates
+     asynchronously. *)
+  match find_cluster_host ~__context ~host:localhost with
+  | Some self ->
+      assert_cluster_host_quorate ~__context ~self
+  | None ->
+      warn "%s: coordinator %s has no cluster_host; skipping quorum check"
+        __FUNCTION__ (Ref.string_of localhost)
+
 let assert_cluster_host_is_enabled_for_matching_sms ~__context ~host ~sr_sm_type
     =
   match get_required_cluster_stacks ~__context ~sr_sm_type with

ocaml/xapi/xapi_clustering.mli

@@ -52,6 +52,9 @@ val get_network_internal :
 val assert_cluster_host_enabled :
   __context:Context.t -> self:[`Cluster_host] Ref.t -> expected:bool -> unit
 
+val assert_pool_ready_for_corosync_ha :
+  __context:Context.t -> cluster_stack:string -> unit
+
 val assert_operation_host_target_is_localhost :
   __context:Context.t -> host:[`host] Ref.t -> unit
 

ocaml/xapi/xapi_ha.ml

@@ -1908,6 +1908,8 @@ let enable __context heartbeat_srs configuration =
   let cluster_stack =
     Cluster_stack_constraints.choose_cluster_stack ~__context
   in
+  if cluster_stack = Constants.Ha_cluster_stack.(to_string Corosync) then
+    Xapi_clustering.assert_pool_ready_for_corosync_ha ~__context ~cluster_stack ;
   Db.Pool.set_ha_cluster_stack ~__context ~self:pool ~value:cluster_stack ;
   Localdb.put Constants.ha_cluster_stack cluster_stack ;
   (* Steps from 8.7 Enabling HA in Marathon spec: