fix: cargo hack, incorporate review comments

Author: nikhilsinhaparseable

Cargo.lock

@@ -117,11 +117,6 @@ zip = { version = "2.2.0", default-features = false, features = ["deflate"] }
 url = "2.4.0"
 prost-build = "0.13.3"
 
-[dev-dependencies]
-maplit = "1.0"
-rstest = "0.23.0"
-arrow = "53.0.0"
-
 [package.metadata.parseable_ui]
 assets-url = "https://github.com/parseablehq/console/releases/download/v0.9.12/build.zip"
 assets-sha1 = "9d5a45f204d709a2dd96f6a5e0b21b3834ee0e36"

Cargo.toml

@@ -18,9 +18,11 @@
 
 use super::logstream::error::{CreateStreamError, StreamError};
 use super::modal::utils::ingest_utils::{flatten_and_push_logs, push_logs};
+use super::otel::logs::flatten_otel_logs;
+use super::otel::metrics::flatten_otel_metrics;
+use super::otel::traces::flatten_otel_traces;
 use super::users::dashboards::DashboardError;
 use super::users::filters::FiltersError;
-use super::{otel_logs, otel_metrics, otel_traces};
 use crate::event::{
     self,
     error::EventError,
@@ -119,7 +121,7 @@ pub async fn handle_otel_logs_ingestion(
         create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
 
         //custom flattening required for otel logs
-        let mut json = otel_logs::flatten_otel_logs(&body);
+        let mut json = flatten_otel_logs(&body);
         for record in json.iter_mut() {
             let body: Bytes = serde_json::to_vec(record).unwrap().into();
             push_logs(&stream_name, &req, &body).await?;
@@ -146,7 +148,7 @@ pub async fn handle_otel_metrics_ingestion(
         create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
 
         //custom flattening required for otel metrics
-        let mut json = otel_metrics::flatten_otel_metrics(&body);
+        let mut json = flatten_otel_metrics(&body);
         for record in json.iter_mut() {
             let body: Bytes = serde_json::to_vec(record).unwrap().into();
             push_logs(&stream_name, &req, &body).await?;
@@ -173,7 +175,7 @@ pub async fn handle_otel_traces_ingestion(
         create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
 
         //custom flattening required for otel traces
-        let mut json = otel_traces::flatten_otel_traces(&body);
+        let mut json = flatten_otel_traces(&body);
         for record in json.iter_mut() {
             let body: Bytes = serde_json::to_vec(record).unwrap().into();
             push_logs(&stream_name, &req, &body).await?;

src/handlers/http/ingest.rs

@@ -38,9 +38,6 @@ pub mod middleware;
 pub mod modal;
 pub mod oidc;
 pub mod otel;
-pub mod otel_logs;
-pub mod otel_metrics;
-pub mod otel_traces;
 pub mod query;
 pub mod rbac;
 pub mod role;

src/handlers/http/mod.rs

@@ -33,7 +33,13 @@ use crate::{
         format::{self, EventFormat},
     },
     handlers::{
-        http::{ingest::PostError, kinesis, otel_logs, otel_metrics, otel_traces},
+        http::{
+            ingest::PostError,
+            kinesis,
+            otel::{
+                logs::flatten_otel_logs, metrics::flatten_otel_metrics, traces::flatten_otel_traces,
+            },
+        },
         LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, LOG_SOURCE_OTEL_LOGS, LOG_SOURCE_OTEL_METRICS,
         LOG_SOURCE_OTEL_TRACES, PREFIX_META, PREFIX_TAGS, SEPARATOR,
     },
@@ -56,17 +62,17 @@ pub async fn flatten_and_push_logs(
 
             //custom flattening required for otel logs
             LOG_SOURCE_OTEL_LOGS => {
-                json = otel_logs::flatten_otel_logs(&body);
+                json = flatten_otel_logs(&body);
             }
 
             //custom flattening required for otel metrics
             LOG_SOURCE_OTEL_METRICS => {
-                json = otel_metrics::flatten_otel_metrics(&body);
+                json = flatten_otel_metrics(&body);
             }
 
             //custom flattening required for otel traces
             LOG_SOURCE_OTEL_TRACES => {
-                json = otel_traces::flatten_otel_traces(&body);
+                json = flatten_otel_traces(&body);
             }
             _ => {
                 tracing::warn!("Unknown log source: {}", log_source);

src/handlers/http/modal/utils/ingest_utils.rs

@@ -15,7 +15,11 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  */
+pub mod logs;
+pub mod metrics;
+#[allow(clippy::all)]
 pub mod proto;
+pub mod traces;
 use proto::common::v1::KeyValue;
 use serde_json::Value;
 use std::collections::BTreeMap;

src/handlers/http/otel.rs

@@ -191,7 +191,6 @@ pub struct Metric {
     /// name of the metric.
     pub name: Option<String>,
     /// description of the metric, which can be used in documentation.
-    
     pub description: Option<String>,
     /// unit in which the metric value is reported. Follows the format
     /// described by <http://unitsofmeasure.org/ucum.html.>
@@ -235,10 +234,8 @@ pub struct Sum {
     pub data_points: Option<Vec<NumberDataPoint>>,
     /// aggregation_temporality describes if the aggregator reports delta changes
     /// since last report time, or cumulative changes since a fixed start time.
-    
     pub aggregation_temporality: Option<i32>,
     /// If "true" means that the sum is monotonic.
-    
     pub is_monotonic: Option<bool>,
 }
 /// Histogram represents the type of a metric that is calculated by aggregating
@@ -250,7 +247,6 @@ pub struct Histogram {
     pub data_points: Option<Vec<HistogramDataPoint>>,
     /// aggregation_temporality describes if the aggregator reports delta changes
     /// since last report time, or cumulative changes since a fixed start time.
-    
     pub aggregation_temporality: Option<i32>,
 }
 /// ExponentialHistogram represents the type of a metric that is calculated by aggregating
@@ -261,7 +257,6 @@ pub struct ExponentialHistogram {
     pub data_points: Option<Vec<ExponentialHistogramDataPoint>>,
     /// aggregation_temporality describes if the aggregator reports delta changes
     /// since last report time, or cumulative changes since a fixed start time.
-    
     pub aggregation_temporality: Option<i32>,
 }
 /// Summary metric data are used to convey quantile summaries,
@@ -560,7 +555,7 @@ pub mod summary_data_point {
     fn deserialize_f64_or_nan<'de, D>(deserializer: D) -> Result<Option<f64>, D::Error> where D: Deserializer<'de>, 
     { 
         struct StringOrFloatVisitor; 
-        impl<'de> serde::de::Visitor<'de> for StringOrFloatVisitor 
+        impl serde::de::Visitor<'_> for StringOrFloatVisitor 
         {
              type Value = Option<f64>; 
              fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result 
@@ -617,68 +612,8 @@ pub struct Exemplar {
 /// which they are aggregated.
 #[repr(i32)]
 pub enum AggregationTemporality {
-    /// UNSPECIFIED is the default AggregationTemporality, it MUST not be used.
     Unspecified = 0,
-    /// DELTA is an AggregationTemporality for a metric aggregator which reports
-    /// changes since last report time. Successive metrics contain aggregation of
-    /// values from continuous and non-overlapping intervals.
-    ///
-    /// The values for a DELTA metric are based only on the time interval
-    /// associated with one measurement cycle. There is no dependency on
-    /// previous measurements like is the case for CUMULATIVE metrics.
-    ///
-    /// For example, consider a system measuring the number of requests that
-    /// it receives and reports the sum of these requests every second as a
-    /// DELTA metric:
-    ///
-    ///    1. The system starts receiving at time=t_0.
-    ///    2. A request is received, the system measures 1 request.
-    ///    3. A request is received, the system measures 1 request.
-    ///    4. A request is received, the system measures 1 request.
-    ///    5. The 1 second collection cycle ends. A metric is exported for the
-    ///       number of requests received over the interval of time t_0 to
-    ///       t_0+1 with a value of 3.
-    ///    6. A request is received, the system measures 1 request.
-    ///    7. A request is received, the system measures 1 request.
-    ///    8. The 1 second collection cycle ends. A metric is exported for the
-    ///       number of requests received over the interval of time t_0+1 to
-    ///       t_0+2 with a value of 2.
     Delta = 1,
-    /// CUMULATIVE is an AggregationTemporality for a metric aggregator which
-    /// reports changes since a fixed start time. This means that current values
-    /// of a CUMULATIVE metric depend on all previous measurements since the
-    /// start time. Because of this, the sender is required to retain this state
-    /// in some form. If this state is lost or invalidated, the CUMULATIVE metric
-    /// values MUST be reset and a new fixed start time following the last
-    /// reported measurement time sent MUST be used.
-    ///
-    /// For example, consider a system measuring the number of requests that
-    /// it receives and reports the sum of these requests every second as a
-    /// CUMULATIVE metric:
-    ///
-    ///    1. The system starts receiving at time=t_0.
-    ///    2. A request is received, the system measures 1 request.
-    ///    3. A request is received, the system measures 1 request.
-    ///    4. A request is received, the system measures 1 request.
-    ///    5. The 1 second collection cycle ends. A metric is exported for the
-    ///       number of requests received over the interval of time t_0 to
-    ///       t_0+1 with a value of 3.
-    ///    6. A request is received, the system measures 1 request.
-    ///    7. A request is received, the system measures 1 request.
-    ///    8. The 1 second collection cycle ends. A metric is exported for the
-    ///       number of requests received over the interval of time t_0 to
-    ///       t_0+2 with a value of 5.
-    ///    9. The system experiences a fault and loses state.
-    ///    10. The system recovers and resumes receiving at time=t_1.
-    ///    11. A request is received, the system measures 1 request.
-    ///    12. The 1 second collection cycle ends. A metric is exported for the
-    ///       number of requests received over the interval of time t_1 to
-    ///       t_0+1 with a value of 1.
-    ///
-    /// Note: Even though, when reporting changes since last report time, using
-    /// CUMULATIVE is valid, it is not recommended. This may cause problems for
-    /// systems that do not use start_time to determine when the aggregation
-    /// value was reset (e.g. Prometheus).
     Cumulative = 2,
 }
 impl AggregationTemporality {