New health checks: 'protocol_listener', 'port_listener'

Closes #35.
Closes #36.

Author: michaelklishin

Cargo.lock

@@ -15,7 +15,7 @@ reqwest = { version = "0.12", features = [
     "multipart",
     "default-tls"
 ] }
-rabbitmq_http_client = { version = "0.13.0", features = ["core", "blocking", "tabled"] }
+rabbitmq_http_client = { version = "0.14.0", features = ["core", "blocking", "tabled"] }
 serde = { version = "1.0", features = ["derive", "std"] }
 serde_json = "1"
 tabled = "0.17"

Cargo.toml

@@ -16,7 +16,9 @@ use std::path::PathBuf;
 use super::constants::*;
 use super::static_urls::*;
 use clap::{Arg, ArgAction, Command};
-use rabbitmq_http_client::commons::{BindingDestinationType, ExchangeType, QueueType};
+use rabbitmq_http_client::commons::{
+    BindingDestinationType, ExchangeType, QueueType, SupportedProtocol,
+};
 
 pub fn parser() -> Command {
     let after_help = color_print::cformat!(
@@ -911,7 +913,7 @@ fn purge_subcommands() -> [Command; 1] {
         )]
 }
 
-fn health_check_subcommands() -> [Command; 4] {
+fn health_check_subcommands() -> [Command; 6] {
     let node_is_quorum_critical_after_help = color_print::cformat!(
         r#"
 <bold>Doc guides</bold>:
@@ -936,11 +938,42 @@ fn health_check_subcommands() -> [Command; 4] {
             DEPRECATED_FEATURE_GUIDE_URL
         ));
 
+    let port_listener = Command::new("port_listener")
+        .about(
+            "verifies that there's a reachable TCP listener on the given port on the target node",
+        )
+        .arg(
+            Arg::new("port")
+                .long("port")
+                .value_parser(clap::value_parser!(u16)),
+        )
+        .after_long_help(color_print::cformat!(
+            "<bold>Doc guide</bold>: {}",
+            HEALTH_CHECK_GUIDE_URL
+        ));
+
+    let protocol_listener = Command::new("protocol_listener")
+        .about(
+            "verifies that there's a reachable TCP listener on the given protocol alias on the target node",
+        )
+        .arg(
+            Arg::new("protocol")
+                .long("protocol")
+                .value_parser(clap::value_parser!(SupportedProtocol))
+                .long_help("An alias for one of the protocols that RabbitMQ supports, with or without TLS: 'amqp', 'amqp/ssl', 'stream', 'stream/ssl', 'mqtt', 'mqtt/ssl', 'stomp', 'stomp/ssl', 'http/web-mqtt', 'https/web-mqtt', 'http/web-stomp', 'https/web-stomp', 'http/prometheus', 'https/prometheus', 'http', 'https'"),
+        )
+        .after_long_help(color_print::cformat!(
+            "<bold>Doc guide</bold>: {}",
+            HEALTH_CHECK_GUIDE_URL
+        ));
+
     [
         local_alarms,
         cluster_wide_alarms,
         node_is_quorum_critical,
         deprecated_features_in_use,
+        port_listener,
+        protocol_listener,
     ]
 }
 

src/cli.rs

@@ -15,9 +15,9 @@
 
 use clap::ArgMatches;
 use rabbitmq_http_client::commons;
-use rabbitmq_http_client::commons::ExchangeType;
 use rabbitmq_http_client::commons::UserLimitTarget;
 use rabbitmq_http_client::commons::VirtualHostLimitTarget;
+use rabbitmq_http_client::commons::{ExchangeType, SupportedProtocol};
 use std::fs;
 use std::process;
 
@@ -591,6 +591,27 @@ pub fn health_check_node_is_quorum_critical(client: APIClient) -> ClientResult<(
     client.health_check_if_node_is_quorum_critical()
 }
 
+pub fn health_check_port_listener(
+    client: APIClient,
+    command_args: &ArgMatches,
+) -> ClientResult<()> {
+    // the flag is required
+    let port = command_args.get_one::<u16>("port").cloned().unwrap();
+    client.health_check_port_listener(port)
+}
+
+pub fn health_check_protocol_listener(
+    client: APIClient,
+    command_args: &ArgMatches,
+) -> ClientResult<()> {
+    // the flag is required
+    let proto = command_args
+        .get_one::<SupportedProtocol>("protocol")
+        .cloned()
+        .unwrap();
+    client.health_check_protocol_listener(proto)
+}
+
 pub fn close_connection(client: APIClient, command_args: &ArgMatches) -> ClientResult<()> {
     // the flag is required
     let name = command_args.get_one::<String>("name").unwrap();

src/commands.rs

@@ -348,6 +348,14 @@ fn dispatch_subcommand(
             let result = commands::health_check_node_is_quorum_critical(client);
             res_handler.health_check_result(result);
         }
+        ("health_check", "port_listener") => {
+            let result = commands::health_check_port_listener(client, command_args);
+            res_handler.health_check_result(result);
+        }
+        ("health_check", "protocol_listener") => {
+            let result = commands::health_check_protocol_listener(client, command_args);
+            res_handler.health_check_result(result);
+        }
         ("rebalance", "queues") => {
             let result = commands::rebalance_queues(client);
             res_handler.no_output_on_success(result);

src/main.rs

@@ -187,7 +187,8 @@ impl ResultHandler {
 
                 println!("{}", table);
             }
-            _ => {
+            Err(e) => {
+                println!("Error: {:?}", e);
                 self.exit_code = Some(ExitCode::Unavailable);
             }
         }

src/output.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 use rabbitmq_http_client::blocking_api::HttpClientError;
-use rabbitmq_http_client::responses::{HealthCheckFailureDetails, Overview};
+use rabbitmq_http_client::responses::{
+    ClusterAlarmCheckDetails, HealthCheckFailureDetails, Overview, QuorumCriticalityCheckDetails,
+};
 use reqwest::StatusCode;
 use tabled::settings::Panel;
 use tabled::{Table, Tabled};
@@ -135,20 +137,18 @@ pub fn failure_details(error: &HttpClientError) -> Table {
                 },
             ];
 
-            match details {
-                HealthCheckFailureDetails::AlarmCheck(details) => {
-                    data.push(RowOfTwoStrings {
-                        key: "reason",
-                        value: details.reason.as_str(),
-                    });
-                }
-                HealthCheckFailureDetails::NodeIsQuorumCritical(details) => {
-                    data.push(RowOfTwoStrings {
-                        key: "reason",
-                        value: details.reason.as_str(),
-                    });
+            let reason = match details {
+                HealthCheckFailureDetails::AlarmCheck(details) => details.reason.clone(),
+                HealthCheckFailureDetails::NodeIsQuorumCritical(details) => details.reason.clone(),
+                HealthCheckFailureDetails::NoActivePortListener(details) => details.reason.clone(),
+                HealthCheckFailureDetails::NoActiveProtocolListener(details) => {
+                    details.reason.clone()
                 }
             };
+            data.push(RowOfTwoStrings {
+                key: "reason",
+                value: reason.as_str(),
+            });
 
             let tb = Table::builder(data);
             tb.build()
@@ -281,6 +281,8 @@ pub fn health_check_failure(
     let reason = match details {
         HealthCheckFailureDetails::AlarmCheck(ref details) => details.reason.clone(),
         HealthCheckFailureDetails::NodeIsQuorumCritical(ref details) => details.reason.clone(),
+        HealthCheckFailureDetails::NoActivePortListener(ref details) => details.reason.clone(),
+        HealthCheckFailureDetails::NoActiveProtocolListener(ref details) => details.reason.clone(),
     };
     let code_str = format!("{}", status_code);
 
@@ -304,24 +306,32 @@ pub fn health_check_failure(
     ];
     let mut tb = Table::builder(vec);
     match details {
-        HealthCheckFailureDetails::AlarmCheck(
-            rabbitmq_http_client::responses::ClusterAlarmCheckDetails { reason: _, alarms },
-        ) => {
+        HealthCheckFailureDetails::AlarmCheck(ClusterAlarmCheckDetails { reason: _, alarms }) => {
             for alarm in alarms {
                 let key = format!("alarm in effect on node {}", alarm.node);
                 let value = alarm.resource;
                 tb.push_record([key.as_str(), value.as_str()]);
             }
         }
-        HealthCheckFailureDetails::NodeIsQuorumCritical(
-            rabbitmq_http_client::responses::QuorumCriticalityCheckDetails { reason: _, queues },
-        ) => {
+        HealthCheckFailureDetails::NodeIsQuorumCritical(QuorumCriticalityCheckDetails {
+            reason: _,
+            queues,
+        }) => {
             for q in queues {
                 let key = "affected queue, stream or internal component";
                 let value = q.readable_name;
                 tb.push_record([key, value.as_str()]);
             }
         }
+        HealthCheckFailureDetails::NoActivePortListener(details) => {
+            tb.push_record(["inactive port", details.inactive_port.to_string().as_str()]);
+        }
+        HealthCheckFailureDetails::NoActiveProtocolListener(details) => {
+            tb.push_record([
+                "inactive protocol",
+                details.inactive_protocol.to_string().as_str(),
+            ]);
+        }
     };
 
     tb.build()

src/tables.rs

@@ -14,7 +14,7 @@
 use predicates::prelude::*;
 
 mod test_helpers;
-use test_helpers::run_succeeds;
+use test_helpers::{run_fails, run_succeeds};
 
 #[test]
 fn test_health_check_local_alarms() -> Result<(), Box<dyn std::error::Error>> {
@@ -30,3 +30,47 @@ fn test_health_check_cluster_wide_alarms() -> Result<(), Box<dyn std::error::Err
 
     Ok(())
 }
+
+#[test]
+fn test_health_check_port_listener_succeeds() -> Result<(), Box<dyn std::error::Error>> {
+    run_succeeds(["health_check", "port_listener", "--port", "15672"])
+        .stdout(predicate::str::contains("passed"));
+
+    Ok(())
+}
+
+#[test]
+fn test_health_check_port_listener_fails() -> Result<(), Box<dyn std::error::Error>> {
+    run_fails(["health_check", "port_listener", "--port", "15679"])
+        .stdout(predicate::str::contains("failed"));
+
+    Ok(())
+}
+
+#[test]
+fn test_health_check_protocol_listener_succeeds() -> Result<(), Box<dyn std::error::Error>> {
+    run_succeeds(["health_check", "protocol_listener", "--protocol", "amqp"])
+        .stdout(predicate::str::contains("passed"));
+
+    Ok(())
+}
+
+#[test]
+fn test_health_check_protocol_listener_fails() -> Result<(), Box<dyn std::error::Error>> {
+    run_fails([
+        "health_check",
+        "protocol_listener",
+        "--protocol",
+        "https/prometheus",
+    ])
+    .stdout(predicate::str::contains("failed"));
+    run_fails([
+        "health_check",
+        "protocol_listener",
+        "--protocol",
+        "unknown/proto",
+    ])
+    .stdout(predicate::str::contains("failed"));
+
+    Ok(())
+}