logger: initial commit of klog functionality for segcache-rs

Adds a new logger which has command logger specific functionality.

Adds command logging to pelikan_segcache_rs.

Changes to the memcache wire protocol to support providing the
necessary context for command logging.

Fixes `cas` operation in seg storage implementation.

Cargo.toml

@@ -5,6 +5,7 @@ members = [
     "src/rust/config",
     "src/rust/core/server",
     "src/rust/entrystore",
+    "src/rust/logger",
     "src/rust/metrics",
     "src/rust/protocol",
     "src/rust/queues",

config/pingserver.toml

@@ -26,7 +26,27 @@ nevent = 1024
 [buf]
 
 [debug]
+# choose from: error, warn, info, debug, trace
 log_level = "info"
+# optionally, log to the file below instead of standard out
+# log_file = "pingserver.log"
+# backup file name for use with log rotation
+log_backup = "pingserver.log.old"
+# trigger log rotation when the file grows beyond this size (in bytes). Set this
+# option to '0' to disable log rotation.
+log_max_size = 1073741824
+
+[klog]
+# optionally, log commands to the file below
+# file = "pingserver.cmd"
+# backup file name for use with log rotation
+backup = "pingserver.cmd.old"
+# trigger log rotation when the file grows beyond this size (in bytes). Set this
+# option to '0' to disable log rotation.
+max_size = 1073741824
+# specify the sampling ratio, 1 in N commands will be logged. Setting to '0'
+# will disable command logging.
+sample = 100
 
 # NOTE: not currently implemented
 [sockio]

config/segcache.toml

@@ -48,9 +48,34 @@ time_type = "Memcache"
 [debug]
 # choose from: error, warn, info, debug, trace
 log_level = "info"
+# optionally, log to the file below instead of standard out
+# log_file = "segcache.log"
+# backup file name for use with log rotation
+log_backup = "segcache.log.old"
+# trigger log rotation when the file grows beyond this size (in bytes). Set this
+# option to '0' to disable log rotation.
+log_max_size = 1073741824
+
+[klog]
+# optionally, log commands to the file below
+# file = "segcache.cmd"
+# backup file name for use with log rotation
+backup = "segcache.cmd.old"
+# trigger log rotation when the file grows beyond this size (in bytes). Set this
+# option to '0' to disable log rotation.
+max_size = 1073741824
+# specify the sampling ratio, 1 in N commands will be logged. Setting to '0'
+# will disable command logging.
+sample = 100
 
 [sockio]
 
 [tcp]
 
 [tls]
+# certificate chain used to validate client certificate
+# certificate_chain = "client.chain"
+# server certificate
+# certificate = "server.crt"
+# server private key
+# private_key = "server.key"

src/rust/config/src/debug.rs

@@ -2,35 +2,50 @@
 // Licensed under the Apache License, Version 2.0
 // http://www.apache.org/licenses/LICENSE-2.0
 
+use crate::units::*;
 use log::Level;
 use serde::{Deserialize, Serialize};
 
 // constants to define default values
-const DEBUG_LOG_LEVEL: Level = Level::Info;
-const DEBUG_LOG_FILE: Option<String> = None;
-const DEBUG_LOG_NBUF: usize = 0;
+const LOG_LEVEL: Level = Level::Info;
+const LOG_FILE: Option<String> = None;
+const LOG_BACKUP: Option<String> = None;
+const LOG_MAX_SIZE: u64 = GB as u64;
+const LOG_NBUF: usize = 0;
 
 // helper functions
 fn log_level() -> Level {
-    DEBUG_LOG_LEVEL
+    LOG_LEVEL
 }
 
 fn log_file() -> Option<String> {
-    DEBUG_LOG_FILE
+    LOG_FILE
+}
+
+fn log_backup() -> Option<String> {
+    LOG_BACKUP
+}
+
+fn log_max_size() -> u64 {
+    LOG_MAX_SIZE
 }
 
 fn log_nbuf() -> usize {
-    DEBUG_LOG_NBUF
+    LOG_NBUF
 }
 
 // struct definitions
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct DebugConfig {
     #[serde(with = "LevelDef")]
     #[serde(default = "log_level")]
     log_level: Level,
     #[serde(default = "log_file")]
     log_file: Option<String>,
+    #[serde(default = "log_backup")]
+    log_backup: Option<String>,
+    #[serde(default = "log_max_size")]
+    log_max_size: u64,
     #[serde(default = "log_nbuf")]
     log_nbuf: usize,
 }
@@ -57,6 +72,17 @@ impl DebugConfig {
         self.log_file.clone()
     }
 
+    pub fn log_backup(&self) -> Option<String> {
+        match &self.log_backup {
+            Some(path) => Some(path.clone()),
+            None => self.log_file.as_ref().map(|path| format!("{}.old", path)),
+        }
+    }
+
+    pub fn log_max_size(&self) -> u64 {
+        self.log_max_size
+    }
+
     pub fn log_nbuf(&self) -> usize {
         self.log_nbuf
     }
@@ -68,6 +94,8 @@ impl Default for DebugConfig {
         Self {
             log_level: log_level(),
             log_file: log_file(),
+            log_backup: log_backup(),
+            log_max_size: log_max_size(),
             log_nbuf: log_nbuf(),
         }
     }

src/rust/config/src/klog.rs

@@ -0,0 +1,123 @@
+// Copyright 2020 Twitter, Inc.
+// Licensed under the Apache License, Version 2.0
+// http://www.apache.org/licenses/LICENSE-2.0
+
+use crate::units::*;
+use serde::{Deserialize, Serialize};
+
+////////////////////////////////////////////////////////////////////////////////
+// constants to define default values
+////////////////////////////////////////////////////////////////////////////////
+
+// log to the file path
+const FILE: Option<String> = None;
+
+// log will rotate to the given backup path
+const BACKUP: Option<String> = None;
+
+// flush interval in milliseconds
+const INTERVAL: usize = 100;
+
+// max log size before rotate in bytes
+const MAX_SIZE: u64 = GB as u64;
+
+// buffer size in bytes
+const NBUF: usize = 0;
+
+// log 1 in every N commands
+const SAMPLE: usize = 100;
+
+////////////////////////////////////////////////////////////////////////////////
+// helper functions
+////////////////////////////////////////////////////////////////////////////////
+
+fn file() -> Option<String> {
+    FILE
+}
+
+fn backup() -> Option<String> {
+    BACKUP
+}
+
+fn interval() -> usize {
+    INTERVAL
+}
+
+fn max_size() -> u64 {
+    MAX_SIZE
+}
+
+fn nbuf() -> usize {
+    NBUF
+}
+
+fn sample() -> usize {
+    SAMPLE
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// struct definitions
+////////////////////////////////////////////////////////////////////////////////
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct KlogConfig {
+    #[serde(default = "file")]
+    file: Option<String>,
+    #[serde(default = "backup")]
+    backup: Option<String>,
+    #[serde(default = "interval")]
+    interval: usize,
+    #[serde(default = "nbuf")]
+    nbuf: usize,
+    #[serde(default = "sample")]
+    sample: usize,
+    #[serde(default = "max_size")]
+    max_size: u64,
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// implementation
+////////////////////////////////////////////////////////////////////////////////
+
+impl KlogConfig {
+    pub fn file(&self) -> Option<String> {
+        self.file.clone()
+    }
+
+    pub fn backup(&self) -> Option<String> {
+        match &self.backup {
+            Some(path) => Some(path.clone()),
+            None => self.file.as_ref().map(|path| format!("{}.old", path)),
+        }
+    }
+
+    pub fn interval(&self) -> usize {
+        self.interval
+    }
+
+    pub fn max_size(&self) -> u64 {
+        self.max_size
+    }
+
+    pub fn nbuf(&self) -> usize {
+        self.nbuf
+    }
+
+    pub fn sample(&self) -> usize {
+        self.sample
+    }
+}
+
+// trait implementations
+impl Default for KlogConfig {
+    fn default() -> Self {
+        Self {
+            file: file(),
+            backup: backup(),
+            interval: interval(),
+            max_size: max_size(),
+            nbuf: nbuf(),
+            sample: sample(),
+        }
+    }
+}

src/rust/config/src/lib.rs

@@ -10,6 +10,7 @@ mod array;
 mod buf;
 mod dbuf;
 mod debug;
+mod klog;
 mod pingserver;
 pub mod seg;
 mod segcache;
@@ -19,13 +20,15 @@ mod stats_log;
 mod tcp;
 pub mod time;
 mod tls;
+mod units;
 mod worker;
 
 pub use admin::AdminConfig;
 pub use array::ArrayConfig;
 pub use buf::BufConfig;
 pub use dbuf::DbufConfig;
 pub use debug::DebugConfig;
+pub use klog::KlogConfig;
 pub use pingserver::PingserverConfig;
 pub use seg::SegConfig;
 pub use segcache::SegcacheConfig;

src/rust/config/src/pingserver.rs

@@ -55,6 +55,8 @@ pub struct PingserverConfig {
     #[serde(default)]
     debug: DebugConfig,
     #[serde(default)]
+    klog: KlogConfig,
+    #[serde(default)]
     sockio: SockioConfig,
     #[serde(default)]
     tcp: TcpConfig,
@@ -114,6 +116,10 @@ impl PingserverConfig {
         &self.debug
     }
 
+    pub fn klog(&self) -> &KlogConfig {
+        &self.klog
+    }
+
     pub fn sockio(&self) -> &SockioConfig {
         &self.sockio
     }
@@ -142,6 +148,7 @@ impl Default for PingserverConfig {
 
             buf: Default::default(),
             debug: Default::default(),
+            klog: Default::default(),
             sockio: Default::default(),
             tcp: Default::default(),
             tls: Default::default(),

src/rust/config/src/segcache.rs

@@ -57,6 +57,8 @@ pub struct SegcacheConfig {
     #[serde(default)]
     debug: DebugConfig,
     #[serde(default)]
+    klog: KlogConfig,
+    #[serde(default)]
     sockio: SockioConfig,
     #[serde(default)]
     tcp: TcpConfig,
@@ -120,6 +122,10 @@ impl SegcacheConfig {
         &self.debug
     }
 
+    pub fn klog(&self) -> &KlogConfig {
+        &self.klog
+    }
+
     pub fn sockio(&self) -> &SockioConfig {
         &self.sockio
     }
@@ -153,6 +159,7 @@ impl Default for SegcacheConfig {
 
             buf: Default::default(),
             debug: Default::default(),
+            klog: Default::default(),
             sockio: Default::default(),
             tcp: Default::default(),
             tls: Default::default(),

src/rust/config/src/units.rs

@@ -0,0 +1,4 @@
+pub const B: usize = 1;
+pub const KB: usize = 1024 * B;
+pub const MB: usize = 1024 * KB;
+pub const GB: usize = 1024 * MB;

src/rust/core/server/Cargo.toml

@@ -20,6 +20,7 @@ common = { path = "../../common" }
 config = { path = "../../config" }
 crossbeam-channel = "0.5.0"
 libc = "0.2.83"
+logger = { path = "../../logger" }
 metrics = { path = "../../metrics" }
 mio = { version = "0.7.7", features = ["os-poll", "tcp"] }
 protocol = { path = "../../protocol" }

src/rust/core/server/src/process/builder.rs

@@ -10,6 +10,7 @@ use config::ServerConfig;
 use config::TlsConfig;
 use config::WorkerConfig;
 use entrystore::EntryStore;
+use logger::PelikanLogReceiver;
 use protocol::{Compose, Execute, Parse};
 use queues::QueuePairs;
 
@@ -46,13 +47,14 @@ where
         storage: Storage,
         max_buffer_size: usize,
         parser: Parser,
+        logger: PelikanLogReceiver,
     ) -> Self {
         // initialize admin
         let ssl_context = common::ssl::ssl_context(tls_config).unwrap_or_else(|e| {
             error!("failed to initialize TLS: {}", e);
             std::process::exit(1);
         });
-        let admin = Admin::new(admin_config, ssl_context).unwrap_or_else(|e| {
+        let admin = Admin::new(admin_config, ssl_context, logger).unwrap_or_else(|e| {
             error!("failed to initialize admin: {}", e);
             std::process::exit(1);
         });