fix(sink): OTLP sink panics when event is invalid or cannot be detected

The current implementation of the OTLP sink contains some
critical flaws which allows algorithm to panic. This could lead
to a vector partition to be crashed.
This PR change the OTLP sink algorithm to handle all the error correctly,
no panic, no expect, no unwrap and logging all internal errors to make
debugging process and customers life easier.

Ref: LOG-19721

Author: seeruyy

src/sinks/opentelemetry/config.rs

@@ -53,6 +53,49 @@ impl SinkBatchSettings for OpentelemetryDefaultBatchSettings {
     const TIMEOUT_SECS: f64 = 1.0;
 }
 
+#[derive(Debug)]
+pub struct OpentelemetrySinkEndpointError {
+    message: String,
+}
+
+impl OpentelemetrySinkEndpointError {
+    pub fn new(msg: &str) -> Self {
+        OpentelemetrySinkEndpointError {
+            message: String::from(msg),
+        }
+    }
+}
+
+impl From<InvalidUri> for OpentelemetrySinkEndpointError {
+    fn from(error: InvalidUri) -> Self {
+        Self::new(&error.to_string())
+    }
+}
+
+impl From<&str> for OpentelemetrySinkEndpointError {
+    fn from(error: &str) -> Self {
+        Self::new(error)
+    }
+}
+
+impl From<http::Error> for OpentelemetrySinkEndpointError {
+    fn from(error: http::Error) -> Self {
+        Self::new(&error.to_string())
+    }
+}
+
+impl std::fmt::Display for OpentelemetrySinkEndpointError {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "{}", self.message)
+    }
+}
+
+impl std::error::Error for OpentelemetrySinkEndpointError {
+    fn description(&self) -> &str {
+        &self.message
+    }
+}
+
 #[derive(Clone, Debug, Default)]
 pub struct OpentelemetryEndpoint {
     healthcheck_uri: Uri,
@@ -62,66 +105,69 @@ pub struct OpentelemetryEndpoint {
 }
 
 impl OpentelemetryEndpoint {
-    pub fn new(endpoint: Uri) -> Self {
-        let scheme = endpoint.scheme_str().unwrap();
-        let authority = endpoint.authority().map(|a| a.as_str()).unwrap();
-        let host = endpoint.host().unwrap().to_owned();
+    pub fn healthcheck(&self) -> Uri {
+        self.healthcheck_uri.clone()
+    }
+
+    pub fn endpoint(&self, model_type: OpentelemetryModelType) -> Option<Uri> {
+        match model_type {
+            OpentelemetryModelType::Logs => Some(self.logs_uri.clone()),
+            OpentelemetryModelType::Metrics => Some(self.metrics_uri.clone()),
+            OpentelemetryModelType::Traces => Some(self.traces_uri.clone()),
+            OpentelemetryModelType::Unknown => None,
+        }
+    }
+}
+
+impl TryFrom<String> for OpentelemetryEndpoint {
+    type Error = OpentelemetrySinkEndpointError;
+
+    fn try_from(endpoint: String) -> Result<Self, Self::Error> {
+        let uri = endpoint
+            .parse::<Uri>()
+            .map_err(OpentelemetrySinkEndpointError::from)?;
+
+        let scheme = uri.scheme_str().ok_or("Endpoint scheme is invalid")?;
+        let authority = uri
+            .authority()
+            .map(|a| a.as_str())
+            .ok_or("Endpoint authority is invalid")?;
+        let host = uri.host().ok_or("Endpoint host is invalid")?;
 
         let healthcheck_uri = Uri::builder()
             .scheme(scheme)
-            .authority(host + ":" + OPENTELEMETRY_HEALTHCHECK_PORT)
+            .authority(host.to_owned() + ":" + OPENTELEMETRY_HEALTHCHECK_PORT)
             .path_and_query("/")
             .build()
-            .unwrap();
+            .map_err(OpentelemetrySinkEndpointError::from)?;
 
         let logs_uri = Uri::builder()
             .scheme(scheme)
             .authority(authority)
             .path_and_query("/v1/logs")
             .build()
-            .unwrap();
+            .map_err(OpentelemetrySinkEndpointError::from)?;
 
         let metrics_uri = Uri::builder()
             .scheme(scheme)
             .authority(authority)
             .path_and_query("/v1/metrics")
             .build()
-            .unwrap();
+            .map_err(OpentelemetrySinkEndpointError::from)?;
 
         let traces_uri = Uri::builder()
             .scheme(scheme)
             .authority(authority)
             .path_and_query("/v1/traces")
             .build()
-            .unwrap();
+            .map_err(OpentelemetrySinkEndpointError::from)?;
 
-        Self {
+        Ok(Self {
             healthcheck_uri,
             logs_uri,
             metrics_uri,
             traces_uri,
-        }
-    }
-
-    pub fn healthcheck(&self) -> Uri {
-        self.healthcheck_uri.clone()
-    }
-
-    pub fn endpoint(&self, model_type: OpentelemetryModelType) -> Uri {
-        match model_type {
-            OpentelemetryModelType::Logs => self.logs_uri.clone(),
-            OpentelemetryModelType::Metrics => self.metrics_uri.clone(),
-            OpentelemetryModelType::Traces => self.traces_uri.clone(),
-        }
-    }
-}
-
-impl TryFrom<String> for OpentelemetryEndpoint {
-    type Error = InvalidUri;
-
-    fn try_from(endpoint: String) -> Result<Self, Self::Error> {
-        let uri = endpoint.parse::<Uri>()?;
-        Ok(Self::new(uri))
+        })
     }
 }
 
@@ -224,7 +270,7 @@ impl SinkConfig for OpentelemetrySinkConfig {
                     client,
                     auth,
                 },
-                ctx.mezmo_ctx,
+                ctx.mezmo_ctx.clone(),
             ));
 
         let compression = self.compression;
