Observable Gauge Callback

[raaynaldo]

I'm still new in Rust, but I'm required to implement OTEL in my rust project. I'm trying to call an async function in the Observable Callback function, but I got this error message

thread 'main' panicked at /Users/zee678/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.1/src/runtime/blocking/shutdown.rs:51:21:
Cannot drop a runtime in a context where blocking is not allowed. This happens when a runtime is dropped from within an asynchronous context.
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

async fn create_observable_gauge() {
    let meter = global::meter("mylibraryname");
    let rt = Runtime::new().unwrap();

    let _observable_gauge = meter
        .f64_observable_gauge("create_observable_gauge")
        .with_description("Testing gauge")
        .with_unit("myunit")
        .with_callback(move |observer| {
            println!("with callback");

            // tokio::spawn(future)
            rt.block_on(async {
                let val = async_function().await; // I need to call this function to get the value and put in the observer
                println!("val hello: {}", val);
                observer.observe(val, &[KeyValue::new("new key from gauge", "myvalue1")]);
            });
        })
        .init();
}

Any thought could be helpful! Thank you!

[utpilla]

@raaynaldo Could you try using futures_executor::block_on() for calling that async function?

async fn create_observable_gauge() {
    let meter = global::meter("mylibraryname");
    
    let _observable_gauge = meter
        .f64_observable_gauge("create_observable_gauge")
        .with_description("Testing gauge")
        .with_unit("myunit")
        .with_callback(move |observer| {
            println!("with callback");

            let val = futures_executor::block_on(async_function());
            println!("val hello: {}", val);
            observer.observe(val, &[KeyValue::new("new key from gauge", "myvalue1")]);
        })
        .init();
}

[raaynaldo]

@utpilla, I tried this approach, but it looks like my app is stuck. I just want to make sure, are you referring to this crate? https://docs.rs/futures-executor/latest/futures_executor/fn.block_on.html

async fn observable_gauge() -> Result<impl Reply, Rejection> {
    let meter = global::meter("observable_gauge");

    let _observable_gauge = meter
        .f64_observable_gauge("observable_gauge")
        .with_callback(move |observer| {
            println!("with callback");

            let val = futures_executor::block_on(async_function());
            println!("val: {}", val);
            observer.observe(val, &[]);
        })
        .init();

    Ok((warp::reply::json(&"Ok")))
}

async fn async_function() -> f64 {
    println!("Task started... ");

    sleep(Duration::from_secs(2)).await;

    println!("Task completed after 3 seconds! ");

    10.0
}

I also couldn't call another API endpoint after I ran the observable_gauge function.

[cijothomas]

Unfortunately this is a limitation with current model. Using futures::block_on don't play well when you are inside an async runtime like tokio already. This is explained seanmonstar/reqwest#1215 (comment) and will be addressed by #2142 (Added a test to confirm it indeed fixes this issue)
Please take a look at tests starting with "async_inside_observable_callback" in #2142

[raaynaldo]

Thank you for the information @cijothomas !

[cijothomas]

See #2063 (reply in thread)

[cijothomas]

https://github.com/open-telemetry/opentelemetry-rust/pull/2428/files Take a look at this PR which shows how to do asynchronous calls from inside Observable. It does show using tokio also.