Add verbosity level to log messages

This commit adds a verbosity level to log messages. The verbosity level
can be configure in the .cnf file or else Note will be assumed by
default. This is to avoid the scheduler from flooding the logs with
messages that indicate the scheduler is running as expected.

There are 4 log levels:
- Info
- Note
- Warning
- Error

The last two are obvious, but the first two are as follows:
- Info: This is just and information, the action that follow this will
not change the state of the proxy.
- Note: This is a note to the user that the action that follows this
will most likely change the state of the proxy.

Closes #16

.gitignore

@@ -10,3 +10,6 @@ target/
 *.pdb
 
 .vscode/
+
+# Demo config
+config_demo.cnf

Cargo.toml

@@ -12,6 +12,7 @@ toml = "0.8.12"
 serde = "1.0"
 chrono = "0.4.35"
 file-guard = "0.2.0"
+once_cell = "1.10.0"
 
 
 [package.metadata.generate-rpm]

build/docker-compose.yml

@@ -13,7 +13,7 @@ services:
     environment:
       - MYSQL_ROOT_PASSWORD=noria
       - MYSQL_DATABASE=noria
-    command: --server-id=1 
+    command: --server-id=1 --local-infile=1
     ports:
       - "3306:3306"
     healthcheck:
@@ -28,7 +28,7 @@ services:
       - "3307:3307"
       - "6034:6034"
     environment:
-      - UPSTREAM_DB_URL=mysql://root:[email protected]:3306/noria
+      - UPSTREAM_DB_URL=mysql://root:[email protected]:3306/imdb
       - LISTEN_ADDRESS=0.0.0.0:3307
     depends_on:
       mysql-master:

readyset_proxysql_scheduler.cnf

@@ -8,6 +8,7 @@ source_hostgroup = 11
 readyset_hostgroup = 99
 warmup_time_s = 60
 lock_file = '/tmp/readyset_scheduler.lock'
-operation_mode="All"
+operation_mode='All'
 number_of_queries=10
-query_discovery_mode='SumTime'
+query_discovery_mode='SumTime'
+log_verbosity='Note'

src/config.rs

@@ -4,7 +4,9 @@ use std::{
     io::Read,
 };
 
-#[derive(serde::Deserialize, Clone, Copy, PartialEq, PartialOrd, Default)]
+use crate::messages::MessageType;
+
+#[derive(serde::Deserialize, Clone, Copy, PartialEq, PartialOrd, Default, Debug)]
 pub enum OperationMode {
     HealthCheck,
     QueryDiscovery,
@@ -33,7 +35,7 @@ impl Display for OperationMode {
     }
 }
 
-#[derive(serde::Deserialize, Clone, Copy, PartialEq, PartialOrd, Default)]
+#[derive(serde::Deserialize, Clone, Copy, PartialEq, PartialOrd, Default, Debug)]
 pub enum QueryDiscoveryMode {
     #[default]
     CountStar,
@@ -66,7 +68,7 @@ impl From<String> for QueryDiscoveryMode {
     }
 }
 
