Add x-ms-client-request-id pipeline policy (Azure#2406)

* Add `x-ms-client-request-id` pipeline policy

Resolves Azure#2240. This also supports setting a custom header name and, thus, must be added by each client to the per-call pipeline policies instead of being added automatically.

* Resolve meeting feedback

* Always add the policy, but support custom policies.
* For now, caller can add the existing `ClientRequestId` header to `ClientMethodOptions::context`.

* Resolve PR feedback

Author: heaths

.vscode/cspell.json

@@ -64,6 +64,7 @@
     "seekable",
     "servicebus",
     "stylesheet",
+    "telemetered",
     "typespec",
     "undelete",
     "upvote",

Cargo.lock

@@ -7,6 +7,7 @@
 - Added `azure_core::process::Executor` to run commands asynchronously.
   The `tokio` feature is disabled by default so `std::process::Command` is used; otherwise, if enabled, `tokio::process::Command` is used.
 - Added `http` module containing all functions, modules, and types from `typespec_client_core::http`.
+- Added `azure_core::http::policies::ClientRequestIdPolicy` to every pipeline. Client libraries can add with custom header name instead.
 - Moved `Pager` from `typespec_client_core::http` to `azure_core::http` module since it is Azure-specific.
 - Re-exported `Body`, `Request`, and `RequestContent` from `http::request` module.
 - Re-exported `create_enum`, `create_extensible_enum` macros from `typespec_client_core`.

sdk/core/azure_core/CHANGELOG.md

@@ -33,6 +33,7 @@ typespec_client_core = { workspace = true, features = ["http", "json"] }
 rustc_version.workspace = true
 
 [dev-dependencies]
+azure_core_test.workspace = true
 azure_identity.workspace = true
 azure_security_keyvault_secrets.path = "../../keyvault/azure_security_keyvault_secrets"
 thiserror.workspace = true

sdk/core/azure_core/Cargo.toml

@@ -1,18 +1,9 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-use typespec_client_core::setters;
-
 /// Telemetry options.
 #[derive(Clone, Debug, Default)]
 pub struct TelemetryOptions {
-    /// Optional application ID to telemetry.
-    pub(crate) application_id: Option<String>,
-}
-
-impl TelemetryOptions {
-    setters! {
-        #[doc = "Set the application ID to telemetry."]
-        application_id: String => Some(application_id),
-    }
+    /// Set the application ID in the `User-Agent` header that can be telemetered
+    pub application_id: Option<String>,
 }

sdk/core/azure_core/src/http/options/telemetry.rs

@@ -1,8 +1,13 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
+use super::policies::ClientRequestIdPolicy;
 use crate::http::{options::TelemetryOptions, policies::TelemetryPolicy};
-use std::{ops::Deref, sync::Arc};
+use std::{
+    any::{Any, TypeId},
+    ops::Deref,
+    sync::Arc,
+};
 use typespec_client_core::http::{self, policies::Policy};
 
 /// Execution pipeline.
@@ -12,15 +17,16 @@ use typespec_client_core::http::{self, policies::Policy};
 /// 1. Client library-specified per-call policies are executed. Per-call policies can fail and bail out of the pipeline
 ///    immediately.
 /// 2. User-specified per-call policies are executed.
-/// 3. Telemetry policy.
-/// 4. Retry policy. It allows to re-execute the following policies.
+/// 3. Built-in per-call policies are executed. If a [`ClientRequestIdPolicy`] was not added to `per_call_policies`,
+///    the default will be added automatically.
+/// 4. The retry policy is executed. It allows to re-execute the following policies.
 /// 5. Client library-specified per-retry policies. Per-retry polices are always executed at least once but are re-executed
 ///    in case of retries.
 /// 6. User-specified per-retry policies are executed.
-/// 7. Authorization policy. Authorization can depend on the HTTP headers and/or the request body so it
+/// 7. The authorization policy is executed. Authorization can depend on the HTTP headers and/or the request body so it
 ///    must be executed right before sending the request to the transport. Also, the authorization
 ///    can depend on the current time so it must be executed at every retry.
-/// 8. Transport policy. Transport policy is always the last policy and is the policy that
+/// 8. The transport policy is executed. Transport policy is always the last policy and is the policy that
 ///    actually constructs the `Response` to be passed up the pipeline.
 ///
 /// A pipeline is immutable. In other words a policy can either succeed and call the following
@@ -45,13 +51,20 @@ impl Pipeline {
     ) -> Self {
         let mut per_call_policies = per_call_policies.clone();
 
+        if per_call_policies
+            .iter()
+            .all(|policy| TypeId::of::<ClientRequestIdPolicy>() != policy.type_id())
+        {
+            per_call_policies.push(Arc::new(ClientRequestIdPolicy::default()));
+        }
+
         let telemetry_policy = TelemetryPolicy::new(
             crate_name,
             crate_version,
             // TODO: &options.telemetry.unwrap_or_default(),
             &TelemetryOptions::default(),
         );
-        per_call_policies.insert(0, Arc::new(telemetry_policy));
+        per_call_policies.push(Arc::new(telemetry_policy));
 
         Self(http::Pipeline::new(
             options,
@@ -68,3 +81,133 @@ impl Deref for Pipeline {
         &self.0
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        http::{
+            headers::{self, HeaderName, Headers},
+            policies::Policy,
+            request::options::ClientRequestId,
+            Context, Method, Request, Response, StatusCode, TransportOptions,
+        },
+        Bytes,
+    };
+    use azure_core_test::http::MockHttpClient;
+    use futures::FutureExt as _;
+    use std::sync::Arc;
+    use typespec_client_core::http::ClientOptions;
+
+    #[tokio::test]
+    async fn pipeline_with_custom_client_request_id_policy() {
+        // Arrange
+        const CUSTOM_HEADER_NAME: &str = "x-custom-request-id";
+        const CUSTOM_HEADER: HeaderName = HeaderName::from_static(CUSTOM_HEADER_NAME);
+        const CLIENT_REQUEST_ID: &str = "custom-request-id";
+
+        let mut ctx = Context::new();
+        ctx.insert(ClientRequestId::new(CLIENT_REQUEST_ID.to_string()));
+
+        let transport = TransportOptions::new(Arc::new(MockHttpClient::new(|req| {
+            async {
+                // Assert
+                let header_value = req
+                    .headers()
+                    .get_optional_str(&CUSTOM_HEADER)
+                    .expect("Custom header should be present");
+                assert_eq!(
+                    header_value, CLIENT_REQUEST_ID,
+                    "Custom header value should match the client request ID"
+                );
+
+                Ok(Response::from_bytes(
+                    StatusCode::Ok,
+                    Headers::new(),
+                    Bytes::new(),
+                ))
+            }
+            .boxed()
+        })));
+        let options = ClientOptions {
+            transport: Some(transport),
+            ..Default::default()
+        };
+
+        let per_call_policies: Vec<Arc<dyn Policy>> =
+            vec![
+                Arc::new(ClientRequestIdPolicy::with_header_name(CUSTOM_HEADER_NAME))
+                    as Arc<dyn Policy>,
+            ];
+        let per_retry_policies = vec![];
+
+        let pipeline = Pipeline::new(
+            Some("test-crate"),
+            Some("1.0.0"),
+            options,
+            per_call_policies,
+            per_retry_policies,
+        );
+
+        let mut request = Request::new("https://example.com".parse().unwrap(), Method::Get);
+
+        // Act
+        pipeline
+            .send::<()>(&ctx, &mut request)
+            .await
+            .expect("Pipeline execution failed");
+    }
+
+    #[tokio::test]
+    async fn pipeline_without_client_request_id_policy() {
+        // Arrange
+        const CLIENT_REQUEST_ID: &str = "default-request-id";
+
+        let mut ctx = Context::new();
+        ctx.insert(ClientRequestId::new(CLIENT_REQUEST_ID.to_string()));
+
+        let transport = TransportOptions::new(Arc::new(MockHttpClient::new(|req| {
+            async {
+                // Assert
+                let header_value = req
+                    .headers()
+                    .get_optional_str(&headers::CLIENT_REQUEST_ID)
+                    .expect("Default header should be present");
+                assert_eq!(
+                    header_value, CLIENT_REQUEST_ID,
+                    "Default header value should match the client request ID"
+                );
+
+                Ok(Response::from_bytes(
+                    StatusCode::Ok,
+                    Headers::new(),
+                    Bytes::new(),
+                ))
+            }
+            .boxed()
+        })));
+        let options = ClientOptions {
+            transport: Some(transport),
+            ..Default::default()
+        };
+
+        let per_call_policies = vec![]; // No ClientRequestIdPolicy added
+        let per_retry_policies = vec![];
+
+        let pipeline = Pipeline::new(
+            Some("test-crate"),
+            Some("1.0.0"),
+            options,
+            per_call_policies,
+            per_retry_policies,
+        );
+
+        let mut request = Request::new("https://example.com".parse().unwrap(), Method::Get);
+
+        // Act
+        pipeline
+            .send::<()>(&ctx, &mut request)
+            .await
+            .expect("Pipeline execution failed");
+    }
+}