inbound: Improve policy metrics

Cargo.lock

@@ -668,7 +668,6 @@ dependencies = [
  "linkerd-duplex",
  "linkerd-errno",
  "linkerd-error",
- "linkerd-error-metrics",
  "linkerd-error-respond",
  "linkerd-exp-backoff",
  "linkerd-http-classify",
@@ -750,6 +749,7 @@ dependencies = [
  "linkerd-tonic-watch",
  "linkerd-tracing",
  "linkerd2-proxy-api",
+ "parking_lot",
  "thiserror",
  "tokio",
  "tokio-test",
@@ -801,6 +801,7 @@ dependencies = [
  "linkerd-identity",
  "linkerd-io",
  "linkerd-tracing",
+ "parking_lot",
  "pin-project",
  "thiserror",
  "tokio",
@@ -910,17 +911,6 @@ dependencies = [
  "futures",
 ]
 
-[[package]]
-name = "linkerd-error-metrics"
-version = "0.1.0"
-dependencies = [
- "futures",
- "linkerd-metrics",
- "parking_lot",
- "pin-project",
- "tower",
-]
-
 [[package]]
 name = "linkerd-error-respond"
 version = "0.1.0"

Cargo.toml

@@ -22,7 +22,6 @@ members = [
     "linkerd/duplex",
     "linkerd/error",
     "linkerd/errno",
-    "linkerd/error-metrics",
     "linkerd/error-respond",
     "linkerd/exp-backoff",
     "linkerd/http-box",

linkerd/app/admin/src/stack.rs

@@ -7,9 +7,10 @@ use linkerd_app_core::{
     serve,
     svc::{self, ExtractParam, InsertParam, Param},
     tls, trace,
-    transport::{self, listen::Bind, ClientAddr, Local, Remote, ServerAddr},
-    Error,
+    transport::{self, listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
+    Error, Result,
 };
+use linkerd_app_inbound as inbound;
 use std::{pin::Pin, time::Duration};
 use thiserror::Error;
 use tokio::sync::mpsc;
@@ -64,7 +65,7 @@ impl Config {
         bind: B,
         identity: Option<LocalCrtKey>,
         report: R,
-        metrics: metrics::Proxy,
+        metrics: inbound::Metrics,
         trace: trace::Handle,
         drain: drain::Watch,
         shutdown: mpsc::UnboundedSender<()>,
@@ -79,11 +80,11 @@ impl Config {
         let (ready, latch) = crate::server::Readiness::new();
         let admin = crate::server::Admin::new(report, ready, shutdown, trace);
         let admin = svc::stack(move |_| admin.clone())
-            .push(metrics.http_endpoint.to_layer::<classify::Response, _, Http>())
+            .push(metrics.proxy.http_endpoint.to_layer::<classify::Response, _, Http>())
+            .push(metrics.http_errors.to_layer())
             .push_on_service(
                 svc::layers()
-                    .push(metrics.http_errors.clone())
-                    .push(errors::layer())
+                    .push(errors::respond::layer())
                     .push(http::BoxResponse::layer()),
             )
             .push(http::NewServeHttp::layer(Default::default(), drain.clone()))
@@ -130,8 +131,10 @@ impl Config {
             )
             .push(svc::BoxNewService::layer())
             .push(detect::NewDetectService::layer(detect::Config::<http::DetectHttp>::from_timeout(DETECT_TIMEOUT)))
-            .push(transport::metrics::NewServer::layer(metrics.transport))
-            .push_map_target(|(tls, addrs): (tls::ConditionalServerTls, B::Addrs)| {
+            .push(transport::metrics::NewServer::layer(metrics.proxy.transport))
+            .push_map_target(move |(tls, addrs): (tls::ConditionalServerTls, B::Addrs)| {
+                // TODO(ver): We should enforce policy here; but we need to permit liveness probes
+                // for destination pods to startup...
                 Tcp {
                     tls,
                     client: addrs.param(),
@@ -161,8 +164,7 @@ impl Param<transport::labels::Key> for Tcp {
             self.tls.clone(),
             self.addr.into(),
             // TODO(ver) enforce policies on the proxy's admin port.
-            Default::default(),
-            Default::default(),
+            metrics::ServerLabel("default:admin".to_string()),
         )
     }
 }
@@ -175,13 +177,28 @@ impl Param<http::Version> for Http {
     }
 }
 
+impl Param<OrigDstAddr> for Http {
+    fn param(&self) -> OrigDstAddr {
+        OrigDstAddr(self.tcp.addr.into())
+    }
+}
+
+impl Param<metrics::ServerLabel> for Http {
+    fn param(&self) -> metrics::ServerLabel {
+        metrics::ServerLabel("default:admin".to_string())
+    }
+}
+
 impl Param<metrics::EndpointLabels> for Http {
     fn param(&self) -> metrics::EndpointLabels {
         metrics::InboundEndpointLabels {
             tls: self.tcp.tls.clone(),
             authority: None,
             target_addr: self.tcp.addr.into(),
-            policy: Default::default(),
+            policy: metrics::AuthzLabels {
+                server: self.param(),
+                authz: "default:all-unauthenticated".to_string(),
+            },
         }
         .into()
     }

linkerd/app/core/Cargo.toml

@@ -28,7 +28,6 @@ linkerd-detect = { path = "../../detect" }
 linkerd-duplex = { path = "../../duplex" }
 linkerd-errno = { path = "../../errno" }
 linkerd-error = { path = "../../error" }
-linkerd-error-metrics = { path = "../../error-metrics" }
 linkerd-error-respond = { path = "../../error-respond" }
 linkerd-exp-backoff = { path = "../../exp-backoff" }
 linkerd-http-classify = { path = "../../http-classify" }

linkerd/app/core/src/control.rs

@@ -79,7 +79,7 @@ impl Config {
 
         svc::stack(ConnectTcp::new(self.connect.keepalive))
             .push(tls::Client::layer(identity))
-            .push_timeout(self.connect.timeout)
+            .push_connect_timeout(self.connect.timeout)
             .push(self::client::layer())
             .push_on_service(svc::MapErrLayer::new(Into::into))
             .into_new_service()

linkerd/app/core/src/errors/mod.rs

@@ -0,0 +1,44 @@
+pub mod respond;
+
+use linkerd_error::Error;
+pub use linkerd_timeout::{FailFastError, ResponseTimeout};
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+#[error("connect timed out after {0:?}")]
+pub(crate) struct ConnectTimeout(pub std::time::Duration);
+
+#[derive(Debug, Error)]
+#[error("{source}")]
+pub struct HttpError {
+    #[source]
+    source: Error,
+    http_status: http::StatusCode,
+    grpc_status: tonic::Code,
+}
+
+impl HttpError {
+    pub fn bad_request(source: impl Into<Error>) -> Self {
+        Self {
+            source: source.into(),
+            http_status: http::StatusCode::BAD_REQUEST,
+            grpc_status: tonic::Code::InvalidArgument,
+        }
+    }
+
+    pub fn forbidden(source: impl Into<Error>) -> Self {
+        Self {
+            source: source.into(),
+            http_status: http::StatusCode::FORBIDDEN,
+            grpc_status: tonic::Code::PermissionDenied,
+        }
+    }
+
+    pub fn loop_detected(source: impl Into<Error>) -> Self {
+        Self {
+            source: source.into(),
+            http_status: http::StatusCode::LOOP_DETECTED,
+            grpc_status: tonic::Code::Aborted,
+        }
+    }
+}