chore(deps): Upgrade to Rust 1.50.0

lib/remap-lang/src/expression/assignment.rs

@@ -122,6 +122,9 @@ impl Expression for Assignment {
     fn execute(&self, state: &mut state::Program, object: &mut dyn Object) -> Result<Value> {
         let value = self.value.execute(state, object);
 
+        // ignoring the unnecessariy wrap as this whole parser is going away momentarily and this
+        // matches up better with the other *_assignment methods anyway
+        #[allow(clippy::unnecessary_wraps)]
         fn var_assignment<'a>(
             state: &mut state::Program,
             var: &Variable,

lib/remap-lang/src/parser.rs

@@ -379,6 +379,9 @@ impl<'a> Parser<'a> {
         Ok((start..end, nodes).into())
     }
 
+    // ignoring the unnecessariy wrap as this whole parser is going away momentarily and this
+    // matches up better with the other *_from_* methods anyway
+    #[allow(clippy::unnecessary_wraps)]
     fn pairs_from_str<'b>(&mut self, rule: R, source: &'b str) -> IResult<Pairs<'b, R>> {
         use pest::Parser;
 

rust-toolchain

@@ -1 +1 @@
-1.49.0
+1.50.0

src/api/schema/components/sink.rs

@@ -109,7 +109,7 @@ impl Sink {
 
     /// Sink metrics
     pub async fn metrics(&self) -> metrics::SinkMetrics {
-        metrics::by_component_name(self.get_name()).to_sink_metrics(self.get_component_type())
+        metrics::by_component_name(self.get_name()).into_sink_metrics(self.get_component_type())
     }
 }
 

src/api/schema/components/source.rs

@@ -107,7 +107,7 @@ impl Source {
 
     /// Source metrics
     pub async fn metrics(&self) -> metrics::SourceMetrics {
-        metrics::by_component_name(&self.get_name()).to_source_metrics(&self.get_component_type())
+        metrics::by_component_name(&self.get_name()).into_source_metrics(&self.get_component_type())
     }
 }
 

src/api/schema/components/transform.rs

@@ -88,7 +88,7 @@ impl Transform {
 
     /// Transform metrics
     pub async fn metrics(&self) -> metrics::TransformMetrics {
-        metrics::by_component_name(&self.0.name).to_transform_metrics(&self.get_component_type())
+        metrics::by_component_name(&self.0.name).into_transform_metrics(&self.get_component_type())
     }
 }
 

src/api/schema/metrics/sink/mod.rs

@@ -14,11 +14,11 @@ pub enum SinkMetrics {
 }
 
 pub trait IntoSinkMetrics {
-    fn to_sink_metrics(self, component_type: &str) -> SinkMetrics;
+    fn into_sink_metrics(self, component_type: &str) -> SinkMetrics;
 }
 
 impl IntoSinkMetrics for Vec<Metric> {
-    fn to_sink_metrics(self, _component_type: &str) -> SinkMetrics {
+    fn into_sink_metrics(self, _component_type: &str) -> SinkMetrics {
         SinkMetrics::GenericSinkMetrics(generic::GenericSinkMetrics::new(self))
     }
 }

src/api/schema/metrics/source/mod.rs

@@ -16,11 +16,11 @@ pub enum SourceMetrics {
 }
 
 pub trait IntoSourceMetrics {
-    fn to_source_metrics(self, component_type: &str) -> SourceMetrics;
+    fn into_source_metrics(self, component_type: &str) -> SourceMetrics;
 }
 
 impl IntoSourceMetrics for Vec<Metric> {
-    fn to_source_metrics(self, component_type: &str) -> SourceMetrics {
+    fn into_source_metrics(self, component_type: &str) -> SourceMetrics {
         match component_type {
             "file" => SourceMetrics::FileSourceMetrics(file::FileSourceMetrics::new(self)),
             _ => SourceMetrics::GenericSourceMetrics(generic::GenericSourceMetrics::new(self)),

src/api/schema/metrics/transform/mod.rs

@@ -14,11 +14,11 @@ pub enum TransformMetrics {
 }
 
 pub trait IntoTransformMetrics {
-    fn to_transform_metrics(self, component_type: &str) -> TransformMetrics;
+    fn into_transform_metrics(self, component_type: &str) -> TransformMetrics;
 }
 
 impl IntoTransformMetrics for Vec<Metric> {
-    fn to_transform_metrics(self, _component_type: &str) -> TransformMetrics {
+    fn into_transform_metrics(self, _component_type: &str) -> TransformMetrics {
         TransformMetrics::GenericTransformMetrics(generic::GenericTransformMetrics::new(self))
     }
 }

src/internal_events/open.rs

@@ -69,13 +69,13 @@ fn gauge_add(gauge: &AtomicUsize, add: isize, emitter: impl Fn(usize)) {
         emitter(new_value);
         // Try to update gauge to new value and releasing writes to gauge metric in the process.
         // Otherwise acquire new writes to gauge metric.
-        let latest = gauge.compare_and_swap(value, new_value, Ordering::AcqRel);
-        if value == latest {
+        value = match gauge.compare_exchange(value, new_value, Ordering::AcqRel, Ordering::Acquire)
+        {
             // Success
-            break;
+            Ok(_) => break,
+            // Try again with new value
+            Err(v) => v,
         }
-        // Try again with new value
-        value = latest;
     }
 
     // In the worst case scenario we will emit `n^2 / 2` times when there are `n` parallel

src/mapping/query/function/split.rs

@@ -23,6 +23,7 @@ impl SplitFn {
 }
 
 impl Function for SplitFn {
+    #[allow(clippy::collapsible_match)] // I expect this file to be going away shortly
     fn execute(&self, ctx: &Event) -> Result<QueryValue> {
         let string = {
             let bytes = required_value!(ctx, self.path, Value::Bytes(v) => v);
@@ -71,10 +72,7 @@ impl Function for SplitFn {
             },
             Parameter {
                 keyword: "pattern",
-                accepts: |v| {
-                    matches!(v, QueryValue::Value(Value::Bytes(_))
-                             | QueryValue::Regex(_))
-                },
+                accepts: |v| matches!(v, QueryValue::Value(Value::Bytes(_)) | QueryValue::Regex(_)),
                 required: true,
             },
             Parameter {

src/sinks/aws_kinesis_firehose.rs

@@ -162,7 +162,7 @@ impl KinesisFirehoseService {
                 cx.acker(),
             )
             .sink_map_err(|error| error!(message = "Fatal kinesis firehose sink error.", %error))
-            .with_flat_map(move |e| stream::iter(encode_event(e, &encoding)).map(Ok));
+            .with_flat_map(move |e| stream::iter(Some(encode_event(e, &encoding))).map(Ok));
 
         Ok(sink)
     }
@@ -240,7 +240,7 @@ enum HealthcheckError {
     StreamNamesMismatch { name: String, stream_name: String },
 }
 
-fn encode_event(mut event: Event, encoding: &EncodingConfig<Encoding>) -> Option<Record> {
+fn encode_event(mut event: Event, encoding: &EncodingConfig<Encoding>) -> Record {
     encoding.apply_rules(&mut event);
     let log = event.into_log();
     let data = match encoding.codec() {
@@ -254,7 +254,7 @@ fn encode_event(mut event: Event, encoding: &EncodingConfig<Encoding>) -> Option
 
     let data = Bytes::from(data);
 
-    Some(Record { data })
+    Record { data }
 }
 
 #[cfg(test)]
@@ -270,7 +270,7 @@ mod tests {
     #[test]
     fn firehose_encode_event_text() {
         let message = "hello world".to_string();
-        let event = encode_event(message.clone().into(), &Encoding::Text.into()).unwrap();
+        let event = encode_event(message.clone().into(), &Encoding::Text.into());
 
         assert_eq!(&event.data[..], message.as_bytes());
     }
@@ -280,7 +280,7 @@ mod tests {
         let message = "hello world".to_string();
         let mut event = Event::from(message.clone());
         event.as_mut_log().insert("key", "value");
-        let event = encode_event(event, &Encoding::Json.into()).unwrap();
+        let event = encode_event(event, &Encoding::Json.into());
 
         let map: BTreeMap<String, String> = serde_json::from_slice(&event.data[..]).unwrap();
 

src/sinks/nats.rs

@@ -85,18 +85,16 @@ impl SinkConfig for NatsSinkConfig {
 }
 
 impl NatsSinkConfig {
-    fn to_nats_options(&self) -> crate::Result<nats::Options> {
+    fn to_nats_options(&self) -> nats::Options {
         // Set reconnect_buffer_size on the nats client to 0 bytes so that the
         // client doesn't buffer internally (to avoid message loss).
-        let options = nats::Options::new()
+        nats::Options::new()
             .with_name(&self.name)
-            .reconnect_buffer_size(0);
-
-        Ok(options)
+            .reconnect_buffer_size(0)
     }
 
     async fn connect(&self) -> crate::Result<nats::asynk::Connection> {
-        self.to_nats_options()?
+        self.to_nats_options()
             .connect_async(&self.url)
             .map_err(|e| e.into())
             .await

src/sinks/papertrail.rs

@@ -65,7 +65,7 @@ impl SinkConfig for PapertrailConfig {
 
         let sink_config = TcpSinkConfig::new(address, self.keepalive, tls, self.send_buffer_bytes);
 
-        sink_config.build(cx, move |event| encode_event(event, pid, &encoding))
+        sink_config.build(cx, move |event| Some(encode_event(event, pid, &encoding)))
     }
 
     fn input_type(&self) -> DataType {
@@ -77,7 +77,7 @@ impl SinkConfig for PapertrailConfig {
     }
 }
 
-fn encode_event(mut event: Event, pid: u32, encoding: &EncodingConfig<Encoding>) -> Option<Bytes> {
+fn encode_event(mut event: Event, pid: u32, encoding: &EncodingConfig<Encoding>) -> Bytes {
     let host = if let Some(host) = event.as_mut_log().remove(log_schema().host_key()) {
         Some(host.to_string_lossy())
     } else {
@@ -110,7 +110,7 @@ fn encode_event(mut event: Event, pid: u32, encoding: &EncodingConfig<Encoding>)
 
     s.push(b'\n');
 
-    Some(Bytes::from(s))
+    Bytes::from(s)
 }
 
 #[cfg(test)]
@@ -137,8 +137,7 @@ mod tests {
                 except_fields: Some(vec!["magic".into()]),
                 timestamp_format: None,
             },
-        )
-        .unwrap();
+        );
 
         let msg =
             bytes.slice(String::from_utf8_lossy(&bytes).find(": ").unwrap() + 2..bytes.len() - 1);

src/sinks/util/buffer/metrics.rs

@@ -265,7 +265,7 @@ impl MetricSet {
     pub fn make_absolute(&mut self, metric: Metric) -> Option<Metric> {
         match metric.data.kind {
             MetricKind::Absolute => Some(metric),
-            MetricKind::Incremental => self.incremental_to_absolute(metric),
+            MetricKind::Incremental => Some(self.incremental_to_absolute(metric)),
         }
     }
 
@@ -281,7 +281,7 @@ impl MetricSet {
     /// Convert the incremental metric into an absolute one, using the
     /// state buffer to keep track of the value throughout the entire
     /// application uptime.
-    fn incremental_to_absolute(&mut self, metric: Metric) -> Option<Metric> {
+    fn incremental_to_absolute(&mut self, metric: Metric) -> Metric {
         let mut entry = MetricEntry(metric.into_absolute());
         let mut existing = self.0.take(&entry).unwrap_or_else(|| {
             // Start from zero value if the entry is not found.
@@ -290,7 +290,7 @@ impl MetricSet {
         existing.data.value.add(&entry.data.value);
         entry.data.value = existing.data.value.clone();
         self.0.insert(existing);
-        Some(entry.0)
+        entry.0
     }
 
     /// Convert the absolute metric into an incremental by calculating

src/sinks/vector.rs

@@ -74,7 +74,7 @@ impl SinkConfig for VectorSinkConfig {
             self.send_buffer_bytes,
         );
 
-        sink_config.build(cx, encode_event)
+        sink_config.build(cx, |event| Some(encode_event(event)))
     }
 
     fn input_type(&self) -> DataType {
@@ -92,7 +92,7 @@ enum HealthcheckError {
     ConnectError { source: std::io::Error },
 }
 
-fn encode_event(event: Event) -> Option<Bytes> {
+fn encode_event(event: Event) -> Bytes {
     let event = proto::EventWrapper::from(event);
     let event_len = event.encoded_len();
     let full_len = event_len + 4;
@@ -101,7 +101,7 @@ fn encode_event(event: Event) -> Option<Bytes> {
     out.put_u32(event_len as u32);
     event.encode(&mut out).unwrap();
 
-    Some(out.into())
+    out.into()
 }
 
 #[cfg(test)]