Merge branch 'main' into feat/log-kvs

Author: cijothomas

RELEASING.md

@@ -16,7 +16,7 @@ A draft PR can be created, but before releasing consider the following:
 
 * Are there any pending pull requests which should be included in the next release?
   * Are they blockers?
-* Are there any unresolved issues which should be resolved before the next release?
+* Are there any unresolved issues which should be resolved before the next release? Check the release [blockers milestone](https://github.com/open-telemetry/opentelemetry-rust/milestones) for every release
 * Bring it up at a SIG meeting, this can usually get some of these questions answered sooner than later. It will also
   help establish a person to perform the release. Ideally this can be someone different each time to ensure that the
   process is documented.

opentelemetry-otlp/CHANGELOG.md

@@ -8,6 +8,10 @@
 
 ### Added
 
+- Aded `http/json` support for all signals ([#1585])
+
+[#1585]: https://github.com/open-telemetry/opentelemetry-rust/pull/1585
+
 - Added `DeltaTemporalitySelector` ([#1568])
 - Add `webkpi-roots` features to `reqwest` and `tonic` backends
 

opentelemetry-otlp/Cargo.toml

@@ -42,6 +42,7 @@ reqwest = { workspace = true, optional = true }
 http = { workspace = true, optional = true }
 serde = { workspace = true, features = ["derive"], optional = true }
 thiserror = { workspace = true }
+serde_json = { workspace = true, optional = true }
 
 [dev-dependencies]
 tokio-stream = { workspace = true, features = ["net"] }
@@ -58,7 +59,7 @@ metrics = ["opentelemetry/metrics", "opentelemetry_sdk/metrics", "opentelemetry-
 logs = ["opentelemetry/logs", "opentelemetry_sdk/logs", "opentelemetry-proto/logs"]
 
 # add ons
-serialize = ["serde"]
+serialize = ["serde", "serde_json"]
 
 default = ["grpc-tonic", "trace"]
 
@@ -71,6 +72,8 @@ tls-webkpi-roots = ["tls", "tonic/tls-webpki-roots"]
 
 # http binary
 http-proto = ["prost", "opentelemetry-http", "opentelemetry-proto/gen-tonic-messages", "http", "trace", "metrics"]
+# http json
+http-json = ["serde_json", "prost", "opentelemetry-http", "opentelemetry-proto/gen-tonic-messages", "opentelemetry-proto/with-serde", "http", "trace", "metrics"]
 reqwest-blocking-client = ["reqwest/blocking", "opentelemetry-http/reqwest"]
 reqwest-client = ["reqwest", "opentelemetry-http/reqwest"]
 reqwest-rustls = ["reqwest", "opentelemetry-http/reqwest-rustls"]

opentelemetry-otlp/examples/basic-otlp-http/Cargo.toml

@@ -9,7 +9,7 @@ publish = false
 once_cell = { workspace = true }
 opentelemetry = { path = "../../../opentelemetry" }
 opentelemetry_sdk = { path = "../../../opentelemetry-sdk", features = ["rt-tokio", "metrics", "logs"] }
-opentelemetry-otlp = { path = "../..", features = ["http-proto", "reqwest-client", "logs"] }
+opentelemetry-otlp = { path = "../..", features = ["http-proto", "http-json", "reqwest-client", "logs"] }
 opentelemetry-appender-tracing = { path = "../../../opentelemetry-appender-tracing", default-features = false}
 opentelemetry-semantic-conventions = { path = "../../../opentelemetry-semantic-conventions" }
 

opentelemetry-otlp/src/exporter/http/logs.rs

@@ -19,7 +19,7 @@ impl LogExporter for OtlpHttpClient {
                 _ => Err(LogError::Other("exporter is already shut down".into())),
             })?;
 
-        let (body, content_type) = build_body(batch)?;
+        let (body, content_type) = self.build_logs_export_body(batch)?;
         let mut request = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
@@ -51,24 +51,3 @@ impl LogExporter for OtlpHttpClient {
         let _ = self.client.lock().map(|mut c| c.take());
     }
 }
-
-#[cfg(feature = "http-proto")]
-fn build_body(logs: Vec<LogData>) -> LogResult<(Vec<u8>, &'static str)> {
-    use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
-    use prost::Message;
-
-    let req = ExportLogsServiceRequest {
-        resource_logs: logs.into_iter().map(Into::into).collect(),
-    };
-    let mut buf = vec![];
-    req.encode(&mut buf).map_err(crate::Error::from)?;
-
-    Ok((buf, "application/x-protobuf"))
-}
-
-#[cfg(not(feature = "http-proto"))]
-fn build_body(logs: Vec<LogData>) -> LogResult<(Vec<u8>, &'static str)> {
-    Err(LogsError::Other(
-        "No http protocol configured. Enable one via `http-proto`".into(),
-    ))
-}

opentelemetry-otlp/src/exporter/http/metrics.rs

@@ -21,7 +21,7 @@ impl MetricsClient for OtlpHttpClient {
                 _ => Err(MetricsError::Other("exporter is already shut down".into())),
             })?;
 
-        let (body, content_type) = build_body(metrics)?;
+        let (body, content_type) = self.build_metrics_export_body(metrics)?;
         let mut request = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
@@ -47,22 +47,3 @@ impl MetricsClient for OtlpHttpClient {
         Ok(())
     }
 }
-
-#[cfg(feature = "http-proto")]
-fn build_body(metrics: &mut ResourceMetrics) -> Result<(Vec<u8>, &'static str)> {
-    use prost::Message;
-
-    let req: opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest =
-        (&*metrics).into();
-    let mut buf = vec![];
-    req.encode(&mut buf).map_err(crate::Error::from)?;
-
-    Ok((buf, "application/x-protobuf"))
-}
-
-#[cfg(not(feature = "http-proto"))]
-fn build_body(metrics: &mut ResourceMetrics) -> Result<(Vec<u8>, &'static str)> {
-    Err(MetricsError::Other(
-        "No http protocol configured. Enable one via `http-proto`".into(),
-    ))
-}

opentelemetry-otlp/src/exporter/http/mod.rs

@@ -1,17 +1,23 @@
+use super::{default_headers, default_protocol, parse_header_string};
 use crate::{
     ExportConfig, Protocol, OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_HEADERS,
     OTEL_EXPORTER_OTLP_TIMEOUT,
 };
 use http::{HeaderName, HeaderValue, Uri};
 use opentelemetry_http::HttpClient;
+#[cfg(feature = "logs")]
+use opentelemetry_sdk::export::logs::LogData;
+#[cfg(feature = "trace")]
+use opentelemetry_sdk::export::trace::SpanData;
+#[cfg(feature = "metrics")]
+use opentelemetry_sdk::metrics::data::ResourceMetrics;
+use prost::Message;
 use std::collections::HashMap;
 use std::env;
 use std::str::FromStr;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
-use super::{default_headers, parse_header_string};
-
 #[cfg(feature = "metrics")]
 mod metrics;
 
@@ -22,7 +28,7 @@ mod logs;
 mod trace;
 
 /// Configuration of the http transport
-#[cfg(feature = "http-proto")]
+#[cfg(any(feature = "http-proto", feature = "http-json"))]
 #[derive(Debug)]
 #[cfg_attr(
     all(
@@ -98,7 +104,7 @@ impl Default for HttpExporterBuilder {
     fn default() -> Self {
         HttpExporterBuilder {
             exporter_config: ExportConfig {
-                protocol: Protocol::HttpBinary,
+                protocol: default_protocol(),
                 ..ExportConfig::default()
             },
             http_config: HttpConfig {
@@ -110,6 +116,12 @@ impl Default for HttpExporterBuilder {
 }
 
 impl HttpExporterBuilder {
+    /// Specify the OTLP protocol to be used by the exporter
+    pub fn with_protocol(mut self, protocol: Protocol) -> Self {
+        self.exporter_config.protocol = protocol;
+        self
+    }
+
     /// Assign client implementation
     pub fn with_http_client<T: HttpClient + 'static>(mut self, client: T) -> Self {
         self.http_config.client = Some(Arc::new(client));
@@ -181,7 +193,13 @@ impl HttpExporterBuilder {
             add_header_from_string(&input, &mut headers);
         }
 
-        Ok(OtlpHttpClient::new(http_client, endpoint, headers, timeout))
+        Ok(OtlpHttpClient::new(
+            http_client,
+            endpoint,
+            headers,
+            self.exporter_config.protocol,
+            timeout,
+        ))
     }
 
     /// Create a log exporter with the current configuration
@@ -254,6 +272,7 @@ struct OtlpHttpClient {
     client: Mutex<Option<Arc<dyn HttpClient>>>,
     collector_endpoint: Uri,
     headers: HashMap<HeaderName, HeaderValue>,
+    protocol: Protocol,
     _timeout: Duration,
 }
 
@@ -263,15 +282,76 @@ impl OtlpHttpClient {
         client: Arc<dyn HttpClient>,
         collector_endpoint: Uri,
         headers: HashMap<HeaderName, HeaderValue>,
+        protocol: Protocol,
         timeout: Duration,
     ) -> Self {
         OtlpHttpClient {
             client: Mutex::new(Some(client)),
             collector_endpoint,
             headers,
+            protocol,
             _timeout: timeout,
         }
     }
+
+    #[cfg(feature = "trace")]
+    fn build_trace_export_body(
+        &self,
+        spans: Vec<SpanData>,
+    ) -> opentelemetry::trace::TraceResult<(Vec<u8>, &'static str)> {
+        use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
+
+        let req = ExportTraceServiceRequest {
+            resource_spans: spans.into_iter().map(Into::into).collect(),
+        };
+        match self.protocol {
+            #[cfg(feature = "http-json")]
+            Protocol::HttpJson => match serde_json::to_string_pretty(&req) {
+                Ok(json) => Ok((json.into(), "application/json")),
+                Err(e) => Err(opentelemetry::trace::TraceError::from(e.to_string())),
+            },
+            _ => Ok((req.encode_to_vec(), "application/x-protobuf")),
+        }
+    }
+
+    #[cfg(feature = "logs")]
+    fn build_logs_export_body(
+        &self,
+        logs: Vec<LogData>,
+    ) -> opentelemetry::logs::LogResult<(Vec<u8>, &'static str)> {
+        use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
+
+        let req = ExportLogsServiceRequest {
+            resource_logs: logs.into_iter().map(Into::into).collect(),
+        };
+        match self.protocol {
+            #[cfg(feature = "http-json")]
+            Protocol::HttpJson => match serde_json::to_string_pretty(&req) {
+                Ok(json) => Ok((json.into(), "application/json")),
+                Err(e) => Err(opentelemetry::logs::LogError::from(e.to_string())),
+            },
+            _ => Ok((req.encode_to_vec(), "application/x-protobuf")),
+        }
+    }
+
+    #[cfg(feature = "metrics")]
+    fn build_metrics_export_body(
+        &self,
+        metrics: &mut ResourceMetrics,
+    ) -> opentelemetry::metrics::Result<(Vec<u8>, &'static str)> {
+        use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
+
+        let req: ExportMetricsServiceRequest = (&*metrics).into();
+
+        match self.protocol {
+            #[cfg(feature = "http-json")]
+            Protocol::HttpJson => match serde_json::to_string_pretty(&req) {
+                Ok(json) => Ok((json.into(), "application/json")),
+                Err(e) => Err(opentelemetry::metrics::MetricsError::Other(e.to_string())),
+            },
+            _ => Ok((req.encode_to_vec(), "application/x-protobuf")),
+        }
+    }
 }
 
 fn build_endpoint_uri(endpoint: &str, path: &str) -> Result<Uri, crate::Error> {

opentelemetry-otlp/src/exporter/http/trace.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use futures_core::future::BoxFuture;
 use http::{header::CONTENT_TYPE, Method};
-use opentelemetry::trace::{TraceError, TraceResult};
+use opentelemetry::trace::TraceError;
 use opentelemetry_sdk::export::trace::{ExportResult, SpanData, SpanExporter};
 
 use super::OtlpHttpClient;
@@ -21,7 +21,7 @@ impl SpanExporter for OtlpHttpClient {
             Err(err) => return Box::pin(std::future::ready(Err(err))),
         };
 
-        let (body, content_type) = match build_body(batch) {
+        let (body, content_type) = match self.build_trace_export_body(batch) {
             Ok(body) => body,
             Err(e) => return Box::pin(std::future::ready(Err(e))),
         };
@@ -67,24 +67,3 @@ impl SpanExporter for OtlpHttpClient {
         let _ = self.client.lock().map(|mut c| c.take());
     }
 }
-
-#[cfg(feature = "http-proto")]
-fn build_body(spans: Vec<SpanData>) -> TraceResult<(Vec<u8>, &'static str)> {
-    use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
-    use prost::Message;
-
-    let req = ExportTraceServiceRequest {
-        resource_spans: spans.into_iter().map(Into::into).collect(),
-    };
-    let mut buf = vec![];
-    req.encode(&mut buf).map_err(crate::Error::from)?;
-
-    Ok((buf, "application/x-protobuf"))
-}
-
-#[cfg(not(feature = "http-proto"))]
-fn build_body(spans: Vec<SpanData>) -> TraceResult<(Vec<u8>, &'static str)> {
-    Err(TraceError::Other(
-        "No http protocol configured. Enable one via `http-proto`".into(),
-    ))
-}