Add TelemetryPolicy

Author: heaths

sdk/core/Cargo.toml

@@ -13,24 +13,26 @@ categories = ["api-bindings"]
 edition = "2018"
 
 [dependencies]
+async-trait = "0.1"
+bytes = "1.0"
 chrono = "0.4"
-http = "0.2"
+clap = "2.33.3"
+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 }
+rustc_version = "0.3.3"
 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"
 
 [dev-dependencies]
 env_logger = "0.8"

sdk/core/src/pipeline.rs

@@ -1,4 +1,4 @@
-use crate::policies::{Policy, PolicyResult};
+use crate::policies::{Policy, PolicyResult, TelemetryPolicy};
 use crate::{Context, Request, Response};
 use std::sync::Arc;
 
@@ -8,10 +8,10 @@ use std::sync::Arc;
 ///
 /// 1. 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. Retry policy. It allows to re-execute the following policies.
+/// 3. 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
+/// 4. 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
@@ -31,7 +31,10 @@ impl Pipeline {
         transport_policy: Arc<dyn Policy>,
     ) -> Self {
         let mut pipeline =
-            Vec::with_capacity(per_call_policies.len() + per_retry_policies.len() + 2);
+            Vec::<Arc<dyn Policy>>::with_capacity(per_call_policies.len() + per_retry_policies.len() + 3);
+
+        // TODO: Create pipeline from ClientOptions which should contain user-specified policies + client-added policies.
+        pipeline.push(Arc::new(TelemetryPolicy::default()));
 
         pipeline.extend_from_slice(&per_call_policies);
         pipeline.push(retry);

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,67 @@
+use crate::{Context, Request, Response};
+use crate::policies::{Policy, PolicyResult};
+
+use http::{HeaderValue, header::USER_AGENT};
+use rustc_version::{Version, version};
+use std::env::consts::{ARCH, OS};
+use std::sync::Arc;
+
+#[derive(Clone, Debug, Default)]
+pub struct TelemetryOptions {
+    application_id: Option<String>,
+}
+
+impl TelemetryOptions {
+    pub fn new(application_id: Option<String>) -> Self {
+        Self { application_id }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct TelemetryPolicy {
+    header: String,
+}
+
+const EMPTY_VERSION: Version = Version {
+    major: 0,
+    minor: 0,
+    patch: 0,
+    pre: Vec::new(),
+    build: Vec::new(),
+};
+
+impl TelemetryPolicy {
+    pub fn new(options: TelemetryOptions) -> Self {
+        let platform_info = format!("({}; {}; {})", version().unwrap_or(EMPTY_VERSION), OS, ARCH);
+        if let Some(application_id) = options.application_id {
+            TelemetryPolicy {
+                header: format!("{} azsdk-rust-{}/{} {}", application_id, clap::crate_name!(), clap::crate_version!(), platform_info),
+            }
+        } else {
+            TelemetryPolicy {
+                header: format!("azsdk-rust-{}/{} {}", clap::crate_name!(), clap::crate_version!(), platform_info),
+            }
+        }
+    }
+}
+
+impl Default for TelemetryPolicy {
+    fn default() -> Self {
+        TelemetryPolicy::new(TelemetryOptions::default())
+    }
+}
+
+#[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
+    }
+}

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
     }