Cargo.lock

@@ -1110,6 +1110,21 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "instrumentation"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "http",
+ "once_cell",
+ "opentelemetry",
+ "opentelemetry-otlp",
+ "opentelemetry-semantic-conventions",
+ "opentelemetry_sdk",
+ "tracing-opentelemetry",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "ipnet"
 version = "2.10.1"
@@ -1512,7 +1527,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "clap",
- "patagia-common",
+ "instrumentation",
  "progenitor",
  "reqwest",
  "schemars",
@@ -1521,20 +1536,6 @@ dependencies = [
  "tracing",
 ]
 
-[[package]]
-name = "patagia-common"
-version = "0.1.0"
-dependencies = [
- "anyhow",
- "once_cell",
- "opentelemetry",
- "opentelemetry-otlp",
- "opentelemetry-semantic-conventions",
- "opentelemetry_sdk",
- "tracing-opentelemetry",
- "tracing-subscriber",
-]
-
 [[package]]
 name = "patagia-controller"
 version = "0.1.0"
@@ -1542,12 +1543,14 @@ dependencies = [
  "anyhow",
  "clap",
  "dropshot",
- "patagia-common",
+ "http",
+ "instrumentation",
  "schemars",
  "serde",
  "slog",
  "slog-async",
  "tokio",
+ "trace-request",
  "tracing",
  "tracing-slog",
 ]
@@ -2738,6 +2741,18 @@ version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"
 
+[[package]]
+name = "trace-request"
+version = "0.1.0"
+dependencies = [
+ "dropshot",
+ "http",
+ "proc-macro2",
+ "quote",
+ "syn",
+ "tracing",
+]
+
 [[package]]
 name = "tracing"
 version = "0.1.41"

Cargo.toml

@@ -2,14 +2,16 @@
 resolver = "2"
 members = [
   "agent",
-  "common",
   "controller",
+  "instrumentation",
+  "trace-request",
   "xtask",
 ]
 default-members = [
   "agent",
-  "common",
   "controller",
+  "instrumentation",
+  "trace-request",
   "xtask",
 ]
 

agent/Cargo.toml

@@ -7,7 +7,7 @@ license = "MPL-2.0"
 [dependencies]
 anyhow.workspace = true
 clap.workspace = true
-patagia-common = { path = "../common" }
+instrumentation = { path = "../instrumentation" }
 progenitor.workspace = true
 reqwest.workspace = true
 schemars.workspace = true

agent/src/main.rs

@@ -3,7 +3,6 @@ use clap::Parser;
 use tokio::time::{sleep, Duration};
 
 mod patagia_api;
-use patagia_common::instrumentation;
 
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]

common/src/lib.rs

@@ -1 +0,0 @@
-pub mod instrumentation;

controller/Cargo.toml

@@ -9,11 +9,16 @@ license = "MPL-2.0"
 anyhow.workspace = true
 clap.workspace = true
 dropshot.workspace = true
-patagia-common = { path = "../common" }
+http.workspace = true
+instrumentation = { path = "../instrumentation" }
 schemars.workspace = true
 serde.workspace = true
 slog-async.workspace = true
 slog.workspace = true
 tokio.workspace = true
+trace-request = { path = "../trace-request" }
 tracing-slog.workspace = true
 tracing.workspace = true
+
+[package.metadata.cargo-machete]
+ignored = ["http"]

controller/src/bin/patagia-controller.rs

@@ -9,7 +9,6 @@ use std::net::SocketAddr;
 use std::str::FromStr;
 use std::sync::Arc;
 
-use patagia_common::instrumentation;
 use patagia_controller::api;
 use patagia_controller::context::ControllerContext;
 

controller/src/version.rs

@@ -1,7 +1,7 @@
 use dropshot::{endpoint, HttpError, HttpResponseOk, RequestContext};
 use schemars::JsonSchema;
 use serde::Serialize;
-use tracing::Instrument;
+use trace_request::trace_request;
 
 use std::sync::Arc;
 
