Live metrics

Cargo.toml

@@ -30,6 +30,7 @@ reqwest-client-vendored-tls = ["opentelemetry-http/reqwest", "reqwest/native-tls
 reqwest-client-rustls = ["opentelemetry-http/reqwest", "reqwest/rustls-tls"]
 surf-client = ["opentelemetry-http/surf"]
 metrics = ["opentelemetry/metrics"]
+live-metrics = ["futures-channel", "futures-util", "sysinfo"]
 
 [dependencies]
 async-trait = "0.1"
@@ -38,6 +39,8 @@ chrono = "0.4"
 flate2 = "1"
 http = "0.2"
 once_cell = "1"
+futures-channel = { version = "0.3", default-features = false, optional = true }
+futures-util = { version = "0.3", default-features = false, optional = true }
 opentelemetry = "0.20"
 opentelemetry-http = "0.9"
 opentelemetry-semantic-conventions = "0.12"
@@ -46,14 +49,15 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde_repr = "0.1"
 thiserror = "1"
+sysinfo = { version = "0.29", optional = true }
 
 [dev-dependencies]
 async-std = { version = "1.10.0", features = ["attributes"] }
 doc-comment = "0.3.3"
 env_logger = "0.10.0"
 insta = "1.13.0"
 opentelemetry = { version = "0.20", features = ["rt-async-std", "rt-tokio", "rt-tokio-current-thread"] }
-opentelemetry-application-insights = { path = ".", features = ["reqwest-client", "reqwest-blocking-client"] }
+opentelemetry-application-insights = { path = ".", features = ["reqwest-client", "reqwest-blocking-client", "live-metrics"] }
 rand = "0.8.5"
 regex = "1.5.5"
 surf = "2.3.2"
@@ -80,6 +84,10 @@ required-features = ["reqwest-client"]
 name = "http_client_surf"
 required-features = ["surf-client", "opentelemetry/rt-async-std"]
 
+[[example]]
+name = "live_metrics"
+required-features = ["live-metrics"]
+
 [[example]]
 name = "metrics"
 required-features = ["metrics"]

examples/live_metrics.rs

@@ -0,0 +1,56 @@
+use opentelemetry::trace::{Span, SpanKind, Status, Tracer as _};
+use opentelemetry_semantic_conventions as semcov;
+use rand::{thread_rng, Rng};
+use std::{error::Error, time::Duration};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+    env_logger::init();
+
+    let tracer = opentelemetry_application_insights::new_pipeline_from_env()?
+        .with_client(reqwest::Client::new())
+        .with_live_metrics(true)
+        .install_batch(opentelemetry::runtime::Tokio);
+
+    print!("Simulating requests. Press Ctrl+C to stop.");
+
+    let mut rng = thread_rng();
+    loop {
+        let success = rng.gen_ratio(9, 10);
+        let _request = tracer
+            .span_builder("request")
+            .with_kind(SpanKind::Server)
+            .with_status(if success {
+                Status::Ok
+            } else {
+                Status::error("")
+            })
+            .with_attributes(vec![
+                semcov::trace::HTTP_REQUEST_METHOD.string("GET"),
+                semcov::trace::URL_SCHEME.string("https"),
+                semcov::trace::URL_PATH.string("/hello/world"),
+                semcov::trace::URL_QUERY.string("name=marry"),
+                semcov::trace::HTTP_RESPONSE_STATUS_CODE.i64(200),
+            ])
+            .start(&tracer);
+        {
+            let mut db = tracer
+                .span_builder("db")
+                .with_kind(SpanKind::Client)
+                .with_status(if success {
+                    Status::Ok
+                } else {
+                    Status::error("")
+                })
+                .start(&tracer);
+            if !success {
+                let err: Box<dyn std::error::Error> = "An error".into();
+                db.record_error(err.as_ref());
+            }
+
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}

src/connection_string.rs

@@ -1,12 +1,16 @@
 use std::{borrow::Cow, collections::HashMap, convert::TryInto, str::FromStr};
 
 pub(crate) const DEFAULT_BREEZE_ENDPOINT: &str = "https://dc.services.visualstudio.com";
+#[cfg(feature = "live-metrics")]
+pub(crate) const DEFAULT_LIVE_ENDPOINT: &str = "https://rt.services.visualstudio.com";
 const FIELDS_SEPARATOR: char = ';';
 const FIELD_KEY_VALUE_SEPARATOR: char = '=';
 
 #[derive(Debug)]
 pub(crate) struct ConnectionString {
     pub(crate) ingestion_endpoint: http::Uri,
+    #[cfg(feature = "live-metrics")]
+    pub(crate) live_endpoint: http::Uri,
     pub(crate) instrumentation_key: String,
 }
 
@@ -42,19 +46,35 @@ impl FromStr for ConnectionString {
             })
             .collect::<Result<_, _>>()?;
 
-        let ingestion_endpoint: http::Uri =
+        let prefix_suffix = result.remove("endpointsuffix").map(|endpoint_suffix| {
+            let location_prefix = result
+                .remove("location")
+                .map(|x| format!("{}.", x))
+                .unwrap_or_else(|| "".into());
+            (location_prefix, endpoint_suffix)
+        });
+
+        let ingestion_endpoint =
             if let Some(ingestion_endpoint) = result.remove("ingestionendpoint") {
                 sanitize_url(ingestion_endpoint)?
-            } else if let Some(endpoint_suffix) = result.remove("endpointsuffix") {
-                let location_prefix = result
-                    .remove("location")
-                    .map(|x| format!("{}.", x))
-                    .unwrap_or_else(|| "".into());
+            } else if let Some((location_prefix, endpoint_suffix)) = prefix_suffix.as_ref() {
                 sanitize_url(format!("https://{}dc.{}", location_prefix, endpoint_suffix))?
             } else {
                 http::Uri::from_static(DEFAULT_BREEZE_ENDPOINT)
             };
 
+        #[cfg(feature = "live-metrics")]
+        let live_endpoint = if let Some(live_endpoint) = result.remove("liveendpoint") {
+            sanitize_url(live_endpoint)?
+        } else if let Some((location_prefix, endpoint_suffix)) = prefix_suffix.as_ref() {
+            sanitize_url(format!(
+                "https://{}live.{}",
+                location_prefix, endpoint_suffix
+            ))?
+        } else {
+            http::Uri::from_static(DEFAULT_LIVE_ENDPOINT)
+        };
+
         if let Some(authorization) = result.remove("authorization") {
             if !authorization.eq_ignore_ascii_case("ikey") {
                 return Err(ParseError::UnsupportedAuthorization);
@@ -66,6 +86,8 @@ impl FromStr for ConnectionString {
 
         Ok(ConnectionString {
             ingestion_endpoint,
+            #[cfg(feature = "live-metrics")]
+            live_endpoint,
             instrumentation_key,
         })
     }
@@ -90,43 +112,56 @@ mod tests {
     use test_case::test_case;
 
     #[test_case(
-        "Authorization=ikey;InstrumentationKey=instr_key;IngestionEndpoint=ingest",
+        "Authorization=ikey;InstrumentationKey=instr_key;IngestionEndpoint=ingest;LiveEndpoint=live",
         "ingest",
+        #[cfg(feature = "live-metrics")] "live",
         "instr_key" ; "default")]
     #[test_case(
-        "Authorization=ikey;InstrumentationKey=instr_key;IngestionEndpoint= http://ingest/   ",
+        "Authorization=ikey;InstrumentationKey=instr_key;IngestionEndpoint= http://ingest/  ;LiveEndpoint= http://live/  ",
         "https://ingest",
+        #[cfg(feature = "live-metrics")] "https://live",
         "instr_key" ; "sanitize url")]
     #[test_case(
