commands: remove stop and start commands

They were significantly broken. They had no awareness of failover, and
also did not communicate with the manager process so could do actions
that contradict what the manager might be trying to do.

If their functionality is still desired, they will need to be
re-implemented as programs that communicate via the manager to perform
their actions. They will also need failover awareness.

Author: bertschinger

src/commands/mod.rs

@@ -8,9 +8,7 @@ pub mod fence;
 pub mod manage;
 pub mod power;
 pub mod reset;
-pub mod start;
 pub mod status;
-pub mod stop;
 pub mod validate;
 
 use {
@@ -26,8 +24,6 @@ use {
 
 use clap::{Parser, Subcommand};
 
-use crate::cluster::Cluster;
-
 /// A `HandledError` represents an error that has already been handled. When you call a function
 /// that returns a `HandledError` or `HandledResult`, you don't need to do anything with that error,
 /// other than just be aware that it happened, and return it on to your caller.
@@ -93,8 +89,6 @@ pub struct Cli {
 #[derive(Subcommand, Debug)]
 pub enum Commands {
     Status(StatusArgs),
-    Start,
-    Stop,
     Discover(DiscoverArgs),
     Failback(FailbackArgs),
     Fence(FenceArgs),
@@ -109,31 +103,18 @@ pub enum Commands {
 
 pub fn main(cli: &Cli) -> HandledResult<()> {
     match &cli.command {
-        Commands::Discover(args) => return discover::discover(args),
-        Commands::Failback(args) => return failback::failback(cli, args),
-        Commands::Fence(args) => return fence::fence(cli, args),
-        Commands::Power(args) => return power::power(cli, args),
-        Commands::Validate => return validate::validate(cli),
-        Commands::Status(args) => return status::status(cli, args),
-        Commands::Manage(args) => return manage::manage(cli, args),
-        Commands::Unmanage(args) => return manage::unmanage(cli, args),
-        Commands::Activate(args) => return activate::activate(cli, args),
-        Commands::Deactivate(args) => return activate::deactivate(cli, args),
-        Commands::Reset(args) => return reset::reset(cli, args),
-        _ => {}
+        Commands::Discover(args) => discover::discover(args),
+        Commands::Failback(args) => failback::failback(cli, args),
+        Commands::Fence(args) => fence::fence(cli, args),
+        Commands::Power(args) => power::power(cli, args),
+        Commands::Validate => validate::validate(cli),
+        Commands::Status(args) => status::status(cli, args),
+        Commands::Manage(args) => manage::manage(cli, args),
+        Commands::Unmanage(args) => manage::unmanage(cli, args),
+        Commands::Activate(args) => activate::activate(cli, args),
+        Commands::Deactivate(args) => activate::deactivate(cli, args),
+        Commands::Reset(args) => reset::reset(cli, args),
     }
-
-    let rt = tokio::runtime::Runtime::new()
-        .handle_err(|e| eprintln!("Error launching tokio runtime: {e}"))?;
-
-    rt.block_on(async {
-        let cluster = Cluster::from_config(cli.config.clone())?;
-        match &cli.command {
-            Commands::Start => start::start(cluster).await,
-            Commands::Stop => stop::stop(cluster).await,
-            _ => unreachable!(),
-        }
-    })
 }
 
 /// Convert multiple nodeset strings into a single, deduplicated NodeSet object.

src/commands/start.rs

@@ -3,14 +3,9 @@
 
 use std::{env, io};
 
-use {futures::AsyncReadExt, rustls::pki_types::ServerName};
+use futures::AsyncReadExt;
 
-use crate::{
-    cluster,
-    remote::ocf,
-    resource::{Location, Resource},
-    tls::get_connector,
-};
+use crate::{cluster, remote::ocf, resource::Resource};
 
 use capnp_rpc::{rpc_twoparty_capnp, twoparty, RpcSystem};
 
@@ -57,7 +52,7 @@ impl From<capnp::Error> for AgentError {
     }
 }
 
-/// Sends an OCF request to perform `op` to a remote agent, determined by `res` and `loc`.
+/// Sends an OCF request to perform `op` to the remote agent reached by `client`.
 ///
 /// Returns a `Result` that contains whether an error occurred while attempting the remote
 /// operation, or contains the result of the operation if the request was succesful.
@@ -69,18 +64,6 @@ impl From<capnp::Error> for AgentError {
 ///
 /// An `Err(_)` variant means that succesful communication did not occur, so it is unknown whether
 /// the operation was attempted or what the outcome was if it was attempted.
-pub async fn remote_ocf_operation(
-    res: &Resource,
-    loc: Location,
-    op: ocf_resource_agent::Operation,
-) -> Result<AgentReply, AgentError> {
-    let request = get_ocf_request(res, loc, op).await?;
-
-    let reply = request.send().promise.await?;
-
-    Ok(get_status(reply)?)
-}
-
 pub async fn remote_ocf_operation_given_client(
     res: &Resource,
     client: &ocf_resource_agent::Client,
@@ -113,65 +96,6 @@ fn get_status(reply: OcfOperationResults) -> Result<AgentReply, capnp::Error> {
     })
 }
 
-/// Create a capnp RPC client and set up the client to perform the operation() RPC.
-async fn get_ocf_request(
-    res: &Resource,
-    loc: Location,
-    op: ocf_resource_agent::Operation,
-) -> io::Result<OperationRequest> {
-    let hostname = match loc {
-        Location::Home => res.home_node.address(),
-        Location::Away => res
-            .failover_node
-            .as_ref()
-            .expect("Called operation on failover node for resource without failover node")
-            .address(),
-    };
-    let stream = tokio::net::TcpStream::connect(hostname).await?;
-    stream.set_nodelay(true).expect("Setting nodelay failed.");
-
-    if res.args.mtls {
-        // Create mtls connector
-        let mtls_connector = get_connector().expect("TODO: handle error here");
-
-        // Set domain/hostname of server we intend to connect to
-        let domain = ServerName::try_from(
-            env::var("HALO_SERVER_DOMAIN_NAME").expect("HALO_SERVER_DOMAIN_NAME not set."),
-        )
-        .unwrap();
-
-        // Perform mtls handshake
-        let mtls_stream = mtls_connector.connect(domain, stream).await?;
-
-        Ok(__get_ocf_request(mtls_stream, res, op))
-    } else {
-        Ok(__get_ocf_request(stream, res, op))
-    }
-}
-
-fn __get_ocf_request<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + 'static>(
-    stream: S,
-    res: &Resource,
-    op: ocf_resource_agent::Operation,
-) -> OperationRequest {
-    let (reader, writer) = tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split();
-    let rpc_network = Box::new(twoparty::VatNetwork::new(
-        futures::io::BufReader::new(reader),
-        futures::io::BufWriter::new(writer),
-        rpc_twoparty_capnp::Side::Client,
-        Default::default(),
-    ));
-    let mut rpc_system = RpcSystem::new(rpc_network, None);
-    let client: ocf_resource_agent::Client = rpc_system.bootstrap(rpc_twoparty_capnp::Side::Server);
-
-    tokio::task::spawn_local(rpc_system);
-
-    let mut request = client.operation_request();
-    prep_request(&mut request, res, op);
-
-    request
-}
-
 /// Prepare a capnp operation RPC request.
 fn prep_request(request: &mut OperationRequest, res: &Resource, op: ocf_resource_agent::Operation) {
     let mut request = request.get();

src/commands/stop.rs

@@ -472,34 +472,6 @@ impl Resource {
         remote_ocf_operation_given_client(self, client, ocf_resource_agent::Operation::Stop).await
     }
 
-    /// Perform a monitor RPC for this resource.
-    pub async fn monitor(&self, loc: Location) -> Result<AgentReply, AgentError> {
-        tokio::task::LocalSet::new()
-            .run_until(async {
-                remote_ocf_operation(self, loc, ocf_resource_agent::Operation::Monitor).await
-            })
-            .await
-    }
-
-    /// Perform a start RPC for this resource.
-    pub async fn start(&self, loc: Location) -> Result<AgentReply, AgentError> {
-        tokio::task::LocalSet::new()
-            .run_until(async {
-                remote_ocf_operation(self, loc, ocf_resource_agent::Operation::Start).await
-            })
-            .await
-    }
-
-    /// Perform a stop RPC for this resource.
-    pub async fn stop(&self) -> Result<AgentReply, AgentError> {
-        tokio::task::LocalSet::new()
-            .run_until(async {
-                remote_ocf_operation(self, Location::Home, ocf_resource_agent::Operation::Stop)
-                    .await
-            })
-            .await
-    }
-
     pub fn status(&self) -> ResourceStatus {
         self.status.lock().unwrap().clone()
     }

src/halo_capnp.rs

@@ -5,11 +5,7 @@
 mod tests {
     use std::sync::Arc;
 
-    use tokio::runtime::Runtime;
-
-    use halo_lib::{
-        halo_capnp::AgentReply, host::FenceCommand, remote::ocf, resource::Location, test_env::*,
-    };
+    use halo_lib::{host::FenceCommand, test_env::*};
 
     /// Create a TestEnvironment for a test.
     ///
@@ -24,79 +20,6 @@ mod tests {
         )
     }
 
-    #[test]
-    fn simple() {
-        let mut env = test_env_helper("simple");
-
-        let agent = TestAgent::new(halo_lib::remote_port(), None);
-
-        let _agent = env.start_remote_agents(vec![agent]);
-
-        let cluster = env.cluster();
-
-        let rt = Runtime::new().unwrap();
-        rt.block_on(async {
-            for res in cluster.resources() {
-                assert!(matches!(
-                    res.start(Location::Home).await,
-                    Ok(AgentReply::Success(ocf::Status::Success))
-                ));
-
-                env.assert_agent_next_line(&agent_expected_line("start", res));
-
-                assert!(matches!(
-                    res.monitor(Location::Home).await,
-                    Ok(AgentReply::Success(ocf::Status::Success))
-                ));
-
-                env.assert_agent_next_line(&agent_expected_line("monitor", res));
-
-                assert!(matches!(
-                    res.stop().await,
-                    Ok(AgentReply::Success(ocf::Status::Success))
-                ));
-                env.assert_agent_next_line(&agent_expected_line("stop", res));
-            }
-        });
-    }
-
-    #[test]
-    fn multi_agent() {
-        let mut env = test_env_helper("multiagent");
-
-        let _agents = env.start_remote_agents(vec![
-            TestAgent::new(8001, Some("mds01".to_string())),
-            TestAgent::new(8002, Some("oss01".to_string())),
-        ]);
-
-        let cluster = env.cluster();
-
-        let rt = Runtime::new().unwrap();
-        rt.block_on(async {
-            for res in cluster.resources() {
-                assert!(matches!(
-                    res.start(Location::Home).await,
-                    Ok(AgentReply::Success(ocf::Status::Success))
-                ));
-
-                env.assert_agent_next_line(&agent_expected_line("start", res));
-
-                assert!(matches!(
-                    res.monitor(Location::Home).await,
-                    Ok(AgentReply::Success(ocf::Status::Success))
-                ));
-
-                env.assert_agent_next_line(&agent_expected_line("monitor", res));
-
-                assert!(matches!(
-                    res.stop().await,
-                    Ok(AgentReply::Success(ocf::Status::Success))
-                ));
-                env.assert_agent_next_line(&agent_expected_line("stop", res));
-            }
-        });
-    }
-
     #[test]
     fn fencing() {
         let env = test_env_helper("fencing");