Import simple telemetry implementation without native_deps and language interface implementation

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "ddprof-exporter",
     "ddprof-ffi",
     "ddprof-profiles",
+    "telemetry",
 ]
 
 [profile.dev]

ddprof-exporter/src/lib.rs

@@ -14,7 +14,7 @@ use hyper_multipart_rfc7578::client::multipart;
 use tokio::runtime::Runtime;
 
 mod connector;
-mod container_id;
+pub mod container_id;
 mod errors;
 
 #[cfg(unix)]

rustfmt.toml

@@ -0,0 +1 @@
+format_macro_matchers = true

telemetry/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+edition = "2018"
+name = "telemetry"
+license = "Apache 2.0"
+version = "0.0.1"
+
+[dependencies]
+anyhow = { version = "1.0" }
+lazy_static = { version = "1.4" }
+regex = { version = "1" }
+reqwest = { version = "0.11.4", features = [
+    "blocking",
+    "json",
+    "rustls-tls",
+], default-features = false }
+http = "0.2"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = { version = "1.0" }
+sys-info = { version = "0.9.0" }
+uuid = { version = "0.8.2", features = ["v4"] }
+futures = { version = "0.3" }
+ddprof-exporter = { path = "../ddprof-exporter"}
+
+[dev-dependencies]
+tokio = { version = "1.17", features=["macros"] }

telemetry/examples/tm-ping.rs

@@ -0,0 +1,19 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+// Simple worker that sends app-started telemetry request to the backend then exits
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let mut header = Default::default();
+    let telemetry = telemetry::build_full(&mut header).await;
+
+    println!(
+        "Payload to be sent: {}",
+        serde_json::to_string_pretty(&telemetry).unwrap()
+    );
+
+    telemetry::push_telemetry(&telemetry).await?;
+
+    println!("Telemetry submitted correctly");
+    Ok(())
+}

telemetry/examples/tm-worker-test.rs

@@ -0,0 +1,62 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+use telemetry::{data, worker};
+
+macro_rules! timeit {
+    ($op_name:literal, $op:block) => {{
+        let start = std::time::Instant::now();
+        let res = $op;
+        let delta = std::time::Instant::now().duration_since(start);
+        println!(
+            concat!($op_name, " took {} ms"),
+            delta.as_secs_f64() * 1000.0
+        );
+        res
+    }};
+}
+
+fn main() {
+    let handle = worker::TelemetryWorkerBuilder::new(
+        "paul-mac".into(),
+        "test_rust".into(),
+        "rust".into(),
+        "1.56".into(),
+        "none".into(),
+    )
+    .run();
+
+    handle.send_start().unwrap();
+    std::thread::sleep(std::time::Duration::from_secs(10));
+
+    handle
+        .add_log(
+            "init.log",
+            "Hello there!".into(),
+            data::LogLevel::Debug,
+            None,
+        )
+        .unwrap();
+    handle
+        .add_log(
+            "init.log",
+            "Another log, with the same logging identifier".into(),
+            data::LogLevel::Debug,
+            None,
+        )
+        .unwrap();
+    handle
+        .add_log(
+            "exception.log",
+            "Something really bad happened".into(),
+            data::LogLevel::Error,
+            Some("At line 56".into()),
+        )
+        .unwrap();
+
+    // About 200ms (the time it takes to send a app-closing request)
+    timeit!("shutdown", {
+        handle.send_stop().unwrap();
+        handle.wait_for_shutdown();
+    });
+}

telemetry/src/comms/mod.rs

@@ -0,0 +1,29 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+use reqwest::{header, Body, IntoUrl, Response};
+
+// TODO: extract the reqwest to allow exchange for alternative implementations, in cases like wasm
+pub async fn request<B: Into<Body>, T: IntoUrl>(
+    url: T,
+    body: B,
+    api_key: Option<&str>,
+) -> anyhow::Result<Response> {
+    let client = reqwest::ClientBuilder::new()
+        .danger_accept_invalid_certs(true)
+        .build()?;
+    let mut req = client
+        .post(url)
+        .header(
+            header::CONTENT_TYPE,
+            header::HeaderValue::from_static("application/json"),
+        )
+        .body(body);
+    if let Some(api_key) = api_key {
+        req = req.header("DD-API-KEY", api_key)
+    }
+
+    let res = client.execute(req.build()?).await?;
+
+    Ok(res)
+}

telemetry/src/config.rs