-#[derive(serde::Deserialize, Clone)]
+#[derive(serde::Deserialize, Clone, Debug)]
 pub struct Config {
     pub proxysql_user: String,
     pub proxysql_password: String,
@@ -83,6 +85,7 @@ pub struct Config {
     pub query_discovery_mode: Option<QueryDiscoveryMode>,
     pub query_discovery_min_execution: Option<u64>,
     pub query_discovery_min_row_sent: Option<u64>,
+    pub log_verbosity: Option<MessageType>,
 }
 
 pub fn read_config_file(path: &str) -> Result<String, std::io::Error> {

src/main.rs

@@ -7,6 +7,7 @@ mod queries;
 use clap::Parser;
 use config::read_config_file;
 use file_guard::Lock;
+use messages::MessageType;
 use mysql::{Conn, OptsBuilder};
 use proxysql::ProxySQL;
 use std::fs::OpenOptions;
@@ -22,10 +23,11 @@ struct Args {
 }
 
 fn main() {
-    messages::print_info("Running readyset_scheduler");
     let args = Args::parse();
     let config_file = read_config_file(&args.config).expect("Failed to read config file");
     let config = config::parse_config_file(&config_file).expect("Failed to parse config file");
+    messages::set_log_verbosity(config.clone().log_verbosity.unwrap_or(MessageType::Note));
+    messages::print_info("Running readyset_scheduler");
     let file = match OpenOptions::new()
         .read(true)
         .write(true)

src/messages.rs

@@ -1,24 +1,72 @@
 use std::process;
 
 use chrono::{DateTime, Local};
+use once_cell::sync::Lazy;
+use std::sync::Mutex;
 
-enum MessageType {
+#[derive(Clone, Copy, serde::Deserialize, Debug, Default, PartialEq, PartialOrd)]
+pub enum MessageType {
+    /// Information message, this will not result in any action
     Info,
+    /// Note message, this will result in some action that changes state
+    #[default]
+    Note,
+    /// Warning message
     Warning,
+    /// Error message
     Error,
 }
+
+impl std::fmt::Display for MessageType {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match self {
+            MessageType::Info => write!(f, "Info"),
+            MessageType::Note => write!(f, "Note"),
+            MessageType::Warning => write!(f, "Warning"),
+            MessageType::Error => write!(f, "Error"),
+        }
+    }
+}
+
+static LOG_VERBOSITY: Lazy<Mutex<MessageType>> = Lazy::new(|| Mutex::new(MessageType::default()));
+
+pub fn set_log_verbosity(level: MessageType) {
+    let mut verbosity = LOG_VERBOSITY.lock().unwrap();
+    *verbosity = level;
+}
+
+pub fn get_log_verbosity() -> MessageType {
+    let verbosity = LOG_VERBOSITY.lock().unwrap();
+    *verbosity
+}
+
 fn print_message_with_ts(message: &str, message_type: MessageType) {
     let datetime_now: DateTime<Local> = Local::now();
     let date_formatted = datetime_now.format("%Y-%m-%d %H:%M:%S");
     let pid = process::id();
     match message_type {
-        MessageType::Info => println!("{} [INFO] Readyset[{}]: {}", date_formatted, pid, message),
-        MessageType::Warning => println!(
-            "{} [WARNING] Readyset[{}]: {}",
-            date_formatted, pid, message
-        ),
+        MessageType::Info => {
+            if MessageType::Info >= get_log_verbosity() {
+                println!("{} [INFO] Readyset[{}]: {}", date_formatted, pid, message);
+            }
+        }
+        MessageType::Note => {
+            if MessageType::Note >= get_log_verbosity() {
+                println!("{} [NOTE] Readyset[{}]: {}", date_formatted, pid, message);
+            }
+        }
+        MessageType::Warning => {
+            if MessageType::Warning >= get_log_verbosity() {
+                eprintln!(
+                    "{} [WARNING] Readyset[{}]: {}",
+                    date_formatted, pid, message
+                );
+            }
+        }
         MessageType::Error => {
-            eprintln!("{} [ERROR] Readyset[{}]: {}", date_formatted, pid, message)
+            if MessageType::Error >= get_log_verbosity() {
+                eprintln!("{} [ERROR] Readyset[{}]: {}", date_formatted, pid, message);
+            }
         }
     }
 }
@@ -27,6 +75,10 @@ pub fn print_info(message: &str) {
     print_message_with_ts(message, MessageType::Info);
 }
 
+pub fn print_note(message: &str) {
+    print_message_with_ts(message, MessageType::Note);
+}
+
 pub fn print_warning(message: &str) {
     print_message_with_ts(message, MessageType::Warning);
 }

src/proxysql.rs

@@ -76,10 +76,10 @@ impl ProxySQL {
         let date_formatted = datetime_now.format("%Y-%m-%d %H:%M:%S");
         if self.warmup_time_s > 0 {
             self.conn.query_drop(format!("INSERT INTO mysql_query_rules (username, mirror_hostgroup, active, digest, apply, comment) VALUES ('{}', {}, 1, '{}', 1, '{}: {}')", query.get_user(), self.readyset_hostgroup, query.get_digest(), MIRROR_QUERY_TOKEN, date_formatted)).expect("Failed to insert into mysql_query_rules");
-            messages::print_info("Inserted warm-up rule");
+            messages::print_note("Inserted warm-up rule");
         } else {
             self.conn.query_drop(format!("INSERT INTO mysql_query_rules (username, destination_hostgroup, active, digest, apply, comment) VALUES ('{}', {}, 1, '{}', 1, '{}: {}')", query.get_user(), self.readyset_hostgroup, query.get_digest(), DESTINATION_QUERY_TOKEN, date_formatted)).expect("Failed to insert into mysql_query_rules");
-            messages::print_info("Inserted destination rule");
+            messages::print_note("Inserted destination rule");
         }
         Ok(true)
     }
@@ -147,7 +147,7 @@ impl ProxySQL {
                     comment, date_formatted
                 );
                 self.conn.query_drop(format!("UPDATE mysql_query_rules SET mirror_hostgroup = NULL, destination_hostgroup = {}, comment = '{}' WHERE rule_id = {}", self.readyset_hostgroup, comment, rule_id)).expect("Failed to update rule");
-                messages::print_info(
+                messages::print_note(
                     format!("Updated rule ID {} from warmup to destination", rule_id).as_str(),
                 );
                 updated_rules = true;
@@ -168,7 +168,7 @@ impl ProxySQL {
                     if ready {
                         status_changes.push((host, HostStatus::Online));
                     } else {
-                        messages::print_info("Readyset is still running Snapshot.");
+                        messages::print_note("Readyset is still running Snapshot.");
                         status_changes.push((host, HostStatus::Shunned));
                     }
                 }
@@ -187,7 +187,7 @@ impl ProxySQL {
                     host.get_hostname(),
                     host.get_port()
                 );
-                messages::print_info(
+                messages::print_note(
                     format!(
                         "Server HG: {}, Host: {}, Port: {} is currently {}. Changing to {}",
                         self.readyset_hostgroup,

src/queries.rs

@@ -178,7 +178,7 @@ impl QueryDiscovery {
                     break;
                 }
                 let digest_text = self.replace_placeholders(query.get_digest_text());
-                messages::print_info(
+                messages::print_note(
                     format!("Going to test query support for {}", digest_text).as_str(),
                 );
                 let supported = proxysql
@@ -187,7 +187,7 @@ impl QueryDiscovery {
                     .check_query_support(&digest_text, query.get_schema()); // Safe to unwrap because we checked if hosts is empty
                 match supported {
                     Ok(true) => {
-                        messages::print_info(
+                        messages::print_note(
                             "Query is supported, adding it to proxysql and readyset"
                                 .to_string()
                                 .as_str(),
@@ -203,7 +203,7 @@ impl QueryDiscovery {
                         current_queries_digest.push(query.get_digest().to_string());
                     }
                     Ok(false) => {
-                        messages::print_info("Query is not supported");
+                        messages::print_note("Query is not supported");
                     }
                     Err(err) => {
                         messages::print_warning(