Tokio 1.0 upgrade (#152)

Author: RAnders00

Cargo.lock

@@ -14,41 +14,40 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-tokio = { version = "0.2.22", features = ["full"] }
-serde = { version = "1.0.118", features = ["derive"] }
-twitch-irc = { version = "1.0.0", features = ["transport-tcp", "metrics-collection"] }
-futures = "0.3.12"
-toml = "0.5.8"
-structopt = "0.3.21"
-thiserror = "1.0.23"
-log = "0.4.14"
-metrics = "0.12.1"
-metrics-runtime = "0.13.0"
-metrics-observer-prometheus = "0.1.4"
-metrics-core = "0.5.2"
-warp = { git = "https://github.com/RAnders00/warp", branch = "v025-backports" }
-http = "0.2.2"
-hyper = "~0.13"
-chrono = { version = "0.4.19", features = ["serde"] }
-serde_json = "1.0.61"
-itertools = "0.10.0"
-humantime-serde = "1.0.1"
-refinery = { version = "0.4.0", features = ["tokio-postgres"] }
-regex = "1.4.3"
-lazy_static = "1.4.0"
-humantime = "2.1.0"
-derivative = "2.2.0"
-env_logger = "0.8.2"
-simple-process-stats = "=0.2.0"
-mobc = { version = "0.5.12", features = ["tokio"] }
-mobc-postgres = "0.5.0"
-tokio-postgres = { version = "0.5.5", features = ["with-chrono-0_4"] }
-rmp-serde = "0.15.1"
-reqwest = { version = "0.10.10", features = ["json"] }
-rand = "0.8.3"
+async-stream = "0.3.0"
+chrono = { version = "0.4", features = ["serde"] }
+derivative = "2.1"
+env_logger = "0.8"
+futures = "0.3"
+http = "0.2"
+humantime = "2.1"
+humantime-serde = "1.0"
+hyper = "0.14"
+itertools = "0.10"
+lazy_static = "1.4"
+log = "0.4"
+metrics = "0.14"
+metrics-exporter-prometheus = { version = "0.3", default-features = false }
+mobc = { version = "0.7", features = ["tokio"] }
+mobc-postgres = "0.7"
+rand = "0.8"
+refinery = { version = "0.5", features = ["tokio-postgres"] }
+regex = "1.4"
+reqwest = { version = "0.11", features = ["json"] }
+rmp-serde = "0.15"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+simple-process-stats = "1.0"
+structopt = "0.3"
+thiserror = "1.0"
+tokio = { version = "1.0", features = ["full"] }
+tokio-postgres = { version = "0.7", features = ["with-chrono-0_4"] }
+toml = "0.5"
+twitch-irc = { version = "2.2", features = ["transport-tcp", "metrics-collection"] }
+warp = { git = "https://github.com/RAnders00/warp", branch = "v030-backports" }
 
 [target.'cfg(unix)'.dependencies]
-rlimit = "0.5.3"
+rlimit = "0.5"
 
 [profile.release]
 lto = "fat"

Cargo.toml

@@ -356,7 +356,7 @@ WHERE access_token = $1",
             channel_messages.pop_front();
         } else {
             let new_gauge_value = self.messages_stored.fetch_add(1, Ordering::SeqCst) + 1;
-            metrics::gauge!("recent_messages_messages_stored", new_gauge_value as i64);
+            metrics::gauge!("recent_messages_messages_stored", new_gauge_value as f64);
         }
     }
 
@@ -433,7 +433,7 @@ WHERE access_token = $1",
                         .messages_stored
                         .fetch_sub(messages_deleted, Ordering::SeqCst)
                         - messages_deleted;
-                    metrics::gauge!("recent_messages_messages_stored", new_gauge_value as i64);
+                    metrics::gauge!("recent_messages_messages_stored", new_gauge_value as f64);
 
                     // remove the mapping from the map if there are no more messages.
                     if channel_messages.len() == 0 {
@@ -497,7 +497,7 @@ WHERE access_token = $1",
                 .messages_stored
                 .fetch_add(messages_added, Ordering::SeqCst)
                 + messages_added;
-            metrics::gauge!("recent_messages_messages_stored", new_gauge_value as i64);
+            metrics::gauge!("recent_messages_messages_stored", new_gauge_value as f64);
 
             messages_map.insert(channel_login, Arc::new(Mutex::new(channel_messages)));
         }

src/db.rs

@@ -1,6 +1,5 @@
 use crate::config::Config;
 use crate::db::DataStorage;
-use futures::prelude::*;
 use std::borrow::Cow;
 use tokio::sync::mpsc;
 use twitch_irc::login::StaticLoginCredentials;
