Access Logging

Cargo.lock

@@ -125,6 +125,19 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
+[[package]]
+name = "chrono"
+version = "0.4.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "670ad68c9088c2a963aaa298cb369688cf3f9465ce5e2d4ca10e6e0098a1ce73"
+dependencies = [
+ "libc",
+ "num-integer",
+ "num-traits",
+ "time",
+ "winapi",
+]
+
 [[package]]
 name = "cmake"
 version = "0.1.45"
@@ -143,6 +156,26 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "crossbeam-channel"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "06ed27e177f16d65f0f0c22a213e17c696ace5dd64b14258b52f9417ccb52db4"
+dependencies = [
+ "cfg-if",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d82cfc11ce7f2c3faef78d8a684447b40d503d9681acebed6cb728d45940c4db"
+dependencies = [
+ "cfg-if",
+ "lazy_static",
+]
+
 [[package]]
 name = "data-encoding"
 version = "2.3.2"
@@ -416,9 +449,9 @@ dependencies = [
 
 [[package]]
 name = "html-escape"
-version = "0.2.9"
+version = "0.2.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "816ea801a95538fc5f53c836697b3f8b64a9d664c4f0b91efe1fe7c92e4dbcb7"
+checksum = "d348900ce941b7474395ba922ed3735a517df4546a2939ddb416ce85eeaa988e"
 dependencies = [
  "utf8-width",
 ]
@@ -602,6 +635,21 @@ version = "0.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7fb9b38af92608140b86b693604b9ffcc5824240a484d1ecd4795bacb2fe88f3"
 
+[[package]]
+name = "linkerd-access-log"
+version = "0.1.0"
+dependencies = [
+ "chrono",
+ "futures",
+ "http",
+ "linkerd-identity",
+ "linkerd-proxy-transport",
+ "linkerd-stack",
+ "linkerd-tracing",
+ "pin-project",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-addr"
 version = "0.1.0"
@@ -645,6 +693,7 @@ dependencies = [
  "hyper",
  "indexmap",
  "ipnet",
+ "linkerd-access-log",
  "linkerd-addr",
  "linkerd-cache",
  "linkerd-concurrency-limit",
@@ -1428,6 +1477,7 @@ dependencies = [
  "tokio",
  "tokio-trace",
  "tracing",
+ "tracing-appender",
  "tracing-log",
  "tracing-subscriber",
 ]
@@ -1597,6 +1647,16 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "num-integer"
+version = "0.1.44"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2cc698a63b549a70bc047073d2949cce27cd1c7b0a4a862d08a8031bc2801db"
+dependencies = [
+ "autocfg",
+ "num-traits",
+]
+
 [[package]]
 name = "num-traits"
 version = "0.2.14"
@@ -2108,6 +2168,16 @@ dependencies = [
  "once_cell",
 ]
 
+[[package]]
+name = "time"
+version = "0.1.43"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca8a50ef2360fbd1eeb0ecd46795a87a19024eb4b53c5dc916ca1fd95fe62438"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.1.1"
@@ -2312,6 +2382,17 @@ dependencies = [
  "tracing-core",
 ]
 
+[[package]]
+name = "tracing-appender"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9965507e507f12c8901432a33e31131222abac31edd90cabbcf85cf544b7127a"
+dependencies = [
+ "chrono",
+ "crossbeam-channel",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "tracing-attributes"
 version = "0.1.13"
@@ -2484,9 +2565,9 @@ dependencies = [
 
 [[package]]
 name = "utf8-width"
-version = "0.1.5"
+version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7cf7d77f457ef8dfa11e4cd5933c5ddb5dc52a94664071951219a97710f0a32b"
+checksum = "9071ac216321a4470a69fb2b28cfc68dcd1a39acd877c8be8e014df6772d8efa"
 
 [[package]]
 name = "want"

Cargo.toml

@@ -5,6 +5,7 @@ resolver = "2"
 
 members = [
     "hyper-balance",
+    "linkerd/access-log",
     "linkerd/addr",
     "linkerd/app/core",
     "linkerd/app/gateway",

linkerd/access-log/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "linkerd-access-log"
+version = "0.1.0"
+authors = ["Linkerd Developers <cncf-linkerd-dev@lists.cncf.io>"]
+license = "Apache-2.0"
+edition = "2018"
+publish = false
+
+[dependencies]
+chrono = "0.4.15"
+futures = "0.3"
+http = "0.2"
+pin-project = "1"
+linkerd-stack = { path = "../stack" }
+linkerd-identity = { path = "../identity" }
+linkerd-proxy-transport = { path = "../proxy/transport" }
+linkerd-tracing = { path = "../tracing" }
+tracing = "0.1.19"

linkerd/access-log/src/lib.rs

@@ -0,0 +1,188 @@
+#![deny(warnings, rust_2018_idioms)]
+
+use futures::TryFuture;
+use linkerd_identity as identity;
+use linkerd_proxy_transport::{ClientAddr, Remote};
+use linkerd_stack as svc;
+use linkerd_tracing::access_log::TRACE_TARGET;
+use pin_project::pin_project;
+use std::{
+    future::Future,
+    net::SocketAddr,
+    pin::Pin,
+    task::{Context, Poll},
+    time::{Duration, Instant},
+};
+use svc::{NewService, Param};
+use tracing::{field, span, Level, Span};
+
+#[derive(Clone, Debug)]
+pub struct NewAccessLog<N> {
+    inner: N,
+}
+
+#[derive(Clone, Debug)]
+pub struct AccessLogContext<S> {
+    inner: S,
+    client_addr: SocketAddr,
+    client_id: Option<identity::Name>,
+}
+
+struct ResponseFutureInner {
+    span: Span,
+    start: Instant,
+    processing: Duration,
+}
+
+#[pin_project]
+pub struct AccessLogFuture<F> {
+    data: Option<ResponseFutureInner>,
+
+    #[pin]
+    inner: F,
+}
+
+impl<N> NewAccessLog<N> {
+    #[inline]
+    pub fn layer() -> impl svc::layer::Layer<N, Service = Self> {
+        svc::layer::mk(|inner| NewAccessLog { inner })
+    }
+}
+
+impl<N, T> NewService<T> for NewAccessLog<N>
+where
+    T: Param<Option<identity::Name>> + Param<Remote<ClientAddr>>,
+    N: NewService<T>,
+{
+    type Service = AccessLogContext<N::Service>;
+
+    fn new_service(&mut self, target: T) -> Self::Service {
+        let Remote(ClientAddr(client_addr)) = target.param();
+        let client_id = target.param();
+        let inner = self.inner.new_service(target);
+        AccessLogContext {
+            inner,
+            client_addr,
+            client_id,
+        }
+    }
+}
+
+impl<S, B1, B2> svc::Service<http::Request<B1>> for AccessLogContext<S>
+where
+    S: svc::Service<http::Request<B1>, Response = http::Response<B2>>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = AccessLogFuture<S::Future>;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, request: http::Request<B1>) -> Self::Future {
+        let get_header = |name: http::header::HeaderName| {
+            request
+                .headers()
+                .get(name)
+                .and_then(|x| x.to_str().ok())
+                .unwrap_or_default()
+        };
+
+        let trace_id = || {
+            let headers = request.headers();
+            headers
+                .get("x-b3-traceid")
+                .or_else(|| headers.get("x-request-id"))
+                .or_else(|| headers.get("x-amzn-trace-id"))
+                .and_then(|x| x.to_str().ok())
+                .unwrap_or_default()
+        };
+
+        let now = || {
+            chrono::Utc::now().format_with_items(
+                [chrono::format::Item::Fixed(chrono::format::Fixed::RFC3339)].iter(),
+            )
+        };
+
+        let span = span!(target: TRACE_TARGET, Level::INFO, "http",
+            timestamp = %now(),
+            client.addr = %self.client_addr,
+            client.id = self.client_id.as_ref().map(identity::Name::as_ref).unwrap_or_default(),
+            processing_ns = field::Empty,
+            total_ns = field::Empty,
+            method = request.method().as_str(),
+            uri =  %request.uri(),
+            version = ?request.version(),
+            user_agent = get_header(http::header::USER_AGENT),
+            host = get_header(http::header::HOST),
+            trace_id = trace_id(),
+            status = field::Empty,
+            request_bytes = get_header(http::header::CONTENT_LENGTH),
+            response_bytes = field::Empty
+        );
+
+        if span.is_disabled() {
+            return AccessLogFuture {
+                data: None,
+                inner: self.inner.call(request),
+            };
+        }
+
+        AccessLogFuture {
+            data: Some(ResponseFutureInner {
+                span,
+                start: Instant::now(),
+                processing: Duration::from_secs(0),
+            }),
+            inner: self.inner.call(request),
+        }
+    }
+}
+
+impl<F, B2> Future for AccessLogFuture<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 mut this = self.project();
+
+        let data: &mut ResponseFutureInner = match &mut this.data {
+            Some(data) => data,
+            None => return this.inner.try_poll(cx),
+        };
+
+        let _enter = data.span.enter();
+        let poll_start = Instant::now();
+
+        let response: http::Response<B2> = match this.inner.try_poll(cx) {
+            Poll::Pending => {
+                data.processing += Instant::now().duration_since(poll_start);
+                return Poll::Pending;
+            }
+            Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
+            Poll::Ready(Ok(response)) => response,
+        };
+
+        let now = Instant::now();
+        let total_ns = now.duration_since(data.start).as_nanos();
+        let processing_ns = (now.duration_since(poll_start) + data.processing).as_nanos();
+
+        let span = &data.span;
+
+        response
+            .headers()
+            .get(http::header::CONTENT_LENGTH)
+            .and_then(|x| x.to_str().ok())
+            .map(|x| span.record("response_bytes", &x));
+
+        span.record("status", &response.status().as_u16());
+        span.record("total_ns", &field::display(total_ns));
+        span.record("processing_ns", &field::display(processing_ns));
+
+        Poll::Ready(Ok(response))
+    }
+}