Export traces via OLTP

This exports traces and logs to Jaeger or OTEL via the opentelemetry OLTP exporter.

Author: Restioson

Cargo.lock

@@ -25,7 +25,7 @@ resolver = "2"
 xtra = { git = "https://github.com/Restioson/xtra", rev = "afff02dd0fc8b92ae264db8a5457773f8af95487" } # Unreleased
 maia = { git = "https://github.com/comit-network/maia", rev = "fc6b78b98407b10b55f8cfd152062ad77f98cd9f" } # Unreleased
 maia-core = { git = "https://github.com/comit-network/maia", tag = "0.1.1", package = "maia-core" } # Pinned to support maia 0.1 and 0.2
-xtra_productivity = { git = "https://github.com/comit-network/xtra-productivity", rev = "bebe45425ae44980186df7b96b41f70cad58a4bb" } # Unreleased
+xtra_productivity = { git = "https://github.com/Restioson/xtra-productivity", rev = "dcf76b6622c89dc6b5630605d8395d8265968406" } # Unreleased
 electrum-client = { git = "https://github.com/comit-network/rust-electrum-client/", branch = "do-not-ignore-empty-lines" }
 
 [profile.dev.package.sqlx-macros]

Cargo.toml

@@ -22,7 +22,7 @@ tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "net"]
 tokio-tasks = { path = "../tokio-tasks", features = ["xtra"] }
 tracing = { version = "0.1" }
 tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "ansi", "env-filter", "local-time", "tracing-log", "json"] }
-xtra = { version = "0.6" }
+xtra = { version = "0.6", features = ["instrumentation"] }
 xtra-bitmex-price-feed = { path = "../xtra-bitmex-price-feed" }
 xtra-libp2p = { path = "../xtra-libp2p" }
-xtra_productivity = { version = "0.1" }
+xtra_productivity = { version = "0.1", features = ["tracing"] }

daemon-tests/Cargo.toml

@@ -55,7 +55,7 @@ xtra-bitmex-price-feed = { path = "../xtra-bitmex-price-feed" }
 xtra-libp2p = { path = "../xtra-libp2p" }
 xtra-libp2p-offer = { path = "../xtra-libp2p-offer" }
 xtra-libp2p-ping = { path = "../xtra-libp2p-ping" }
-xtra_productivity = { version = "0.1.0" }
+xtra_productivity = { version = "0.1.0", features = ["tracing"] }
 xtras = { path = "../xtras" }
 
 [dev-dependencies]

daemon/Cargo.toml

@@ -0,0 +1,7 @@
+# Run jaeger for viewing traces when debugging locally
+docker run \
+  -p 5778:5778 \
+  -p 16686:16686 \
+  -p 13133:13133 \
+  -p 4317:55680 \
+  jaegertracing/opentelemetry-all-in-one:latest

jaeger.sh

@@ -40,5 +40,5 @@ xtra-bitmex-price-feed = { path = "../xtra-bitmex-price-feed" }
 xtra-libp2p = { path = "../xtra-libp2p" }
 xtra-libp2p-offer = { path = "../xtra-libp2p-offer" }
 xtra-libp2p-ping = { path = "../xtra-libp2p-ping" }
-xtra_productivity = { version = "0.1.0" }
+xtra_productivity = { version = "0.1.0", features = ["tracing"] }
 xtras = { path = "../xtras" }

maker/Cargo.toml

