X-Git-Url: http://git.euphorik.ch/?a=blobdiff_plain;f=src%2Fmain.rs;h=08039fb26eb442810a7783296bae277554d5cf70;hb=8ff916896124ea900ea2c6b66c87e6cc72ffe1eb;hp=84dd95cd42861f93edc4dcf9f3a1eb0f31507e6f;hpb=a98ad1b7b21ff82a01fd27b73ad4e9095a714bb2;p=stakingWatchdog.git

diff --git a/src/main.rs b/src/main.rs
index 84dd95c..08039fb 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -3,6 +3,9 @@
  */
 
 use std::{
+    fmt,
+    net::UdpSocket,
+    sync::{Arc, Mutex},
     thread,
     time::{self, Duration},
 };
@@ -17,11 +20,11 @@ use serde::Deserialize;
 use crate::config::Config;
 
 mod config;
-// mod error;
 
 const FILE_CONF: &str = "config.ron";
 const CHECK_PERIOD: Duration = Duration::from_secs(10); // 10 s.
-const EMAIL_RESEND_PERIOD: Duration = Duration::from_secs(6 * 60 * 60); // 6 h.
+const PING_TIMEOUT: Duration = Duration::from_secs(10); // 10 s.
+const EMAIL_RESEND_PERIOD: Duration = Duration::from_secs(2 * 60 * 60); // 2 h.
 const STATE_PRINT_PERIOD: Duration = Duration::from_secs(15 * 60); // 15 min.
 const BASE_URI: &str = "http://localhost:5052/eth/v1/";
 
@@ -38,6 +41,8 @@ fn main() -> Result<()> {
         }
     );
 
+    let check_alive_error_mutex = start_check_alive_thread();
+
     let mut time_last_email_send = time::Instant::now() - EMAIL_RESEND_PERIOD;
     let mut time_last_state_printed = time::Instant::now() - STATE_PRINT_PERIOD;
     let mut error_state = false;
@@ -45,15 +50,19 @@ fn main() -> Result<()> {
     loop {
         let time_beginning_loop = time::Instant::now();
 
-        if let Err(error) = check_validators(&config.pub_keys) {
+        if let Err(error) = check_validators(&config.pub_keys)
+            .as_ref()
+            .and(check_alive_error_mutex.lock().unwrap().as_ref())
+        {
             error_state = true;
             println!("Error: {:?}", error);
             if time::Instant::now() - time_last_email_send >= EMAIL_RESEND_PERIOD {
                 // Send e-mail.
                 println!("Sending email...");
                 match send_email(
-                    "Staking ERROR",
-                    &format!("Error: {:?}", error),
+                    "Watchdog: Staking error",
+                    &format!("Error: {}", error),
+                    &config.smtp_relay_address,
                     &config.smtp_login,
                     &config.smtp_password,
                 ) {
@@ -95,6 +104,59 @@ enum CheckError {
     ReqwestError(reqwest::Error),
     ValidatorError { pub_key: String, message: String },
     ValidatorStatusError { pub_key: String, message: String },
+    CheckAlive,
+}
+
+impl fmt::Display for CheckError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            CheckError::HttpError(text) => {
+                write!(
+                    f,
+                    "Beacon node health check can't be reached (HTTP error): {}",
+                    text
+                )
+            }
+            CheckError::NotSync => {
+                write!(
+                    f,
+                    "Beacon node health check is syncing (currently not sync)"
+                )
+            }
+            CheckError::InvalidSyncStatus => {
+                write!(f, "Beacon node health check returns an invalid status code")
+            }
+            CheckError::NodeHavingIssues => {
+                write!(
+                    f,
+                    "Beacon node health check is not initilized or having issue"
+                )
+            }
+            CheckError::UnknownCodeFromHealthCheck(code) => {
+                write!(
+                    f,
+                    "Beacon node health check returns an unknown code: {}",
+                    code
+                )
+            }
+            CheckError::ReqwestError(error) => {
+                write!(f, "Error from reqwest: {}", error)
+            }
+            CheckError::ValidatorError { pub_key, message } => {
+                write!(f, "Validator '{}' returns an error: {}", pub_key, message)
+            }
+            CheckError::ValidatorStatusError { pub_key, message } => {
+                write!(
+                    f,
+                    "Validator '{}' returns a status error: {}",
+                    pub_key, message
+                )
+            }
+            CheckError::CheckAlive => {
+                write!(f, "Check alive ping hasn't been received")
+            }
+        }
+    }
 }
 
 impl From<reqwest::Error> for CheckError {
@@ -184,7 +246,13 @@ fn check_validators(pub_keys: &[String]) -> std::result::Result<(), CheckError>
     Ok(())
 }
 
-fn send_email(title: &str, body: &str, login: &str, pass: &str) -> Result<()> {
+fn send_email(
+    title: &str,
+    body: &str,
+    smtp_relay_address: &str,
+    login: &str,
+    pass: &str,
+) -> Result<()> {
     let email = Message::builder()
         .message_id(None)
         .from("Staking Watchdog <redmine@d-lan.net>".parse()?)
@@ -195,15 +263,46 @@ fn send_email(title: &str, body: &str, login: &str, pass: &str) -> Result<()> {
 
     let creds = Credentials::new(login.to_string(), pass.to_string());
 
-    // Open a remote connection to gmail
-    let mailer = SmtpTransport::relay("mail.gandi.net")?
+    let mailer = SmtpTransport::relay(smtp_relay_address)?
         .credentials(creds)
         .build();
 
-    // Send the email
     let response = mailer.send(&email)?;
 
     println!("{:?}", response);
 
     Ok(())
 }
+
+fn start_check_alive_thread() -> Arc<Mutex<std::result::Result<(), CheckError>>> {
+    let check_alive_error_mutex: Arc<Mutex<std::result::Result<(), CheckError>>> =
+        Arc::new(Mutex::new(Ok(())));
+    let check_alive_error_mutex_copy = check_alive_error_mutex.clone();
+
+    let _thread_check_alive_handle = thread::spawn(move || {
+        let socket = UdpSocket::bind("0.0.0.0:8739").unwrap();
+        socket.set_read_timeout(Some(PING_TIMEOUT)).unwrap();
+
+        let mut buffer = [0u8; 8];
+
+        loop {
+            match socket.recv_from(&mut buffer) {
+                Ok((size, src)) => {
+                    let mut check_alive_error = check_alive_error_mutex.lock().unwrap();
+                    if size == 8 {
+                        *check_alive_error = Ok(());
+                        socket.send_to(&buffer, &src).unwrap();
+                    } else {
+                        *check_alive_error = Err(CheckError::CheckAlive);
+                    }
+                }
+                Err(_error) => {
+                    let mut check_alive_error = check_alive_error_mutex.lock().unwrap();
+                    *check_alive_error = Err(CheckError::CheckAlive);
+                }
+            }
+        }
+    });
+
+    check_alive_error_mutex_copy
+}