mutate/VRL: allow setting .mtype for simple metrics (#113)

Fixes BIT-8327

Signed-off-by: Matt Klein <mklein@bitdrift.io>

Author: mattklein123

.github/workflows/build.yaml

@@ -17,7 +17,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Cache
-        uses: Swatinem/rust-cache@v2.7.3
+        uses: Swatinem/rust-cache@v2.9.1
 
       - name: Rust Toolchain
         uses: actions-rs/toolchain@v1
@@ -71,7 +71,7 @@ jobs:
           profile: minimal
 
       - name: Cache
-        uses: Swatinem/rust-cache@v2.7.3
+        uses: Swatinem/rust-cache@v2.9.1
 
       - name: Build
         env:

pulse-metrics/src/pipeline/processor/mutate/mod_test.rs

@@ -244,6 +244,122 @@ async fn flatten_summary() {
     .await;
 }
 
+#[tokio::test]
+async fn change_mtype_counter_to_gauge() {
+  let mut helper = Helper::new(
+    r#"
+.mtype = "gauge"
+    "#,
+  );
+
+  helper
+    .expect_send_and_receive(
+      make_abs_counter("foo", &[("tag", "value")], 0, 1.0),
+      vec![make_gauge("foo", &[("tag", "value")], 0, 1.0)],
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn change_mtype_gauge_to_delta_gauge() {
+  let mut helper = Helper::new(
+    r#"
+.mtype = "delta_gauge"
+    "#,
+  );
+
+  helper
+    .expect_send_and_receive(
+      make_gauge("foo", &[("tag", "value")], 0, 1.0),
+      vec![make_metric_ex(
+        "foo",
+        &[("tag", "value")],
+        0,
+        Some(MetricType::DeltaGauge),
+        None,
+        MetricValue::Simple(1.0),
+        MetricSource::PromRemoteWrite,
+        DownstreamId::LocalOrigin,
+        None,
+      )],
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn reject_histogram_mtype_change() {
+  let helper = Helper::new(
+    r#"
+.mtype = "counter"
+    "#,
+  );
+
+  helper
+    .processor
+    .clone()
+    .recv_samples(vec![make_metric_ex(
+      "foo",
+      &[],
+      0,
+      Some(MetricType::Histogram),
+      None,
+      MetricValue::Histogram(HistogramData {
+        buckets: vec![HistogramBucket {
+          le: 3.0,
+          count: 1.0,
+        }],
+        sample_count: 1.0,
+        sample_sum: 2.0,
+      }),
+      MetricSource::PromRemoteWrite,
+      DownstreamId::LocalOrigin,
+      None,
+    )])
+    .await;
+
+  helper
+    .helper
+    .stats_helper
+    .assert_counter_eq(1, "processor:drop_error", &labels! {});
+}
+
+#[tokio::test]
+async fn reject_summary_mtype_change() {
+  let helper = Helper::new(
+    r#"
+.mtype = "counter"
+    "#,
+  );
+
+  helper
+    .processor
+    .clone()
+    .recv_samples(vec![make_metric_ex(
+      "foo",
+      &[],
+      0,
+      Some(MetricType::Summary),
+      None,
+      MetricValue::Summary(SummaryData {
+        quantiles: vec![SummaryBucket {
+          quantile: 3.0,
+          value: 1.0,
+        }],
+        sample_count: 1.0,
+        sample_sum: 2.0,
+      }),
+      MetricSource::PromRemoteWrite,
+      DownstreamId::LocalOrigin,
+      None,
+    )])
+    .await;
+
+  helper
+    .helper
+    .stats_helper
+    .assert_counter_eq(1, "processor:drop_error", &labels! {});
+}
+
 #[tokio::test]
 async fn modify_all_tags() {
   let mut helper = Helper::new(
@@ -768,6 +884,19 @@ fn editable_parsed_metric() {
     assert_matches!(metric.cached_metric(), CachedMetric::NotInitialized);
   }
 
+  {
+    let mut metric = make_abs_counter("prod:some_service:test:name", &[], 0, 1.0);
+    metric.initialize_cache(&metric_cache);
+    let mut editable_metric = EditableParsedMetric::new(&mut metric);
+    editable_metric.change_mtype(b"gauge").unwrap();
+    drop(editable_metric);
+    assert_eq!(
+      metric,
+      make_gauge("prod:some_service:test:name", &[], 0, 1.0)
+    );
+    assert_matches!(metric.cached_metric(), CachedMetric::NotInitialized);
+  }
+
   {
     let mut metric = make_abs_counter(
       "prod:some_service:test:name",

pulse-metrics/src/protos/metric.rs

@@ -618,6 +618,7 @@ pub struct EditableParsedMetric<'a> {
   metric: &'a mut ParsedMetric,
   tag_insertion_index: Option<usize>,
   name_changed: bool,
+  mtype_changed: bool,
   deleted_tags: Option<Vec<bool>>,
 }
 
@@ -627,6 +628,7 @@ impl<'a> EditableParsedMetric<'a> {
       metric,
       tag_insertion_index: None,
       name_changed: false,
+      mtype_changed: false,
       deleted_tags: None,
     }
   }
@@ -736,6 +738,28 @@ impl<'a> EditableParsedMetric<'a> {
     self.name_changed = true;
   }
 
+  pub fn change_mtype(&mut self, mtype: &[u8]) -> Result<(), &'static str> {
+    if !matches!(self.metric.metric.value, MetricValue::Simple(_)) {
+      return Err("assigning to mtype is only supported for simple metrics");
+    }
+
+    let mtype = match mtype {
+      b"counter" => MetricType::Counter(CounterType::Absolute),
+      b"gauge" => MetricType::Gauge,
+      b"delta_gauge" => MetricType::DeltaGauge,
+      b"direct_gauge" => MetricType::DirectGauge,
+      b"timer" => MetricType::Timer,
+      _ => return Err("assigning to mtype requires a supported metric type string"),
+    };
+
+    if self.metric.metric.id.mtype() != Some(mtype) {
+      self.metric.metric.id.set_mtype(mtype);
+      self.mtype_changed = true;
+    }
+
+    Ok(())
+  }
+
   #[must_use]
   pub fn metric(&self) -> &ParsedMetric {
     self.metric
@@ -765,7 +789,11 @@ impl Drop for EditableParsedMetric<'_> {
       self.metric.metric.id.tags.sort_unstable();
     }
 
-    if self.tag_insertion_index.is_some() || self.name_changed || self.deleted_tags.is_some() {
+    if self.tag_insertion_index.is_some()
+      || self.name_changed
+      || self.mtype_changed
+      || self.deleted_tags.is_some()
+    {
       self.metric.cached_metric = CachedMetric::NotInitialized;
     }
   }

pulse-metrics/src/vrl/mod.rs

@@ -18,6 +18,7 @@ use vrl::compiler::state::{ExternalEnv, RuntimeState};
 use vrl::compiler::{
   CompileConfig,
   Context,
+  ExpressionError,
   OwnedValueOrRef,
   Program,
   Resolved,
@@ -43,13 +44,15 @@ mod pulse_log;
 struct EditableMetricVrlTarget<'a> {
   metric: EditableParsedMetric<'a>,
   flatten_prom_histogram_and_summary: bool,
+  runtime_error: Option<&'static str>,
 }
 
 impl<'a> EditableMetricVrlTarget<'a> {
   const fn new(metric: EditableParsedMetric<'a>) -> Self {
     Self {
       metric,
       flatten_prom_histogram_and_summary: false,
+      runtime_error: None,
     }
   }
 }
@@ -90,6 +93,18 @@ impl Target for EditableMetricVrlTarget<'_> {
                 .clone(),
             );
           },
+          [mtype] if mtype == "mtype" => {
+            let Some(mtype) = value.as_bytes() else {
+              self
+                .runtime_error
+                .get_or_insert("assigning to mtype requires a string");
+              return Ok(());
+            };
+
+            if let Err(error) = self.metric.change_mtype(mtype) {
+              self.runtime_error.get_or_insert(error);
+            }
+          },
           [tags, tag_name] if tags == "tags" => {
             self.metric.add_or_change_tag(TagValue {
               tag: tag_name.to_bytes(),
@@ -368,10 +383,19 @@ impl ProgramWrapper {
       Context::new(&mut target, &mut state, &timezone).with_dynamic_state(&mut dynamic_data);
     let resolved = self.program.resolve(&mut ctx);
     let flatten_prom_histogram_and_summary = target.flatten_prom_histogram_and_summary;
+    let runtime_error = target.runtime_error;
 
     // Drop target to release mutable borrow of sample before applying downstream ID suffix.
     drop(target);
 
+    let resolved = match (resolved, runtime_error) {
+      (
+        Ok(_) | Err(ExpressionError::Return { .. } | ExpressionError::Abort { .. }),
+        Some(error),
+      ) => Err(ExpressionError::from(error)),
+      (resolved, _) => resolved,
+    };
+
     // Apply downstream ID suffix if set by VRL program.
     if let Some(suffix) = dynamic_data.downstream_id_suffix {
       sample.append_to_downstream_id(&suffix);