OTLP MetricExporter deadlock issue

[cijothomas]

https://github.com/open-telemetry/opentelemetry-rust/blob/main/opentelemetry-otlp/src/exporter/tonic/metrics.rs#L58

Tonic MetricExporter uses std::Mutex to protect inner client, and the lock is held across await points. This can cause deadlock. The fix would be to use tokio::Mutex, but then shutdown/cleanup cannot be easily done, like already faced by LogExporter. But that is better than having code that deadlocks.

Opening an issue to track fixing this.

[cijothomas]

Looks like same issue in HTTP also https://github.com/open-telemetry/opentelemetry-rust/blob/main/opentelemetry-otlp/src/exporter/http/mod.rs#L271

[fraillt]

I might be missing something, but current code doesn't hold the lock across the await point... Given that inner is accessed in two places (export and shutdown), I pretty sure that code is correct and it's not possible to deadlock... What do I miss here?

[cijothomas]

.and_then(|g| match &*g { Some(client) => Ok(Arc::clone(client)), _ => Err(OTelSdkError::AlreadyShutdown), })?;

if the guard is dropped here itself, then yes, this looks correct. (I was incorrectly assuming the lock guard is not dropped until export() method ends..

[jakubdyszkiewicz]

Hey @cijothomas 👋

I need to dynamically reconfigure the exporter during binary execution, but the SDK seems to be stuck on shutdown. Is this the same issue?

Here is a minimal repro

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_otel_recorder_shutdown() -> anyhow::Result<()> {
        let telemetry_url = Url::parse("http://localhost:8090/otlp/v1/metrics").expect("should parse");
        let exporter = opentelemetry_otlp::MetricExporter::builder()
            .with_http()
            .with_endpoint(telemetry_url)
            .with_protocol(Protocol::HttpBinary)
            .with_timeout(Duration::from_secs(1))
            .with_http_client(reqwest::Client::builder().build().expect("should build"))
            .build()
            .expect("should build");

        let reader = PeriodicReader::builder(exporter, runtime::Tokio)
            .with_interval(Duration::from_secs(10))
            .build();

        let sdk = SdkMeterProvider::builder()
            .with_reader(reader)
            .build();

        let _ = sdk.shutdown(); // called on drop(sdk);

        Ok(())
    }
}

The test never finishes, it's stuck on the shutdown().

[brendanhay]

I've run into the same issue as @jakubdyszkiewicz with shutdown() completely blocking the program exit, not only in tests, but anywhere the tokio current thread runtime is used. You can omit shutdown() and Dropping the SdkMeterProvider will also block program exit.

FWIW #[tokio::test] uses the current_thread runtime by default. If you use #[tokio::test(flavor = "multi_thread")] with the appropriate tokio features shutdown() will make progress and exit.

@jakubdyszkiewicz tangentially, due to how the MetricExporter::builder() works the timeout specified to with_timeout will not be passed to the http client specified in with_http_client. The timeout is only used if one of the internal feature gated http clients is created/used.

[cijothomas]

Can you share a minimal repro of the shutdown issue? (do include the cargo.toml file as well so we can which the feature flags that are active.)

[scottgerring]

Have confirmed the behaviour from @jakubdyszkiewicz's reproduce above; this is what's happening because the test is using a single-threaded tokio runtime:

fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
      // ...
      let (sender, receiver) = oneshot::channel();
      inner.message_sender.try_send(Message::Shutdown(sender))?;
      
      // We end up blocked here, waiting for our message to come back.
      // But the worker task that would send the response is a tokio task
      // that needs this same thread to run - deadlock.
      let shutdown_result = futures_executor::block_on(receiver)?;  // line 417
      // ...
  }

As far as I can tell this is known behaviour the project warns against (but a bit of a footgun!) :

https://github.com/open-telemetry/opentelemetry-rust/blob/65969fd40e328049ddabe5d55a588b2d55dcc4af/opentelemetry-sdk/src/metrics/periodic_reader.rs?plain=1#L103-L105

https://github.com/open-telemetry/opentelemetry-rust/blob/65969fd40e328049ddabe5d55a588b2d55dcc4af/opentelemetry-sdk/src/logs/batch_log_processor.rs?plain=1#L99-L102

The advice we give is to use the runtime:CurrentRuntime runtime:

https://github.com/open-telemetry/opentelemetry-rust/blob/65969fd40e328049ddabe5d55a588b2d55dcc4af/opentelemetry-sdk/src/trace/span_processor_with_async_runtime.rs?plain=1#L31-L44

