Merge pull request #47 from frigus02/compression

gzip compression for requests

Author: frigus02

CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+- Use gzip compression (`Content-Encoding: gzip`) for `POST /v2/track` HTTP request.
+
 ## [0.19.0] - 2022-01-23
 
 - Upgrade to `v0.17.0` of `opentelemetry`.

Cargo.toml

@@ -34,6 +34,7 @@ metrics = ["opentelemetry/metrics", "ureq"]
 async-trait = "0.1"
 bytes = "1"
 chrono = "0.4"
+flate2 = "1"
 http = "0.2"
 once_cell = "1"
 opentelemetry = "0.17"

src/lib.rs

@@ -511,6 +511,13 @@ pub enum Error {
     #[error("serializing upload request failed with {0}")]
     UploadSerializeRequest(serde_json::Error),
 
+    /// Application Insights telemetry data failed serialize or compress. Telemetry reporting failed
+    /// because of this.
+    ///
+    /// Note: This is an error in this crate. If you spot this, please open an issue.
+    #[error("compressing upload request failed with {0}")]
+    UploadCompressRequest(std::io::Error),
+
     /// Application Insights telemetry response failed to deserialize from JSON.
     ///
     /// Telemetry reporting may have worked. But since we could not look into the response, we

src/uploader.rs

@@ -1,9 +1,11 @@
 use crate::{models::Envelope, Error, HttpClient};
 use bytes::Bytes;
+use flate2::{write::GzEncoder, Compression};
 use http::{Request, Response, Uri};
 use serde::Deserialize;
 #[cfg(feature = "metrics")]
 use std::io::Read;
+use std::io::Write;
 
 const STATUS_OK: u16 = 200;
 const STATUS_PARTIAL_CONTENT: u16 = 206;
@@ -33,9 +35,10 @@ pub(crate) async fn send(
     endpoint: &Uri,
     items: Vec<Envelope>,
 ) -> Result<(), Error> {
-    let payload = serde_json::to_vec(&items).map_err(Error::UploadSerializeRequest)?;
+    let payload = serialize_request_body(items)?;
     let request = Request::post(endpoint)
         .header(http::header::CONTENT_TYPE, "application/json")
+        .header(http::header::CONTENT_ENCODING, "gzip")
         .body(payload)
         .expect("request should be valid");
 
@@ -50,11 +53,12 @@ pub(crate) async fn send(
 /// Sends a telemetry items to the server.
 #[cfg(feature = "metrics")]
 pub(crate) fn send_sync(endpoint: &Uri, items: Vec<Envelope>) -> Result<(), Error> {
-    let payload = serde_json::to_vec(&items).map_err(Error::UploadSerializeRequest)?;
+    let payload = serialize_request_body(items)?;
 
     // TODO Implement retries
     let response = match ureq::post(&endpoint.to_string())
         .set(http::header::CONTENT_TYPE.as_str(), "application/json")
+        .set(http::header::CONTENT_ENCODING.as_str(), "gzip")
         .send_bytes(&payload)
     {
         Ok(response) => response,
@@ -79,6 +83,20 @@ pub(crate) fn send_sync(endpoint: &Uri, items: Vec<Envelope>) -> Result<(), Erro
     )
 }
 
+fn serialize_request_body(items: Vec<Envelope>) -> Result<Vec<u8>, Error> {
+    // Weirdly gzip_encoder.write_all(serde_json::to_vec()) seems to be faster than
+    // serde_json::to_writer(gzip_encoder). In a local test operating on items that result in
+    // ~13MiB of JSON, this is what I've seen:
+    // gzip_encoder.write_all(serde_json::to_vec()): 159ms
+    // serde_json::to_writer(gzip_encoder):          247ms
+    let serialized = serde_json::to_vec(&items).map_err(Error::UploadSerializeRequest)?;
+    let mut gzip_encoder = GzEncoder::new(Vec::new(), Compression::default());
+    gzip_encoder
+        .write_all(&serialized)
+        .map_err(Error::UploadCompressRequest)?;
+    gzip_encoder.finish().map_err(Error::UploadCompressRequest)
+}
+
 fn handle_response(response: Response<Bytes>) -> Result<(), Error> {
     match response.status().as_u16() {
         STATUS_OK => Ok(()),

tests/http_requests.rs

@@ -206,6 +206,7 @@ mod tick {
 }
 
 mod format {
+    use flate2::read::GzDecoder;
     use http::Request;
     use regex::Regex;
 
@@ -250,7 +251,9 @@ mod format {
     }
 
     fn pretty_print_json(body: &[u8]) -> String {
-        let json: serde_json::Value = serde_json::from_slice(body).expect("body is valid json");
+        let gzip_decoder = GzDecoder::new(body);
+        let json: serde_json::Value =
+            serde_json::from_reader(gzip_decoder).expect("body is valid json");
         serde_json::to_string_pretty(&json).unwrap()
     }
 }

tests/snapshots/http_requests__traces_batch_async_std.snap

@@ -1,12 +1,13 @@
 ---
 source: tests/http_requests.rs
-assertion_line: 131
+assertion_line: 100
 expression: traces_batch_async_std
 
 ---
 POST /v2/track HTTP/1.1
 host: dc.services.visualstudio.com
 content-type: application/json
+content-encoding: gzip
 
 [
   {

tests/snapshots/http_requests__traces_batch_tokio.snap

@@ -1,12 +1,13 @@
 ---
 source: tests/http_requests.rs
-assertion_line: 145
+assertion_line: 114
 expression: traces_batch_tokio
 
 ---
 POST /v2/track HTTP/1.1
 host: dc.services.visualstudio.com
 content-type: application/json
+content-encoding: gzip
 
 [
   {

tests/snapshots/http_requests__traces_simple.snap

@@ -7,6 +7,7 @@ expression: traces_simple
 POST /v2/track HTTP/1.1
 host: dc.services.visualstudio.com
 content-type: application/json
+content-encoding: gzip
 
 [
   {
@@ -92,6 +93,7 @@ content-type: application/json
 POST /v2/track HTTP/1.1
 host: dc.services.visualstudio.com
 content-type: application/json
+content-encoding: gzip
 
 [
   {