Feature/opentelemetry 0.28 (#15)

* opentelemetry 0.28
* Use cache mount in Dockerfile

Author: hardbyte

Cargo.lock

@@ -6,9 +6,9 @@ members = [
 ]
 
 [workspace.dependencies]
-axum = { version = "0.8" }
+axum = { version = "0.8.1" }
 tower = "0.5.1"
 reqwest = "0.12.7"
-opentelemetry = "0.27.1"
+opentelemetry = "0.28.0"
 tracing = { version = "0.1.41" }
-tracing-opentelemetry = { version = "0.28.0" }
+tracing-opentelemetry = { version = "0.29.0" }

Cargo.toml

@@ -1,14 +1,22 @@
-ARG RUST_VERSION=1.84.0
+ARG RUST_VERSION=1.84
 
 FROM rust:${RUST_VERSION}-bookworm AS builder
 WORKDIR /usr/src/bookapp
-COPY . .
 ENV SQLX_OFFLINE=true
-RUN cargo install --path bookapp
+
+# Copy the full source and build the app in release mode
+COPY . .
+
+RUN --mount=type=cache,target=/usr/local/cargo/registry \
+    --mount=type=cache,target=/usr/src/bookapp/target \
+    cargo build --release --package bookapp && \
+    mv /usr/src/bookapp/target/release/bookapp /bookapp
+
 
 FROM debian:bookworm-slim
 RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates wget && rm -rf /var/lib/apt/lists/*
-COPY --from=builder /usr/local/cargo/bin/bookapp /usr/local/bin/bookapp
+
+COPY --from=builder /bookapp /usr/local/bin/bookapp
 
 ENV RUST_LOG="info,sqlx=info,bookapp=debug,backend=debug"
 CMD ["bookapp"]

Dockerfile

@@ -6,18 +6,19 @@ edition = "2021"
 
 [dependencies]
 client = { path = "../client" }
-tower-otel-http-metrics = { version = "0.10.0", features = ["axum"] }
+tower-otel-http-metrics = { version = "0.13.0", features = ["axum"] }
 
-anyhow = "1.0.95"
+anyhow = "1.0.97"
 axum = { workspace = true, features = ["macros", "matched-path", "tracing" ] }
 dotenv = "0.15.0"
 opentelemetry = { workspace = true}
-opentelemetry-otlp = { version="0.27.0" , features = [
+opentelemetry-otlp = { version="0.28.0" , features = [
     "serialize",
     "reqwest-client",
+    "grpc-tonic",
 ]}
-opentelemetry_sdk = { version="0.27.1", features = ["trace", "opentelemetry-http", "rt-tokio", "tracing"]}
-serde = "1.0.217"
+opentelemetry_sdk = { version="0.28.0", features = ["trace", "opentelemetry-http", "rt-tokio", "tracing"]}
+serde = "1.0.218"
 sqlx = { version = "0.8.3", features = ["runtime-tokio", "postgres"] }
 tokio = { version = "1.43", features = ["full", "tracing"] }
 tower = {workspace = true}
@@ -27,23 +28,24 @@ tracing-opentelemetry = { workspace = true , features=["async-trait"]}
 tracing-subscriber = { version = "0.3.19", features=["fmt", "env-filter", "json", "tracing-log"] }
 reqwest = {workspace = true}
 
-reqwest-middleware = { version="0.4.0", features = ["json"] }
-reqwest-tracing = { version="0.5.5",  features = ["opentelemetry_0_27"] }
+reqwest-middleware = { version="0.4.1", features = ["json"] }
+reqwest-tracing = { version="0.5.6",  features = ["opentelemetry_0_28"] }
 
 futures = "0.3.31"
 
-axum-tracing-opentelemetry = { version = "0.25", features = [] }
-tracing-opentelemetry-instrumentation-sdk = { version = "0.24.2", features = ["tracing_level_info"] }
+axum-tracing-opentelemetry = { version = "0.26.1", features = [] }
+
+tracing-opentelemetry-instrumentation-sdk = { version = "0.26.0", features = ["tracing_level_info"] }
 
 
 rdkafka = { version = "0.37.0", features = ["tokio"] }
-serde_json = "1.0.137"
-opentelemetry-http = "0.27.0"
+serde_json = "1.0.140"
+opentelemetry-http = "0.28.0"
 tracing-loki = "0.2.6"
 hostname = "0.4.0"
-opentelemetry-appender-tracing = "0.27.0"
+opentelemetry-appender-tracing = "0.28.1"
 hyper = "1.6.0"
 rand = "0.9.0"
 matchit = "^0.8"
 console-subscriber = "0.4.1"
-async-trait = "0.1.86"
+async-trait = "0.1.87"

bookapp/Cargo.toml

@@ -176,7 +176,7 @@ pub async fn run_consumer() -> Result<()> {
                 };
 
                 // Create a new root span via tracing:
-                let span = tracing::info_span!("book_ingestion", "otel.kind" = "consumer",);
+                let span = tracing::info_span!("book_ingestion", "otel.kind" = "Consumer");
 
                 // Extract tracing context from headers
                 let headers = m.headers();

bookapp/src/book_ingestion.rs

@@ -8,7 +8,7 @@ use tracing::{debug, info};
 pub struct BookCreateIn {
     pub title: String,
     pub author: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(skip_serializing_if = "Option::is_none", default)]
     pub status: Option<BookStatus>,
 }
 
@@ -78,22 +78,24 @@ pub async fn get_book(connection_pool: &PgPool, id: i32) -> Result<Book> {
     .await?)
 }
 
+#[tracing::instrument(skip(connection_pool), name = "create_book_in_db")]
 pub async fn create_book(
     connection_pool: &PgPool,
     author: String,
     title: String,
-    status: Option<BookStatus>,
+    status: BookStatus,
 ) -> Result<i32> {
     Ok(sqlx::query!(
         r#"insert into books (title, author, status) VALUES ($1, $2, $3) returning id"#,
         title,
         author,
-        status as Option<BookStatus>,
+        status as BookStatus,
     )
     .fetch_one(connection_pool)
     .await?
     .id)
 }
+
 pub async fn delete_book(connection_pool: &PgPool, id: i32) -> Result<()> {
     sqlx::query!("delete from books where id=$1", id)
         .execute(connection_pool)

bookapp/src/db.rs

@@ -325,8 +325,8 @@ pub async fn error_injection_middleware(
         tracing::Span::current().record("error_rate", &config.error_rate);
 
         // Generate a random number between 0.0 and 1.0
-        let mut rng = rand::thread_rng();
-        let random_value: f64 = rng.gen();
+        let mut rng = rand::rng();
+        let random_value: f64 = rng.random();
 
         if random_value < config.error_rate {
             tracing::debug!(
@@ -414,6 +414,6 @@ pub fn error_injection_service(
 ) -> Router {
     Router::new()
         .route("/", get(get_all_configs_handler).post(create_config))
-        .route("/:id", put(update_config).delete(delete_config))
+        .route("/{id}", put(update_config).delete(delete_config))
         .layer(Extension(error_injection_store))
 }

bookapp/src/error_injection_middleware.rs

@@ -36,14 +36,14 @@ fn router(connection_pool: PgPool, producer: FutureProducer) -> Router {
         .layer(Extension(producer))
         // Our custom error injection layer can inject errors
         // This layer itself can be traced - so needs to be added before our OtelAxumLayer
-        // .layer(axum::middleware::from_fn_with_state(
-        //     error_injection_store.clone(),
-        //     error_injection_middleware::error_injection_middleware,
-        // ))
-        // .nest_service(
-        //     "/error-injection",
-        //     error_injection_middleware::error_injection_service(error_injection_store.clone()),
-        // )
+        .layer(axum::middleware::from_fn_with_state(
+            error_injection_store.clone(),
+            error_injection_middleware::error_injection_middleware,
+        ))
+        .nest_service(
+            "/error-injection",
+            error_injection_middleware::error_injection_service(error_injection_store.clone()),
+        )
         .layer(Extension(connection_pool))
         // This layer creates a new Tracing span called "request" for each request,
         // it logs headers etc but on its own doesn't do the OTEL trace context propagation.
@@ -60,7 +60,7 @@ fn router(connection_pool: PgPool, producer: FutureProducer) -> Router {
         // as long as not filtered out!
         .layer(OtelAxumLayer::default())
         .layer(
-            tower_otel_http_metrics::HTTPMetricsLayerBuilder::new()
+            tower_otel_http_metrics::HTTPMetricsLayerBuilder::builder()
                 .with_meter(opentelemetry::global::meter(env!("CARGO_CRATE_NAME")))
                 .build()
                 .expect("Failed to build otel metrics layer"),
@@ -79,7 +79,7 @@ async fn main() -> Result<()> {
     let enable_kafka_producer =
         std::env::var("ENABLE_KAFKA_PRODUCER").unwrap_or_else(|_| "false".to_string()) == "true";
 
-    tracing_config::init_tracing();
+    let (trace_provider, meter_provider, log_provider) = tracing_config::init_tracing();
 
     // Init db
     info!("Setting up Database");
@@ -129,7 +129,9 @@ async fn main() -> Result<()> {
 
     info!("Shutting down OpenTelemetry");
 
-    global::shutdown_tracer_provider();
+    trace_provider.shutdown()?;
+    meter_provider.shutdown()?;
+    log_provider.shutdown()?;
 
     Ok(())
 }

bookapp/src/main.rs

@@ -5,7 +5,7 @@ use axum::http::StatusCode;
 use axum::routing::{delete, get, patch, post};
 use axum::{extract, http::Request, Extension, Json, Router};
 
-use opentelemetry::trace::TraceContextExt;
+use opentelemetry::trace::{TraceContextExt, SpanKind};
 use rdkafka::producer::FutureProducer;
 use sqlx::PgPool;
 use tracing::{Instrument, Level};
@@ -56,7 +56,7 @@ async fn get_all_books(Extension(con): Extension<PgPool>) -> Result<Json<Vec<Boo
     }
 }
 
-#[tracing::instrument(fields(otel.kind = "client"))]
+#[tracing::instrument(fields(book_id, otel.kind = "Client"))]
 async fn get_book_details_with_progenitor_client(
     book_id: i32,
 ) -> Result<client::ResponseValue<client::types::Book>, client::Error> {
@@ -137,16 +137,16 @@ async fn create_book(
     Extension(producer): Extension<FutureProducer>,
     Json(book): Json<BookCreateIn>,
 ) -> Result<Json<i32>, StatusCode> {
-    if let Ok(new_id) = db::create_book(&con, book.author, book.title, book.status).await {
+    let status = book.status.unwrap_or(BookStatus::Available);
+    if let Ok(new_id) = db::create_book(&con, book.author, book.title, status).await {
         queue_background_ingestion_task(&producer, new_id).await;
-
         Ok(Json(new_id))
     } else {
-        Err(StatusCode::NOT_FOUND)
+        Err(StatusCode::INTERNAL_SERVER_ERROR)
     }
 }
 
-#[tracing::instrument(skip(producer), fields(otel.kind = "producer"))]
+#[tracing::instrument(skip(producer), fields(otel.kind = "Producer"))]
 async fn queue_background_ingestion_task(producer: &FutureProducer, new_id: i32) {
     // Prepare message
     let book_message = crate::book_ingestion::BookIngestionMessage { book_id: new_id };

bookapp/src/rest.rs

@@ -1,26 +1,26 @@
 use opentelemetry::trace::TracerProvider;
 use opentelemetry_otlp::{LogExporter, WithExportConfig};
-use opentelemetry_sdk::logs::LoggerProvider;
-use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
+use opentelemetry_sdk::logs::SdkLoggerProvider;
+use opentelemetry_sdk::metrics::{SdkMeterProvider};
 use opentelemetry_sdk::propagation::TraceContextPropagator;
+use opentelemetry_sdk::trace::SdkTracerProvider;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::Layer;
 
 fn init_meter_provider(
-) -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, opentelemetry_sdk::metrics::MetricError> {
+) -> Result<SdkMeterProvider, opentelemetry_sdk::metrics::MetricError> {
     let exporter = opentelemetry_otlp::MetricExporter::builder()
         .with_tonic()
         .with_timeout(std::time::Duration::from_secs(10))
         .build()?;
 
-    let reader = PeriodicReader::builder(exporter, opentelemetry_sdk::runtime::Tokio).build();
-
     let provider = SdkMeterProvider::builder()
-        .with_reader(reader)
-        .with_resource(opentelemetry_sdk::Resource::default())
-        .with_resource(opentelemetry_sdk::Resource::new(vec![
-            opentelemetry::KeyValue::new("service.name", "bookapp"),
-        ]))
+        .with_periodic_exporter(exporter)
+        .with_resource(
+            opentelemetry_sdk::Resource::builder()
+                .with_attributes(vec![opentelemetry::KeyValue::new("service.name", "bookapp")])
+                .build()
+        )
         .build();
 
     let cloned_provider = provider.clone();
@@ -29,38 +29,46 @@ fn init_meter_provider(
 }
 
 fn init_logger_provider(
-) -> Result<opentelemetry_sdk::logs::LoggerProvider, opentelemetry_sdk::logs::LogError> {
+) -> Result<opentelemetry_sdk::logs::SdkLoggerProvider, opentelemetry_sdk::logs::LogError> {
     // Note Opentelemetry does not provide a global API to manage the logger provider.
-    let exporter = LogExporter::builder().with_tonic().build()?;
-
-    let provider = LoggerProvider::builder()
-        .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
-        .build();
+    let exporter = LogExporter::builder()
+        .with_tonic()
+        .build()?;
 
-    Ok(provider)
+    Ok(SdkLoggerProvider::builder()
+        //.with_resource()
+        .with_batch_exporter(exporter)
+        .build())
 }
 
-pub fn init_tracing() {
+pub fn init_tracing() -> (
+    SdkTracerProvider, SdkMeterProvider, SdkLoggerProvider
+) {
     opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
 
     // Metrics
     let meter_provider = init_meter_provider().unwrap();
-    let opentelemetry_metrics_layer = tracing_opentelemetry::MetricsLayer::new(meter_provider);
+    let opentelemetry_metrics_layer = tracing_opentelemetry::MetricsLayer::new(meter_provider.clone());
 
     // Tracing
     // Uses OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
-    // Assumes a GRPC endpoint (e.g port 4317)
+    // Assumes a GRPC endpoint (e.g., port 4317)
     let exporter = opentelemetry_otlp::SpanExporter::builder()
         .with_tonic()
         .build()
-        .unwrap();
+        .expect("Failed to create OTLP span exporter");
 
-    let tracer_provider = opentelemetry_sdk::trace::TracerProvider::builder()
-        .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
+    let tracer_provider = SdkTracerProvider::builder()
+        .with_batch_exporter(exporter)
         //.with_resource(opentelemetry_sdk::Resource::default())
         .build();
 
     // Explicitly set the tracer provider globally
+    // Setting global tracer provider is required if other parts of the application
+    // uses global::tracer() or global::tracer_with_version() to get a tracer.
+    // Cloning simply creates a new reference to the same tracer provider. It is
+    // important to hold on to the tracer_provider here, to invoke
+    // shutdown on it when application ends.
     opentelemetry::global::set_tracer_provider(tracer_provider.clone());
 
     // Filter the tracing layer - we can add custom filters that only impact the tracing layer
@@ -127,4 +135,7 @@ pub fn init_tracing() {
 
     // Set the subscriber as the global default
     tracing::subscriber::set_global_default(subscriber).expect("Failed to set subscriber");
+
+    // Return the tracer, meter and logger provider as a tuple for shutdown
+    (tracer_provider, meter_provider, log_provider)
 }

bookapp/src/tracing_config.rs

@@ -8,8 +8,8 @@ edition = "2021"
 [dependencies]
 tracing = { workspace = true }
 reqwest = { workspace = true, features = ["json", "stream"] }
-serde = { version = "1.0.217", features = ["derive", "rc"] }
-serde_json = { version = "1.0.138" }
+serde = { version = "1.0.218", features = ["derive", "rc"] }
+serde_json = { version = "1.0.140" }
 opentelemetry = { workspace = true}
 tracing-opentelemetry = { workspace = true, features=["async-trait"] }
 progenitor-client = { version = "0.9.1" }

client/Cargo.toml

@@ -81,7 +81,7 @@ services:
 
 
   telemetry:
-    image: grafana/otel-lgtm:0.8.4
+    image: grafana/otel-lgtm:0.8.5
     depends_on:
       db:
         condition: service_healthy

docker-compose.yaml

@@ -4,7 +4,7 @@ GET http://localhost:8000/books
 Accept: application/json
 
 ### GET a book
-GET http://localhost:8000/books/98
+GET http://localhost:8000/books/1
 Accept: application/json
 
 ### Create a book