@@ -31,7 +31,7 @@ use xtras::supervisor::always_restart;
 async fn main() -> Result<()> {
     let opts = Opts::parse();
 
-    logger::init(opts.log_level, opts.json).context("initialize logger")?;
+    logger::init(opts.log_level, opts.json, "maker").context("initialize logger")?;
     tracing::info!("Running version: {}", daemon::version::version());
     let settlement_interval_hours = SETTLEMENT_INTERVAL.whole_hours();
 

maker/src/main.rs

@@ -13,9 +13,12 @@ atty = "0.2"
 daemon = { path = "../daemon" }
 http-api-problem = { version = "0.53.0", features = ["rocket"] }
 model = { path = "../model" }
+opentelemetry = { version = "0.17.0", features = ["rt-tokio"] }
+opentelemetry-otlp = { version = "0.10.0" }
 rocket = { version = "0.5.0-rc.1", features = ["json"] }
 rocket-basicauth = { path = "../rocket-basicauth" }
 serde = { version = "1", features = ["derive"] }
 time = "0.3.11"
 tracing = { version = "0.1" }
+tracing-opentelemetry = "0.17.3"
 tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt", "ansi", "env-filter", "local-time", "tracing-log", "json"] }

shared-bin/Cargo.toml

@@ -1,15 +1,28 @@
-use anyhow::anyhow;
+use anyhow::Context;
 use anyhow::Result;
+use opentelemetry::sdk::propagation::TraceContextPropagator;
+use opentelemetry::sdk::trace;
+use opentelemetry::sdk::Resource;
+use opentelemetry::KeyValue;
 use time::macros::format_description;
 use tracing_subscriber::fmt::time::UtcTime;
 use tracing_subscriber::EnvFilter;
+use tracing_subscriber::Layer;
+use tracing_subscriber::Registry;
 
 pub use tracing_subscriber::filter::LevelFilter;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
 
 const RUST_LOG_ENV: &str = "RUST_LOG";
 
 #[allow(clippy::print_stdout)] // because the logger is only initialized at the end of this function but we want to print a warning
-pub fn init(level: LevelFilter, json_format: bool) -> Result<()> {
+pub fn init(
+    level: LevelFilter,
+    json_format: bool,
+    instrumentation: bool,
+    service_name: &'static str,
+) -> Result<()> {
     if level == LevelFilter::OFF {
         return Ok(());
     }
@@ -31,23 +44,45 @@ pub fn init(level: LevelFilter, json_format: bool) -> Result<()> {
     };
     let filter = filter.add_directive(format!("{level}").parse()?);
 
-    let builder = tracing_subscriber::fmt()
-        .with_env_filter(filter)
+    let fmt_layer = tracing_subscriber::fmt::layer()
         .with_writer(std::io::stderr)
         .with_ansi(is_terminal);
 
-    let result = if json_format {
-        builder.json().with_timer(UtcTime::rfc_3339()).try_init()
+    let fmt_layer = if json_format {
+        fmt_layer.json().with_timer(UtcTime::rfc_3339()).boxed()
     } else {
-        builder
+        fmt_layer
             .compact()
             .with_timer(UtcTime::new(format_description!(
                 "[year]-[month]-[day] [hour]:[minute]:[second]"
             )))
-            .try_init()
+            .boxed()
     };
 
-    result.map_err(|e| anyhow!("Failed to init logger: {e}"))?;
+    let fmt_layer = fmt_layer.with_filter(filter);
+
+    let telemetry = if instrumentation {
+        opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
+        let cfg = trace::Config::default()
+            .with_resource(Resource::new([KeyValue::new("service.name", service_name)]));
+
+        let tracer = opentelemetry_otlp::new_pipeline()
+            .tracing()
+            .with_trace_config(cfg)
+            .with_exporter(opentelemetry_otlp::new_exporter().tonic())
+            .install_batch(opentelemetry::runtime::Tokio)
+            .context("Failed to initialise OTLP exporter")?;
+
+        Some(tracing_opentelemetry::layer().with_tracer(tracer))
+    } else {
+        None
+    };
+
+    Registry::default()
+        .with(telemetry)
+        .with(fmt_layer)
+        .try_init()
+        .context("Failed to init logger")?;
 
     tracing::info!("Initialized logger");
 

shared-bin/src/logger.rs

@@ -30,7 +30,7 @@ tracing = { version = "0.1" }
 uuid = "0.8"
 webbrowser = "0.7.1"
 x25519-dalek = "1.1"
-xtra = { version = "0.6" }
+xtra = { version = "0.6", features = ["instrumentation"] }
 xtra-bitmex-price-feed = { path = "../xtra-bitmex-price-feed" }
 xtra-libp2p = { path = "../xtra-libp2p" }
 

taker/Cargo.toml

@@ -85,6 +85,10 @@ struct Opts {
     #[clap(short, long)]
     json: bool,
 
+    /// If enabled, traces will be exported to the OTEL collector
+    #[clap(short, long)]
+    instrumentation: bool,
+
     /// Configure the log level, e.g.: one of Error, Warn, Info, Debug, Trace
     #[clap(short, long, default_value = "Debug")]
     log_level: LevelFilter,
@@ -284,7 +288,8 @@ async fn main() -> Result<()> {
     let network = opts.network();
     let (maker_url, maker_id, maker_peer_id) = opts.maker()?;
 
-    logger::init(opts.log_level, opts.json).context("initialize logger")?;
+    logger::init(opts.log_level, opts.json, opts.instrumentation, "taker")
+        .context("initialize logger")?;
     tracing::info!("Running version: {}", daemon::version::version());
     let settlement_interval_hours = SETTLEMENT_INTERVAL.whole_hours();
 

taker/src/main.rs

@@ -36,6 +36,7 @@ use std::borrow::Cow;
 use std::path::PathBuf;
 use tokio::select;
 use tokio::sync::watch;
+use tracing::instrument;
 
 type Taker = TakerActorSystem<
     oracle::Actor,
@@ -54,6 +55,7 @@ pub struct IdentityInfo {
 }
 
 #[rocket::get("/feed")]
+#[instrument(name = "GET /feed", skip(rx))]
 pub async fn feed(
     rx: &State<Feeds>,
     rx_wallet: &State<watch::Receiver<Option<WalletInfo>>>,
@@ -148,6 +150,7 @@ pub struct CfdOrderRequest {
 }
 
 #[rocket::post("/cfd/order", data = "<cfd_order_request>")]
+#[instrument(name = "POST /cfd/order", skip(taker))]
 pub async fn post_order_request(
     cfd_order_request: Json<CfdOrderRequest>,
     taker: &State<Taker>,
@@ -170,6 +173,7 @@ pub async fn post_order_request(
 }
 
 #[rocket::post("/cfd/<id>/<action>")]
+#[instrument(name = "POST /cfd/<id>/<action>", skip(taker))]
 pub async fn post_cfd_action(
     id: Uuid,
     action: String,
@@ -205,6 +209,7 @@ pub async fn post_cfd_action(
 }
 
 #[rocket::get("/alive")]
+#[instrument(name = "GET /alive")]
 pub fn get_health_check() {}
 
 #[derive(Debug, Clone, Copy, Deserialize)]
@@ -233,12 +238,14 @@ pub struct MarginResponse {
 struct Asset;
 
 #[rocket::get("/assets/<file..>")]
+#[instrument(name = "GET /assets/<file>")]
 pub fn dist<'r>(file: PathBuf, _auth: Authenticated) -> impl Responder<'r, 'static> {
     let filename = format!("assets/{}", file.display());
     Asset::get(&filename).into_response(file)
 }
 
 #[rocket::get("/<_paths..>", format = "text/html")]
+#[instrument(name = "GET /<_paths>")]
 pub fn index<'r>(_paths: PathBuf, _auth: Authenticated) -> impl Responder<'r, 'static> {
     let asset = Asset::get("index.html").ok_or(Status::NotFound)?;
     Ok::<(ContentType, Cow<[u8]>), Status>((ContentType::HTML, asset.data))
@@ -253,6 +260,7 @@ pub struct WithdrawRequest {
 }
 
 #[rocket::post("/withdraw", data = "<withdraw_request>")]
+#[instrument(name = "POST /withdraw", skip(taker))]
 pub async fn post_withdraw_request(
     withdraw_request: Json<WithdrawRequest>,
     taker: &State<Taker>,
@@ -279,6 +287,7 @@ pub async fn post_withdraw_request(
 }
 
 #[rocket::get("/metrics")]
+#[instrument(name = "GET /metrics")]
 pub async fn get_metrics<'r>(_auth: Authenticated) -> Result<String, HttpApiProblem> {
     let metrics = prometheus::TextEncoder::new()
         .encode_to_string(&prometheus::gather())
@@ -292,6 +301,7 @@ pub async fn get_metrics<'r>(_auth: Authenticated) -> Result<String, HttpApiProb
 }
 
 #[rocket::put("/sync")]
+#[instrument(name = "PUT /sync", skip(taker))]
 pub async fn put_sync_wallet(
     taker: &State<Taker>,
     _auth: Authenticated,
@@ -311,6 +321,7 @@ pub struct HealthCheck {
 }
 
 #[rocket::get("/version")]
+#[instrument(name = "GET /version")]
 pub async fn get_version() -> Json<HealthCheck> {
     Json(HealthCheck {
         daemon_version: daemon::version::version().to_string(),

taker/src/routes.rs

@@ -19,7 +19,7 @@ tokio = "1"
 tokio-tasks = { path = "../tokio-tasks" }
 tracing = "0.1"
 xtra = "0.6"
-xtra_productivity = { version = "0.1.0" }
+xtra_productivity = { version = "0.1.0", features = ["tracing"] }
 
 [dev-dependencies]
 rust_decimal_macros = "1"

xtra-bitmex-price-feed/Cargo.toml

@@ -9,6 +9,8 @@ use async_trait::async_trait;
 use libp2p_core::Multiaddr;
 use libp2p_core::PeerId;
 use std::time::Duration;
+use tracing::instrument;
+use tracing::Instrument;
 use xtra::Address;
 use xtra_productivity::xtra_productivity;
 use xtras::SendAsyncSafe;
@@ -21,6 +23,7 @@ pub const CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
 /// Polls Endpoint at startup to check whether connection got established correctly, and
 /// then listens for ConnectionDropped message to stop itself.
 /// Should be used in conjunction with supervisor maintaining resilient connection.
+#[derive(Debug)]
 pub struct Actor {
     endpoint: Address<Endpoint>,
     connect_address: Multiaddr,
@@ -38,6 +41,7 @@ impl Actor {
         }
     }
 
+    #[instrument(err)]
     async fn connect(&self) -> Result<(), Error> {
         self.endpoint
             .send(Connect(self.connect_address.clone()))
@@ -86,6 +90,7 @@ impl Actor {
             .expect("to always have peer id if successfully started")
     }
 
+    #[instrument(ret, err)]
     async fn is_connection_established(&self) -> Result<bool> {
         Ok(self
             .endpoint
@@ -95,6 +100,7 @@ impl Actor {
             .contains(&self.peer_id()))
     }
 
+    #[instrument(err)]
     async fn dial(&self) -> Result<()> {
         if self.is_connection_established().await? {
             tracing::info!("Connection is already established, no need to connect");
@@ -106,7 +112,12 @@ impl Actor {
         }
 
         // Only check the connection again after it had enough time to be established
-        tokio::time::sleep(CONNECTION_TIMEOUT).await;
+        tokio::time::sleep(CONNECTION_TIMEOUT)
+            .instrument(tracing::debug_span!(
+                "Wait connection timeout",
+                timeout_secs = CONNECTION_TIMEOUT.as_secs()
+            ))
+            .await;
 
         anyhow::ensure!(
             self.is_connection_established().await?,

xtra-libp2p/src/dialer.rs

@@ -14,7 +14,7 @@ tokio-tasks = { path = "../tokio-tasks" }
 tracing = { version = "0.1" }
 uuid = { version = "1.1", features = ["v4"] }
 xtra = { version = "0.6", features = ["instrumentation"] }
-xtra_productivity = { version = "0.1.0" }
+xtra_productivity = { version = "0.1.0", features = ["tracing"] }
 
 [dev-dependencies]
 tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] }