@@ -233,6 +279,7 @@ impl SinkConfig for OpentelemetrySinkConfig {
             encoder: OpentelemetryEncoder,
             compression,
             batcher_settings,
+            mezmo_ctx: ctx.mezmo_ctx,
         };
         Ok((VectorSink::from_event_streamsink(sink), healthcheck))
     }

src/sinks/opentelemetry/encoding.rs

@@ -19,9 +19,9 @@ impl Encoder<Result<OpentelemetryModel, OpentelemetrySinkError>> for Opentelemet
         writer: &mut dyn io::Write,
     ) -> io::Result<(usize, GroupedCountByteSize)> {
         let output = match input? {
-            OpentelemetryModel::Logs(log_model) => encode_log(&log_model)?,
-            OpentelemetryModel::Metrics(metric_model) => encode_metrics(&metric_model)?,
-            OpentelemetryModel::Traces(trace_model) => encode_traces(trace_model)?,
+            OpentelemetryModel::Logs(log_models) => encode_log(log_models)?,
+            OpentelemetryModel::Metrics(metric_models) => encode_metrics(metric_models)?,
+            OpentelemetryModel::Traces(trace_models) => encode_traces(trace_models)?,
         };
         let size = as_tracked_write::<_, _, io::Error>(writer, &output, |writer, output| {
             writer.write_all(output)?;

src/sinks/opentelemetry/logs/encoding.rs

@@ -4,11 +4,9 @@ use crate::sinks::opentelemetry::{
 use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
 use prost::Message;
 
-pub fn encode(model: &OpentelemetryLogsModel) -> Result<Vec<u8>, OpentelemetrySinkError> {
-    let logs = model.0[0].get("logs").unwrap().clone();
-
+pub fn encode(models: Vec<OpentelemetryLogsModel>) -> Result<Vec<u8>, OpentelemetrySinkError> {
     let req = ExportLogsServiceRequest {
-        resource_logs: logs.into_iter().map(Into::into).collect(),
+        resource_logs: models.into_iter().map(|model| model.0.into()).collect(),
     };
 
     let mut buf = vec![];