@@ -19,16 +19,7 @@ struct VersionInfo {
     method = GET,
     path = "/version",
 }]
-#[tracing::instrument(
-  skip(rqctx),
-  fields(
-        http.method=rqctx.request.method().as_str(),
-        http.path=rqctx.request.uri().path(),
-        http.remote_ip=rqctx.request.remote_addr().ip().to_string(),
-        request_id = rqctx.request_id,
-    ),
-    err(Debug),
-)]
+#[trace_request]
 pub(crate) async fn version(
     rqctx: RequestContext<Arc<ControllerContext>>,
 ) -> Result<HttpResponseOk<VersionInfo>, HttpError> {
@@ -45,5 +36,6 @@ pub(crate) async fn version(
     }
     .instrument(tracing::info_span!("Let's do the thing...."))
     .await;
+
     Ok(HttpResponseOk(ver))
 }

instrumentation/Cargo.toml

@@ -1,12 +1,12 @@
 [package]
-name = "patagia-common"
-description = "Common control plane modules"
+name = "instrumentation"
 version = "0.1.0"
 edition = "2021"
 license = "MPL-2.0"
 
 [dependencies]
 anyhow.workspace = true
+http.workspace = true
 once_cell.workspace = true
 opentelemetry-otlp.workspace = true
 opentelemetry_sdk.workspace = true

instrumentation/src/lib.rs

@@ -0,0 +1,102 @@
+use anyhow::{anyhow, Result};
+use once_cell::sync::Lazy;
+use opentelemetry::{trace::TracerProvider as _, KeyValue};
+use opentelemetry_sdk::{
+    metrics::{MeterProviderBuilder, PeriodicReader, SdkMeterProvider},
+    runtime,
+    trace::{RandomIdGenerator, Sampler, TracerProvider},
+    Resource,
+};
+use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+
+static RESOURCE: Lazy<Resource> = Lazy::new(|| {
+    Resource::new(vec![
+        KeyValue::new(
+            opentelemetry_semantic_conventions::resource::SERVICE_NAME,
+            env!("CARGO_PKG_NAME"),
+        ),
+        KeyValue::new(
+            opentelemetry_semantic_conventions::resource::SERVICE_VERSION,
+            env!("CARGO_PKG_VERSION"),
+        ),
+    ])
+});
+
+// Construct MeterProvider for MetricsLayer
+fn init_meter_provider() -> Result<SdkMeterProvider> {
+    let exporter = opentelemetry_otlp::MetricExporter::builder()
+        .with_tonic()
+        .with_temporality(opentelemetry_sdk::metrics::Temporality::default())
+        .build()
+        .map_err(|e| anyhow!("Error creating OTLP metric exporter: {:?}", e))?;
+
+    let meter_provider = MeterProviderBuilder::default()
+        .with_resource(RESOURCE.clone())
+        .with_reader(
+            PeriodicReader::builder(exporter, runtime::Tokio)
+                .with_interval(std::time::Duration::from_secs(10))
+                .build(),
+        )
+        .build();
+
+    opentelemetry::global::set_meter_provider(meter_provider.clone());
+
+    Ok(meter_provider)
+}
+
+// Construct TracerProvider for OpenTelemetryLayer
+fn init_tracer_provider() -> Result<TracerProvider> {
+    let exporter = opentelemetry_otlp::SpanExporter::builder()
+        .with_tonic()
+        .build()
+        .map_err(|e| anyhow!("Error creating OTLP span exporter: {:?}", e))?;
+
+    let tracer_provider = opentelemetry_sdk::trace::TracerProvider::builder()
+        .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
+            1.0,
+        ))))
+        .with_resource(RESOURCE.clone())
+        .with_id_generator(RandomIdGenerator::default())
+        .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
+        .build();
+
+    Ok(tracer_provider)
+}
+
+// Initialize tracing-subscriber and return OtelGuard for opentelemetry-related termination processing
+pub fn init_tracing_subscriber() -> Result<OtelGuard> {
+    let meter_provider = init_meter_provider()?;
+    let tracer_provider = init_tracer_provider()?;
+
+    let tracer = tracer_provider.tracer("tracing-otel-subscriber");
+
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::EnvFilter::from_default_env())
+        .with(tracing_subscriber::fmt::layer())
+        .with(MetricsLayer::new(meter_provider.clone()))
+        .with(OpenTelemetryLayer::new(tracer))
+        .init();
+
+    Ok(OtelGuard {
+        meter_provider,
+        tracer_provider,
+    })
+}
+
+pub struct OtelGuard {
+    meter_provider: SdkMeterProvider,
+    tracer_provider: TracerProvider,
+}
+
+impl Drop for OtelGuard {
+    fn drop(&mut self) {
+        if let Err(err) = self.tracer_provider.shutdown() {
+            eprintln!("{err:?}");
+        }
+        if let Err(err) = self.meter_provider.shutdown() {
+            eprintln!("{err:?}");
+        }
+    }
+}

