WIP

Cargo.lock

@@ -716,6 +716,22 @@ version = "0.5.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8dd5a6d5999d9907cda8ed67bbd137d3af8085216c2ac62de5be860bd41f304a"
 
+[[package]]
+name = "linkerd2-access-log"
+version = "0.1.0"
+dependencies = [
+ "base64 0.10.1",
+ "bytes 0.5.4",
+ "futures 0.3.5",
+ "hex",
+ "http 0.2.1",
+ "linkerd2-error",
+ "pin-project",
+ "tokio",
+ "tower",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd2-addr"
 version = "0.1.0"
@@ -739,6 +755,7 @@ name = "linkerd2-app"
 version = "0.1.0"
 dependencies = [
  "bytes 0.5.4",
+ "chrono",
  "futures 0.3.5",
  "h2 0.2.5",
  "http 0.2.1",
@@ -755,6 +772,7 @@ dependencies = [
  "linkerd2-opencensus",
  "linkerd2-proxy-api",
  "net2",
+ "pin-project",
  "quickcheck",
  "regex 1.0.0",
  "ring",
@@ -783,6 +801,7 @@ dependencies = [
  "hyper",
  "indexmap",
  "libc",
+ "linkerd2-access-log",
  "linkerd2-addr",
  "linkerd2-admit",
  "linkerd2-buffer",

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "hyper-balance",
+    "linkerd/access-log",
     "linkerd/addr",
     "linkerd/admit",
     "linkerd/app/core",

linkerd/access-log/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "linkerd2-access-log"
+version = "0.1.0"
+authors = ["Linkerd Developers <[email protected]>"]
+edition = "2018"
+publish = false
+
+[dependencies]
+base64 = "0.10.1"
+bytes = "0.5"
+futures = "0.3"
+hex = "0.3.2"
+http = "0.2"
+linkerd2-error = { path = "../error" }
+tower = { version = "0.3", default-features = false }
+tracing = "0.1.2"
+tokio = {version = "0.2", features = ["sync"]}
+pin-project = "0.4"

linkerd/access-log/src/layer.rs

@@ -0,0 +1,99 @@
+use crate::tracker::{ResponseTracker, TrackerState};
+use crate::AccessLog;
+use futures::{ready, TryFuture};
+use pin_project::pin_project;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use tokio::sync::mpsc;
+
+/// A layer that adds access logging
+#[derive(Clone)]
+pub struct AccessLogLayer {
+    shared: Option<Arc<TrackerState>>,
+}
+
+#[derive(Clone)]
+pub struct AccessLogContext<Svc> {
+    inner: Svc,
+    shared: Option<Arc<TrackerState>>,
+}
+
+#[pin_project]
+pub struct ResponseFuture<F> {
+    tracker: Option<ResponseTracker>,
+
+    #[pin]
+    inner: F,
+}
+
+impl<Svc> tower::layer::Layer<Svc> for AccessLogLayer {
+    type Service = AccessLogContext<Svc>;
+
+    fn layer(&self, inner: Svc) -> Self::Service {
+        Self::Service {
+            inner,
+            shared: self.shared.clone(),
+        }
+    }
+}
+
+impl<Svc, B1, B2> tower::Service<http::Request<B1>> for AccessLogContext<Svc>
+where
+    Svc: tower::Service<http::Request<B1>, Response = http::Response<B2>>,
+{
+    type Response = Svc::Response;
+    type Error = Svc::Error;
+    type Future = ResponseFuture<Svc::Future>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Svc::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut request: http::Request<B1>) -> Self::Future {
+        let shared = match &self.shared {
+            Some(shared) => shared,
+            None => {
+                return ResponseFuture {
+                    tracker: None,
+                    inner: self.inner.call(request),
+                }
+            }
+        };
+
+        let (request_tracker, response_tracker) = shared.tracker();
+        request_tracker.register(&mut request);
+
+        ResponseFuture {
+            inner: self.inner.call(request),
+            tracker: Some(response_tracker),
+        }
+    }
+}
+
+impl AccessLogLayer {
+    pub fn new(sink: Option<mpsc::Sender<AccessLog>>) -> Self {
+        Self {
+            shared: sink.map(|s| Arc::new(TrackerState::new(s))),
+        }
+    }
+}
+
+impl<F, B2> Future for ResponseFuture<F>
+where
+    F: TryFuture<Ok = http::Response<B2>>,
+{
+    type Output = Result<F::Ok, F::Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        let mut inner: http::Response<B2> = ready!(this.inner.try_poll(cx))?;
+
+        if let Some(tracker) = this.tracker.take() {
+            tracker.register(&mut inner);
+        }
+
+        Poll::Ready(Ok(inner))
+    }
+}

linkerd/access-log/src/lib.rs

@@ -0,0 +1,49 @@
+#![deny(warnings, rust_2018_idioms)]
+
+use linkerd2_error::Error;
+use tokio::sync::mpsc;
+
+pub mod layer;
+mod tracker;
+
+use http::{HeaderValue, Method, StatusCode, Uri};
+pub use layer::{AccessLogContext, AccessLogLayer};
+
+#[derive(Debug)]
+pub struct AccessLog {
+    pub uri: Uri,
+    pub method: Method,
+    pub status: StatusCode,
+    pub host: Option<HeaderValue>,
+    pub user_agent: Option<HeaderValue>,
+    pub trace_id: Option<HeaderValue>,
+    pub request_processing_time_us: u64,
+    pub upstream_processing_time_us: u64,
+    pub response_processing_time_us: u64,
+}
+
+impl Default for AccessLog {
+    fn default() -> Self {
+        AccessLog {
+            uri: Default::default(),
+            method: Default::default(),
+            status: Default::default(),
+            host: None,
+            user_agent: None,
+            trace_id: None,
+            request_processing_time_us: 0,
+            upstream_processing_time_us: 0,
+            response_processing_time_us: 0,
+        }
+    }
+}
+
+pub trait AccessLogSink {
+    fn try_send(&mut self, log: AccessLog) -> Result<(), Error>;
+}
+
+impl AccessLogSink for mpsc::Sender<AccessLog> {
+    fn try_send(&mut self, span: AccessLog) -> Result<(), Error> {
+        self.try_send(span).map_err(Into::into)
+    }
+}