cmd/logcatcher: match mincatcher output

Signed-off-by: Christine Dodrill <me@christine.website>

Cargo.lock

@@ -712,12 +712,15 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "axum",
+ "chrono",
  "futures",
  "headers",
  "hyper",
  "lazy_static",
  "logtail",
  "prometheus",
+ "serde",
+ "serde_json",
  "structopt",
  "tokio",
  "tower",

cmd/logcatcher/Cargo.toml

@@ -10,6 +10,7 @@ anyhow = "1"
 futures = "0.3"
 headers = "0.3"
 lazy_static = "1"
+serde_json = "1"
 structopt = "0.3"
 tower = "0.4"
 tracing = "0.1"
@@ -23,6 +24,10 @@ logtail = { path = "../../crates/logtail" }
 version = "0.2"
 features = [ "headers" ]
 
+[dependencies.chrono]
+version = "0.4"
+features = [ "serde" ]
+
 [dependencies.hyper]
 version = "0.14"
 features = [ "full" ]
@@ -31,6 +36,10 @@ features = [ "full" ]
 version = "0.12"
 features = [ "process" ]
 
+[dependencies.serde]
+version = "1"
+features = [ "derive" ]
+
 [dependencies.tokio]
 version = "1"
 features = [ "full" ]

cmd/logcatcher/src/server/mod.rs

@@ -4,10 +4,12 @@ use axum::{
     handler::{get, post},
     Router,
 };
+use chrono::prelude::*;
 use headers::{Header, HeaderName, HeaderValue};
 use hyper::http::StatusCode;
+use logtail::PublicID;
 use prometheus::{Encoder, IntGauge};
-use std::{convert::TryInto, io::Write, net::SocketAddr, path::PathBuf};
+use std::{convert::TryInto, net::SocketAddr, path::PathBuf};
 use tokio::sync::Semaphore;
 
 lazy_static! {
@@ -108,7 +110,28 @@ async fn put_logs(
     let compressed_body = hyper::body::to_bytes(body).await.unwrap();
     zstd::block::decompress_to_buffer(&compressed_body, &mut decompressed_body).unwrap();
 
-    std::io::stdout().lock().write(&decompressed_body).unwrap();
+    let vals: Vec<serde_json::Value> = serde_json::from_slice(&decompressed_body).unwrap();
+
+    #[derive(serde::Serialize)]
+    pub struct Envelope {
+        log_id: PublicID,
+        collection: String,
+        data: serde_json::Value,
+        server_time: DateTime<Utc>,
+    }
+
+    for data in vals {
+        serde_json::to_writer(
+            std::io::stdout().lock(),
+            &Envelope {
+                log_id: private_id.as_public(),
+                collection: collection.clone(),
+                data,
+                server_time: Utc::now(),
+            },
+        )
+        .unwrap();
+    }
 
     StatusCode::NO_CONTENT
 }