Report profiling data in v2.4 intake format; compress files (#53)

* Report profiling data in v2.4 intake format

This was tested with both the Ruby (agent and agentless modes) and
the PHP profilers.

This also introduces a breaking API change: the
`ddog_ProfileExporter_build` / `ProfileExporter::build` functions
now take two additional arguments -- the profiling library name
and version.

Other than that change, using the v2.4 intake format is transparent
to the libdatadog users.

Thanks to @morrisonlevi for pairing with me on this.

Note that this does not (yet) include support for including
attributes in the reporting data. I'll leave that for a separate PR.

* Adjust profiler_tags encoding

* Remove data[] from the name of the files

* Add lz4 compression to files

* Don't compress event.json

* Rename profile_library_[name|version] to profiling_library_[name|version]

* Test for DD-EVP-ORIGIN*

* Move profiling_library_{name,version} to constructor

* Document some intake details

Co-authored-by: Levi Morrison <levi.morrison@datadoghq.com>

examples/ffi/exporter.cpp

@@ -90,8 +90,13 @@ int main(int argc, char *argv[]) {
 
   ddog_PushTagResult_drop(tag_result);
 
-  ddog_NewProfileExporterResult exporter_new_result =
-      ddog_ProfileExporter_new(DDOG_CHARSLICE_C("native"), &tags, endpoint);
+  ddog_NewProfileExporterResult exporter_new_result = ddog_ProfileExporter_new(
+      DDOG_CHARSLICE_C("exporter-example"),
+      DDOG_CHARSLICE_C("1.2.3"),
+      DDOG_CHARSLICE_C("native"),
+      &tags,
+      endpoint
+  );
   ddog_Vec_tag_drop(tags);
 
   if (exporter_new_result.tag == DDOG_NEW_PROFILE_EXPORTER_RESULT_ERR) {
@@ -109,8 +114,14 @@ int main(int argc, char *argv[]) {
 
   ddog_Slice_file files = {.ptr = files_, .len = sizeof files_ / sizeof *files_};
 
-  ddog_Request *request = ddog_ProfileExporter_build(exporter, encoded_profile->start,
-                                                     encoded_profile->end, files, nullptr, 30000);
+  ddog_Request *request = ddog_ProfileExporter_build(
+    exporter,
+    encoded_profile->start,
+    encoded_profile->end,
+    files,
+    nullptr,
+    30000
+  );
 
   ddog_CancellationToken *cancel = ddog_CancellationToken_new();
   ddog_CancellationToken *cancel_for_background_thread = ddog_CancellationToken_clone(cancel);

profiling-ffi/src/exporter.rs

@@ -116,15 +116,25 @@ unsafe fn try_to_endpoint(endpoint: Endpoint) -> anyhow::Result<exporter::Endpoi
 #[must_use]
 #[export_name = "ddog_ProfileExporter_new"]
 pub extern "C" fn profile_exporter_new(
+    profiling_library_name: CharSlice,
+    profiling_library_version: CharSlice,
     family: CharSlice,
     tags: Option<&ddcommon_ffi::Vec<Tag>>,
     endpoint: Endpoint,
 ) -> NewProfileExporterResult {
     match || -> anyhow::Result<ProfileExporter> {
+        let library_name = unsafe { profiling_library_name.to_utf8_lossy() }.into_owned();
+        let library_version = unsafe { profiling_library_version.to_utf8_lossy() }.into_owned();
         let family = unsafe { family.to_utf8_lossy() }.into_owned();
         let converted_endpoint = unsafe { try_to_endpoint(endpoint)? };
         let tags = tags.map(|tags| tags.iter().map(Tag::clone).collect());
-        ProfileExporter::new(family, tags, converted_endpoint)
+        ProfileExporter::new(
+            library_name,
+            library_version,
+            family,
+            tags,
+            converted_endpoint,
+        )
     }() {
         Ok(exporter) => NewProfileExporterResult::Ok(Box::into_raw(Box::new(exporter))),
         Err(err) => NewProfileExporterResult::Err(err.into()),
@@ -316,6 +326,14 @@ mod test {
     use crate::exporter::*;
     use ddcommon_ffi::Slice;
 
+    fn profiling_library_name() -> CharSlice<'static> {
+        CharSlice::from("dd-trace-foo")
+    }
+
+    fn profiling_library_version() -> CharSlice<'static> {
+        CharSlice::from("1.2.3")
+    }
+
     fn family() -> CharSlice<'static> {
         CharSlice::from("native")
     }
@@ -334,22 +352,34 @@ mod test {
         let host = Tag::new("host", "localhost").expect("static tags to be valid");
         tags.push(host);
 
-        let result = profile_exporter_new(family(), Some(&tags), endpoint_agent(endpoint()));
+        let result = profile_exporter_new(
+            profiling_library_name(),
+            profiling_library_version(),
+            family(),
+            Some(&tags),
+            endpoint_agent(endpoint()),
+        );
 
         match result {
             NewProfileExporterResult::Ok(exporter) => unsafe {
                 profile_exporter_delete(Some(Box::from_raw(exporter)))
             },
             NewProfileExporterResult::Err(message) => {
-                std::mem::drop(message);
+                drop(message);
                 panic!("Should not occur!")
             }
         }
     }
 
     #[test]
     fn test_build() {
-        let exporter_result = profile_exporter_new(family(), None, endpoint_agent(endpoint()));
+        let exporter_result = profile_exporter_new(
+            profiling_library_name(),
+            profiling_library_version(),
+            family(),
+            None,
+            endpoint_agent(endpoint()),
+        );
 
         let exporter = match exporter_result {
             NewProfileExporterResult::Ok(exporter) => unsafe {

profiling/Cargo.toml

@@ -28,8 +28,11 @@ hyper = {version = "0.14", features = ["client"], default-features = false}
 hyper-multipart-rfc7578 = "0.7.0"
 indexmap = "1.8"
 libc = "0.2"
+lz4_flex = { version = "0.9", default-features = false, features = ["std", "safe-encode", "frame"] }
+mime = "0.3.16"
 mime_guess = {version = "2.0", default-features = false}
 percent-encoding = "2.1"
 prost = "0.10"
+serde_json = {version = "1.0"}
 tokio = {version = "1.8", features = ["rt", "macros"]}
 tokio-util = "0.7.1"

profiling/src/exporter/config.rs

@@ -61,7 +61,7 @@ pub fn agentless<AsStrRef: AsRef<str>, IntoCow: Into<Cow<'static, str>>>(
     site: AsStrRef,
     api_key: IntoCow,
 ) -> anyhow::Result<Endpoint> {
-    let intake_url: String = format!("https://intake.profile.{}/v1/input", site.as_ref());
+    let intake_url: String = format!("https://intake.profile.{}/api/v2/profile", site.as_ref());
 
     Ok(Endpoint {
         url: Uri::from_str(intake_url.as_str())?,

profiling/src/exporter/mod.rs

@@ -10,6 +10,10 @@ pub use chrono::{DateTime, Utc};
 pub use ddcommon::tag::Tag;
 pub use hyper::Uri;
 use hyper_multipart_rfc7578::client::multipart;
+use lz4_flex::frame::FrameEncoder;
+use mime;
+use serde_json::json;
+use std::io::Write;
 use tokio::runtime::Runtime;
 use tokio_util::sync::CancellationToken;
 
@@ -40,6 +44,8 @@ pub struct ProfileExporter {
     exporter: Exporter,
     endpoint: Endpoint,
     family: Cow<'static, str>,
+    profiling_library_name: Cow<'static, str>,
+    profiling_library_version: Cow<'static, str>,
     tags: Option<Vec<Tag>>,
 }
 
@@ -106,15 +112,24 @@ impl Request {
 }
 
 impl ProfileExporter {
-    pub fn new<IntoCow: Into<Cow<'static, str>>>(
-        family: IntoCow,
+    pub fn new<F, N, V>(
+        profiling_library_name: N,
+        profiling_library_version: V,
+        family: F,
         tags: Option<Vec<Tag>>,
         endpoint: Endpoint,
-    ) -> anyhow::Result<ProfileExporter> {
+    ) -> anyhow::Result<ProfileExporter>
+    where
+        F: Into<Cow<'static, str>>,
+        N: Into<Cow<'static, str>>,
+        V: Into<Cow<'static, str>>,
+    {
         Ok(Self {
             exporter: Exporter::new()?,
             endpoint,
             family: family.into(),
+            profiling_library_name: profiling_library_name.into(),
+            profiling_library_version: profiling_library_version.into(),
             tags,
         })
     }
@@ -130,30 +145,58 @@ impl ProfileExporter {
     ) -> anyhow::Result<Request> {
         let mut form = multipart::Form::default();
 
-        form.add_text("version", "3");
-        form.add_text("start", start.format("%Y-%m-%dT%H:%M:%S%.9fZ").to_string());
-        form.add_text("end", end.format("%Y-%m-%dT%H:%M:%S%.9fZ").to_string());
-        form.add_text("family", self.family.as_ref());
-
-        for tags in self.tags.as_ref().iter().chain(additional_tags.iter()) {
-            for tag in tags.iter() {
-                form.add_text("tags[]", tag.to_string());
-            }
+        // combine tags and additional_tags
+        let mut tags_profiler = String::new();
+        let other_tags = additional_tags.into_iter();
+        for tag in self.tags.iter().chain(other_tags).flatten() {
+            tags_profiler.push_str(tag.as_ref());
+            tags_profiler.push(',');
         }
+        tags_profiler.pop(); // clean up the trailing comma
+
+        let attachments: Vec<String> = files.iter().map(|file| file.name.to_owned()).collect();
+
+        let event = json!({
+            "attachments": attachments,
+            "tags_profiler": tags_profiler,
+            "start": start.format("%Y-%m-%dT%H:%M:%S%.9fZ").to_string(),
+            "end": end.format("%Y-%m-%dT%H:%M:%S%.9fZ").to_string(),
+            "family": self.family.as_ref(),
+            "version": "4",
+        })
+        .to_string();
+
+        form.add_reader_file_with_mime(
+            // Intake does not look for filename=event.json, it looks for name=event.
+            "event",
+            // this one shouldn't be compressed
+            Cursor::new(event),
+            "event.json",
+            mime::APPLICATION_JSON,
+        );
 
         for file in files {
-            form.add_reader_file(
-                format!("data[{}]", file.name),
-                Cursor::new(file.bytes.to_owned()),
-                file.name,
-            )
+            let mut encoder = FrameEncoder::new(Vec::new());
+            encoder.write_all(file.bytes)?;
+            let encoded = encoder.finish()?;
+            /* The Datadog RFC examples strip off the file extension, but the exact behavior isn't
+             * specified. This does the simple thing of using the filename without modification for
+             * the form name because intake does not care about these name of the form field for
+             * these attachments.
+             */
+            form.add_reader_file(file.name, Cursor::new(encoded), file.name)
         }
 
         let builder = self
             .endpoint
             .into_request_builder(concat!("DDProf/", env!("CARGO_PKG_VERSION")))?
             .method(http::Method::POST)
-            .header("Connection", "close");
+            .header("Connection", "close")
+            .header("DD-EVP-ORIGIN", self.profiling_library_name.as_ref())
+            .header(
+                "DD-EVP-ORIGIN-VERSION",
+                self.profiling_library_version.as_ref(),
+            );
 
         Ok(
             Request::from(form.set_body_convert::<hyper::Body, multipart::Body>(builder)?)

profiling/tests/form.rs

@@ -55,10 +55,18 @@ mod tests {
 
     #[test]
     fn multipart_agent() {
+        let profiling_library_name = "dd-trace-foo";
+        let profiling_library_version = "1.2.3";
         let base_url = "http://localhost:8126".parse().expect("url to parse");
         let endpoint = config::agent(base_url).expect("endpoint to construct");
-        let exporter = ProfileExporter::new("php", Some(default_tags()), endpoint)
-            .expect("exporter to construct");
+        let exporter = ProfileExporter::new(
+            profiling_library_name,
+            profiling_library_version,
+            "php",
+            Some(default_tags()),
+            endpoint,
+        )
+        .expect("exporter to construct");
 
         let request = multipart(&exporter);
 
@@ -69,27 +77,50 @@ mod tests {
 
         let actual_headers = request.headers();
         assert!(!actual_headers.contains_key("DD-API-KEY"));
+        assert_eq!(
+            actual_headers.get("DD-EVP-ORIGIN").unwrap(),
+            profiling_library_name
+        );
+        assert_eq!(
+            actual_headers.get("DD-EVP-ORIGIN-VERSION").unwrap(),
+            profiling_library_version
+        );
     }
 
     #[test]
     fn multipart_agentless() {
+        let profiling_library_name = "dd-trace-foo";
+        let profiling_library_version = "1.2.3";
         let api_key = "1234567890123456789012";
         let endpoint = config::agentless("datadoghq.com", api_key).expect("endpoint to construct");
-        let exporter = ProfileExporter::new("php", Some(default_tags()), endpoint)
-            .expect("exporter to construct");
+        let exporter = ProfileExporter::new(
+            profiling_library_name,
+            profiling_library_version,
+            "php",
+            Some(default_tags()),
+            endpoint,
+        )
+        .expect("exporter to construct");
 
         let request = multipart(&exporter);
 
         assert_eq!(
             request.uri().to_string(),
-            "https://intake.profile.datadoghq.com/v1/input"
+            "https://intake.profile.datadoghq.com/api/v2/profile"
         );
 
         let actual_headers = request.headers();
 
+        assert_eq!(actual_headers.get("DD-API-KEY").unwrap(), api_key);
+
+        assert_eq!(
+            actual_headers.get("DD-EVP-ORIGIN").unwrap(),
+            profiling_library_name
+        );
+
         assert_eq!(
-            actual_headers.get("DD-API-KEY").expect("api key to exist"),
-            api_key
+            actual_headers.get("DD-EVP-ORIGIN-VERSION").unwrap(),
+            profiling_library_version
         );
     }
 }