Add TelemetryPolicy

sdk/core/Cargo.toml

@@ -13,24 +13,27 @@ categories = ["api-bindings"]
 edition = "2018"
 
 [dependencies]
+async-trait = "0.1"
+bytes = "1.0"
 chrono = "0.4"
-http = "0.2"
+dyn-clone = "1.0"
 futures = "0.3"
+http = "0.2"
 hyper = { version = "0.14", optional = true }
+hyper-rustls = { version = "0.22", optional = true }
 log = "0.4"
-thiserror = "1.0"
+oauth2 = "4.0.0"
+rand = "0.7"
+reqwest = { version = "0.11", features = ["stream"], optional = true }
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
+thiserror = "1.0"
 url = "2.2"
 uuid = { version = "0.8" }
-bytes = "1.0"
-hyper-rustls = { version = "0.22", optional = true }
-async-trait = "0.1"
-oauth2 = "4.0.0"
-reqwest = { version = "0.11", features = ["stream"], optional = true }
-rand = "0.7"
-dyn-clone = "1.0"
+
+[build-dependencies]
+rustc_version = "0.3.3"
 
 [dev-dependencies]
 env_logger = "0.8"

sdk/core/build.rs

@@ -0,0 +1,9 @@
+use rustc_version::version;
+
+fn main() {
+    let version = match version() {
+        Ok(version) => version.to_string(),
+        Err(_) => "unknown".to_string(),
+    };
+    println!("cargo:rustc-env=AZSDK_RUSTC_VERSION={}", version);
+}

sdk/core/src/client_options.rs

@@ -0,0 +1,12 @@
+use crate::policies::{Policy, TelemetryOptions};
+use std::sync::Arc;
+
+/// Options passed clients to customer policies, telemetry, etc.
+#[derive(Clone, Debug, Default)]
+pub struct ClientOptions {
+    // TODO: Expose retry options and transport overrides.
+    pub per_call_policies: Vec<Arc<dyn Policy>>,
+    pub per_retry_policies: Vec<Arc<dyn Policy>>,
+
+    pub telemetry: TelemetryOptions,
+}

sdk/core/src/lib.rs

@@ -10,6 +10,7 @@ extern crate serde_derive;
 mod macros;
 
 mod bytes_stream;
+pub mod client_options;
 mod constants;
 mod context;
 mod errors;
@@ -35,6 +36,7 @@ use std::fmt::Debug;
 use uuid::Uuid;
 
 pub use bytes_stream::*;
+pub use client_options::ClientOptions;
 pub use constants::*;
 pub use context::Context;
 pub use errors::*;

sdk/core/src/pipeline.rs

@@ -1,17 +1,20 @@
-use crate::policies::{Policy, PolicyResult};
-use crate::{Context, Request, Response};
+use crate::policies::{Policy, PolicyResult, TelemetryPolicy};
+use crate::{ClientOptions, Context, Request, Response};
 use std::sync::Arc;
 
 /// Execution pipeline.
 ///
 /// A pipeline follows a precise flow:
 ///
-/// 1. Per call policies are executed. Per call policies can fail and bail out of the pipeline
+/// 1. Client library-specified per-call policies are executed. Per-call policies can fail and bail out of the pipeline
 ///    immediately.
