feat(config): Global Default Log Schemas

Cargo.toml

@@ -142,6 +142,8 @@ warp = "0.1.20"
 evmap = { version = "7", features = ["bytes"] }
 logfmt = "0.0.2"
 notify = "4.0.14"
+once_cell = "1.3"
+getset = "0.1.0"
 
 [target.'cfg(unix)'.dependencies]
 atty = "0.2"

benches/bench.rs

@@ -623,7 +623,7 @@ fn bench_elasticsearch_index(c: &mut Criterion) {
                     let mut event = Event::from("hello world");
                     event
                         .as_mut_log()
-                        .insert(event::TIMESTAMP.clone(), Utc::now());
+                        .insert(event::log_schema().timestamp_key().clone(), Utc::now());
 
                     (Template::from("index-%Y.%m.%d"), event)
                 },
@@ -640,7 +640,7 @@ fn bench_elasticsearch_index(c: &mut Criterion) {
                     let mut event = Event::from("hello world");
                     event
                         .as_mut_log()
-                        .insert(event::TIMESTAMP.clone(), Utc::now());
+                        .insert(event::log_schema().timestamp_key().clone(), Utc::now());
 
                     (Template::from("index"), event)
                 },

benches/event.rs

@@ -73,7 +73,9 @@ fn benchmark_event(c: &mut Criterion) {
 fn create_event(json: Value) -> LogEvent {
     let s = serde_json::to_string(&json).unwrap();
     let mut event = Event::new_empty_log();
-    event.as_mut_log().insert(event::MESSAGE.clone(), s);
+    event
+        .as_mut_log()
+        .insert(event::log_schema().message_key().clone(), s);
 
     let mut parser = JsonParser::from(JsonParserConfig::default());
     parser.transform(event).unwrap().into_log()

src/event/mod.rs

@@ -1,9 +1,11 @@
 use self::proto::{event_wrapper::Event as EventProto, metric::Value as MetricProto, Log};
 use bytes::Bytes;
 use chrono::{DateTime, SecondsFormat, TimeZone, Utc};
+use getset::{Getters, Setters};
 use lazy_static::lazy_static;
 use metric::{MetricKind, MetricValue};
-use serde::{Serialize, Serializer};
+use once_cell::sync::OnceCell;
+use serde::{Deserialize, Serialize, Serializer};
 use std::collections::{hash_map::Drain, HashMap};
 use std::iter::FromIterator;
 use string_cache::DefaultAtom as Atom;
@@ -22,10 +24,9 @@ pub mod proto {
     include!(concat!(env!("OUT_DIR"), "/event.proto.rs"));
 }
 
+pub static LOG_SCHEMA: OnceCell<LogSchema> = OnceCell::new();
+
 lazy_static! {
-    pub static ref MESSAGE: Atom = Atom::from("message");
-    pub static ref HOST: Atom = Atom::from("host");
-    pub static ref TIMESTAMP: Atom = Atom::from("timestamp");
     pub static ref PARTIAL: Atom = Atom::from("_partial");
 }
 
@@ -159,6 +160,40 @@ impl<K: Into<Atom>, V: Into<Value>> FromIterator<(K, V)> for LogEvent {
     }
 }
 
+pub fn log_schema() -> &'static LogSchema {
+    // TODO: Help Rust project support before_each
+    // Support uninitialized schemas in tests to help our contributors.
+    // Don't do it in release because that is scary.
+    #[cfg(debug_assertions)]
+    {
+        if LOG_SCHEMA.get().is_none() {
+            error!("You are not initializing a schema in this test -- This could fail in release");
+            LOG_SCHEMA.set(LogSchema::default()).ok(); // If this fails it means some other test set it while we were trying to.
+        }
+    }
+    LOG_SCHEMA.get().expect("Schema was not initialized")
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Getters, Setters)]
+pub struct LogSchema {
+    #[getset(get = "pub", set = "pub")]
+    message_key: Atom,
+    #[getset(get = "pub", set = "pub")]
+    timestamp_key: Atom,
+    #[getset(get = "pub", set = "pub")]
+    host_key: Atom,
+}
+
+impl Default for LogSchema {
+    fn default() -> Self {
+        LogSchema {
+            message_key: Atom::from("message"),
+            timestamp_key: Atom::from("timestamp"),
+            host_key: Atom::from("host"),
+        }
+    }
+}
+
 #[derive(PartialEq, Debug, Clone)]
 pub enum Value {
     Bytes(Bytes),
@@ -483,7 +518,7 @@ impl From<Event> for Vec<u8> {
     fn from(event: Event) -> Vec<u8> {
         event
             .into_log()
-            .remove(&MESSAGE)
+            .remove(&log_schema().message_key())
             .unwrap()
             .as_bytes()
             .to_vec()
@@ -496,8 +531,12 @@ impl From<Bytes> for Event {
             fields: HashMap::new(),
         });
 
-        event.as_mut_log().insert(MESSAGE.clone(), message);
-        event.as_mut_log().insert(TIMESTAMP.clone(), Utc::now());
+        event
+            .as_mut_log()
+            .insert(log_schema().message_key().clone(), message);
+        event
+            .as_mut_log()
+            .insert(log_schema().timestamp_key().clone(), Utc::now());
 
         event
     }
@@ -565,7 +604,7 @@ mod test {
             "message": "raw log line",
             "foo": "bar",
             "bar": "baz",
-            "timestamp": event.as_log().get(&super::TIMESTAMP),
+            "timestamp": event.as_log().get(&super::log_schema().timestamp_key()),
         });
 
         let actual_all = serde_json::to_value(event.as_log().all_fields()).unwrap();

