closes #70

Signed-off-by: Valerian Saliou <valerian@valeriansaliou.name>

Author: valeriansaliou

PROTOCOL.md

@@ -170,6 +170,7 @@ _The Sonic Channel Control mode is used for administration purposes. Once in thi
 **➡️ Available commands:**
 
 * `TRIGGER`: trigger an action (syntax: `TRIGGER [<action>]?`; time complexity: `O(1)`)
+* `INFO`: get server information (syntax: `INFO`; time complexity: `O(1)`)
 * `PING`: ping server (syntax: `PING`; time complexity: `O(1)`)
 * `HELP`: show help (syntax: `HELP [<manual>]?`; time complexity: `O(1)`)
 * `QUIT`: stop connection (syntax: `QUIT`; time complexity: `O(1)`)

src/channel/command.rs

@@ -11,6 +11,7 @@ use std::str::{self, SplitWhitespace};
 use std::vec::Vec;
 
 use super::format::unescape;
+use super::statistics::ChannelStatistics;
 use crate::query::builder::{QueryBuilder, QueryBuilderResult};
 use crate::query::types::{QueryGenericLang, QuerySearchLimit, QuerySearchOffset};
 use crate::store::fst::StoreFSTPool;
@@ -62,7 +63,7 @@ lazy_static! {
     pub static ref COMMANDS_MODE_INGEST: Vec<&'static str> =
         vec!["PUSH", "POP", "COUNT", "FLUSHC", "FLUSHB", "FLUSHO", "PING", "HELP", "QUIT"];
     pub static ref COMMANDS_MODE_CONTROL: Vec<&'static str> =
-        vec!["TRIGGER", "PING", "HELP", "QUIT"];
+        vec!["TRIGGER", "INFO", "PING", "HELP", "QUIT"];
     pub static ref CONTROL_TRIGGER_ACTIONS: Vec<&'static str> = vec!["consolidate"];
     static ref MANUAL_MODE_SEARCH: HashMap<&'static str, &'static Vec<&'static str>> =
         [("commands", &*COMMANDS_MODE_SEARCH)]
@@ -794,12 +795,34 @@ impl ChannelCommandControl {
                         _ => Err(ChannelCommandError::NotFound),
                     }
                 } else {
-                    Err(ChannelCommandError::InvalidFormat("HELP [<action>]?"))
+                    Err(ChannelCommandError::InvalidFormat("TRIGGER [<action>]?"))
                 }
             }
         }
     }
 
+    pub fn dispatch_info(_parts: SplitWhitespace) -> ChannelResult {
+        if let Ok(statistics) = ChannelStatistics::gather() {
+            let statistics_result = format!(
+                "uptime({}) clients_connected({}) commands_total({}) \
+                 command_latency_best({}) command_latency_worst({}) \
+                 kv_open_count({}) fst_open_count({}) fst_consolidate_count({})",
+                statistics.uptime,
+                statistics.clients_connected,
+                statistics.commands_total,
+                statistics.command_latency_best,
+                statistics.command_latency_worst,
+                statistics.kv_open_count,
+                statistics.fst_open_count,
+                statistics.fst_consolidate_count
+            );
+
+            Ok(vec![ChannelCommandResponse::Result(statistics_result)])
+        } else {
+            Err(ChannelCommandError::InternalError)
+        }
+    }
+
     pub fn dispatch_help(parts: SplitWhitespace) -> ChannelResult {
         ChannelCommandBase::generic_dispatch_help(parts, &*MANUAL_MODE_CONTROL)
     }

src/channel/handle.rs

@@ -16,6 +16,7 @@ use super::message::{
     ChannelMessageResult,
 };
 use super::mode::ChannelMode;
+use super::statistics::CLIENTS_CONNECTED;
 use crate::APP_CONF;
 use crate::LINE_FEED;
 
@@ -72,6 +73,9 @@ impl ChannelHandle {
         // Send connected banner
         write!(stream, "{}{}", *CONNECTED_BANNER, LINE_FEED).expect("write failed");
 
+        // Increment connected clients count
+        *CLIENTS_CONNECTED.write().unwrap() += 1;
+
         // Ensure channel mode is set
         match Self::ensure_start(&stream) {
             Ok(mode) => {
@@ -95,6 +99,9 @@ impl ChannelHandle {
                 write!(stream, "ENDED {}{}", err.to_str(), LINE_FEED).expect("write failed");
             }
         }
+
+        // Decrement connected clients count
+        *CLIENTS_CONNECTED.write().unwrap() -= 1;
     }
 
     fn configure_stream(stream: &TcpStream, is_established: bool) {

src/channel/message.rs

@@ -14,6 +14,7 @@ use super::command::{
     ChannelCommandResponse, ChannelCommandSearch, COMMANDS_MODE_CONTROL, COMMANDS_MODE_INGEST,
     COMMANDS_MODE_SEARCH,
 };
+use super::statistics::{COMMANDS_TOTAL, COMMAND_LATENCY_BEST, COMMAND_LATENCY_WORST};
 use crate::LINE_FEED;
 
 pub struct ChannelMessage;
@@ -107,6 +108,27 @@ impl ChannelMessage {
             );
         }
 
+        // Update command statistics
+        {
+            // Update performance measures
+            // Notice: commands that take 0ms are not accounted for there (ie. those are usually \
+            //   commands that do no work or I/O; they would make statistics less accurate)
+            let command_took_millis = command_took.as_millis() as u32;
+
+            if command_took_millis > *COMMAND_LATENCY_WORST.read().unwrap() {
+                *COMMAND_LATENCY_WORST.write().unwrap() = command_took_millis;
+            }
+            if command_took_millis > 0
+                && (*COMMAND_LATENCY_BEST.read().unwrap() == 0
+                    || command_took_millis < *COMMAND_LATENCY_BEST.read().unwrap())
+            {
+                *COMMAND_LATENCY_BEST.write().unwrap() = command_took_millis;
+            }
+
+            // Increment total commands
+            *COMMANDS_TOTAL.write().unwrap() += 1;
+        }
+
         result
     }
 
