Import simple telemetry implementation without native_deps and language interface implementation

Cargo.toml

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

ddcommon/Cargo.toml

@@ -0,0 +1,18 @@
+# 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.
+
+[package]
+edition = "2018"
+license = "Apache-2.0"
+name = "ddcommon"
+version = "0.5.0-rc.1"
+
+[lib]
+crate-type = ["lib"]
+
+[dependencies]
+lazy_static = "1.4"
+regex = "1.5"
+
+[dev-dependencies]
+maplit = "1.0"

ddcommon/NOTICE

@@ -0,0 +1 @@
+../NOTICE

ddcommon/src/lib.rs

@@ -0,0 +1,4 @@
+// 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.
+
+pub mod container_id;

ddprof-exporter/Cargo.toml

@@ -16,9 +16,7 @@ bytes = "1.0"
 chrono = "0.4"
 futures = "0.3"
 http = "0.2"
-lazy_static = "1.4"
 libc = "0.2"
-regex = "1.5"
 hyper = { version = "0.14", features = ["http1", "client", "tcp", "stream"], default-features = false }
 tokio = { version = "1.8", features = ["rt", "macros"]}
 tokio-rustls = { version = "0.23" }
@@ -34,6 +32,7 @@ rustls-native-certs = { version = "0.6" }
 hyper-rustls = { version = "0.23", default-features = false, features = ["native-tokio", "http1", "tls12"] }
 hex = "0.4"
 hyper-multipart-rfc7578 = "0.7.0"
+ddcommon = { path = "../ddcommon" }
 
 [dev-dependencies]
 maplit = "1.0"

ddprof-exporter/src/lib.rs

@@ -16,7 +16,6 @@ use tokio::runtime::Runtime;
 use tokio_util::sync::CancellationToken;
 
 mod connector;
-mod container_id;
 mod errors;
 pub mod tag;
 
@@ -220,7 +219,7 @@ impl ProfileExporterV3 {
             );
         }
 
-        if let Some(container_id) = container_id::get_container_id() {
+        if let Some(container_id) = ddcommon::container_id::get_container_id() {
             builder = builder.header(DATADOG_CONTAINER_ID_HEADER, container_id);
         }
 

ddtelemetry/Cargo.toml

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

ddtelemetry/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 = ddtelemetry::build_full(&mut header).await;
+
+    println!(
+        "Payload to be sent: {}",
+        serde_json::to_string_pretty(&telemetry).unwrap()
+    );
+
+    ddtelemetry::push_telemetry(&telemetry).await?;
+
+    println!("Telemetry submitted correctly");
+    Ok(())
+}

ddtelemetry/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 ddtelemetry::{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();
+    });
+}

ddtelemetry/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)
+}

ddtelemetry/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
+    }
+}

ddtelemetry/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,
+        }
+    }
+}

ddtelemetry/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),
+}