9.1.1 (#278)

* fix: fix tracing span names and missing fields

CHANGELOG.md

@@ -1,17 +1,21 @@
+## 9.1.1
+
+* Fix tracing span names and missing fields
+
 ## 9.1.0
 
 * Add [RediSearch](https://github.com/RediSearch/RediSearch) interface.
 * Adapt testing and CI processes to test Redis and Valkey
 * Add `FromIterator` impl to `RedisMap`
 * Add `ExclusivePool` client
-* Support `redis+unix` config URLs for Unix domain sockets.
+* Support `redis+unix` config URLs for Unix domain sockets
 * Add `PEXPIRE` and `PEXPIREAT`
 * Replace `trust-dns-resolver` with `hickory-resolver`
 
 ## 9.0.3
 
 * Fix `bytes_utils` min version
-* Fix rustls reexports with `enable-rustls-ring`.
+* Fix rustls reexports with `enable-rustls-ring`
 
 ## 9.0.2
 

Cargo.toml

@@ -11,7 +11,7 @@ name = "fred"
 readme = "README.md"
 repository = "https://github.com/aembke/fred.rs"
 rust-version = "1.75"
-version = "9.1.0"
+version = "9.1.1"
 
 [package.metadata.docs.rs]
 all-features = true

bin/inf_loop/Cargo.toml

@@ -15,8 +15,20 @@ pretty_env_logger = "0.5"
 tokio = { version = "1", features = ["full"] }
 futures = "0.3"
 rand = "0.8"
+opentelemetry = { version = "0.18.0", features = ["rt-tokio", "trace"] }
+opentelemetry-jaeger = { version = "0.17.0", features = ["tokio", "isahc_collector_client", "isahc", "collector_client", "rt-tokio"] }
+tracing-attributes = "0.1.23"
+tracing-opentelemetry = "0.18.0"
+tracing-core = "0.1.30"
+tracing-subscriber = "0.3.16"
+tracing = "0.1.37"
 
 [dependencies.fred]
 #path = "../.."
 path = "/fred"
 features = ["network-logs", "debug-ids", "replicas", "i-all"]
+
+[features]
+stdout-tracing = ["fred/partial-tracing"]
+partial-tracing = ["fred/partial-tracing"]
+full-tracing = ["fred/full-tracing"]

bin/inf_loop/cli.yml

@@ -7,6 +7,10 @@ args:
       long: cluster
       help: Whether to use a clustered deployment.
       takes_value: false
+  - tracing:
+      long: tracing
+      help: Whether to enable tracing.
+      takes_value: false
   - replicas:
       long: replicas
       help: Whether to use `GET` with replicas instead of `INCR` with primary nodes.

bin/inf_loop/docker-compose.yml

@@ -13,7 +13,7 @@ services:
         REDIS_VERSION: "${REDIS_VERSION}"
     networks:
       - fred-tests
-    entrypoint: "cargo run --release -- ${TEST_ARGV}"
+    entrypoint: "cargo run --release --features partial-tracing -- ${TEST_ARGV}"
     environment:
       RUST_LOG: "${RUST_LOG}"
       REDIS_VERSION: "${REDIS_VERSION}"

bin/inf_loop/src/main.rs

@@ -14,9 +14,18 @@ use fred::{
   prelude::*,
   types::{ReplicaConfig, UnresponsiveConfig},
 };
+use opentelemetry::{
+  global,
+  sdk::{
+    export::trace::stdout,
+    runtime::{Runtime, Tokio},
+    trace::{self, RandomIdGenerator, Sampler, TraceRuntime},
+  },
+};
 use rand::{self, distributions::Alphanumeric, Rng};
 use std::{default::Default, time::Duration};
 use tokio::time::sleep;
+use tracing_subscriber::{layer::SubscriberExt, Layer, Registry};
 
 #[derive(Debug)]
 struct Argv {
@@ -28,13 +37,15 @@ struct Argv {
   pub interval: u64,
   pub wait:     u64,
   pub auth:     String,
+  pub tracing:  bool,
 }
 
 fn parse_argv() -> Argv {
   let yaml = load_yaml!("../cli.yml");
   let matches = App::from_yaml(yaml).get_matches();
   let cluster = matches.is_present("cluster");
   let replicas = matches.is_present("replicas");
+  let tracing = matches.is_present("tracing");
 
   let host = matches
     .value_of("host")
@@ -67,16 +78,80 @@ fn parse_argv() -> Argv {
     interval,
     wait,
     replicas,
+    tracing,
   }
 }
 
+#[cfg(all(
+  not(feature = "partial-tracing"),
+  not(feature = "stdout-tracing"),
+  not(feature = "full-tracing")
+))]
+pub fn setup_tracing(enable: bool) {}
+
+#[cfg(feature = "stdout-tracing")]
+pub fn setup_tracing(enable: bool) {
+  if enable {
+    info!("Starting stdout tracing...");
+    let layer = tracing_subscriber::fmt::layer()
+      .with_writer(std::io::stdout)
+      .with_ansi(false)
+      .event_format(tracing_subscriber::fmt::format().pretty())
+      .with_thread_names(true)
+      .with_level(true)
+      .with_line_number(true)
+      .with_filter(tracing_subscriber::filter::LevelFilter::TRACE);
+    let subscriber = Registry::default().with(layer);
+    tracing::subscriber::set_global_default(subscriber).expect("Failed to set global tracing subscriber");
+  }
+}
+
+#[cfg(any(feature = "partial-tracing", feature = "full-tracing"))]
+pub fn setup_tracing(enable: bool) {
+  let sampler = if enable {
+    info!("Starting tracing...");
+    Sampler::AlwaysOn
+  } else {
+    Sampler::AlwaysOff
+  };
+
+  global::set_text_map_propagator(opentelemetry_jaeger::Propagator::new());
+  let jaeger_install = opentelemetry_jaeger::new_agent_pipeline()
+    .with_service_name("fred-inf-loop")
+    .with_endpoint("jaeger:6831")
+    .with_trace_config(
+      trace::config()
+        .with_sampler(sampler)
+        .with_id_generator(RandomIdGenerator::default())
+        .with_max_attributes_per_span(32),
+    )
+    .install_simple();
+
+  let tracer = match jaeger_install {
+    Ok(t) => t,
+    Err(e) => panic!("Fatal error initializing tracing: {:?}", e),
+  };
+
+  let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
+  let subscriber = Registry::default().with(telemetry);
+  tracing::subscriber::set_global_default(subscriber).expect("Failed to set global tracing subscriber");
+
+  info!("Initialized opentelemetry-jaeger pipeline.");
+}
+
 #[tokio::main]
 async fn main() -> Result<(), RedisError> {
   pretty_env_logger::init_timed();
   let argv = parse_argv();
   info!("Running with configuration: {:?}", argv);
+  setup_tracing(argv.tracing);
 
   let config = RedisConfig {
+    #[cfg(any(feature = "partial-tracing", feature = "stdout-tracing", feature = "full-tracing"))]
+    tracing: TracingConfig {
+      enabled: argv.tracing,
+      ..Default::default()
+    },
     server: if argv.cluster {
       ServerConfig::new_clustered(vec![(&argv.host, argv.port)])
     } else {
@@ -118,7 +193,7 @@ async fn main() -> Result<(), RedisError> {
 
   info!("Connecting to {}:{}...", argv.host, argv.port);
   pool.init().await?;
-  info!("Connected to {}:{}.", argv.host, argv.port);
+  info!("Connected to {}:{}", argv.host, argv.port);
   pool.flushall_cluster().await?;
 
   if argv.wait > 0 {

src/trace/README.md

@@ -4,7 +4,7 @@ Tracing
 Tracing is implemented via the [tracing](https://github.com/tokio-rs/tracing) crate. This page describes the spans used
 by the client and the fields emitted on each of the spans.
 
-![](../../tests/screenshot.png)
+![](../../tests/jaeger-2024.jpg)
 
 See the [benchmark](../../bin/benchmark) application for an example showing how to configure tracing with a
 local Jaeger instance. This crate ships with a [small example](../../tests/docker/compose/jaeger.yml) that

src/trace/enabled.rs

@@ -79,7 +79,7 @@ pub fn create_args_span(_parent: Option<TraceId>, _inner: &Arc<RedisClientInner>
 #[cfg(feature = "full-tracing")]
 pub fn create_queued_span(parent: Option<TraceId>, inner: &Arc<RedisClientInner>) -> Span {
   let buf_len = inner.counters.read_cmd_buffer_len();
-  span_lvl!(inner.full_tracing_span_level(), parent: parent, "queued", buf_len)
+  span_lvl!(inner.full_tracing_span_level(), parent: parent, "fred.queued", buf_len)
 }
 
 #[cfg(not(feature = "full-tracing"))]

src/utils.rs

@@ -481,8 +481,8 @@ where
     args_span.record("num_args", &command.args().len());
     (command, rx, req_size)
   };
-  cmd_span.record("cmd", &command.kind.to_str_debug());
-  cmd_span.record("req_size", &req_size);
+  cmd_span.record("cmd.name", &command.kind.to_str_debug());
+  cmd_span.record("cmd.req", &req_size);
 
   let queued_span = trace::create_queued_span(cmd_span.id(), inner);
   let timed_out = command.timed_out.clone();

tests/docker/compose/jaeger.yml

@@ -1,8 +1,15 @@
 version: "3"
+
+networks:
+  fred-tests:
+    driver: bridge
+
 services:
   jaeger:
     container_name: "jaeger"
     image: "jaegertracing/all-in-one:latest"
+    networks:
+      - fred-tests
     environment:
       - "COLLECTOR_ZIPKIN_HOST_PORT=:9411"
       - "COLLECTOR_OTLP_ENABLED=true"