@@ -149,6 +171,7 @@ impl ChannelMessageMode for ChannelMessageModeControl {
     fn handle(message: &str) -> Result<Vec<ChannelCommandResponse>, ChannelCommandError> {
         gen_channel_message_mode_handle!(message, COMMANDS_MODE_CONTROL, {
             "TRIGGER" => ChannelCommandControl::dispatch_trigger,
+            "INFO" => ChannelCommandControl::dispatch_info,
             "HELP" => ChannelCommandControl::dispatch_help,
         })
     }

src/channel/mod.rs

@@ -14,3 +14,4 @@ mod message;
 mod mode;
 
 pub mod listen;
+pub mod statistics;

src/channel/statistics.rs

@@ -0,0 +1,60 @@
+// Sonic
+//
+// Fast, lightweight and schema-less search backend
+// Copyright: 2019, Valerian Saliou <valerian@valeriansaliou.name>
+// License: Mozilla Public License v2.0 (MPL v2.0)
+
+use std::ops::Deref;
+use std::sync::RwLock;
+use std::time::Instant;
+
+use crate::store::fst::StoreFSTPool;
+use crate::store::kv::StoreKVPool;
+
+lazy_static! {
+    static ref START_TIME: Instant = Instant::now();
+    pub static ref CLIENTS_CONNECTED: RwLock<u32> = RwLock::new(0);
+    pub static ref COMMANDS_TOTAL: RwLock<u64> = RwLock::new(0);
+    pub static ref COMMAND_LATENCY_BEST: RwLock<u32> = RwLock::new(0);
+    pub static ref COMMAND_LATENCY_WORST: RwLock<u32> = RwLock::new(0);
+}
+
+#[derive(Default)]
+pub struct ChannelStatistics {
+    pub uptime: u64,
+    pub clients_connected: u32,
+    pub commands_total: u64,
+    pub command_latency_best: u32,
+    pub command_latency_worst: u32,
+    pub kv_open_count: usize,
+    pub fst_open_count: usize,
+    pub fst_consolidate_count: usize,
+}
+
+pub fn ensure_states() {
+    // Ensure all statics are initialized (a `deref` is enough to lazily initialize them)
+    let (_, _, _, _, _) = (
+        START_TIME.deref(),
+        CLIENTS_CONNECTED.deref(),
+        COMMANDS_TOTAL.deref(),
+        COMMAND_LATENCY_BEST.deref(),
+        COMMAND_LATENCY_WORST.deref(),
+    );
+}
+
+impl ChannelStatistics {
+    pub fn gather() -> Result<ChannelStatistics, ()> {
+        let (kv_count, fst_count) = (StoreKVPool::count(), StoreFSTPool::count());
+
+        Ok(ChannelStatistics {
+            uptime: START_TIME.elapsed().as_secs(),
+            clients_connected: *CLIENTS_CONNECTED.read().unwrap(),
+            commands_total: *COMMANDS_TOTAL.read().unwrap(),
+            command_latency_best: *COMMAND_LATENCY_BEST.read().unwrap(),
+            command_latency_worst: *COMMAND_LATENCY_WORST.read().unwrap(),
+            kv_open_count: kv_count,
+            fst_open_count: fst_count.0,
+            fst_consolidate_count: fst_count.1,
+        })
+    }
+}

src/main.rs

@@ -51,6 +51,7 @@ use graceful::SignalGuard;
 use log::LevelFilter;
 
 use channel::listen::ChannelListenBuilder;
+use channel::statistics::ensure_states as ensure_states_channel_statistics;
 use config::config::Config;
 use config::logger::ConfigLogger;
 use config::reader::ConfigReader;
@@ -137,6 +138,9 @@ fn make_app_args() -> AppArgs {
 fn ensure_states() {
     // Ensure all statics are valid (a `deref` is enough to lazily initialize them)
     let (_, _) = (APP_ARGS.deref(), APP_CONF.deref());
+
+    // Ensure per-module states
+    ensure_states_channel_statistics();
 }
 
 fn main() {

src/store/fst.rs

@@ -93,6 +93,13 @@ impl StoreFSTPathMode {
 }
 
 impl StoreFSTPool {
+    pub fn count() -> (usize, usize) {
+        (
+            GRAPH_POOL.read().unwrap().len(),
+            GRAPH_CONSOLIDATE.read().unwrap().len(),
+        )
+    }
+
     pub fn acquire<'a, T: Into<&'a str>>(collection: T, bucket: T) -> Result<StoreFSTBox, ()> {
         let (collection_str, bucket_str) = (collection.into(), bucket.into());
 

src/store/kv.rs

@@ -64,6 +64,10 @@ lazy_static! {
 }
 
 impl StoreKVPool {
+    pub fn count() -> usize {
+        STORE_POOL.read().unwrap().len()
+    }
+
     pub fn acquire<'a, T: Into<&'a str>>(
         mode: StoreKVAcquireMode,
         collection: T,