Moved otel support for polling into poller (#3279)

Removed custom code from `create_certificate` API code and moved the
corresponding logic into the `Poller`.

This involved adding an additional parameter to the `callback` for
`Poller::with_callback` that provides the `ctx` used for the pipeline.

Bonus: Instead of using a tuple to maintain the state of the "stream"
created by the `Poller`, encapsulate the state in a structure to make it
easier to modify the state of the "stream" moving forward (when there
were 3 fields in the state, it wasn't a huge deal, but with 5 fields, it
was getting unwieldy).

NOTE: This is still a draft PR, the next step is to work on similar
changes for pagers.

Author: LarryOsterman

.gitignore

@@ -20,6 +20,7 @@ vcpkg_installed/
 
 # Editor user customizations.
 .vscode/launch.json
+.vscode/mcp.json
 .idea/
 
 # Editor temp files.

sdk/core/azure_core/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Features Added
 
+- Added `Context::to_owned()` to create a newly owned `Context` from an existing `Context`.
 - Added `ItemIterator::continuation_token()` and `with_continuation_token()` to resume paging items. The current page is restarted until _after_ all items have been iterated.
 - Added `PipelineOptions::retry_status_codes` for configuring which status codes should trigger a retry.
 - Added `Response<T, F>::body(&self) -> &ResponseBody`.
@@ -14,10 +15,12 @@
 
 ### Breaking Changes
 
+- Added `Context` field to `PollerOptions`. Client methods which return `Poller` objects should accept a `PollerOptions` in their `method_options` field instead of a `ClientMethodOptions`.
 - Added `F: Format` type parameter to `Poller` and `PollerResult`.
 - Added `Format` associated type to `StatusMonitor`.
 - Added `Format::deserialize()` function to `Format` trait.
 - Added `S` type parameter to `xml::from_xml` congruent with `json::from_json()`.
+- Changed `PollerOptions::frequency` from `Option<Duration>` to `Duration`.
 - Moved deserializers and serializers for optional base64-encoded bytes to `base64::option` module. `base64` module now deserializes or serializes non-optional fields congruent with the `time` module.
 - Removed `constants` module.
 - Removed `credentials::DEFAULT_SCOPE_SUFFIX`.

sdk/core/azure_core/src/http/pager.rs

@@ -5,7 +5,11 @@
 
 use crate::{
     error::ErrorKind,
-    http::{headers::HeaderName, response::Response, Context, DeserializeWith, Format, JsonFormat},
+    http::{
+        headers::HeaderName, policies::create_public_api_span, response::Response, Context,
+        DeserializeWith, Format, JsonFormat,
+    },
+    tracing::{Span, SpanStatus},
 };
 use async_trait::async_trait;
 use futures::{stream::unfold, FutureExt, Stream};
@@ -314,7 +318,7 @@ impl<P: Page> ItemIterator<P> {
     /// }
     /// let url = "https://example.com/my_paginated_api".parse().unwrap();
     /// let mut base_req = Request::new(url, Method::Get);
-    /// let pager = ItemIterator::from_callback(move |next_link: PagerState<Url>, ctx| {
+    /// let pager = ItemIterator::from_callback(move |next_link: PagerState<Url>, ctx: Context| {
     ///     // The callback must be 'static, so you have to clone and move any values you want to use.
     ///     let pipeline = pipeline.clone();
     ///     let api_version = api_version.clone();
@@ -830,6 +834,7 @@ struct StreamState<'a, C, F> {
     make_request: F,
     continuation_token: Arc<Mutex<Option<String>>>,
     ctx: Context<'a>,
+    added_span: bool,
 }
 
 fn iter_from_callback<
@@ -855,10 +860,12 @@ where
             make_request,
             continuation_token,
             ctx,
+            added_span: false,
         },
         |mut stream_state| async move {
             // Get the `continuation_token` to pick up where we left off, or None for the initial page,
             // but don't override the terminal `State::Done`.
+
             if stream_state.state != State::Done {
                 let result = match stream_state.continuation_token.lock() {
                     Ok(next_token) => match next_token.as_deref() {
@@ -888,6 +895,12 @@ where
             let result = match stream_state.state {
                 State::Init => {
                     tracing::debug!("initial page request");
+                    // At the very start of polling, create a span for the entire request, and attach it to the context
+                    let span = create_public_api_span(&stream_state.ctx, None, None);
+                    if let Some(ref s) = span {
+                        stream_state.added_span = true;
+                        stream_state.ctx = stream_state.ctx.with_value(s.clone());
+                    }
                     (stream_state.make_request)(PagerState::Initial, stream_state.ctx.clone()).await
                 }
                 State::More(n) => {
@@ -905,6 +918,17 @@ where
             };
             let (item, next_state) = match result {
                 Err(e) => {
+                    if stream_state.added_span {
+                        if let Some(span) = stream_state.ctx.value::<Arc<dyn Span>>() {
+                            // Mark the span as an error with an appropriate description.
+                            span.set_status(SpanStatus::Error {
+                                description: e.to_string(),
+                            });
+                            span.set_attribute("error.type", e.kind().to_string().into());
+                            span.end();
+                        }
+                    }
+
                     stream_state.state = State::Done;
                     return Some((Err(e), stream_state));
                 }
@@ -923,6 +947,15 @@ where
                     if let Ok(mut token) = stream_state.continuation_token.lock() {
                         *token = None;
                     }
+                    // When the result is done, finalize the span. Note that we only do that if we created the span in the first place,
+                    // otherwise it is the responsibility of the caller to end their span.
+                    if stream_state.added_span {
+                        if let Some(span) = stream_state.ctx.value::<Arc<dyn Span>>() {
+                            // P is unconstrained, so it's not possible to retrieve the status code for now.
+
+                            span.end();
+                        }
+                    }
                     (Ok(response), State::Done)
                 }
             };