trace-request/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "trace-request"
+version = "0.1.0"
+edition = "2021"
+license = "MPL-2.0"
+
+[lib]
+proc-macro = true
+
+[dependencies]
+proc-macro2 = "1"
+quote = "1"
+syn = { version = "2.0", features = ["full"] }
+
+[dev-dependencies]
+dropshot = { workspace = true }
+http = { workspace = true }
+tracing = { workspace = true }

trace-request/src/lib.rs

@@ -0,0 +1,79 @@
+// Original source: https://github.com/oxidecomputer/rfd-api/blob/main/trace-request/src/lib.rs
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+#![allow(dead_code, unused_imports)]
+
+extern crate proc_macro;
+#[macro_use]
+extern crate quote;
+
+use std::result::Iter;
+
+use proc_macro::TokenStream;
+use proc_macro2::{Delimiter, Group, Span, TokenTree};
+use quote::ToTokens;
+use syn::{
+    bracketed,
+    parse::{Parse, ParseStream, Parser},
+    parse_macro_input,
+    punctuated::Punctuated,
+    Block, DeriveInput, Ident, ItemFn, LitStr, Result, Token, Type,
+};
+
+#[proc_macro_attribute]
+pub fn trace_request(_attr: TokenStream, input: TokenStream) -> TokenStream {
+    let mut input = parse_macro_input!(input as ItemFn);
+    let body_block = input.block;
+    let fn_name = &input.sig.ident;
+
+    let token_stream = quote! {
+        {
+            use tracing::Instrument;
+
+            fn get_status<T>(res: &Result<T, HttpError>) -> http::StatusCode where T: dropshot::HttpCodedResponse {
+                match res {
+                    Ok(_) => T::STATUS_CODE,
+                    Err(err) => err.status_code.as_status(),
+                }
+            }
+
+            let request_id = &rqctx.request_id;
+            let req = &rqctx.request;
+
+            async {
+                let result = async #body_block.await;
+                let status = get_status(&result);
+
+                let span = tracing::Span::current();
+                span.record("http.status", &status.as_str());
+                if let Err(err) = &result {
+                    span.record("error.external", &err.external_message);
+                    span.record("error.internal", &err.internal_message);
+                }
+
+                result
+            }.instrument(
+                tracing::info_span!(
+                    stringify!(handler.#fn_name),
+                    "request.id" = request_id,
+                    "http.method" = req.method().as_str(),
+                    "http.uri" = req.uri().to_string(),
+                    "http.status" = tracing::field::Empty,
+                    "error.external" = tracing::field::Empty,
+                    "error.internal" = tracing::field::Empty
+                )
+                ).await
+        }
+    };
+    let wrapped_body_block: TokenStream = token_stream.into();
+
+    input.block = Box::new(parse_macro_input!(wrapped_body_block as Block));
+
+    quote! {
+        #input
+    }
+    .into()
+}