Rust 1.73.0 update

Cargo.toml

@@ -183,7 +183,7 @@ prost-types = "0.9.0"
 tremor-otelapis = { version = "=0.2.4" }
 
 # aws-s3
-aws-sdk-s3 = "0.30"
+aws-sdk-s3 = "0.33"
 aws-types = "0.56"
 aws-config = "0.56"
 aws-smithy-http = "0.56"

Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.72-bullseye as builder
+FROM rust:1.73-bullseye as builder
 
 # Avoid warnings by switching to noninteractive
 ENV DEBIAN_FRONTEND=noninteractive

Dockerfile.learn

@@ -1,4 +1,4 @@
-FROM rust:1.72-bullseye as builder
+FROM rust:1.73-bullseye as builder
 
 RUN cargo install --features=ssl websocat
 

Dockerfile.native

@@ -1,4 +1,4 @@
-FROM rust:1.72-bullseye as builder
+FROM rust:1.73-bullseye as builder
 
 # Avoid warnings by switching to noninteractive
 ENV DEBIAN_FRONTEND=noninteractive

rust-toolchain

@@ -1 +1 @@
-1.72.0
+1.73.0

src/codec/tremor.rs

@@ -420,6 +420,8 @@ mod test {
     }
     proptest! {
         #[test]
+        #[allow(clippy::ignored_unit_patterns)]
+
         fn prop_round_trip(v1 in arb_tremor_value()) {
             let mut v: Vec<u8> = Vec::new();
             Tremor::encode(&v1, &mut v).expect("failed to encode");

src/connectors/impls/crononome/handler.rs

@@ -99,7 +99,7 @@ impl<I> PartialEq for TemporalItem<I> {
 
 impl<I> PartialOrd for TemporalItem<I> {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        self.at.partial_cmp(&other.at)
+        Some(self.at.cmp(&other.at))
     }
 }
 

src/connectors/impls/gcl.rs

@@ -253,8 +253,7 @@
 //! end;
 //! deploy flow main;
 //! ```
-
-use rand::Rng;
+use tremor_common::rand;
 use tremor_script::{tremor_fn, Registry};
 use tremor_value::Value;
 
@@ -265,21 +264,11 @@ pub(crate) mod writer;
 //  uses: a 16-character hexadecimal encoding of an 8-byte array,
 // such as 000000000000004a
 fn random_span_id_value(ingest_ns_seed: u64) -> Value<'static> {
-    let mut rng = tremor_common::rand::make_prng(ingest_ns_seed);
-    let span_id: String = (0..8)
-        .map(|_| rng.gen::<u8>())
-        .map(|b| format!("{b:02x}"))
-        .collect();
-    Value::from(span_id)
+    Value::from(tremor_common::rand::octet_string(8, ingest_ns_seed))
 }
 
 fn random_trace_id_value(ingest_ns_seed: u64) -> Value<'static> {
-    let mut rng = tremor_common::rand::make_prng(ingest_ns_seed);
-    let span_id: String = (0..16)
-        .map(|_| rng.gen::<u8>())
-        .map(|b| format!("{b:02x}"))
-        .collect();
-    Value::from(span_id)
+    Value::from(rand::octet_string(16, ingest_ns_seed))
 }
 
 /// Extend function registry with `GCP Cloud Logging` support

src/connectors/impls/kafka/consumer.rs

