Move turborepo-otel to the experimental periodic_reader_with_async_runtime feature from opentelemetry_sdk

Author: bkonkle

crates/turborepo-lib/src/run/summary/mod.rs

@@ -27,9 +27,10 @@ use turborepo_env::EnvironmentVariableMap;
 use turborepo_repository::package_graph::{PackageGraph, PackageName};
 use turborepo_scm::SCM;
 use turborepo_task_id::TaskId;
-use turborepo_ui::{color, cprintln, cwriteln, ColorConfig, BOLD, BOLD_CYAN, GREY};
+use turborepo_ui::{BOLD, BOLD_CYAN, ColorConfig, GREY, color, cprintln, cwriteln};
 
-pub(crate) use self::task::TaskSummary; // Re-exported for use in observability/otel.rs
+/// Re-exported for use in observability/otel.rs
+pub(crate) use self::task::{CacheStatus, TaskSummary};
 use self::{
     execution::TaskState, task::SinglePackageTaskSummary, task_factory::TaskSummaryFactory,
 };

crates/turborepo-otel/Cargo.toml

@@ -12,7 +12,12 @@ opentelemetry-otlp = { version = "0.31", features = [
   "metrics",
 ] }
 opentelemetry-semantic-conventions = "0.31"
-opentelemetry_sdk = { version = "0.31", features = ["rt-tokio", "metrics"] }
+opentelemetry_sdk = { version = "0.31", features = [
+  "metrics",
+  "rt-tokio",
+  "experimental_async_runtime",
+  "experimental_metrics_periodicreader_with_async_runtime",
+] }
 serde = { workspace = true, optional = true }
 serde_json = { workspace = true, optional = true }
 thiserror = { workspace = true }

crates/turborepo-otel/examples/local-collector/README.md

@@ -0,0 +1,137 @@
+# Local OTEL collector example
+
+This example shows how to build the `turbo` binary with the **OpenTelemetry (OTEL)** feature enabled and send metrics to a local collector running via **Docker Compose**. It’s intended as a lightweight, local integration harness for the OTEL exporter.
+
+## 1. Prerequisites
+
+- **Docker & docker compose** installed and running
+- **Rust toolchain** (matching this repo’s `rust-toolchain.toml`)
+- **pnpm** installed (per `CONTRIBUTING.md`)
+
+All commands below assume the repo root is `turborepo/`.
+
+## 2. Build `turbo` with the OTEL feature
+
+From the repo root:
+
+```bash
+pnpm install
+cargo build -p turbo --features otel
+```
+
+This produces an OTEL-enabled `turbo` binary at: `./target/debug/turbo`
+
+## 3. Start the local collector stack
+
+From this example directory:
+
+```bash
+cd crates/turborepo-otel/examples/local-collector
+docker compose up -d
+```
+
+This starts:
+
+- **`otel-collector`** (OTLP receiver + debug + Prometheus exporter)
+- **`prometheus`** (scrapes metrics from the collector)
+- **`grafana`** (optional visualization)
+
+Ports:
+
+- **OTLP gRPC**: `4317`
+- **OTLP HTTP**: `4318`
+- **Collector metrics / Prometheus exporter**: `8888`, `8889`
+- **Prometheus UI**: `9090`
+- **Grafana UI**: `3000`
+
+You can confirm the collector is ready:
+
+```bash
+docker compose logs otel-collector
+```
+
+You should see a line similar to:
+
+```text
+Everything is ready. Begin running and processing data.
+```
+
+## 4. Configure `turbo` to send metrics to the local collector
+
+In a new shell, from the repo root, export the OTEL env vars:
+
+```bash
+cd /path/to/turborepo
+
+export TURBO_EXPERIMENTAL_OTEL_ENABLED=1
+export TURBO_EXPERIMENTAL_OTEL_PROTOCOL=grpc
+export TURBO_EXPERIMENTAL_OTEL_ENDPOINT=http://127.0.0.1:4317
+export TURBO_EXPERIMENTAL_OTEL_RESOURCE="service.name=turborepo,env=local"
+# Optional (defaults shown)
+export TURBO_EXPERIMENTAL_OTEL_METRICS_RUN_SUMMARY=1
+export TURBO_EXPERIMENTAL_OTEL_METRICS_TASK_DETAILS=0
+```
+
+These environment variables bypass `turbo.json` and directly configure the OTEL exporter.
+
+## 5. Run a task and emit metrics
+
+Use the **locally built** OTEL-enabled binary:
+
+```bash
+./target/debug/turbo run lint --filter=turbo-ignore
+```
+
+You can replace `lint --filter=turbo-ignore` with any real task in this repo; the important part is that the command finishes so a run summary can be exported.
+
+## 6. Verify metrics reached the collector
+
+- **Collector logs (debug exporter)**:
+
+  ```bash
+  cd crates/turborepo-otel/examples/local-collector
+  docker compose logs --tail=100 otel-collector
+  ```
+
+  You should see entries like:
+
+  ```text
+  Metrics {"otelcol.component.id": "debug", "otelcol.signal": "metrics", "resource metrics": 1, "metrics": 4, "data points": 4}
+  Resource attributes:
+       -> env: Str(local)
+       -> service.name: Str(turborepo)
+  Metric #0
+       -> Name: turbo.run.duration_ms
+  Metric #1
+       -> Name: turbo.run.tasks.attempted
+  Metric #2
+       -> Name: turbo.run.tasks.failed
+  Metric #3
+       -> Name: turbo.run.tasks.cached
+  ```
+
+- **Prometheus UI** (optional):
+
+  - Open `http://localhost:9090`
+  - Query for metrics such as:
+    - `turbo.run.duration_ms`
+    - `turbo.run.tasks.attempted`
+    - `turbo.run.tasks.failed`
+    - `turbo.run.tasks.cached`
+
+- **Grafana UI** (optional):
+
+  - Open `http://localhost:3000` (default credentials are usually `admin` / `admin`)
+  - Add a Prometheus data source pointing at `http://prometheus:9090`
+  - Build dashboards using the `turbo.*` metrics
+
+## 7. Cleanup
+
+To stop the local collector stack:
+
+```bash
+cd crates/turborepo-otel/examples/local-collector
+docker compose down
+```
+
+The OTEL-enabled `turbo` binary remains available at `./target/debug/turbo`. You can continue using it with the same environment variables to send metrics to this collector or to another OTLP-compatible backend.