@@ -0,0 +1,94 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+use lazy_static::lazy_static;
+use std::env;
+
+pub const DEFAULT_DD_SITE: &str = "datadoghq.com";
+pub const PROD_INTAKE_FORMAT_PREFIX: &str = "https://instrumentation-telemetry-intake";
+
+pub const STAGING_INTAKE: &str = "https://all-http-intake.logs.datad0g.com";
+const DIRECT_TELEMETRY_URL_PATH: &str = "/api/v2/apmtelemetry";
+const AGENT_TELEMETRY_URL_PATH: &str = "/telemetry/proxy/api/v2/apmtelemetry";
+
+const DEFAULT_AGENT_HOST: &str = "localhost";
+const DEFAULT_AGENT_PORT: u16 = 8126;
+
+pub struct Config {
+    api_key: Option<String>,
+    #[allow(dead_code)]
+    agent_url: String,
+    telemetry_url: String,
+    telemetry_debug_logging_enabled: bool,
+}
+
+fn get_agent_base_url() -> String {
+    let agent_port = env::var("DD_AGENT_PORT")
+        .ok()
+        .and_then(|p| p.parse::<u16>().ok())
+        .unwrap_or(DEFAULT_AGENT_PORT);
+    let agent_host = env::var("DD_AGENT_HOST").unwrap_or_else(|_| String::from(DEFAULT_AGENT_HOST));
+
+    format!("http://{}:{}", agent_host, agent_port)
+}
+
+fn get_intake_base_url() -> String {
+    //TODO: support dd_site and additional endpoitns configuration
+    if let Some(url) = env::var("DD_APM_TELEMETRY_DD_URL")
+        .ok()
+        .filter(|s| !s.is_empty())
+    {
+        return url;
+    }
+
+    if let Ok(dd_site) = env::var("DD_SITE") {
+        if dd_site.is_empty() {
+            format!("{}.{}", PROD_INTAKE_FORMAT_PREFIX, DEFAULT_DD_SITE)
+        } else {
+            format!("{}.{}", PROD_INTAKE_FORMAT_PREFIX, dd_site)
+        }
+    } else {
+        String::from(STAGING_INTAKE)
+    }
+}
+
+impl Config {
+    pub fn get() -> &'static Self {
+        lazy_static! {
+            static ref CFG: Config = Config::read_env_config();
+        }
+        &CFG
+    }
+    pub fn read_env_config() -> Self {
+        let api_key = env::var("DD_API_KEY").ok().filter(|p| !p.is_empty());
+        let agent_url = get_agent_base_url();
+        let telemetry_url = if api_key.is_some() {
+            let telemetry_intake_base_url = get_intake_base_url();
+            format!("{}{}", telemetry_intake_base_url, DIRECT_TELEMETRY_URL_PATH)
+        } else {
+            format!("{}{}", &agent_url, AGENT_TELEMETRY_URL_PATH)
+        };
+        Config {
+            api_key,
+            agent_url,
+            telemetry_url,
+            telemetry_debug_logging_enabled: false,
+        }
+    }
+
+    pub fn is_telemetry_debug_logging_enabled(&self) -> bool {
+        self.telemetry_debug_logging_enabled
+    }
+
+    pub fn api_key(&self) -> Option<&str> {
+        self.api_key.as_deref()
+    }
+
+    pub fn telemetry_url(&self) -> &str {
+        &self.telemetry_url
+    }
+
+    pub fn is_direct(&self) -> bool {
+        self.api_key.is_some() // If API key is provided call directly
+    }
+}

telemetry/src/data/common.rs

@@ -0,0 +1,70 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+use serde::{Deserialize, Serialize};
+
+use crate::data::*;
+
+#[derive(Serialize, Deserialize, Debug)]
+pub enum ApiVersion {
+    #[serde(rename = "v1")]
+    V1,
+}
+
+#[derive(Serialize, Debug)]
+pub struct Telemetry<'a> {
+    pub api_version: ApiVersion,
+    pub tracer_time: u64,
+    pub runtime_id: &'a str,
+    pub seq_id: u64,
+    pub application: &'a Application,
+    pub host: &'a Host,
+    #[serde(flatten)]
+    pub payload: Payload,
+}
+
+#[derive(Serialize, Deserialize, Default, Debug)]
+pub struct Application {
+    pub service_name: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub service_version: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub env: Option<String>,
+    pub language_name: String,
+    pub language_version: String,
+    pub tracer_version: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub runtime_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub runtime_version: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub runtime_patches: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Default, Debug)]
+pub struct Host {
+    pub hostname: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub container_id: Option<String>,
+    pub os: Option<String>,
+    pub os_version: Option<String>,
+    pub kernel_name: Option<String>,
+    pub kernel_release: Option<String>,
+    pub kernel_version: Option<String>,
+}
+
+impl Application {
+    pub fn new_rust_app() -> Self {
+        Self {
+            service_name: String::from(env!("CARGO_PKG_NAME")),
+            service_version: Some(String::from(env!("CARGO_PKG_VERSION"))),
+            env: None,
+            language_name: String::from("rust"),
+            language_version: String::from("n/a"),
+            tracer_version: String::from("n/a"),
+            runtime_name: None,
+            runtime_version: None,
+            runtime_patches: None,
+        }
+    }
+}

telemetry/src/data/metrics.rs

@@ -0,0 +1,21 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct CounterGauge {
+    metric: String,
+    points: Vec<(u64, f64)>,
+    tags: Vec<String>,
+    common: bool,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(tag = "type")]
+pub enum Metric {
+    #[serde(rename = "gauge")]
+    Gauge(CounterGauge),
+    #[serde(rename = "gauge")]
+    Counter(CounterGauge),
+}

telemetry/src/data/mod.rs

@@ -0,0 +1,11 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+mod common;
+mod payloads;
+
+pub use common::*;
+pub use payload::*;
+pub use payloads::*;
+pub mod metrics;
+pub mod payload;

telemetry/src/data/payload.rs

@@ -0,0 +1,24 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2021-Present Datadog, Inc.
+
+use crate::data::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(tag = "request_type", content = "payload")]
+pub enum Payload {
+    #[serde(rename = "app-started")]
+    AppStarted(AppStarted),
+    #[serde(rename = "app-dependencies-loaded")]
+    AppDependenciesLoaded(AppDependenciesLoaded),
+    #[serde(rename = "app-integrations-change")]
+    AppIntegrationsChange(AppIntegrationsChange),
+    #[serde(rename = "app-heartbeat")]
+    AppHearbeat(()),
+    #[serde(rename = "app-closing")]
+    AppClosing(()),
+    #[serde(rename = "generate-metrics")]
+    GenerateMetrics(GenerateMetrics),
+    #[serde(rename = "logs")]
+    Logs(Vec<Log>),
+}