Applied to the test above, that would look like this, which does not deadlock:

 #[tokio::test]
  async fn test_otel_recorder_shutdown() {
      let exporter = opentelemetry_otlp::MetricExporter::builder()
          .with_http()
          .with_endpoint("http://localhost:8090/otlp/v1/metrics")
          .with_protocol(Protocol::HttpBinary)
          .with_timeout(Duration::from_secs(1))
          .with_http_client(reqwest::Client::builder().build().expect("should build"))
          .build()
          .expect("should build");

      // FIX: Use TokioCurrentThread instead of Tokio
      // note that this requires the rt-tokio-current-thread feature on opentelemetry_sdk
      let reader = PeriodicReader::builder(exporter, runtime::TokioCurrentThread)
          .with_interval(Duration::from_secs(10))
          .build();

      let sdk = SdkMeterProvider::builder()
          .with_reader(reader)
          .build();

      let _ = sdk.shutdown(); // No longer deadlocks
  }

better failure mode?

Why don't we detect in the shutdown that we're running this unsupported combination of things, and actively shift the work off onto a thread? Something like this ...

fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
      // ... 

      let on_current_thread_runtime = tokio::runtime::Handle::try_current()
          .map(|h| matches!(h.runtime_flavor(), tokio::runtime::RuntimeFlavor::CurrentThread))
          .unwrap_or(false) && not_using_otel_tokio_current_thread_runtime; // Need to get at this too 
       
      let shutdown_result = if on_current_thread_runtime {
          // Single-threaded runtime - move the wait to a separate thread, otherwise we'll deadlock
          warn!("Consider please switch to runtime::TokioCurrentThread!"); 
          std::thread::scope(|s| {
              s.spawn(|| futures_executor::block_on(receiver)).join().unwrap()
          })
      } else {
          // Current behavior for all other cases
          futures_executor::block_on(receiver)
      }
      .map_err(|err| OTelSdkError::InternalFailure(err.to_string()))?;

      // ... rest of shutdown ...
  }

[taisho6339]

let on_current_thread_runtime = tokio::runtime::Handle::try_current()
          .map(|h| matches!(h.runtime_flavor(), tokio::runtime::RuntimeFlavor::CurrentThread))
          .unwrap_or(false) && not_using_otel_tokio_current_thread_runtime; // Need to get at this too 

This behavior is not specific to the CurrentThread mode, but the multithread mode with a thread also causes the issue. For example, if you run your application on k8s attached with 1 core CPU, the number of threads should be single even if your specify multi-thread mode.

[scottgerring]

Good point; we’d need to guard against pool size rather than mode and update the documentation I highlighted also.

I’ll hack this up Monday and see where we land.

[scottgerring]

Suggestion above doesn't work; when we join on the thread we create to shift the work off of the tokio runtime ... we block the tokio runtime. Trying a couple other ideas.

[bantonsson]

So from looking at the code a bit I only see one option, and that is that the Runtime trait has all the necessary things including block_on. Any of the tokio variants must have their own tokio::runtime::Handle, either provided by the user or created from defaults, and use that to do spawn, block_on, etc.

[scottgerring]

Here's a bit of a sketch. What if:

• We add block_on to the runtime, so that our shutdowns can defer the blocking decision to the runtime itself (per @bantonsson 's comment above)
• We combine both tokio runtime variants into one that does the right thing - e.g. if it is a single threaded runtime, it spawns a thread for blocking work, if it is not, it doesn't

The latter is a breaking change but helps to address the situation we have today - where a user doesn't realise they need to opt into the OTel current thread runtime (because they are running with a single vCPU or have configured the tokio single-threaded runtime) - by instead making this decision automatically.

@cijothomas @bantonsson what do you think of an approach like this? I'm not convinced i've thought through all the interactions yet. There could be other options (e.g., make these methods async, and push the problem outwards).

[bantonsson]

@scottgerring I think that this proposal is the only sane way forward and there should be constructors that either create a tokio runtime themselves, or get a runtime handle, so they can use an existing one.

• We add block_on to the runtime, so that our shutdowns can defer the blocking decision to the runtime itself (per @bantonsson 's comment above)
• We combine both tokio runtime variants into one that does the right thing - e.g. if it is a single threaded runtime, it spawns a thread for blocking work, if it is not, it doesn't