crates/turborepo-otel/examples/local-collector/docker-compose.yml

@@ -0,0 +1,27 @@
+services:
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:0.139.0
+    volumes:
+      - ./otel-collector.yml:/etc/otelcol-contrib/config.yaml
+    ports:
+      - 1888:1888 # pprof extension
+      - 8888:8888 # Prometheus metrics exposed by the Collector
+      - 8889:8889 # Prometheus exporter metrics
+      - 13133:13133 # health_check extension
+      - 4317:4317 # OTLP gRPC receiver
+      - 4318:4318 # OTLP http receiver
+      - 55679:55679 # zpages extension
+
+  prometheus:
+    image: prom/prometheus:latest
+    volumes:
+      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
+    ports:
+      - "9090:9090"
+    depends_on: [otel-collector]
+
+  grafana:
+    image: grafana/grafana:latest
+    ports:
+      - "3000:3000"
+    depends_on: [prometheus]

crates/turborepo-otel/examples/local-collector/otel-collector.yml

@@ -0,0 +1,21 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+      http:
+        endpoint: 0.0.0.0:4318
+
+exporters:
+  debug:
+    verbosity: detailed
+  prometheus:
+    endpoint: "0.0.0.0:8889"
+    resource_to_telemetry_conversion:
+      enabled: true
+
+service:
+  pipelines:
+    metrics:
+      receivers: [otlp]
+      exporters: [debug, prometheus]

crates/turborepo-otel/examples/local-collector/prometheus.yml

@@ -0,0 +1,7 @@
+global:
+  scrape_interval: 5s
+
+scrape_configs:
+  - job_name: "otel-collector"
+    static_configs:
+      - targets: ["otel-collector:8889"]

crates/turborepo-otel/src/lib.rs

@@ -11,7 +11,8 @@ use opentelemetry::{
 use opentelemetry_otlp::{WithExportConfig, WithHttpConfig, WithTonicConfig};
 use opentelemetry_sdk::{
     Resource,
-    metrics::{PeriodicReader, SdkMeterProvider, Temporality},
+    metrics::{SdkMeterProvider, Temporality, periodic_reader_with_async_runtime},
+    runtime::Tokio,
 };
 use opentelemetry_semantic_conventions::resource::SERVICE_NAME;
 use thiserror::Error;
@@ -138,6 +139,15 @@ impl Handle {
         let meter = provider.meter("turborepo");
         let instruments = Arc::new(create_instruments(&meter));
 
+        tracing::debug!(
+            target: "turborepo_otel",
+            "initialized otel exporter: endpoint={} protocol={:?} run_summary={} task_details={}",
+            config.endpoint,
+            config.protocol,
+            config.metrics.run_summary,
+            config.metrics.task_details
+        );
+
         Ok(Self {
             inner: Arc::new(HandleInner {
                 provider,
@@ -148,6 +158,14 @@ impl Handle {
     }
 
     pub fn record_run(&self, payload: &RunMetricsPayload) {
+        tracing::debug!(
+            target: "turborepo_otel",
+            "record_run payload: run_id={} attempted={} failed={} cached={}",
+            payload.run_id,
+            payload.attempted_tasks,
+            payload.failed_tasks,
+            payload.cached_tasks
+        );
         if self.inner.metrics.run_summary {
             self.inner.instruments.record_run_summary(payload);
         }
@@ -157,6 +175,7 @@ impl Handle {
     }
 
     pub fn shutdown(self) {
+        tracing::debug!(target = "turborepo_otel", "shutting down otel exporter");
         match Arc::try_unwrap(self.inner) {
             Ok(inner) => {
                 if let Err(err) = inner.provider.shutdown() {
@@ -174,6 +193,13 @@ impl Handle {
 
 impl Instruments {
     fn record_run_summary(&self, payload: &RunMetricsPayload) {
+        tracing::debug!(
+            target: "turborepo_otel",
+            "record_run_summary run_id={} duration_ms={} attempted={}",
+            payload.run_id,
+            payload.duration_ms,
+            payload.attempted_tasks
+        );
         let attrs = build_run_attributes(payload);
         self.run_duration.record(payload.duration_ms, &attrs);
         self.run_attempted.add(payload.attempted_tasks, &attrs);
@@ -182,6 +208,12 @@ impl Instruments {
     }
 
     fn record_task_details(&self, payload: &RunMetricsPayload) {
+        tracing::debug!(
+            target: "turborepo_otel",
+            "record_task_details run_id={} task_count={}",
+            payload.run_id,
+            payload.tasks.len()
+        );
         let base_attrs = build_run_attributes(payload);
         for task in payload.tasks.iter() {
             let mut attrs = base_attrs.clone();
@@ -256,7 +288,7 @@ fn build_provider(config: &Config) -> Result<SdkMeterProvider, Error> {
         }
     };
 
-    let reader = PeriodicReader::builder(exporter)
+    let reader = periodic_reader_with_async_runtime::PeriodicReader::builder(exporter, Tokio)
         .with_interval(Duration::from_secs(15))
         .build();