diff --git a/axum/polars-otel-shuttle/Cargo.toml b/axum/polars-otel-shuttle/Cargo.toml
new file mode 100644
index 00000000..5c92f3bf
--- /dev/null
+++ b/axum/polars-otel-shuttle/Cargo.toml
@@ -0,0 +1,43 @@
+[package]
+name = "polars-otel-shuttle"
+version = "0.1.0"
+edition = "2021"
+
+[features]
+bench-cli = []          # local CLI binary
+otel-otlp = ["opentelemetry-otlp"]  # enable real OTLP exporter when you want it
+
+[dependencies]
+anyhow = "1"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+
+# Web (for Shuttle build)
+axum = { version = "0.8", features = ["json"] }
+tower = "0.5"
+tower-http = { version = "0.6", features = ["cors", "trace"] }
+
+# Tracing
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json", "time"] }
+time = { version = "0.3", features = ["formatting", "macros"] }
+tracing-opentelemetry = "0.31"
+
+# OpenTelemetry (core + SDK; OTLP exporter is optional via feature)
+opentelemetry = { version = "0.30", features = ["trace"] }
+opentelemetry_sdk = { version = "0.30", features = ["trace", "rt-tokio"] }
+opentelemetry-otlp = { version = "0.30", features = ["http-proto", "tls"], optional = true }
+
+# Runtime (new enough to satisfy shuttle-runtime)
+tokio = { version = "1.47.0", features = ["rt-multi-thread", "macros"] }
+
+# ETL
+polars = { version = "0.49", features = [
+  "lazy", "csv", "parquet", "fmt", "strings", "dtype-date", "dtype-datetime"
+] }
+chrono = { version = "0.4", features = ["clock"] }
+comfy-table = "7"
+
+# Shuttle (only used on deploy)
+shuttle-runtime = { version = "0.56", features = ["setup-otel-exporter"] }
+shuttle-axum = "0.56"
diff --git a/axum/polars-otel-shuttle/README.md b/axum/polars-otel-shuttle/README.md
new file mode 100644
index 00000000..0c215222
--- /dev/null
+++ b/axum/polars-otel-shuttle/README.md
@@ -0,0 +1,55 @@
+# Polars ETL with OpenTelemetry
+
+An ETL pipeline using Polars for data processing and OpenTelemetry for observability, deployable on Shuttle.
+
+## What it does
+
+- Processes CSV data with Polars (load → clean → aggregate → filter/sort → save)
+- Exposes HTTP endpoints via Axum
+- Emits metrics and traces via OpenTelemetry
+- Integrates with Better Stack for monitoring (when deployed on Shuttle)
+
+## Running locally
+
+**ETL benchmark:**
+```bash
+RUST_LOG=info cargo run --release --features bench-cli
+```
+
+**Web service:**
+```bash
+cargo run --release
+```
+
+Place your CSV file at `./data/yellow_tripdata_2015-01.csv` or set `DATA_PATH=/path/to/file.csv`.
+
+## Deploy to Shuttle
+
+```bash
+shuttle deploy
+```
+
+## Endpoints
+
+- `GET /` - Health check
+- `GET /health` - Health check with metrics
+- `GET /benchmark` - Returns ETL metrics and emits telemetry events
+
+## Observability
+
+The service emits OpenTelemetry metrics:
+- `monotonic_counter.http_requests_total` - Request counts
+- `histogram.request_duration_ms` - Request latency
+- `histogram.*_time_ms` - ETL stage timings
+- `monotonic_counter.rows_processed` - Data throughput
+
+When deployed on Shuttle with Better Stack integration enabled, these automatically appear in your telemetry dashboard.
+
+## Configuration
+
+**Environment variables:**
+- `DATA_PATH` - Path to CSV file (default: `data/yellow_tripdata_2015-01.csv`)
+
+**Features:**
+- `bench-cli` - Enables local ETL benchmark
+- `otel-otlp` - Enables OTLP export (for non-Shuttle deployments)
diff --git a/axum/polars-otel-shuttle/Shuttle.toml b/axum/polars-otel-shuttle/Shuttle.toml
new file mode 100644
index 00000000..c858139b
--- /dev/null
+++ b/axum/polars-otel-shuttle/Shuttle.toml
@@ -0,0 +1,2 @@
+name = "polars-otel-shuttle"
+# Shuttle uses your #[shuttle_runtime::main] entry point in src/main.rs
diff --git a/axum/polars-otel-shuttle/src/etl.rs b/axum/polars-otel-shuttle/src/etl.rs
new file mode 100644
index 00000000..1c194dda
--- /dev/null
+++ b/axum/polars-otel-shuttle/src/etl.rs
@@ -0,0 +1,332 @@
+use polars::prelude::*;
+use std::collections::HashMap;
+use std::time::Instant;
+use tracing::{info, instrument};
+
+#[allow(dead_code)]
+fn rss_mb() -> f64 {
+    if let Ok(s) = std::fs::read_to_string("/proc/self/status") {
+        for line in s.lines() {
+            if let Some(val) = line.strip_prefix("VmRSS:") {
+                let kb: f64 = val
+                    .split_whitespace()
+                    .next()
+                    .unwrap_or("0")
+                    .parse()
+                    .unwrap_or(0.0);
+                return kb / 1024.0;
+            }
+        }
+    }
+    0.0
+}
+
+#[allow(dead_code)]
+fn bump_peak(metrics: &mut HashMap<String, f64>, label: &str) {
+    let m = rss_mb();
+    metrics.insert(format!("{}_memory_mb", label), m);
+    let peak = metrics.get("peak_memory_mb").cloned().unwrap_or(0.0);
+    if m > peak {
+        metrics.insert("peak_memory_mb".into(), m);
+    }
+}
+
+#[allow(dead_code)]
+pub struct PolarsETL {
+    df: Option<LazyFrame>,
+    metrics: HashMap<String, f64>,
+}
+
+#[allow(dead_code)]
+impl PolarsETL {
+    pub fn new() -> Self {
+        Self {
+            df: None,
+            metrics: HashMap::new(),
+        }
+    }
+
+    pub fn get_metrics(&self) -> &HashMap<String, f64> {
+        &self.metrics
+    }
+
+    #[instrument(level = "info", skip(self))]
+    pub fn load_data(&mut self, file_path: &str) -> PolarsResult<&mut Self> {
+        info!("Loading data: {file_path}");
+        let start = Instant::now();
+
+        let lf = LazyCsvReader::new(file_path)
+            .with_has_header(true)
+            .with_infer_schema_length(Some(2000))
+            .map_parse_options(|opts| opts.with_try_parse_dates(false))
+            .finish()?
+            .select([
+                col("pickup_longitude"),
+                col("pickup_latitude"),
+                col("dropoff_longitude"),
+                col("dropoff_latitude"),
+                col("trip_distance"),
+                col("passenger_count"),
+                col("tpep_pickup_datetime"),
+                col("tpep_dropoff_datetime"),
+                col("total_amount"),
+            ]);
+
+        self.df = Some(lf);
+        let t = start.elapsed().as_secs_f64();
+        self.metrics.insert("load_time".into(), t);
+        bump_peak(&mut self.metrics, "after_load");
+        info!(
+            load_time_s = t,
+            peak_mb = self.metrics.get("peak_memory_mb").cloned().unwrap_or(0.0),
+            "Data scan created"
+        );
+        Ok(self)
+    }
+
+    #[instrument(level = "info", skip(self))]
+    pub fn clean_data(&mut self) -> PolarsResult<&mut Self> {
+        info!("Cleaning data");
+        let start = Instant::now();
+
+        if let Some(df) = &self.df {
+            let fmt: PlSmallStr = "%Y-%m-%d %H:%M:%S%.f".into();
+            let to_dt_opts = StrptimeOptions {
+                format: Some(fmt),
+                strict: false,
+                exact: false,
+                cache: true,
+            };
+
+            let cleaned = df
+                .clone()
+                .filter(
+                    col("pickup_longitude")
+                        .neq(lit(0.0))
+                        .and(col("pickup_latitude").neq(lit(0.0)))
+                        .and(col("dropoff_longitude").neq(lit(0.0)))
+                        .and(col("dropoff_latitude").neq(lit(0.0)))
+                        .and(col("trip_distance").gt(lit(0.0)))
+                        .and(col("trip_distance").lt(lit(100.0)))
+                        .and(col("passenger_count").gt(lit(0)))
+                        .and(col("passenger_count").lt_eq(lit(6))),
+                )
+                .with_columns([
+                    col("tpep_pickup_datetime").str().strptime(
+                        DataType::Datetime(TimeUnit::Microseconds, None),
+                        to_dt_opts.clone(),
+                        lit("coerce"),
+                    ),
+                    col("tpep_dropoff_datetime").str().strptime(
+                        DataType::Datetime(TimeUnit::Microseconds, None),
+                        to_dt_opts,
+                        lit("coerce"),
+                    ),
+                ])
+                .with_columns(
+                    [(col("tpep_dropoff_datetime") - col("tpep_pickup_datetime"))
+                        .dt()
+                        .total_minutes()
+                        .alias("trip_duration_minutes")],
+                )
+                .filter(
+                    col("trip_duration_minutes")
+                        .gt(lit(0))
+                        .and(col("trip_duration_minutes").lt(lit(480))),
+                )
+                .cache();
+
+            self.df = Some(cleaned);
+        }
+
+        let t = start.elapsed().as_secs_f64();
+        self.metrics.insert("clean_time".into(), t);
+        bump_peak(&mut self.metrics, "after_clean");
+        info!(
+            clean_time_s = t,
+            peak_mb = self.metrics.get("peak_memory_mb").cloned().unwrap_or(0.0),
+            "Cleaned"
+        );
+        Ok(self)
+    }
+
+    #[instrument(level = "info", skip(self))]
+    pub fn aggregate_data(&mut self) -> PolarsResult<&mut Self> {
+        info!("Aggregating");
+        let start = Instant::now();
+
+        if let Some(df) = &self.df {
+            let df_feats = df.clone().with_columns([
+                col("tpep_pickup_datetime").dt().date().alias("date"),
+                col("tpep_pickup_datetime").dt().hour().alias("hour"),
+                col("tpep_pickup_datetime").dt().weekday().alias("weekday"),
+            ]);
+
+            let _daily = df_feats
+                .clone()
+                .group_by([col("date")])
+                .agg([
+                    col("trip_distance").count().alias("trip_count"),
+                    col("trip_distance").mean().alias("avg_trip_distance"),
+                    col("trip_distance").sum().alias("total_trip_distance"),
+                    col("trip_duration_minutes")
+                        .mean()
+                        .alias("avg_trip_duration"),
+                    col("trip_duration_minutes")
+                        .sum()
+                        .alias("total_trip_duration"),
+                    col("passenger_count").sum().alias("total_passengers"),
+                    col("total_amount").mean().alias("avg_total_amount"),
+                    col("total_amount").sum().alias("total_revenue"),
+                ])
+                .collect()?;
+
+            let _hourly = df_feats
+                .clone()
+                .group_by([col("hour")])
+                .agg([
+                    col("trip_distance").count().alias("trip_count"),
+                    col("trip_distance").mean().alias("avg_trip_distance"),
+                    col("trip_duration_minutes")
+                        .mean()
+                        .alias("avg_trip_duration"),
+                    col("total_amount").mean().alias("avg_total_amount"),
+                ])
+                .collect()?;
+
+            let _dow = df_feats
+                .clone()
+                .group_by([col("weekday")])
+                .agg([
+                    col("trip_distance").count().alias("trip_count"),
+                    col("trip_distance").mean().alias("avg_trip_distance"),
+                    col("total_amount").mean().alias("avg_total_amount"),
+                ])
+                .collect()?;
+
+            let _ = (_daily, _hourly, _dow);
+        }
+
+        let t = start.elapsed().as_secs_f64();
+        self.metrics.insert("aggregate_time".into(), t);
+        bump_peak(&mut self.metrics, "after_aggregate");
+        info!(
+            aggregate_time_s = t,
+            peak_mb = self.metrics.get("peak_memory_mb").cloned().unwrap_or(0.0),
+            "Aggregations done"
+        );
+        Ok(self)
+    }
+
+    #[instrument(level = "info", skip(self))]
+    pub fn sort_and_filter(&mut self) -> PolarsResult<&mut Self> {
+        info!("Sort & filter counts");
+        let start = Instant::now();
+
+        if let Some(df) = &self.df {
+            let counts = df
+                .clone()
+                .with_columns([
+                    col("tpep_pickup_datetime").dt().hour().alias("hour"),
+                    col("tpep_pickup_datetime").dt().weekday().alias("weekday"),
+                ])
+                .select([
+                    col("trip_distance")
+                        .count()
+                        .cast(DataType::Int64)
+                        .alias("rows_after_cleaning"),
+                    col("trip_distance")
+                        .gt(lit(10.0))
+                        .cast(DataType::Int64)
+                        .sum()
+                        .alias("long_trips_count"),
+                    col("total_amount")
+                        .gt(lit(50.0))
+                        .cast(DataType::Int64)
+                        .sum()
+                        .alias("expensive_trips_count"),
+                    (col("hour")
+                        .eq(lit(7))
+                        .or(col("hour").eq(lit(8)))
+                        .or(col("hour").eq(lit(9)))
+                        .or(col("hour").eq(lit(17)))
+                        .or(col("hour").eq(lit(18)))
+                        .or(col("hour").eq(lit(19))))
+                    .cast(DataType::Int64)
+                    .sum()
+                    .alias("rush_hour_trips_count"),
+                    col("weekday")
+                        .gt_eq(lit(6))
+                        .cast(DataType::Int64)
+                        .sum()
+                        .alias("weekend_trips_count"),
+                    (col("trip_distance")
+                        .gt(lit(5.0))
+                        .and(col("total_amount").gt(lit(30.0)))
+                        .and(col("passenger_count").gt_eq(lit(2))))
+                    .cast(DataType::Int64)
+                    .sum()
+                    .alias("premium_trips_count"),
+                ])
+                .collect()?;
+
+            let get_i64 = |name: &str| -> PolarsResult<i64> {
+                Ok(counts.column(name)?.i64()?.get(0).unwrap_or(0))
+            };
+
+            self.metrics.insert(
+                "rows_after_cleaning".into(),
+                get_i64("rows_after_cleaning")? as f64,
+            );
+            self.metrics.insert(
+                "long_trips_count".into(),
+                get_i64("long_trips_count")? as f64,
+            );
+            self.metrics.insert(
+                "expensive_trips_count".into(),
+                get_i64("expensive_trips_count")? as f64,
+            );
+            self.metrics.insert(
+                "rush_hour_trips_count".into(),
+                get_i64("rush_hour_trips_count")? as f64,
+            );
+            self.metrics.insert(
+                "weekend_trips_count".into(),
+                get_i64("weekend_trips_count")? as f64,
+            );
+            self.metrics.insert(
+                "premium_trips_count".into(),
+                get_i64("premium_trips_count")? as f64,
+            );
+
+            info!(
+                long = self.metrics["long_trips_count"],
+                expensive = self.metrics["expensive_trips_count"],
+                "Counts computed"
+            );
+        }
+
+        let t = start.elapsed().as_secs_f64();
+        self.metrics.insert("sort_filter_time".into(), t);
+        bump_peak(&mut self.metrics, "after_sort_filter");
+        info!(
+            sort_filter_time_s = t,
+            peak_mb = self.metrics.get("peak_memory_mb").cloned().unwrap_or(0.0),
+            "Sort & filter done"
+        );
+        Ok(self)
+    }
+
+    #[instrument(level = "info", skip(self))]
+    pub fn save_results(&mut self, output_dir: &str) -> Result<(), Box<dyn std::error::Error>> {
+        let start = Instant::now();
+        std::fs::create_dir_all(output_dir)?;
+        let metrics_json = serde_json::to_string_pretty(&self.metrics)?;
+        std::fs::write(format!("{}/polars_metrics.json", output_dir), metrics_json)?;
+        let t = start.elapsed().as_secs_f64();
+        self.metrics.insert("save_time".into(), t);
+        bump_peak(&mut self.metrics, "after_save");
+        info!(save_time_s = t, "Saved results");
+        Ok(())
+    }
+}
diff --git a/axum/polars-otel-shuttle/src/main.rs b/axum/polars-otel-shuttle/src/main.rs
new file mode 100644
index 00000000..d551ade7
--- /dev/null
+++ b/axum/polars-otel-shuttle/src/main.rs
@@ -0,0 +1,205 @@
+mod etl;
+mod observability;
+
+use axum::{extract::Query, http::StatusCode, response::Json, routing::get, Router};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::time::Instant;
+use tower_http::{cors::CorsLayer, trace::TraceLayer};
+
+//
+// =============== 1) LOCAL CLI BENCH (feature: bench-cli) ===============
+// Build & run locally:  cargo run --release --features bench-cli
+//
+#[cfg(feature = "bench-cli")]
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    observability::init_tracing("polars-etl-benchmark-cli");
+
+    println!("{}", "=".repeat(50));
+    println!("🚀 STARTING POLARS ETL BENCHMARK");
+    println!("{}", "=".repeat(50));
+
+    // Check if data file exists (configurable via env)
+    let data_file = std::env::var("DATA_PATH")
+        .unwrap_or_else(|_| "data/yellow_tripdata_2015-01.csv".to_string());
+
+    if !std::path::Path::new(&data_file).exists() {
+        eprintln!("❌ Data file not found: {}", &data_file);
+        eprintln!(
+            "Tip: put the file at ./data/yellow_tripdata_2015-01.csv or set DATA_PATH=<path>"
+        );
+        return Ok(());
+    }
+
+    let total_start = Instant::now();
+    let mut etl = PolarsETL::new();
+
+    match etl
+        .load_data(&data_file)?
+        .clean_data()?
+        .aggregate_data()?
+        .sort_and_filter()?
+        .save_results("../results")
+    {
+        Ok(_) => {
+            let total_time = total_start.elapsed().as_secs_f64();
+
+            println!("\n{}", "=".repeat(50));
+            println!("🎉 POLARS BENCHMARK COMPLETE!");
+            println!("{}", "=".repeat(50));
+            println!("⏱️  Total time: {:.2} seconds", total_time);
+
+            println!("\n📈 Key Performance Metrics:");
+            let metrics = etl.get_metrics();
+            for (key, value) in metrics {
+                if key.contains("time") {
+                    let formatted_key = key
+                        .replace('_', " ")
+                        .split_whitespace()
+                        .map(|w| {
+                            let mut c = w.chars();
+                            match c.next() {
+                                None => String::new(),
+                                Some(f) => f.to_uppercase().collect::<String>() + c.as_str(),
+                            }
+                        })
+                        .collect::<Vec<_>>()
+                        .join(" ");
+                    println!("  {}: {:.2}s", formatted_key, value);
+                }
+            }
+            println!("{}", "=".repeat(50));
+        }
+        Err(e) => {
+            println!("❌ Error during Polars benchmark: {}", e);
+        }
+    }
+
+    Ok(())
+}
+
+//
+// =============== 2) SHUTTLE WEB APP (default build) ====================
+// Deploy: cargo shuttle deploy
+//
+#[cfg(not(feature = "bench-cli"))]
+#[shuttle_runtime::main]
+async fn shuttle_main() -> shuttle_axum::ShuttleAxum {
+    let app = api_router()
+        .layer(CorsLayer::permissive())
+        .layer(TraceLayer::new_for_http());
+    tracing::info!(event = "startup", message = "Axum app ready");
+    Ok(app.into())
+}
+
+#[derive(Deserialize)]
+struct BenchmarkQuery {
+    #[serde(default)]
+    _sample_size: Option<usize>,
+}
+
+#[derive(Serialize)]
+struct BenchmarkResult {
+    metrics: HashMap<String, f64>,
+    message: String,
+    performance_summary: String,
+}
+
+async fn health() -> &'static str {
+    // emit a cheap heartbeat counter
+    tracing::info!(monotonic_counter.health_checks = 1, "health checked");
+    tracing::info!(endpoint = "/health", status = "ok");
+    "ok"
+}
+
+#[tracing::instrument(name = "benchmark", skip_all)]
+async fn run_benchmark(_q: Query<BenchmarkQuery>) -> Result<Json<BenchmarkResult>, StatusCode> {
+    let t0 = Instant::now();
+
+    // Count requests to this endpoint
+    tracing::info!(
+        monotonic_counter.http_requests_total = 1,
+        route = "/benchmark",
+        "Benchmark hit"
+    );
+
+    tracing::info!(endpoint = "/benchmark", event = "start_demo_benchmark");
+    // NOTE: We return representative metrics here (do not run heavy ETL on the Shuttle dyno).
+    let mut metrics = HashMap::new();
+    metrics.insert("load_time".to_string(), 1.2);
+    metrics.insert("clean_time".to_string(), 0.8);
+    metrics.insert("aggregate_time".to_string(), 0.4);
+    metrics.insert("sort_filter_time".to_string(), 0.3);
+    metrics.insert("save_time".to_string(), 0.1);
+    metrics.insert("total_time".to_string(), 2.8);
+
+    // Export timings as histograms (ms) so Better Stack can graph them
+    tracing::info!(
+        histogram.load_time_ms = 1.2_f64 * 1000.0,
+        stage = "load",
+        "stage timing"
+    );
+    tracing::info!(
+        histogram.clean_time_ms = 0.8_f64 * 1000.0,
+        stage = "clean",
+        "stage timing"
+    );
+    tracing::info!(
+        histogram.aggregate_time_ms = 0.4_f64 * 1000.0,
+        stage = "aggregate",
+        "stage timing"
+    );
+    tracing::info!(
+        histogram.sort_filter_time_ms = 0.3_f64 * 1000.0,
+        stage = "sort_filter",
+        "stage timing"
+    );
+    tracing::info!(
+        histogram.save_time_ms = 0.1_f64 * 1000.0,
+        stage = "save",
+        "stage timing"
+    );
+    tracing::info!(
+        histogram.total_time_ms = 2.8_f64 * 1000.0,
+        stage = "total",
+        "stage timing"
+    );
+
+    // Rows processed (monotonic counter)
+    tracing::info!(
+        monotonic_counter.rows_processed = 12_748_986_i64,
+        "rows processed (demo)"
+    );
+
+    // Request duration histogram
+    let elapsed_ms = t0.elapsed().as_millis() as f64;
+    tracing::info!(
+        histogram.request_duration_ms = elapsed_ms,
+        route = "/benchmark",
+        "Benchmark responded"
+    );
+
+    let rows_per_second = 12_748_986.0 / 2.8;
+    let summary = format!(
+        "🚀 Polars processed ~12.7M taxi records in ~2.8s (demo) → ~{:.0} rows/sec via Rust + Polars + OTEL.",
+        rows_per_second
+    );
+    tracing::info!(
+        endpoint = "/benchmark",
+        event = "end_demo_benchmark",
+        total_time = 2.8
+    );
+
+    Ok(Json(BenchmarkResult {
+        metrics,
+        message: "✅ Demo benchmark complete (server-safe)".into(),
+        performance_summary: summary,
+    }))
+}
+
+fn api_router() -> Router {
+    Router::new()
+        .route("/", get(health))
+        .route("/health", get(health))
+        .route("/benchmark", get(run_benchmark))
+}
diff --git a/axum/polars-otel-shuttle/src/observability.rs b/axum/polars-otel-shuttle/src/observability.rs
new file mode 100644
index 00000000..8bee3156
--- /dev/null
+++ b/axum/polars-otel-shuttle/src/observability.rs
@@ -0,0 +1,99 @@
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
+
+#[allow(dead_code)]
+pub fn init_tracing(_service_name: &str) {
+    // JSON logs with RFC3339 timestamps
+    let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+
+    let fmt_layer = tracing_subscriber::fmt::layer()
+        .with_ansi(false)
+        .json()
+        .with_current_span(true)
+        .with_span_list(true)
+        .with_target(false)
+        .with_file(false)
+        .with_line_number(false)
+        .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339());
+
+    // Base registry (logs only)
+    let registry = tracing_subscriber::registry()
+        .with(env_filter)
+        .with(fmt_layer);
+
+    // If you compile with `--features otel-otlp`, add an OTLP span layer too.
+    #[cfg(feature = "otel-otlp")]
+    {
+        if let Some(layer) = build_otel_layer(service_name) {
+            registry = registry.with(layer);
+        }
+    }
+
+    registry.init();
+}
+
+#[cfg(feature = "otel-otlp")]
+fn build_otel_layer<S>(
+    service_name: &str,
+) -> Option<tracing_opentelemetry::OpenTelemetryLayer<S, sdktrace::Tracer>>
+where
+    S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
+{
+    use opentelemetry_otlp::Protocol;
+
+    // Read endpoint/headers from env (works for BetterStack or any OTLP endpoint)
+    let endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
+        .unwrap_or_else(|_| "http://localhost:4318".into());
+
+    // Headers in the standard comma-separated form: k1=v1,k2=v2
+    let headers = std::env::var("OTEL_EXPORTER_OTLP_HEADERS").ok();
+    let headers_map = parse_headers(headers.as_deref());
+
+    // Resource (service name etc.)
+    let resource = Resource::builder()
+        .with_service_name(service_name.to_string())
+        .with_attributes(vec![KeyValue::new("telemetry.sdk.language", "rust")])
+        .build();
+
+    // Build an OTLP/HTTP span exporter (0.30 API)
+    let exporter = opentelemetry_otlp::SpanExporter::builder()
+        .with_http()
+        .with_endpoint(endpoint)
+        .with_protocol(Protocol::HttpBinary) // or HttpProtobuf; works with most collectors
+        .with_headers(headers_map)
+        .build()
+        .ok()?;
+
+    // Tracer provider with batch exporter on Tokio
+    let provider = sdktrace::TracerProvider::builder()
+        .with_resource(resource)
+        .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
+        .build();
+
+    let tracer = provider.tracer("polars-otel-shuttle");
+
+    // Install as global provider
+    opentelemetry::global::set_tracer_provider(provider);
+
+    // Bridge tracing -> OpenTelemetry
+    Some(tracing_opentelemetry::layer().with_tracer(tracer))
+}
+
+#[cfg(feature = "otel-otlp")]
+fn parse_headers(input: Option<&str>) -> std::collections::HashMap<String, String> {
+    let mut map = std::collections::HashMap::new();
+    if let Some(s) = input {
+        for kv in s.split(',') {
+            if let Some((k, v)) = kv.split_once('=') {
+                map.insert(k.trim().to_string(), v.trim().to_string());
+            }
+        }
+    }
+    map
+}
+
+/// Call this on shutdown if you want to block until spans are exported.
+#[allow(dead_code)]
+pub fn shutdown_tracing() {
+    // No-op for opentelemetry 0.30.
+    // The tracer provider will flush on drop when the process exits.
+}
diff --git a/templates.toml b/templates.toml
index 20ff33b0..e92cc689 100644
--- a/templates.toml
+++ b/templates.toml
@@ -241,6 +241,13 @@ path = "axum/todo-app"
 use_cases = ["Web app", "Storage"]
 tags = ["axum", "postgres", "database"]
 
+[templates.axum-polars-otel-shuttle]
+title = "Polars ETL with OpenTelemetry"
+description = "ETL pipeline with Polars and Open Telemetry monitoring"
+path = "axum/polars-otel-shuttle"
+use_cases = ["ETL", "Data Engineering", "Observability"]
+tags = ["axum", "polars", "opentelemetry", "etl"]
+
 [templates.axum-turso]
 title = "Turso"
 description = "Connect to a Turso DB with shuttle-turso"
@@ -375,6 +382,7 @@ use_cases = ["MCP", "AI", "AI Agents"]
 tags = ["axum", "mcp", "sse", "oauth"]
 
 
+
 ## EXAMPLES ##
 
 [examples.metadata]
@@ -471,3 +479,5 @@ use_cases = ["Web app"]
 tags = ["rocket", "yew"]
 author = "wiseaidev"
 repo = "https://github.com/wiseaidev/rocket-yew-starter-pack"
+
+