-        "Foo=1;InstrumentationKey=instr_key;Bar=2;IngestionEndpoint=ingest;Baz=3",
+        "Foo=1;InstrumentationKey=instr_key;Bar=2;IngestionEndpoint=ingest;LiveEndpoint=live;Baz=3",
         "ingest",
+        #[cfg(feature = "live-metrics")] "live",
         "instr_key" ; "ignore unknown fields")]
     #[test_case(
         "InstrumentationKey=instr_key",
         DEFAULT_BREEZE_ENDPOINT,
+        #[cfg(feature = "live-metrics")] DEFAULT_LIVE_ENDPOINT,
         "instr_key" ; "default endpoint")]
     #[test_case(
         "InstrumentationKey=instr_key;EndpointSuffix=ai.contoso.com",
         "https://dc.ai.contoso.com",
+        #[cfg(feature = "live-metrics")] "https://live.ai.contoso.com",
         "instr_key" ; "endpoint suffix")]
     #[test_case(
         "InstrumentationKey=instr_key;EndpointSuffix=ai.contoso.com;Location=westus2",
         "https://westus2.dc.ai.contoso.com",
+        #[cfg(feature = "live-metrics")] "https://westus2.live.ai.contoso.com",
         "instr_key" ; "endpoint suffix & location")]
     #[test_case(
-        "InstrumentationKey=instr_key;EndpointSuffix=ai.contoso.com;IngestionEndpoint=ingest",
+        "InstrumentationKey=instr_key;EndpointSuffix=ai.contoso.com;IngestionEndpoint=ingest;LiveEndpoint=live",
         "ingest",
+        #[cfg(feature = "live-metrics")] "live",
         "instr_key" ; "endpoint suffix & override")]
     fn parse_succeeds(
         connection_string: &'static str,
         expected_ingestion_endpoint: &'static str,
+        #[cfg(feature = "live-metrics")] expected_live_endpoint: &'static str,
         expected_instrumentation_key: &'static str,
     ) {
         let result: ConnectionString = connection_string.parse().unwrap();
         assert_eq!(
             http::Uri::try_from(expected_ingestion_endpoint).unwrap(),
             result.ingestion_endpoint
         );
+        #[cfg(feature = "live-metrics")]
+        assert_eq!(
+            http::Uri::try_from(expected_live_endpoint).unwrap(),
+            result.live_endpoint
+        );
         assert_eq!(
             expected_instrumentation_key.to_string(),
             result.instrumentation_key