@@ -385,7 +385,7 @@ impl ConsumerContext for TremorRDKafkaContext<SourceContext> {
         offsets: &rdkafka::TopicPartitionList,
     ) {
         match result {
-            Ok(_) => {
+            Ok(()) => {
                 if offsets.count() > 0 {
                     debug!("{} Offsets committed successfully", &self.ctx);
                     if log_enabled!(Debug) {
@@ -995,7 +995,7 @@ impl TopicResolver {
 
 #[cfg(test)]
 mod test {
-
+    #![allow(clippy::ignored_unit_patterns)]
     use super::{Config, Offset, TopicResolver};
     use crate::errors::Result;
     use proptest::prelude::*;

src/connectors/impls/otel/id.rs

@@ -15,8 +15,10 @@
 #![allow(dead_code)]
 
 use crate::errors::Result;
-use rand::Rng;
+use ::rand::Rng;
 use simd_json::ValueAccess;
+use std::fmt::Write;
+use tremor_common::rand;
 use tremor_value::Value;
 
 pub(crate) fn random_span_id_bytes(ingest_ns_seed: u64) -> Vec<u8> {
@@ -26,12 +28,7 @@ pub(crate) fn random_span_id_bytes(ingest_ns_seed: u64) -> Vec<u8> {
 }
 
 pub(crate) fn random_span_id_string(ingest_ns_seed: u64) -> String {
-    let mut rng = tremor_common::rand::make_prng(ingest_ns_seed);
-    let span_id: String = (0..8)
-        .map(|_| rng.gen_range(0_u8..=255_u8))
-        .map(|b| format!("{b:02x}"))
-        .collect();
-    span_id
+    rand::octet_string(8, ingest_ns_seed)
 }
 
 pub(crate) fn random_span_id_array(ingest_ns_seed: u64) -> Value<'static> {
@@ -52,12 +49,7 @@ pub(crate) fn random_trace_id_bytes(ingest_ns_seed: u64) -> Vec<u8> {
 }
 
 pub(crate) fn random_trace_id_string(ingest_ns_seed: u64) -> String {
-    let mut rng = tremor_common::rand::make_prng(ingest_ns_seed);
-    let span_id: String = (0..16)
-        .map(|_| rng.gen_range(0_u8..=255_u8))
-        .map(|b| format!("{b:02x}"))
-        .collect();
-    span_id
+    tremor_common::rand::octet_string(16, ingest_ns_seed)
 }
 
 pub(crate) fn random_trace_id_value(ingest_ns_seed: u64) -> Value<'static> {
@@ -79,20 +71,18 @@ pub(crate) fn hex_span_id_to_pb(data: Option<&Value<'_>>) -> Result<Vec<u8>> {
     hex_id_to_pb("span", data, 8, false)
 }
 
-pub(crate) fn hex_span_id_to_json(data: &[u8]) -> Value<'static> {
-    let hex: String = data.iter().map(|b| format!("{b:02x}")).collect();
-    Value::from(hex)
+pub(crate) fn hex_id_to_json(data: &[u8]) -> Value<'static> {
+    Value::from(data.iter().fold(String::new(), |mut o, b| {
+        // ALLOW: if we can't allocate it's worse, we'd have the same problem with format
+        let _ = write!(o, "{b:02x}");
+        o
+    }))
 }
 
 pub(crate) fn hex_trace_id_to_pb(data: Option<&Value<'_>>) -> Result<Vec<u8>> {
     hex_id_to_pb("trace", data, 16, false)
 }
 
-pub(crate) fn hex_trace_id_to_json(data: &[u8]) -> Value<'static> {
-    let hex: String = data.iter().map(|b| format!("{b:02x}")).collect();
-    Value::from(hex)
-}
-
 fn hex_id_to_pb(
     kind: &str,
     data: Option<&Value<'_>>,
@@ -119,55 +109,38 @@ fn hex_id_to_pb(
 
 #[cfg(test)]
 pub mod test {
+    #![allow(clippy::ignored_unit_patterns)]
     use super::*;
 
     use proptest::prelude::*;
     use proptest::proptest;
 
-    pub(crate) fn pb_span_id_to_json(pb: &[u8]) -> Value {
-        let hex: String = pb.iter().map(|b| format!("{b:02x}")).collect();
-        Value::String(hex.into())
-    }
-
-    pub(crate) fn json_span_id_to_pb(json: Option<&Value<'_>>) -> Result<Vec<u8>> {
-        hex_span_id_to_pb(json)
-    }
-
-    pub(crate) fn pb_trace_id_to_json(pb: &[u8]) -> Value {
-        let hex: String = pb.iter().map(|b| format!("{b:02x}")).collect();
-        Value::String(hex.into())
-    }
-
-    pub(crate) fn json_trace_id_to_pb(json: Option<&Value<'_>>) -> Result<Vec<u8>> {
-        hex_trace_id_to_pb(json)
-    }
-
     #[test]
     fn test_utilities() -> Result<()> {
         let nanos = tremor_common::time::nanotime();
         let span_bytes = random_span_id_bytes(nanos);
         let trace_bytes = random_trace_id_bytes(nanos);
 
-        let span_json = pb_span_id_to_json(&span_bytes);
-        let trace_json = pb_trace_id_to_json(&trace_bytes);
+        let span_json = hex_id_to_json(&span_bytes);
+        let trace_json = hex_id_to_json(&trace_bytes);
 
-        let span_pb = json_span_id_to_pb(Some(&span_json))?;
-        let trace_pb = json_trace_id_to_pb(Some(&trace_json))?;
+        let span_pb = hex_span_id_to_pb(Some(&span_json))?;
+        let trace_pb = hex_trace_id_to_pb(Some(&trace_json))?;
 
         assert_eq!(span_bytes, span_pb);
         assert_eq!(trace_bytes, trace_pb);
 
         let span_array = random_span_id_array(nanos);
         let trace_array = random_trace_id_array(nanos);
 
-        let span_pb = json_span_id_to_pb(Some(&span_array))?;
-        let trace_pb = json_trace_id_to_pb(Some(&trace_array))?;
+        let span_pb = hex_span_id_to_pb(Some(&span_array))?;
+        let trace_pb = hex_trace_id_to_pb(Some(&trace_array))?;
 
-        let span_json = pb_span_id_to_json(&span_pb);
-        let trace_json = pb_span_id_to_json(&trace_pb);
+        let span_json = hex_id_to_json(&span_pb);
+        let trace_json = hex_id_to_json(&trace_pb);
 
-        let span_pb2 = json_span_id_to_pb(Some(&span_json))?;
-        let trace_pb2 = json_trace_id_to_pb(Some(&trace_json))?;
+        let span_pb2 = hex_span_id_to_pb(Some(&span_json))?;
+        let trace_pb2 = hex_trace_id_to_pb(Some(&trace_json))?;
 
         assert_eq!(span_pb2, span_pb);
         assert_eq!(trace_pb2, trace_pb);

src/connectors/impls/otel/logs.rs

@@ -58,8 +58,8 @@ fn log_record_to_json(log: LogRecord) -> Result<Value<'static>> {
         "severity_number": affirm_severity_number_valid(log.severity_number)?,
         "severity_text": log.severity_text.to_string(),
         "flags": affirm_traceflags_valid(log.flags)?,
-        "span_id": id::hex_span_id_to_json(&log.span_id),
-        "trace_id": id::hex_trace_id_to_json(&log.trace_id),
+        "span_id": id::hex_id_to_json(&log.span_id),
+        "trace_id": id::hex_id_to_json(&log.trace_id),
         "attributes": common::key_value_list_to_json(log.attributes),
         "dropped_attributes_count": log.dropped_attributes_count,
         "body": common::maybe_any_value_to_json(log.body),
@@ -198,9 +198,9 @@ mod tests {
     fn instrumentation_library_logs() -> Result<()> {
         let nanos = tremor_common::time::nanotime();
         let span_id_pb = id::random_span_id_bytes(nanos);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanos);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         let pb = vec![InstrumentationLibraryLogs {
             schema_url: "schema_url".into(),
@@ -253,9 +253,9 @@ mod tests {
     fn resource_logs() -> Result<()> {
         let nanos = tremor_common::time::nanotime();
         let span_id_pb = id::random_span_id_bytes(nanos);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanos);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         let pb = ExportLogsServiceRequest {
             resource_logs: vec![ResourceLogs {
@@ -326,9 +326,9 @@ mod tests {
     fn resource_logs_severity_unspecified_regression() -> Result<()> {
         let nanos = tremor_common::time::nanotime();
         let span_id_pb = id::random_span_id_bytes(nanos);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanos);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         let pb = ExportLogsServiceRequest {
             resource_logs: vec![ResourceLogs {

src/connectors/impls/otel/metrics.rs

@@ -40,8 +40,8 @@ pub(crate) fn int_exemplars_to_json(data: Vec<IntExemplar>) -> Value<'static> {
     data.into_iter()
         .map(|exemplar| {
             literal!({
-                "span_id": id::hex_span_id_to_json(&exemplar.span_id),
-                "trace_id": id::hex_trace_id_to_json(&exemplar.trace_id),
+                "span_id": id::hex_id_to_json(&exemplar.span_id),
+                "trace_id": id::hex_id_to_json(&exemplar.trace_id),
                 "filtered_labels": common::string_key_value_to_json(exemplar.filtered_labels),
                 "time_unix_nano": exemplar.time_unix_nano,
                 "value": exemplar.value
@@ -83,8 +83,8 @@ pub(crate) fn double_exemplars_to_json(data: Vec<Exemplar>) -> Value<'static> {
             };
 
             let mut r = literal!({
-                "span_id": id::hex_span_id_to_json(&exemplar.span_id),
-                "trace_id": id::hex_trace_id_to_json(&exemplar.trace_id),
+                "span_id": id::hex_id_to_json(&exemplar.span_id),
+                "trace_id": id::hex_id_to_json(&exemplar.trace_id),
                 "filtered_attributes": filtered_attributes,
                 "time_unix_nano": exemplar.time_unix_nano,
             });
@@ -583,9 +583,9 @@ mod tests {
     fn int_exemplars() -> Result<()> {
         let nanos = tremor_common::time::nanotime();
         let span_id_pb = id::random_span_id_bytes(nanos);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanos);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         let pb = vec![IntExemplar {
             span_id: span_id_pb.clone(),
@@ -620,9 +620,9 @@ mod tests {
     fn double_exemplars() -> Result<()> {
         let nanos = tremor_common::time::nanotime();
         let span_id_pb = id::random_span_id_bytes(nanos);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanos);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         let pb = vec![Exemplar {
             filtered_attributes: vec![],

src/connectors/impls/otel/trace.rs

@@ -85,8 +85,8 @@ pub(crate) fn span_links_to_json(pb: Vec<Link>) -> Value<'static> {
     pb.into_iter()
         .map(|data| {
             literal!({
-                "trace_id": id::hex_trace_id_to_json(&data.trace_id),
-                "span_id": id::hex_span_id_to_json(&data.span_id),
+                "trace_id": id::hex_id_to_json(&data.trace_id),
+                "span_id": id::hex_id_to_json(&data.span_id),
                 "trace_state": data.trace_state,
                 "attributes": common::key_value_list_to_json(data.attributes),
                 "dropped_attributes_count" : data.dropped_attributes_count
@@ -133,9 +133,9 @@ fn span_to_json(span: Span) -> Value<'static> {
         "attributes": common::key_value_list_to_json(span.attributes),
         "events": span_events_to_json(span.events),
         "links": span_links_to_json(span.links),
-        "span_id": id::hex_span_id_to_json(&span.span_id),
-        "parent_span_id": id::hex_span_id_to_json(&span.parent_span_id),
-        "trace_id": id::hex_trace_id_to_json(&span.trace_id),
+        "span_id": id::hex_id_to_json(&span.span_id),
+        "parent_span_id": id::hex_id_to_json(&span.parent_span_id),
+        "trace_id": id::hex_id_to_json(&span.trace_id),
         "start_time_unix_nano": span.start_time_unix_nano,
         "end_time_unix_nano": span.end_time_unix_nano,
         "trace_state": span.trace_state,
@@ -343,9 +343,9 @@ mod tests {
     fn span_link() -> Result<()> {
         let nanotime = tremor_common::time::nanotime();
         let span_id_pb = id::random_span_id_bytes(nanotime);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanotime);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         let pb = vec![Link {
             attributes: vec![],
@@ -384,11 +384,11 @@ mod tests {
     fn instrument_library_spans() -> Result<()> {
         let nanotime = tremor_common::time::nanotime();
         let parent_span_id_json = id::random_span_id_value(nanotime);
-        let parent_span_id_pb = id::test::json_span_id_to_pb(Some(&parent_span_id_json))?;
+        let parent_span_id_pb = id::hex_span_id_to_pb(Some(&parent_span_id_json))?;
         let span_id_pb = id::random_span_id_bytes(nanotime);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanotime);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         let pb = vec![InstrumentationLibrarySpans {
             schema_url: "schema_url".into(),
@@ -460,11 +460,11 @@ mod tests {
     fn resource_spans() -> Result<()> {
         let nanotime = tremor_common::time::nanotime();
         let parent_span_id_json = id::random_span_id_value(nanotime);
-        let parent_span_id_pb = id::test::json_span_id_to_pb(Some(&parent_span_id_json))?;
+        let parent_span_id_pb = id::hex_span_id_to_pb(Some(&parent_span_id_json))?;
         let span_id_pb = id::random_span_id_bytes(nanotime);
-        let span_id_json = id::test::pb_span_id_to_json(&span_id_pb);
+        let span_id_json = id::hex_id_to_json(&span_id_pb);
         let trace_id_json = id::random_trace_id_value(nanotime);
-        let trace_id_pb = id::test::json_trace_id_to_pb(Some(&trace_id_json))?;
+        let trace_id_pb = id::hex_trace_id_to_pb(Some(&trace_id_json))?;
 
         #[allow(deprecated)]
         let pb = ExportTraceServiceRequest {