diff --git a/src/cluster.rs b/src/cluster.rs index 650297a..c293e65 100644 --- a/src/cluster.rs +++ b/src/cluster.rs @@ -93,25 +93,19 @@ impl Cluster { } pub async fn main_loop(self: Arc) { - if self.args.manage_resources { - if self.failover { - let futures: Vec<_> = self.hosts.values().map(|h| h.manage_ha(&self)).collect(); + if self.failover { + let futures: Vec<_> = self.hosts.values().map(|h| h.manage_ha(&self)).collect(); - let _ = future::join_all(futures).await; - } else { - let futures: Vec<_> = self.hosts.values().map(|h| h.manage(&self)).collect(); - - let _ = future::join_all(futures).await; - } - } else if self.failover { - let futures: Vec<_> = self.hosts.values().map(|h| h.observe_ha(&self)).collect(); + let _ = future::join_all(futures).await; + } else if self.args.manage_resources { + let futures: Vec<_> = self.hosts.values().map(|h| h.manage(&self)).collect(); let _ = future::join_all(futures).await; } else { let futures: Vec<_> = self.hosts.values().map(|h| h.observe(&self)).collect(); let _ = future::join_all(futures).await; - }; + } } pub fn resource_groups(&self) -> impl Iterator { @@ -260,6 +254,13 @@ impl Cluster { new.apply_state(); + // If the cluster is running in "observe-only" mode, mark every resource group as unmanaged + if !args.manage_resources { + for rg in new.resource_groups() { + rg.set_managed(false); + } + } + Ok(new) } @@ -306,7 +307,9 @@ impl Cluster { ) }) .collect(); - Resource::from_config(self.me, dependents, host, failover_host, self.id, args) + Resource::from_config( + self.me, dependents, /*host, failover_host,*/ self.id, args, + ) } } @@ -366,7 +369,7 @@ impl Cluster { failover_host.clone(), args.clone(), ); - ResourceGroup::new(root, args.clone()) + ResourceGroup::new(root, args.clone(), Arc::clone(&host), failover_host.clone()) }) .collect() } @@ -387,14 +390,11 @@ impl Cluster { pub fn print_summary(&self) { println!("=== Resource Groups ==="); for rg in &self.resource_groups { + println!("\thome node: {}", rg.home_node().id()); + println!("\tfailover node: {:?}", rg.failover_node().map(|h| h.id())); for res in rg.resources() { print!("{}: ", res.id); println!("{}", res.params_string()); - println!("\thome node: {}", res.home_node.id()); - println!( - "\tfailover node: {:?}", - res.failover_node.as_ref().map(|h| h.id()) - ); } } diff --git a/src/host/ha/manage.rs b/src/host/ha.rs similarity index 84% rename from src/host/ha/manage.rs rename to src/host/ha.rs index 563fd28..08e5985 100644 --- a/src/host/ha/manage.rs +++ b/src/host/ha.rs @@ -6,13 +6,13 @@ use std::{io, mem::take}; use { - futures::{future, stream::FuturesUnordered, StreamExt}, + futures::{stream::FuturesUnordered, StreamExt}, log::{debug, warn}, }; use crate::{ cluster::Cluster, - resource::{ManagementError, ResourceStatus}, + resource::{ManagementError, ResourceGroup, ResourceStatus}, }; use super::*; @@ -125,6 +125,51 @@ impl HostState { } impl Host { + fn ha_failover_partner(&self) -> &Arc { + self.failover_partner() + .expect("Host without failover partner in HA routine.") + } + + /// Sends a message of the given type to self. + async fn send_message_to_self(&self, token: ResourceToken, message: Message) { + self.sender.send(new_message(token, message)).await.unwrap(); + } + + /// Sends the token over to the partner in the given message type. + /// + /// This flips the location field -- the caller should NOT adjust location before calling this! + async fn send_message_to_partner(&self, mut token: ResourceToken, message: Message) { + token.location = token.location.swap(); + + let partner = self.ha_failover_partner(); + + partner + .sender + .send(new_message(token, message)) + .await + .unwrap(); + } + + fn send_message_to_partner_delayed( + &self, + mut token: ResourceToken, + message: Message, + dur: u64, + ) { + token.location = token.location.swap(); + + let partner = Arc::clone(self.ha_failover_partner()); + + tokio::task::spawn(async move { + tokio::time::sleep(tokio::time::Duration::from_millis(dur)).await; + partner + .sender + .send(new_message(token, message)) + .await + .unwrap() + }); + } + /// The main management loop for managing a particular host. /// /// When the system starts, each Host task is responsible for determining the state of its @@ -145,12 +190,7 @@ impl Host { pub async fn manage_ha(&self, cluster: &Arc) { let mut state = HostState::new(); - let my_resources = self.mint_resource_tokens(cluster); - - // Check whether this host's primary resources are running locally, in order to determine if - // they should be managed locally or if the failover partner needs to check if they are - // failed over. - state.manage_these_resources = self.startup(cluster, my_resources).await; + state.check_these_resources = self.mint_resource_tokens(cluster); loop { match self.get_client(cluster).await { @@ -194,6 +234,24 @@ impl Host { /// In the meantime, also listen for messages from the partner host task as well as the admin /// CLI utility, and handle them. async fn remote_disconnected_loop(&self, cluster: &Arc, state: &mut HostState) { + for token in take(&mut state.check_these_resources) { + self.send_message_to_partner(token, Message::CheckResourceGroup) + .await; + } + + for token in take(&mut state.manage_these_resources) { + let rg = cluster.get_resource_group(&token.id); + if !rg.get_managed() { + self.send_message_to_partner_delayed( + token, + Message::CheckResourceGroup, + cluster.args.sleep_time, + ); + } else { + state.manage_these_resources.push(token); + } + } + tokio::select! { _ = self.remote_liveness_check(cluster) => {} _ = self.handle_messages_remote_disconnected(cluster, state) => {} @@ -213,24 +271,45 @@ impl Host { loop { match self.receive_message().await { HostMessage::Resource(message) => { - let check_text = format!("Cannot determine resource status because connection failed to its {} host.", - match &message.resource_group.location { - Location::Home => "home", - Location::Away => "failover", - } - ); + let rg = cluster.get_resource_group(&message.resource_group.id); + + // If the host is known down (i.e., was previously fenced), then we know that + // the resource is stopped, and any requests to manage it here can be bounced + // back to the partner to manage it there. + if rg.is_stopped_at_location(message.resource_group.location) { + self.send_message_to_partner_delayed( + message.resource_group, + Message::ManageResourceGroup, + cluster.args.sleep_time, + ); + + continue; + } + rg.root.set_status_recursive(ResourceStatus::Unknown( + "Connection to remote host lost.".to_string(), + )); + + // If the resource group is unmanaged, the admin might start it on the partner, + // so just check on it over there. + if !rg.get_managed() { + self.send_message_to_partner_delayed( + message.resource_group, + Message::CheckResourceGroup, + cluster.args.sleep_time, + ); + + continue; + } match message.kind { Message::ManageResourceGroup => { - let rg = cluster.get_resource_group(&message.resource_group.id); - rg.root - .set_status_recursive(ResourceStatus::Error(check_text)); state.manage_these_resources.push(message.resource_group); } Message::CheckResourceGroup => { - let rg = cluster.get_resource_group(&message.resource_group.id); - rg.root - .set_status_recursive(ResourceStatus::Error(check_text)); - state.check_these_resources.push(message.resource_group); + self.send_message_to_partner_delayed( + message.resource_group, + Message::CheckResourceGroup, + cluster.args.sleep_time, + ); } other => { panic!( @@ -307,9 +386,6 @@ impl Host { Message::ManageResourceGroup => { self.launch_task(&mut tasks, state, cluster, t, client, Task::Manage) } - Message::ObserveResourceGroup => { - self.launch_task(&mut tasks, state, cluster, t, client, Task::Observe) - } Message::SwitchHost => { self.launch_task(&mut tasks, state, cluster, t, client, Task::Switch) } @@ -571,6 +647,7 @@ impl Host { } self.set_connected(false); + self.update_resource_groups_stopped(cluster); warn!("Host {} has been powered off.", self.id()); @@ -584,6 +661,16 @@ impl Host { } } + fn update_resource_groups_stopped(&self, cluster: &Cluster) { + for rg in cluster.host_home_resource_groups(self) { + rg.has_been_stopped(Location::Home); + } + + for rg in cluster.host_home_resource_groups(self.ha_failover_partner()) { + rg.has_been_stopped(Location::Away); + } + } + /// Initiate an admin fence request - unless a fence request was already in progress, in which /// case, do nothing. fn admin_fence_request(&self, state: &mut HostState) { @@ -614,7 +701,7 @@ impl Host { let rg = cluster.get_resource_group(&task.id); // If a resource is currently home, there is nothing to do for it. - if rg.root.home_node.id() == self.id() { + if rg.home_node().id() == self.id() { return true; } @@ -630,6 +717,14 @@ impl Host { state.outstanding_resource_tasks = still_running; } + // Preconditions to being able to move a resource: + // - the resource must be in managed mode + // - the partner must be active + // - the partner must be connected + fn switch_host_legal(&self, rg: &ResourceGroup) -> bool { + rg.get_managed() && self.ha_failover_partner().active() + } + async fn switch_host( &self, token: &ResourceToken, @@ -639,6 +734,10 @@ impl Host { let id = token.id.clone(); let rg = cluster.get_resource_group(&id); + if !self.switch_host_legal(rg) { + return (WhereTo::Here, Message::ManageResourceGroup); + } + match rg.stop_resources(client).await { Ok(()) => (WhereTo::Partner, Message::ManageResourceGroup), Err(ManagementError::Configuration) => { @@ -652,136 +751,65 @@ impl Host { } } - /// The purpose of this procedure is to perform startup logic to discover the existing state of - /// resources when the management service starts. - /// - /// - for each ResourceGroup: - /// - Query the status of the resoource group on this host - /// - if it is discovered to be running, return the token to the caller, who will arrange - /// for the ResourceGroup management routine to run - /// - if it is not running, send a message to the failover partner to check its status - /// there. - /// - the failover partner will either discover it to be running there and start managing - /// it, or will send a message back to this host to start managing it. - /// - /// If a connection to the remote agent for this Host cannot be established, then just send - /// a message to the failover partner to see if the ResourceGroup is running there. - async fn startup( - &self, - cluster: &Cluster, - my_resources: Vec, - ) -> Vec { - let (manage_these, send_these): (Vec, Vec) = - match self.get_client(cluster).await { - Ok(client) => { - let mut manage_these = Vec::new(); - let mut send_these = Vec::new(); - - let statuses = my_resources - .into_iter() - .map(|token| self.home_startup_check(token, cluster, &client)); - - for (token, is_home) in future::join_all(statuses).await { - if is_home { - manage_these.push(token) - } else { - send_these.push(token) - } - } - - (manage_these, send_these) - } - Err(_) => { - self.set_connected(false); - (Vec::new(), my_resources) - } - }; - - for token in send_these { - self.send_message_to_partner(token, Message::CheckResourceGroup) - .await; - } - - manage_these - } - - /// Perform startup check for a resource on its home node. - /// Returns (_, true) if the resource is running on the home node. - async fn home_startup_check( - &self, - token: ResourceToken, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - ) -> (ResourceToken, bool) { - match is_resource_group_running_here(&token, cluster, client, false).await { - Ok(is_running_here) => (token, is_running_here), - Err(_) => (token, false), - } - } - /// Checks if the resource group appears to be running on the node with the given client. /// /// - If yes, this sends a Message::ManageResourceGroup message to self, to direct this Host /// task to begin managing the resource group. /// - /// - If not, then behavior depends on whether this is runnong on the home node or not. + /// - If not, then behavior depends on whether this is running on the home node or not, and + /// also whether the resource is in managed mode. + /// /// - Home node: send a CheckResourceGroup message to the failover Host, to see if the /// resource might be running there. + /// /// - Failover node: send a ManageResourceGroup back to the home Host, to direct it to begin /// managing the resource group. /// /// - If an error was observed, returns a Message::ResourceError to inform the main Host task of /// the situation. - async fn check_resource_group_managed( + async fn check_resource_group( &self, token: &ResourceToken, cluster: &Cluster, client: &ocf_resource_agent::Client, ) -> (WhereTo, Message) { - match is_resource_group_running_here(token, cluster, client, true).await { + let rg = cluster.get_resource_group(&token.id); + + // If the resource is known to be stopped at the partner, then we can confidently update + // the overall status to stopped if it's stopped here as well. But if the resource state is + // unknown on the partner, the overall resource state must remain unknown. + let update_status_if_stopped = rg.is_stopped_at_location(token.location.swap()); + + match rg + .is_running_here(client, token.location, update_status_if_stopped) + .await + { Ok(is_running_here) => { if is_running_here { (WhereTo::Here, Message::ManageResourceGroup) + } else if rg.get_managed() && rg.is_running_nowhere() { + let whereto = match token.location { + Location::Away => WhereTo::Partner, + Location::Home => WhereTo::Here, + }; + (whereto, Message::ManageResourceGroup) } else { - match token.location { - Location::Away => (WhereTo::Partner, Message::ManageResourceGroup), - Location::Home => (WhereTo::Partner, Message::CheckResourceGroup), - } + // Resource is stopped and unmanaged: See if it's running on partner + tokio::time::sleep(tokio::time::Duration::from_millis(cluster.args.sleep_time)) + .await; + (WhereTo::Partner, Message::CheckResourceGroup) } } Err(ManagementError::Configuration) => (WhereTo::Here, Message::ResourceError), - Err(ManagementError::Connection) => (WhereTo::Here, Message::RequestFailover), - } - } - - /// Perform observation of a resource group when the manager process is in "managed mode", but - /// the resource group itself is set to "managed = false". - /// - /// This checks each Host in a pair in turn, and if the resource becomes managed again, then it - /// needs to redo the startup checks. - async fn observe_resource_group_managed_mode( - &self, - token: &ResourceToken, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - ) -> (WhereTo, Message) { - let id = token.id.clone(); - let rg = cluster.get_resource_group(&id); - - match is_resource_group_running_here(token, cluster, client, true).await { - Ok(is_running_here) => { - if is_running_here { - (WhereTo::Here, Message::ManageResourceGroup) - } else if rg.get_managed() { - (WhereTo::Partner, Message::CheckResourceGroup) + Err(ManagementError::Connection) => { + if rg.get_managed() { + (WhereTo::Here, Message::RequestFailover) } else { tokio::time::sleep(tokio::time::Duration::from_millis(cluster.args.sleep_time)) .await; - (WhereTo::Partner, Message::ObserveResourceGroup) + (WhereTo::Partner, Message::CheckResourceGroup) } } - Err(ManagementError::Configuration) => (WhereTo::Here, Message::ResourceError), - Err(ManagementError::Connection) => (WhereTo::Partner, Message::ObserveResourceGroup), } } @@ -796,7 +824,7 @@ impl Host { ) -> (WhereTo, Message) { let rg = cluster.get_resource_group(&token.id); - if !self.active() { + if !self.active() && self.switch_host_legal(rg) { warn!("Host {} asked to manage resource group {}, but it is inactive. Requesting partner manage it.", self.id(), token.id); return (WhereTo::Here, Message::SwitchHost); } @@ -809,7 +837,7 @@ impl Host { match res { // Resource was stopped, and it is no longer supposed to be managed. // Enter "Observe" mode, starting with a check on the partner host. - Ok(()) => (WhereTo::Partner, Message::ObserveResourceGroup), + Ok(()) => (WhereTo::Partner, Message::CheckResourceGroup), Err(ManagementError::Connection) => { debug!( "{}: broken connection while managing {}", @@ -904,14 +932,7 @@ impl Host { ) -> (WhereTo, Message) { match task { Task::Manage => self.manage_resource_group(cluster, token, client).await, - Task::Observe => { - self.observe_resource_group_managed_mode(token, cluster, client) - .await - } - Task::Check => { - self.check_resource_group_managed(token, cluster, client) - .await - } + Task::Check => self.check_resource_group(token, cluster, client).await, Task::Switch => self.switch_host(token, client, cluster).await, } } @@ -919,7 +940,6 @@ impl Host { enum Task { Manage, - Observe, Check, Switch, } diff --git a/src/host/ha/mod.rs b/src/host/ha/mod.rs deleted file mode 100644 index 36e776f..0000000 --- a/src/host/ha/mod.rs +++ /dev/null @@ -1,77 +0,0 @@ -// SPDX-License-Identifier: MIT -// Copyright 2025. Triad National Security, LLC. - -use futures::stream::FuturesUnordered; - -use crate::{ - cluster::Cluster, - resource::{Location, ManagementError}, -}; - -use super::*; - -pub mod manage; -pub mod observe; - -impl Host { - fn ha_failover_partner(&self) -> &Arc { - self.failover_partner() - .expect("Host without failover partner in HA routine.") - } - - /// Sends a message of the given type to self. - async fn send_message_to_self(&self, token: ResourceToken, message: Message) { - self.sender.send(new_message(token, message)).await.unwrap(); - } - - /// Sends the token over to the partner in the given message type. - /// - /// This flips the location field -- the caller should NOT adjust location before calling this! - async fn send_message_to_partner(&self, mut token: ResourceToken, message: Message) { - token.location = token.location.swap(); - - let partner = self.ha_failover_partner(); - - partner - .sender - .send(new_message(token, message)) - .await - .unwrap(); - } - - fn send_message_to_partner_delayed( - &self, - mut token: ResourceToken, - message: Message, - dur: u64, - ) { - token.location = token.location.swap(); - - let partner = Arc::clone(self.ha_failover_partner()); - - tokio::task::spawn(async move { - tokio::time::sleep(tokio::time::Duration::from_millis(dur)).await; - partner - .sender - .send(new_message(token, message)) - .await - .unwrap() - }); - } -} - -/// Determine if a resource is running on the system connected in the given client. -/// -/// If communication fails for some reason, an answer cannot be given, so Err(_) is returned -/// instead. -async fn is_resource_group_running_here( - token: &ResourceToken, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - update_status_if_stopped: bool, -) -> Result { - let rg = cluster.get_resource_group(&token.id); - - rg.is_running_here(client, token.location, update_status_if_stopped) - .await -} diff --git a/src/host/ha/observe.rs b/src/host/ha/observe.rs deleted file mode 100644 index b9f59a0..0000000 --- a/src/host/ha/observe.rs +++ /dev/null @@ -1,414 +0,0 @@ -// SPDX-License-Identifier: MIT -// Copyright 2025. Triad National Security, LLC. - -//! Observation of a failover cluster with HA pairs. - -use std::mem::take; - -use { - futures::StreamExt, - log::{debug, trace}, -}; - -use crate::{cluster::Cluster, resource::ResourceStatus}; - -use super::*; - -struct HostState { - resources_to_observe: Vec, - resources_with_errors: Vec, - outstanding_resource_tasks: Vec, - exit_connected_loop_requested: bool, - resources_in_transit: Vec, -} - -impl HostState { - fn new() -> Self { - Self { - resources_to_observe: Vec::new(), - resources_with_errors: Vec::new(), - outstanding_resource_tasks: Vec::new(), - exit_connected_loop_requested: false, - resources_in_transit: Vec::new(), - } - } - - fn resource_task_exited(&mut self, id: &str) { - let still_running = take(&mut self.outstanding_resource_tasks) - .into_iter() - .filter(|task| task.id != id) - .collect(); - - self.outstanding_resource_tasks = still_running; - } - - fn lost_connection(&mut self, id: &str) { - self.exit_connected_loop_requested = true; - self.resource_task_exited(id); - for revoke in &self.outstanding_resource_tasks { - revoke.lost_connection.notify_one(); - } - } - - fn should_exit_connected_loop(&mut self) -> bool { - if !self.exit_connected_loop_requested { - return false; - } - - if self.outstanding_resource_tasks.is_empty() { - self.exit_connected_loop_requested = false; - true - } else { - false - } - } -} - -impl Host { - pub async fn observe_ha(&self, cluster: &Cluster) { - let my_resources = self.mint_resource_tokens(cluster); - debug!("host {}: resources: {my_resources:?}", self.id()); - - let mut state = HostState::new(); - - state.resources_to_observe = my_resources; - - loop { - match self.get_client(cluster).await { - Ok(client) => { - debug!( - "Host {} established connection to its remote agent.", - self.id() - ); - self.remote_connected_loop_observe(&mut state, cluster, &client) - .await; - - for rg in take(&mut state.resources_in_transit) { - self.send_message_to_partner(rg, Message::CheckResourceGroup) - .await; - } - } - Err(_) => { - debug!( - "Host {} failed to establish connection to its remote agent.", - self.id() - ); - - for token in take(&mut state.resources_to_observe) { - self.send_message_to_partner(token, Message::CheckResourceGroup) - .await; - } - - self.remote_disconnected_loop_observe(cluster).await; - } - } - } - } - - async fn remote_disconnected_loop_observe(&self, cluster: &Cluster) { - tokio::select! { - _ = self.remote_liveness_check(cluster) => {} - _ = self.handle_messages_remote_disconnected_observe(cluster) => {} - } - } - - async fn handle_messages_remote_disconnected_observe(&self, cluster: &Cluster) { - loop { - match self.receive_message().await { - HostMessage::Command(command) => { - panic!("Unexpected to receive command {command:?} in ha observe mode.") - } - HostMessage::Resource(event) => { - match event.kind { - Message::RequestFailover | Message::SwitchHost | Message::TaskCanceled => { - panic!( - "Unexpected to receive a {:?} event in observe mode.", - event.kind - ); - } - Message::ResourceError => { - panic!("Unexpected to receive a resource error message in disconnected mode."); - } - Message::ManageResourceGroup - | Message::CheckResourceGroup - | Message::ObserveResourceGroup => { - // If we received one of these messages, it means the resource is not - // running on the partner. But we don't know if it's running here since - // we can't reach the remote, so we have to update the status to - // unknown. - cluster - .get_resource_group(&event.resource_group.id) - .root - .set_status_recursive(ResourceStatus::Unknown( - "Connection to remote host lost.".to_string(), - )); - self.send_message_to_partner_delayed( - event.resource_group, - Message::CheckResourceGroup, - cluster.args.sleep_time, - ); - } - } - } - HostMessage::TaskDone(_) => {} - HostMessage::MessageFollows => { - panic!("Unexpect to get MessageFollows in this context.") - } - } - } - } - - async fn remote_connected_loop_observe( - &self, - state: &mut HostState, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - ) { - // Create a set of tasks related to this host's management duties. - let mut tasks: ManagementTasks = FuturesUnordered::new(); - - tasks.push(Box::pin(self.receive_message())); - - for token in take(&mut state.resources_to_observe) { - self.launch_observe_task( - &mut tasks, - state, - cluster, - token, - client, - Task::CheckPartnerUnknown, - ); - } - - while let Some(event) = tasks.next().await { - trace!("Host {} got event: {event:?}", self.id()); - match event { - HostMessage::Command(command) => { - panic!("Unexpected to receive command {command:?} in ha observe mode.") - } - HostMessage::Resource(event) => { - tasks.push(Box::pin(self.receive_message())); - let id = &event.resource_group.id; - match event.kind { - Message::SwitchHost => { - panic!( - "Unexpected to receive a {:?} event in observe mode.", - event.kind - ) - } - // Although this is "observe-only" mode, the Manage message is used to indicate - // that the resource is known to be stopped on the partner - meaning if it is - // stopped here as well, the status should be updated to "stopped" instead of - // being left at "unknown". - Message::ManageResourceGroup => { - self.launch_observe_task( - &mut tasks, - state, - cluster, - event.resource_group, - client, - Task::CheckPartnerKnownStopped, - ); - } - Message::CheckResourceGroup => { - self.launch_observe_task( - &mut tasks, - state, - cluster, - event.resource_group, - client, - Task::CheckPartnerUnknown, - ); - } - Message::ObserveResourceGroup => { - self.launch_observe_task( - &mut tasks, - state, - cluster, - event.resource_group, - client, - Task::Observe, - ); - } - Message::ResourceError => { - state.resource_task_exited(id); - state.resources_with_errors.push(event.resource_group); - } - Message::TaskCanceled => { - state.resource_task_exited(id); - cluster.get_resource_group(id).root.set_status_recursive( - ResourceStatus::Unknown( - "Connection to remote host lost.".to_string(), - ), - ); - state.resources_in_transit.push(event.resource_group); - } - Message::RequestFailover => { - state.lost_connection(id); - state.resources_in_transit.push(event.resource_group); - } - }; - } - HostMessage::TaskDone(id) => state.resource_task_exited(&id), - HostMessage::MessageFollows => {} - } - - if state.should_exit_connected_loop() { - return; - } - } - - unreachable!( - "Tasks loop should not exit, a receive message task should always be registered." - ) - } - - async fn check_resource_group( - &self, - token: &ResourceToken, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - update_status_if_stopped: bool, - ) -> (WhereTo, Message) { - match is_resource_group_running_here(token, cluster, client, update_status_if_stopped).await - { - Ok(is_running_here) => { - if is_running_here { - (WhereTo::Here, Message::ObserveResourceGroup) - } else { - tokio::time::sleep(tokio::time::Duration::from_millis(cluster.args.sleep_time)) - .await; - (WhereTo::Partner, Message::ManageResourceGroup) - } - } - Err(ManagementError::Connection) => { - debug!( - "{}: broken connection while checking {}", - self.id(), - token.id - ); - (WhereTo::Here, Message::RequestFailover) - } - Err(ManagementError::Configuration) => { - debug!( - "host {} got a configuration error when managing resource group {}", - self.id(), - token.id - ); - (WhereTo::Here, Message::ResourceError) - } - } - } - - async fn observe_resource_group_ha( - &self, - token: &ResourceToken, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - ) -> (WhereTo, Message) { - let id = &token.id; - let rg = cluster.get_resource_group(id); - match rg.observe_loop(client, true, token.location).await { - // Resource stopped: need to see if it started running on partner. - Ok(()) => { - tokio::time::sleep(tokio::time::Duration::from_millis(cluster.args.sleep_time)) - .await; - (WhereTo::Partner, Message::ManageResourceGroup) - } - Err(ManagementError::Connection) => { - debug!("{}: broken connection while observing {}", self.id(), id); - (WhereTo::Here, Message::RequestFailover) - } - Err(ManagementError::Configuration) => { - debug!( - "host {} got a configuration error when managing resource group {}", - self.id(), - id - ); - (WhereTo::Here, Message::ResourceError) - } - } - } - - fn launch_observe_task<'a>( - &'a self, - tasks: &mut ManagementTasks<'a>, - state: &mut HostState, - cluster: &'a Cluster, - token: ResourceToken, - client: &'a ocf_resource_agent::Client, - task: Task, - ) { - if state.exit_connected_loop_requested { - state.resource_task_exited(&token.id); - state.resources_in_transit.push(token); - return; - } - - let revoke = ResourceTaskCancel::new(token.id.clone()); - state.outstanding_resource_tasks.push(revoke.clone()); - tasks.push(Box::pin( - self.run_task_with_cancellation(cluster, client, token, revoke, task), - )); - } - - async fn run_task_with_cancellation( - &self, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - token: ResourceToken, - revoke: ResourceTaskCancel, - task: Task, - ) -> HostMessage { - tokio::select! { - biased; - - _ = revoke.lost_connection.notified() => { - self.send_message_to_self(token, Message::TaskCanceled).await; - HostMessage::MessageFollows - } - - _ = revoke.switch_host.notified() => panic!("Unexpected to receive switch_host notification in this context"), - - (whereto, msg) = self.run_task(cluster, client, &token, task) => { - let id = token.id.clone(); - match whereto { - WhereTo::Here => { - self.send_message_to_self(token, msg).await; - HostMessage::MessageFollows - } - WhereTo::Partner => { - self.send_message_to_partner(token, msg).await; - HostMessage::TaskDone(id) - } - } - } - } - } - - async fn run_task( - &self, - cluster: &Cluster, - client: &ocf_resource_agent::Client, - token: &ResourceToken, - task: Task, - ) -> (WhereTo, Message) { - match task { - Task::Observe => self.observe_resource_group_ha(token, cluster, client).await, - Task::CheckPartnerUnknown => { - self.check_resource_group(token, cluster, client, false) - .await - } - Task::CheckPartnerKnownStopped => { - self.check_resource_group(token, cluster, client, true) - .await - } - } - } -} - -enum Task { - Observe, - CheckPartnerUnknown, - CheckPartnerKnownStopped, -} diff --git a/src/host/mod.rs b/src/host/mod.rs index 9be1a81..ee55753 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -409,18 +409,12 @@ struct ResourceMessage { /// The commands that can be sent to a Host management task. #[derive(Debug)] enum Message { - /// Check the status of the resource group to determine if it is running or not. This message - /// carries the assumption that the resource group should be started on the home node, if - /// it is found to not be running on either node in the pair. + /// Check the status of the resource group to determine if it is running or not. CheckResourceGroup, /// Begin management of the resource group. ManageResourceGroup, - /// Begin observation of the resource group -- check on its status, but don't start it if - /// stopped. - ObserveResourceGroup, - /// A resource management task has observed a condition like "connection timed out" and /// failover should be triggered. RequestFailover, diff --git a/src/manager/http.rs b/src/manager/http.rs index ee29188..2047659 100644 --- a/src/manager/http.rs +++ b/src/manager/http.rs @@ -75,7 +75,12 @@ pub struct ResourceJson { } impl ResourceJson { - fn build(res: &Resource, managed: bool) -> Self { + fn build( + res: &Resource, + managed: bool, + home_host: String, + failover_host: Option, + ) -> Self { let mut comment = None; let status = match res.status() { @@ -100,8 +105,8 @@ impl ResourceJson { status, comment, managed, - home_host: res.home_node.id(), - failover_host: res.failover_node.as_ref().map(|h| h.id()), + home_host, + failover_host, } } } @@ -164,8 +169,16 @@ async fn get_status(cluster: Arc) -> Json { .resource_groups() .flat_map(|rg| { let managed = rg.get_managed(); - rg.resources() - .map(move |res| ResourceJson::build(res, managed)) + let home_host = rg.home_node().id(); + let failover_host = rg.failover_node().map(|h| h.id()); + rg.resources().map(move |res| { + ResourceJson::build( + res, + managed, + home_host.to_string(), + failover_host.as_ref().map(|h| h.to_string()), + ) + }) }) .collect(), diff --git a/src/resource.rs b/src/resource.rs index e3a8784..81f4af8 100644 --- a/src/resource.rs +++ b/src/resource.rs @@ -63,14 +63,53 @@ pub struct ResourceGroup { pub root: Resource, managed: Mutex, args: manager::Cli, + home_node: HostKnowledge, + failover_node: Option, +} + +#[derive(Debug)] +struct HostKnowledge { + host: Arc, + state: Mutex, +} + +#[derive(Clone, Copy, Debug, PartialEq)] +enum State { + Running, + Stopped, + Unknown, +} + +impl HostKnowledge { + fn new(host: Arc) -> Self { + Self { + host, + state: Mutex::new(State::Unknown), + } + } + + fn state(&self) -> State { + *self.state.lock().unwrap() + } + + fn set_state(&self, state: State) { + *self.state.lock().unwrap() = state; + } } impl ResourceGroup { - pub fn new(root: Resource, args: manager::Cli) -> Self { + pub fn new( + root: Resource, + args: manager::Cli, + home_node: Arc, + failover_node: Option>, + ) -> Self { Self { root, managed: Mutex::new(true), args, + home_node: HostKnowledge::new(home_node), + failover_node: failover_node.map(HostKnowledge::new), } } @@ -79,7 +118,11 @@ impl ResourceGroup { } pub fn home_node(&self) -> &Arc { - &self.root.home_node + &self.home_node.host + } + + pub fn failover_node(&self) -> Option<&Arc> { + self.failover_node.as_ref().map(|h| &h.host) } /// The host-driven resource management loop manages resources on a given location until @@ -155,9 +198,7 @@ impl ResourceGroup { client: &ocf_resource_agent::Client, loc: Location, ) -> Result<(), ManagementError> { - let futures = self.resources().map(|r| r.update_status(client, loc, true)); - - get_worst_error(future::join_all(futures).await.into_iter()) + self.is_running_here(client, loc, true).await.map(|_| ()) } /// Attempt to start the resources in this resource group on the given location. @@ -166,7 +207,15 @@ impl ResourceGroup { client: &ocf_resource_agent::Client, loc: Location, ) -> Result<(), ManagementError> { - self.root.start_if_needed_recursive(client, loc).await + let res = self.root.start_if_needed_recursive(client, loc).await; + + match res { + Ok(()) => self.set_host_state(State::Running, loc), + Err(ManagementError::Connection) => self.set_host_state(State::Unknown, loc), + _ => {} + }; + + res } /// Attempt to stop the resources in this resource group. @@ -174,7 +223,20 @@ impl ResourceGroup { &self, client: &ocf_resource_agent::Client, ) -> Result<(), ManagementError> { - self.root.stop_recursive(client).await + let loc = match self.root.status() { + ResourceStatus::RunningOnHome => Some(Location::Home), + ResourceStatus::RunningOnAway => Some(Location::Away), + _ => None, + }; + let res = self.root.stop_recursive(client).await; + if let Some(loc) = loc { + match res { + Ok(()) => self.set_host_state(State::Stopped, loc), + Err(ManagementError::Connection) => self.set_host_state(State::Unknown, loc), + _ => {} + } + } + res } pub fn get_overall_status(&self) -> ResourceStatus { @@ -198,9 +260,87 @@ impl ResourceGroup { /// Sets resources group's managed status pub fn set_managed(&self, managed: bool) { let mut managed_status = self.managed.lock().unwrap(); + + // When a resource transitions from being unmanaged to being managed, any assumptions about + // its state MUST be revalidated before the manager can take any actions on it. Therefore, + // clear any out of date knowledge on whether it was known to be running somewhere. + if !*managed_status && managed { + *self.home_node.state.lock().unwrap() = State::Unknown; + if let Some(ref failover_node) = self.failover_node { + *failover_node.state.lock().unwrap() = State::Unknown; + } + } + *managed_status = managed; } + pub fn has_been_stopped(&self, loc: Location) { + self.set_host_state(State::Stopped, loc); + } + + pub fn is_running_nowhere(&self) -> bool { + self.home_node.state() == State::Stopped + && self.failover_node.as_ref().unwrap().state() == State::Stopped + } + + pub fn is_stopped_at_location(&self, loc: Location) -> bool { + let state = match loc { + Location::Home => self.home_node.state(), + Location::Away => self.failover_node.as_ref().expect("must be set").state(), + }; + + state == State::Stopped + } + + fn set_host_state(&self, state: State, loc: Location) { + let host = match loc { + Location::Home => &self.home_node, + Location::Away => self + .failover_node + .as_ref() + .expect("Failover node must be set."), + }; + + host.set_state(state); + + self.assert_knowledge_invariant(); + } + + fn assert_knowledge_invariant(&self) { + if let Some(ref failover_node) = self.failover_node { + let home_state = self.home_node.state(); + let failover_state = failover_node.state(); + if home_state == State::Running && failover_state == State::Running { + panic!("Resource group {} violated invariant: system believes it to be running on both nodes.", + self.id()); + } + } + } + + fn assert_not_running_elsewhere(&self, loc: Location) { + let Some(failover_node) = &self.failover_node else { + // This assertion is irrelevant for non-HA clusters, so just return. + return; + }; + + match loc { + Location::Home => { + if self.root.status() == ResourceStatus::RunningOnAway + || failover_node.state() == State::Running + { + panic!("Manager tried to manage resource {} on its home node while it still believes it to be running on its failover node.", self.id()); + } + } + Location::Away => { + if self.root.status() == ResourceStatus::RunningOnHome + || self.home_node.state() == State::Running + { + panic!("Manager tried to manage resource {} on its failover node while it still believes it to be running on its home node.", self.id()); + } + } + } + } + /// Check if the resource group is running on the system connected via the given Client. /// /// This checks each resource individually for the purpose of updating the status, but it uses @@ -211,15 +351,26 @@ impl ResourceGroup { loc: Location, update_status_if_stopped: bool, ) -> Result { + self.assert_not_running_elsewhere(loc); + let futures = self .resources() .map(|r| r.update_status(client, loc, update_status_if_stopped)); - get_worst_error(future::join_all(futures).await.into_iter())?; + get_worst_error(future::join_all(futures).await.into_iter()).inspect_err(|e| { + if matches!(e, ManagementError::Connection) { + self.set_host_state(State::Unknown, loc); + self.root.set_status_recursive(ResourceStatus::Unknown( + "Connection to remote host lost.".to_string(), + )); + } + })?; if self.root.is_running() { + self.set_host_state(State::Running, loc); Ok(true) } else { + self.set_host_state(State::Stopped, loc); Ok(false) } } @@ -266,8 +417,6 @@ pub struct Resource { // TODO: better privacy here status: Mutex, - pub home_node: Arc, - pub failover_node: Option>, pub args: manager::Cli, } @@ -276,8 +425,6 @@ impl Resource { pub fn from_config( res: crate::config::Resource, dependents: Vec, - home_node: Arc, - failover_node: Option>, id: String, args: manager::Cli, ) -> Self { @@ -288,8 +435,6 @@ impl Resource { status: Mutex::new(ResourceStatus::Unknown( "Manager is starting up".to_string(), )), - home_node, - failover_node, id, args, } diff --git a/tests/ha.rs b/tests/ha.rs index 4a6c467..048174d 100644 --- a/tests/ha.rs +++ b/tests/ha.rs @@ -73,7 +73,7 @@ mod tests { let cluster_status = env.get_status(); for res in cluster_status.resources { - assert_eq!(res.status, "Error"); + assert_eq!(res.status, "Unknown"); } } @@ -100,7 +100,7 @@ mod tests { if res.id.contains("0") { assert_eq!(res.status, "Running"); } else { - assert_eq!(res.status, "Error"); + assert_eq!(res.status, "Unknown"); } } } @@ -146,7 +146,7 @@ mod tests { let cluster_status = env.get_status(); for res in cluster_status.resources { - assert_eq!(res.status, "Error"); + assert_eq!(res.status, "Unknown"); } let _a = env.start_agent(0); @@ -181,7 +181,7 @@ mod tests { let cluster_status = env.get_status(); for res in cluster_status.resources { - assert_eq!(res.status, "Error"); + assert_eq!(res.status, "Unknown"); } let _b = env.start_agent(0); // Now start the agent where the resources are running. @@ -1153,6 +1153,34 @@ mod tests { } } + /// Startup - a host is deactivated, but running an unmanaged resource. + /// + /// The manager should not move the resources (unmanaged dominates deactivated when determining + /// how to treat a resource). + #[test] + fn deactivate5() { + let env = test_env_helper("deactivate5"); + env.start_resource("zpool_0", 0); + env.start_resource("mdt_0", 0); + env.start_resource("zpool_1", 1); + env.start_resource("mdt_1", 1); + + let _a = env.start_agent(0); + let _b = env.start_agent(1); + let _m = env.start_manager(true); + + std::thread::sleep(std::time::Duration::from_secs(1)); + env.unmanage_resource("zpool_1"); + std::thread::sleep(std::time::Duration::from_secs(1)); + env.deactivate_host(1); + + std::thread::sleep(std::time::Duration::from_secs(2)); + let cluster_status = env.get_status(); + for res in cluster_status.resources { + assert_eq!(res.status, "Running"); + } + } + fn deactivate_one_host(env: &HaEnvironment) { std::thread::sleep(std::time::Duration::from_secs(1)); env.deactivate_host(0); @@ -1197,9 +1225,15 @@ mod tests { } let _b = env.start_agent(1); + std::thread::sleep(std::time::Duration::from_secs(1)); env.failback(1).unwrap(); std::thread::sleep(std::time::Duration::from_secs(1)); + let cluster_status = env.get_status(); + for res in cluster_status.resources { + assert_eq!(res.status, "Running"); + } + drop(_b); std::thread::sleep(std::time::Duration::from_secs(1)); let cluster_status = env.get_status(); diff --git a/tests/state.rs b/tests/state.rs index 0e5af6f..6243da5 100644 --- a/tests/state.rs +++ b/tests/state.rs @@ -130,7 +130,7 @@ mod tests { let cluster_status = env.get_status(); for res in cluster_status.resources { if res.id.contains("1") { - assert_eq!(res.status, "Error"); + assert_eq!(res.status, "Unknown"); } else { assert_eq!(res.status, "Running"); } @@ -147,7 +147,7 @@ mod tests { // now fence manually: env.fence(1, true).unwrap(); - std::thread::sleep(std::time::Duration::from_secs(1)); + std::thread::sleep(std::time::Duration::from_secs(2)); let cluster_status = env.get_status(); for res in cluster_status.resources { if res.id.contains("1") {