src/event/unflatten.rs

@@ -398,7 +398,7 @@ mod tests {
         fn unflatten_abirtrary(json in prop::json()) {
             let s = serde_json::to_string(&json).unwrap();
             let mut event = Event::new_empty_log();
-            event.as_mut_log().insert(event::MESSAGE.clone(), s);
+            event.as_mut_log().insert(event::log_schema().message_key().clone(), s);
 
             let mut parser = JsonParser::from(JsonParserConfig::default());
             let event = parser.transform(event).unwrap().into_log();

src/generate.rs

@@ -279,6 +279,11 @@ mod tests {
             Ok(r#"data_dir = "/var/lib/vector/"
 dns_servers = []
 
+[log_schema]
+message_key = "message"
+timestamp_key = "timestamp"
+host_key = "host"
+
 [sources.source0]
 max_length = 102400
 type = "stdin"
@@ -307,6 +312,11 @@ when_full = "block"
             Ok(r#"data_dir = "/var/lib/vector/"
 dns_servers = []
 
+[log_schema]
+message_key = "message"
+timestamp_key = "timestamp"
+host_key = "host"
+
 [sources.source0]
 max_length = 102400
 type = "stdin"
@@ -335,6 +345,11 @@ when_full = "block"
             Ok(r#"data_dir = "/var/lib/vector/"
 dns_servers = []
 
+[log_schema]
+message_key = "message"
+timestamp_key = "timestamp"
+host_key = "host"
+
 [sources.source0]
 max_length = 102400
 type = "stdin"
@@ -357,6 +372,11 @@ when_full = "block"
             Ok(r#"data_dir = "/var/lib/vector/"
 dns_servers = []
 
+[log_schema]
+message_key = "message"
+timestamp_key = "timestamp"
+host_key = "host"
+
 [sinks.sink0]
 healthcheck = true
 inputs = ["TODO"]
@@ -375,6 +395,11 @@ when_full = "block"
             Ok(r#"data_dir = "/var/lib/vector/"
 dns_servers = []
 
+[log_schema]
+message_key = "message"
+timestamp_key = "timestamp"
+host_key = "host"
+
 [transforms.transform0]
 inputs = []
 type = "add_fields"

src/main.rs

@@ -13,7 +13,7 @@ use structopt::{clap::AppSettings, StructOpt};
 use tokio_signal::unix::{Signal, SIGHUP, SIGINT, SIGQUIT, SIGTERM};
 use topology::Config;
 use tracing_futures::Instrument;
-use vector::{config_paths, generate, list, metrics, runtime, topology, trace, unit_test};
+use vector::{config_paths, event, generate, list, metrics, runtime, topology, trace, unit_test};
 
 #[derive(StructOpt, Debug)]
 #[structopt(rename_all = "kebab-case")]
@@ -247,6 +247,9 @@ fn main() {
     let config = config.unwrap_or_else(|| {
         std::process::exit(exitcode::CONFIG);
     });
+    event::LOG_SCHEMA
+        .set(config.global.log_schema.clone())
+        .expect("Couldn't set schema");
 
     let mut rt = {
         let threads = opts.threads.unwrap_or(max(1, num_cpus::get()));

src/sinks/aws_cloudwatch_logs/mod.rs

@@ -247,11 +247,12 @@ impl CloudwatchLogsSvc {
     }
 
     pub fn encode_log(&self, mut log: LogEvent) -> InputLogEvent {
-        let timestamp = if let Some(Value::Timestamp(ts)) = log.remove(&event::TIMESTAMP) {
-            ts.timestamp_millis()
-        } else {
-            chrono::Utc::now().timestamp_millis()
-        };
+        let timestamp =
+            if let Some(Value::Timestamp(ts)) = log.remove(&event::log_schema().timestamp_key()) {
+                ts.timestamp_millis()
+            } else {
+                chrono::Utc::now().timestamp_millis()
+            };
 
         match self.encoding {
             Encoding::Json => {
@@ -260,7 +261,7 @@ impl CloudwatchLogsSvc {
             }
             Encoding::Text => {
                 let message = log
-                    .get(&event::MESSAGE)
+                    .get(&event::log_schema().message_key())
                     .map(|v| v.to_string_lossy())
                     .unwrap_or_else(|| "".into());
                 InputLogEvent { message, timestamp }
@@ -688,7 +689,7 @@ mod tests {
         event.insert("key", "value");
         let encoded = svc(Default::default()).encode_log(event.clone());
 
-        let ts = if let Value::Timestamp(ts) = event[&event::TIMESTAMP] {
+        let ts = if let Value::Timestamp(ts) = event[&event::log_schema().timestamp_key()] {
             ts.timestamp_millis()
         } else {
             panic!()
@@ -707,7 +708,7 @@ mod tests {
         event.insert("key", "value");
         let encoded = svc(config).encode_log(event.clone());
         let map: HashMap<Atom, String> = serde_json::from_str(&encoded.message[..]).unwrap();
-        assert!(map.get(&event::TIMESTAMP).is_none());
+        assert!(map.get(&event::log_schema().timestamp_key()).is_none());
     }
 
     #[test]