@@ -39,7 +38,7 @@ impl IrcListener {
         data_storage: &'static DataStorage,
         max_buffer_size: usize,
     ) {
-        while let Some(message) = incoming_messages.next().await {
+        while let Some(message) = incoming_messages.recv().await {
             tokio::spawn(async move {
                 if let Some(channel_login) = message.channel_login() {
                     let message_source = message.source().as_raw_irc();

src/irc_listener.rs

@@ -11,8 +11,9 @@ mod web;
 
 use crate::config::{Args, Config};
 use crate::db::DataStorage;
+#[cfg(not(unix))]
 use futures::prelude::*;
-use metrics_runtime::Receiver;
+use metrics_exporter_prometheus::PrometheusBuilder;
 use structopt::StructOpt;
 
 #[tokio::main]
@@ -24,12 +25,11 @@ async fn main() {
     increase_nofile_rlimit();
 
     // init metrics system
-    let metrics_receiver = Receiver::builder()
-        .build()
-        .expect("failed to create receiver");
-    let metrics_controller = metrics_receiver.controller();
-    metrics_receiver.install();
+    let prom_recorder = Box::leak(Box::new(PrometheusBuilder::new().build()));
+    let prom_handle = prom_recorder.handle();
+    metrics::set_recorder(prom_recorder).unwrap();
     system_monitoring::spawn_system_monitoring();
+    register_application_metrics();
 
     // args and config parsing
     let args = Args::from_args();
@@ -101,7 +101,7 @@ async fn main() {
     tokio::spawn(data_storage.run_task_vacuum_old_messages(config));
     let web_join_handle = tokio::spawn(web::run(
         listener,
-        metrics_controller,
+        prom_handle,
         data_storage,
         irc_listener,
         config,
@@ -111,10 +111,13 @@ async fn main() {
     let ctrl_c_event = async {
         use tokio::signal::unix::{signal, SignalKind};
 
-        let sigint = signal(SignalKind::interrupt()).unwrap();
-        let sigterm = signal(SignalKind::terminate()).unwrap();
+        let mut sigint = signal(SignalKind::interrupt()).unwrap();
+        let mut sigterm = signal(SignalKind::terminate()).unwrap();
 
-        futures::stream::select(sigint, sigterm).next().await
+        tokio::select! {
+            _ = sigint.recv() => {},
+            _ = sigterm.recv() => {}
+        }
     };
     #[cfg(not(unix))]
     let ctrl_c_event =
@@ -172,3 +175,29 @@ fn increase_nofile_rlimit() {
         log::debug!("NOFILE rlimit: no need to increase (soft limit is not below hard limit)")
     }
 }
+
+/// Register all created metrics to initialize them as zero and give them their description.
+fn register_application_metrics() {
+    metrics::register_counter!(
+        "recent_messages_messages_appended",
+        "Total number of messages appended to storage"
+    );
+    metrics::register_gauge!(
+        "recent_messages_messages_stored",
+        "Number of messages currently stored in storage"
+    );
+    metrics::register_counter!(
+        "recent_messages_messages_vacuumed",
+        "Total number of messages that were removed by the automatic vacuum runner"
+    );
+    metrics::register_counter!(
+        "recent_messages_message_vacuum_runs",
+        "Total number of times the automatic vacuum runner has been started for a certain channel"
+    );
+    metrics::register_histogram!(
+        "http_request_duration_nanoseconds",
+        metrics::Unit::Nanoseconds,
+        "Distribution of how many nanoseconds incoming web requests took to answer them"
+    );
+    metrics::register_counter!("http_request", "Total number of incoming HTTP requests");
+}

src/main.rs

@@ -32,7 +32,10 @@ impl ContainerFrame {
         }
 
         if options.hide_moderation_messages
-            && matches!(self.original_message, ServerMessage::ClearChat(_) | ServerMessage::ClearMsg(_))
+            && matches!(
+                self.original_message,
+                ServerMessage::ClearChat(_) | ServerMessage::ClearMsg(_)
+            )
         {
             return None;
         }
@@ -131,8 +134,15 @@ impl MessageContainer {
             ServerMessage::try_from(IRCMessage::parse(&message.message_source).unwrap()).unwrap();
 
         // we export PRIVMSG, CLEARCHAT, CLEARMSG, USERNOTICE, NOTICE and ROOMSTATE
-        if !matches!(server_message, ServerMessage::Privmsg(_) | ServerMessage::ClearChat(_) | ServerMessage::ClearMsg(_) | ServerMessage::UserNotice(_) | ServerMessage::Notice(_) | ServerMessage::RoomState(_))
-        {
+        if !matches!(
+            server_message,
+            ServerMessage::Privmsg(_)
+                | ServerMessage::ClearChat(_)
+                | ServerMessage::ClearMsg(_)
+                | ServerMessage::UserNotice(_)
+                | ServerMessage::Notice(_)
+                | ServerMessage::RoomState(_)
+        ) {
             return;
         }
 

src/message_export.rs

@@ -4,15 +4,37 @@ use tokio::time::Duration;
 
 /// Provides metrics for CPU and memory usage.
 pub fn spawn_system_monitoring() {
-    metrics::gauge!("process_start_time_seconds", Utc::now().timestamp());
+    metrics::register_gauge!(
+        "process_start_time_seconds",
+        "UTC timestamp (in seconds) of when the process started."
+    );
+    metrics::gauge!("process_start_time_seconds", Utc::now().timestamp() as f64);
+
+    metrics::register_gauge!(
+        "process_cpu_user_seconds_total",
+        metrics::Unit::Seconds,
+        "Cumulative number of seconds spent executing in user mode"
+    );
+    metrics::register_gauge!(
+        "process_cpu_system_seconds_total",
+        metrics::Unit::Seconds,
+        "Cumulative number of seconds spent executing in kernel mode"
+    );
+    metrics::register_gauge!(
+        "process_cpu_seconds_total",
+        metrics::Unit::Seconds,
+        "Cumulative number of seconds spent executing in either kernel or user mode"
+    );
+    metrics::register_gauge!(
+        "process_resident_memory_bytes",
+        metrics::Unit::Bytes,
+        "Resident memory usage size as reported by the kernel, in bytes"
+    );
     tokio::spawn(run_system_monitoring());
 }
 
 async fn run_system_monitoring() {
     let mut interval = tokio::time::interval(Duration::from_secs(10));
-    let mut last_seconds_user: u64 = 0;
-    let mut last_seconds_kernel: u64 = 0;
-    let mut last_seconds_total: u64 = 0;
     loop {
         interval.tick().await;
 
@@ -25,23 +47,16 @@ async fn run_system_monitoring() {
             }
         };
 
-        // we do this retarded delta calculation because the `metrics` crate only has functionality
-        // to _increment_ a counter. Additionally, it only supports whole numbers (i64).
-        // For this reason, we simply calculate how many seconds to add since the last run.
-        let user_seconds = system_stats.cpu_time_user.as_secs() - last_seconds_user;
-        let kernel_seconds = system_stats.cpu_time_kernel.as_secs() - last_seconds_kernel;
-        let total_seconds = (system_stats.cpu_time_user + system_stats.cpu_time_kernel).as_secs()
-            - last_seconds_total;
-        metrics::counter!("process_cpu_user_seconds_total", user_seconds);
-        metrics::counter!("process_cpu_system_seconds_total", kernel_seconds);
-        metrics::counter!("process_cpu_seconds_total", total_seconds);
-        last_seconds_user += user_seconds;
-        last_seconds_kernel += kernel_seconds;
-        last_seconds_total += total_seconds;
-
+        let user_seconds = system_stats.cpu_time_user.as_secs_f64();
+        let kernel_seconds = system_stats.cpu_time_kernel.as_secs_f64();
+        let total_seconds =
+            (system_stats.cpu_time_user + system_stats.cpu_time_kernel).as_secs_f64();
+        metrics::gauge!("process_cpu_user_seconds_total", user_seconds);
+        metrics::gauge!("process_cpu_system_seconds_total", kernel_seconds);
+        metrics::gauge!("process_cpu_seconds_total", total_seconds);
         metrics::gauge!(
             "process_resident_memory_bytes",
-            system_stats.memory_usage_bytes as i64
+            system_stats.memory_usage_bytes as f64
         );
     }
 }

src/system_monitoring.rs

@@ -83,7 +83,7 @@ pub async fn get_recent_messages(
 
         if !is_confirmed_joined {
             // wait 5 seconds then check again
-            tokio::time::delay_for(Duration::from_secs(5)).await;
+            tokio::time::sleep(Duration::from_secs(5)).await;
             is_confirmed_joined = irc_listener.is_join_confirmed(channel_login.clone()).await;
         }
 

src/web/get_metrics.rs

@@ -53,7 +53,7 @@ pub async fn set_ignored(
 
         data_storage.purge_messages(&authorization.user_login).await;
         tokio::spawn(async move {
-            tokio::time::delay_for(Duration::from_secs(3)).await;
+            tokio::time::sleep(Duration::from_secs(3)).await;
             data_storage.purge_messages(&authorization.user_login).await;
         });
     } else {