-/// 2. Retry policy. It allows to reexecute the following policies.
-/// 3. Per retry policies. Per retry polices are always executed at least once but are reexecuted
+/// 2. User-specified per-call policies are executed.
+/// 3. Telemetry policy.
+/// 4. Retry policy. 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.
-/// 4. Transport policy. Transtport policy is always the last policy and is the policy that
+/// 6. User-specified per-retry policies are executed.
+/// 7. Transport policy. 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
@@ -24,18 +27,38 @@ pub struct Pipeline {
 }
 
 impl Pipeline {
+    /// Creates a new pipeline given the client library crate name and version,
+    /// alone with user-specified and client library-specified policies.
+    ///
+    /// Crates can simply pass `option_env!("CARGO_PKG_NAME")` and `option_env!("CARGO_PKG_VERSION")` for the
+    /// `crate_name` and `crate_version` arguments respectively.
     pub fn new(
+        crate_name: Option<&'static str>,
+        crate_version: Option<&'static str>,
+        options: &ClientOptions,
         per_call_policies: Vec<Arc<dyn Policy>>,
         retry: Arc<dyn Policy>,
         per_retry_policies: Vec<Arc<dyn Policy>>,
         transport_policy: Arc<dyn Policy>,
     ) -> Self {
-        let mut pipeline =
-            Vec::with_capacity(per_call_policies.len() + per_retry_policies.len() + 2);
+        let mut pipeline: Vec<Arc<dyn Policy>> = Vec::with_capacity(
+            options.per_call_policies.len()
+                + per_call_policies.len()
+                + options.per_retry_policies.len()
+                + per_retry_policies.len()
+                + 3,
+        );
 
         pipeline.extend_from_slice(&per_call_policies);
+        pipeline.extend_from_slice(&options.per_call_policies);
+        pipeline.push(Arc::new(TelemetryPolicy::new(
+            crate_name,
+            crate_version,
+            &options.telemetry,
+        )));
         pipeline.push(retry);
         pipeline.extend_from_slice(&per_retry_policies);
+        pipeline.extend_from_slice(&options.per_retry_policies);
         pipeline.push(transport_policy);
 
         Self { pipeline }

sdk/core/src/policies/mod.rs

@@ -1,10 +1,12 @@
 mod retry_policies;
+mod telemetry_policy;
 mod transport;
 
 use crate::{Context, Request, Response};
 pub use retry_policies::*;
 use std::error::Error;
 use std::sync::Arc;
+pub use telemetry_policy::*;
 pub use transport::*;
 
 pub type PolicyResult<T> = Result<T, Box<dyn Error + Send + Sync>>;

sdk/core/src/policies/telemetry_policy.rs

@@ -0,0 +1,131 @@
+use crate::policies::{Policy, PolicyResult};
+use crate::{Context, Request, Response};
+
+use http::{header::USER_AGENT, HeaderValue};
+use std::env::consts::{ARCH, OS};
+use std::sync::Arc;
+
+#[derive(Clone, Debug, Default)]
+pub struct TelemetryOptions {
+    pub application_id: Option<String>,
+}
+
+#[derive(Clone, Debug)]
+pub struct TelemetryPolicy {
+    header: String,
+}
+
+/// Sets the User-Agent header with useful information in a typical format for Azure SDKs.
+///
+/// Client libraries should create a `TelemetryPolicy` using `option_env!()` like so:
+/// ```
+/// use azure_core::policies::{TelemetryOptions, TelemetryPolicy};
+/// let policy = TelemetryPolicy::new(option_env!("CARGO_PKG_NAME"), option_env!("CARGO_PKG_VERSION"), &TelemetryOptions::default());
+/// ```
+impl<'a> TelemetryPolicy {
+    pub fn new(
+        crate_name: Option<&'a str>,
+        crate_version: Option<&'a str>,
+        options: &TelemetryOptions,
+    ) -> Self {
+        Self::new_with_rustc_version(
+            crate_name,
+            crate_version,
+            option_env!("AZSDK_RUSTC_VERSION"),
+            options,
+        )
+    }
+
+    fn new_with_rustc_version(
+        crate_name: Option<&'a str>,
+        crate_version: Option<&'a str>,
+        rustc_version: Option<&'static str>,
+        options: &TelemetryOptions,
+    ) -> Self {
+        const UNKNOWN: &'static str = "unknown";
+        let mut crate_name = crate_name.unwrap_or(UNKNOWN);
+        let crate_version = crate_version.unwrap_or(UNKNOWN);
+        let rustc_version = rustc_version.unwrap_or(UNKNOWN);
+        let platform_info = format!("({}; {}; {})", rustc_version, OS, ARCH,);
+
+        if let Some(name) = crate_name.strip_prefix("azure_") {
+            crate_name = name;
+        }
+
+        let header = match &options.application_id {
+            Some(application_id) => format!(
+                "{} azsdk-rust-{}/{} {}",
+                application_id, crate_name, crate_version, platform_info
+            ),
+            None => format!(
+                "azsdk-rust-{}/{} {}",
+                crate_name, crate_version, platform_info
+            ),
+        };
+
+        TelemetryPolicy { header: header }
+    }
+}
+
+#[async_trait::async_trait]
+impl Policy for TelemetryPolicy {
+    async fn send(
+        &self,
+        ctx: &mut Context,
+        request: &mut Request,
+        next: &[Arc<dyn Policy>],
+    ) -> PolicyResult<Response> {
+        request
+            .headers_mut()
+            .insert(USER_AGENT, HeaderValue::from_str(&self.header).unwrap());
+
+        next[0].send(ctx, request, &next[1..]).await
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_without_application_id() {
+        let policy = TelemetryPolicy::new_with_rustc_version(
+            Some("azure_test"), // Tests that "azure_" is removed.
+            Some("1.2.3"),
+            Some("4.5.6"),
+            &TelemetryOptions::default(),
+        );
+        assert_eq!(
+            policy.header,
+            format!("azsdk-rust-test/1.2.3 (4.5.6; {}; {})", OS, ARCH)
+        );
+    }
+
+    #[test]
+    fn test_with_application_id() {
+        let options = TelemetryOptions {
+            application_id: Some("my_app".to_string()),
+        };
+        let policy = TelemetryPolicy::new_with_rustc_version(
+            Some("test"),
+            Some("1.2.3"),
+            Some("4.5.6"),
+            &options,
+        );
+        assert_eq!(
+            policy.header,
+            format!("my_app azsdk-rust-test/1.2.3 (4.5.6; {}; {})", OS, ARCH)
+        );
+    }
+
+    #[test]
+    fn test_missing_env() {
+        // Would simulate if option_env!("CARGO_PKG_NAME"), for example, returned None.
+        let policy =
+            TelemetryPolicy::new_with_rustc_version(None, None, None, &TelemetryOptions::default());
+        assert_eq!(
+            policy.header,
+            format!("azsdk-rust-unknown/unknown (unknown; {}; {})", OS, ARCH)
+        )
+    }
+}

sdk/core/src/request.rs

@@ -45,6 +45,10 @@ impl Request {
         &self.headers
     }
 
+    pub fn headers_mut(&mut self) -> &mut HeaderMap {
+        &mut self.headers
+    }
+
     pub fn body(&self) -> &Body {
         &self.body
     }

sdk/cosmos/src/clients/cosmos_client.rs

@@ -4,6 +4,7 @@ use crate::resources::ResourceType;
 use crate::{headers::*, CosmosError};
 use crate::{requests, ReadonlyString};
 
+use azure_core::client_options::ClientOptions;
 use azure_core::pipeline::Pipeline;
 use azure_core::policies::{LinearRetryPolicy, Policy, TransportOptions, TransportPolicy};
 use azure_core::Context;
@@ -32,6 +33,7 @@ pub struct CosmosClient {
 }
 /// TODO
 pub struct CosmosOptions {
+    options: ClientOptions,
     retry: Arc<dyn Policy>,
     transport: TransportOptions,
 }
@@ -40,6 +42,7 @@ impl CosmosOptions {
     /// TODO
     pub fn with_client(client: Arc<dyn HttpClient>) -> Self {
         Self {
+            options: ClientOptions::default(),
             retry: Arc::new(LinearRetryPolicy::default()), // this defaults to linear backoff
             transport: TransportOptions::new(client),
         }
@@ -52,6 +55,9 @@ fn new_pipeline_from_options(options: CosmosOptions) -> Pipeline {
     let per_retry_policies = Vec::new();
     let transport_policy = TransportPolicy::new(options.transport);
     Pipeline::new(
+        option_env!("CARGO_PKG_NAME"),
+        option_env!("CARGO_PKG_VERSION"),
+        &options.options,
         per_call_policies,
         options.retry,
         per_retry_policies,