vrl: add pulse_log() and pulse_inc_counter() functions

Cargo.toml

@@ -15,14 +15,14 @@ ahash                  = "0.8"
 anyhow                 = "1"
 assert_matches         = "1.5.0"
 async-trait            = "0.1"
-aws-config             = "1.6.3"
+aws-config             = "1.8.0"
 aws-credential-types   = "1.2.3"
-aws-sdk-sqs            = "1.72.0"
-aws-sigv4              = "1.3.2"
+aws-sdk-sqs            = "1.73.0"
+aws-sigv4              = "1.3.3"
 aws-smithy-async       = "1.2.5"
 aws-smithy-http        = "0.62.1"
-aws-smithy-runtime-api = { version = "1.8.0", features = ["test-util"] }
-aws-smithy-types       = "1.3.1"
+aws-smithy-runtime-api = { version = "1.8.1", features = ["test-util"] }
+aws-smithy-types       = "1.3.2"
 axum                   = "0.8.4"
 backoff                = { version = "0.4.0", features = ["tokio"] }
 base64ct               = "1.8.0"
@@ -40,7 +40,7 @@ bd-test-helpers        = { git = "https://github.com/bitdriftlabs/shared-core.gi
 bd-time                = { git = "https://github.com/bitdriftlabs/shared-core.git" }
 built                  = { version = "0.8", features = ["git2"] }
 bytes                  = "1"
-cc                     = "1.2.26"
+cc                     = "1.2.27"
 clap                   = { version = "4.5.40", features = ["derive", "env"] }
 comfy-table            = "7.1.4"
 console-subscriber     = "0.4.1"
@@ -104,13 +104,13 @@ rand              = { version = "0.9", features = ["small_rng"] }
 rand_xoshiro      = "0.7"
 regex             = "1"
 
-reqwest = { version = "0.12.20", default-features = false, features = [
+reqwest = { version = "0.12.22", default-features = false, features = [
   "rustls-tls-webpki-roots",
   "json",
 ] }
 
 reusable-fmt               = "0.2.0"
-rustls                     = "0.23.27"
+rustls                     = "0.23.28"
 serde                      = { version = "1", features = ["derive"] }
 serde_json                 = "1"
 serde_yaml                 = "0.9.34"
@@ -132,7 +132,7 @@ unwrap-infallible          = "0.1.5"
 url                        = "2.5.4"
 uuid                       = { version = "1.17.0", features = ["v4"] }
 
-vrl = { git = "https://github.com/mattklein123/vrl.git", branch = "performance-20240826", default-features = false, features = [
+vrl = { git = "https://github.com/mattklein123/vrl.git", branch = "performance-20250625", default-features = false, features = [
   "compiler",
   "value",
   "diagnostic",

pulse-metrics/src/lru_map.rs

@@ -227,7 +227,7 @@ impl<K: Eq + Hash, V, S: BuildHasher + Clone> LruMap<K, V, S> {
       .from_hash(hash, |key_to_match| key_to_match.key.equivalent(key))
     {
       shard.adjust_lru(key.as_ref());
-      let key: *const K = &key.key;
+      let key: *const K = &raw const key.key;
       let value: *const V = value;
       Some(Ref {
         _guard: map,

pulse-metrics/src/pipeline/inflow/wire/util.rs

@@ -293,6 +293,7 @@ impl SocketHandler {
     }
   }
 
+  #[allow(clippy::cognitive_complexity)]
   pub(super) async fn run<T>(mut self, mut socket: T, mut shutdown: ComponentShutdown)
   where
     T: AsyncRead + AsyncWrite + Unpin,

pulse-metrics/src/pipeline/mod.rs

@@ -458,6 +458,7 @@ impl MetricPipeline {
   }
 
   // Updating an existing configuration in place.
+  #[allow(clippy::cognitive_complexity)]
   pub async fn update_config(&self, config: PipelineConfig) -> anyhow::Result<()> {
     check_routes(&config)?;
 

pulse-metrics/src/pipeline/processor/aggregation/mod.rs

@@ -313,6 +313,7 @@ impl AggregationProcessor {
     drop(shutdown);
   }
 
+  #[allow(clippy::cognitive_complexity)]
   async fn do_flush<JitterType: DurationJitter>(self: &Arc<Self>) {
     let _flush_time = self.stats.flush_time.start_timer();
     log::debug!("performing aggregation flush");

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

@@ -29,7 +29,7 @@ fn make_time_provider(time: i64) -> TestTimeProvider {
   }
 }
 
-#[ignore]
+#[ignore = "stress test, run manually only"]
 #[tokio::test(flavor = "multi_thread")]
 async fn stress() {
   let mut helper = HelperBuilder::default()

pulse-metrics/src/pipeline/processor/cardinality_limiter/mod.rs

@@ -12,10 +12,11 @@ mod mod_test;
 use super::{PipelineProcessor, ProcessorFactoryContext};
 use crate::pipeline::PipelineDispatch;
 use crate::protos::metric::{MetricId, ParsedMetric};
-use crate::vrl::ProgramWrapper;
+use crate::vrl::{ProgramWrapper, PulseDynamicState};
 use ahash::{HashMap, HashMapExt};
 use async_trait::async_trait;
 use bd_log::warn_every;
+use bd_server_stats::stats::Scope;
 use bd_shutdown::{ComponentShutdown, ComponentShutdownTriggerHandle};
 use bd_time::TimeDurationExt;
 use cardinality_limiter_config::per_pod_limit::Override_limit_location;
@@ -157,14 +158,17 @@ impl<H: Hasher + Default + Send + 'static> K8sPodLimiter<H> {
     buckets: usize,
     mut k8s_pods_info: OwnedPodsInfoSingleton,
     shutdown: ComponentShutdown,
+    scope: Scope,
   ) -> anyhow::Result<Self> {
     let pod_limiter_config = K8sPodLimiterConfig {
       default_size_limit: config.default_size_limit,
       vrl_program: config
         .override_limit_location
         .as_ref()
         .map(|override_limit_location| match override_limit_location {
-          Override_limit_location::VrlProgram(vrl_program) => ProgramWrapper::new(vrl_program),
+          Override_limit_location::VrlProgram(vrl_program) => {
+            ProgramWrapper::new(vrl_program, PulseDynamicState::new(scope))
+          },
         })
         .transpose()?,
     };
@@ -298,6 +302,7 @@ impl CardinalityLimiterProcessor {
         config.buckets as usize,
         (context.k8s_watch_factory)().await?.make_owned(),
         context.shutdown_trigger_handle.make_shutdown(),
+        context.scope.clone(),
       )?),
     };
 

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

@@ -20,7 +20,7 @@ use crate::protos::metric::{
   ParsedMetric,
   TagValue,
 };
-use crate::vrl::ProgramWrapper;
+use crate::vrl::{ProgramWrapper, PulseDynamicState};
 use async_trait::async_trait;
 use bd_log::warn_every;
 use bd_server_stats::stats::Scope;
@@ -65,7 +65,10 @@ pub struct MutateProcessor {
 impl MutateProcessor {
   pub fn new(config: &MutateConfig, context: ProcessorFactoryContext) -> anyhow::Result<Self> {
     Ok(Self {
-      program: ProgramWrapper::new(&config.vrl_program)?,
+      program: ProgramWrapper::new(
+        &config.vrl_program,
+        PulseDynamicState::new(context.scope.clone()),
+      )?,
       dispatcher: context.dispatcher,
       stats: MutateStats::new(&context.scope),
     })

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

@@ -39,6 +39,8 @@ use pulse_common::k8s::test::make_node_info;
 use pulse_common::metadata::{Metadata, PodMetadata};
 use pulse_protobuf::protos::pulse::config::processor::v1::mutate::MutateConfig;
 use std::sync::Arc;
+use time::ext::NumericalStdDuration;
+use tokio::time::sleep;
 use vrl::btreemap;
 
 // TODO(mattklein123): The use of env vars in these tests needs to be scoped/randomized as any
@@ -72,6 +74,69 @@ impl Helper {
   }
 }
 
+#[tokio::test]
+async fn pulse_inc_counter() {
+  let mut helper = Helper::new(
+    r#"
+if .name == "kube_job_status_failed" {
+  pulse_inc_counter("hello", 1)
+}
+    "#,
+  );
+
+  helper
+    .expect_send_and_receive(
+      make_abs_counter("kube_job_status_failed", &[], 0, 1.0),
+      vec![make_abs_counter("kube_job_status_failed", &[], 0, 1.0)],
+    )
+    .await;
+  helper
+    .expect_send_and_receive(
+      make_abs_counter("kube_job_status_success", &[], 0, 1.0),
+      vec![make_abs_counter("kube_job_status_success", &[], 0, 1.0)],
+    )
+    .await;
+
+  helper
+    .helper
+    .stats_helper
+    .assert_counter_eq(1, "processor:hello", &labels! {});
+}
+
+#[tokio::test(start_paused = true)]
+async fn pulse_log() {
+  let mut helper = Helper::new(
+    r#"
+if .name == "kube_job_status_failed" {
+  pulse_log("hello world")
+  pulse_log("different rate limit")
+}
+    "#,
+  );
+
+  helper
+    .expect_send_and_receive(
+      make_abs_counter("kube_job_status_failed", &[], 0, 1.0),
+      vec![make_abs_counter("kube_job_status_failed", &[], 0, 1.0)],
+    )
+    .await;
+  helper
+    .expect_send_and_receive(
+      make_abs_counter("kube_job_status_failed", &[], 0, 1.0),
+      vec![make_abs_counter("kube_job_status_failed", &[], 0, 1.0)],
+    )
+    .await;
+
+  sleep(1.std_seconds()).await;
+
+  helper
+    .expect_send_and_receive(
+      make_abs_counter("kube_job_status_failed", &[], 0, 1.0),
+      vec![make_abs_counter("kube_job_status_failed", &[], 0, 1.0)],
+    )
+    .await;
+}
+
 #[tokio::test]
 async fn flatten_histogram() {
   let mut helper = Helper::new(

pulse-metrics/src/vrl/mod.rs

@@ -6,7 +6,10 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 use crate::protos::metric::{EditableParsedMetric, MetricType, ParsedMetric, TagValue};
+use crate::vrl::pulse_inc_counter::PulseIncCounter;
+use crate::vrl::pulse_log::PulseLog;
 use anyhow::{anyhow, bail};
+use bd_server_stats::stats::Scope;
 use itertools::Itertools;
 use pulse_common::metadata::Metadata;
 use vrl::compiler::state::{ExternalEnv, RuntimeState};
@@ -26,6 +29,9 @@ use vrl::path::{OwnedTargetPath, PathPrefix};
 use vrl::value::kind::Collection;
 use vrl::value::{Kind, Value};
 
+mod pulse_inc_counter;
+mod pulse_log;
+
 //
 // EditableMetricVrlTarget
 //
@@ -68,7 +74,11 @@ impl Target for EditableMetricVrlTarget<'_> {
         if path.path.is_root() {
           return Err("assigning to root event is not supported".to_string());
         }
-        match path.path.to_alternative_components(3)[0].as_slice() {
+        let Some(path) = path.path.to_alternative_components(3) else {
+          return Err("path does not refer to a field".to_string());
+        };
+
+        match path.as_slice() {
           [name] if name == "name" => {
             self.metric.change_name(
               value
@@ -133,7 +143,10 @@ impl Target for EditableMetricVrlTarget<'_> {
           // asked for.
           return Ok(None);
         }
-        match path.path.to_alternative_components(3)[0].as_slice() {
+        let Some(path) = path.path.to_alternative_components(3) else {
+          return Ok(None);
+        };
+        match path.as_slice() {
           [name] if name == "name" => Ok(Some(OwnedValueOrRef::Owned(Value::Bytes(
             self.metric.metric().metric().get_id().name().clone(),
           )))),
@@ -197,7 +210,10 @@ impl Target for EditableMetricVrlTarget<'_> {
     match path.prefix {
       PathPrefix::Event => {
         log::trace!("removing: path={path}");
-        let removed = match path.path.to_alternative_components(3)[0].as_slice() {
+        let Some(path) = path.path.to_alternative_components(3) else {
+          return Err("path does not refer to a field".to_string());
+        };
+        let removed = match path.as_slice() {
           [tags, tag_name] if tags == "tags" => self
             .metric
             .delete_tag(tag_name.as_bytes())
@@ -268,6 +284,21 @@ pub struct RunWithMetricResult {
   pub flatten_prom_histogram_and_summary: bool,
 }
 
+//
+// PulseDynamicState
+//
+
+pub struct PulseDynamicState {
+  scope: Scope,
+}
+
+impl PulseDynamicState {
+  #[must_use]
+  pub fn new(scope: Scope) -> Self {
+    Self { scope }
+  }
+}
+
 //
 // ProgramWrapper
 //
@@ -277,7 +308,7 @@ pub struct ProgramWrapper {
 }
 
 impl ProgramWrapper {
-  pub fn new(program: &str) -> anyhow::Result<Self> {
+  pub fn new(program: &str, dynamic_state: PulseDynamicState) -> anyhow::Result<Self> {
     let external = ExternalEnv::new_with_kind(
       Kind::object(
         Collection::empty()
@@ -291,13 +322,15 @@ impl ProgramWrapper {
       Metadata::schema(),
     );
 
-    let result = compile_with_external(
-      program,
-      &vrl::stdlib::all(),
-      &external,
-      CompileConfig::default(),
-    )
-    .map_err(|e| anyhow!("VRL compile error: {}", Formatter::new(program, e)))?;
+    let mut functions = vrl::stdlib::all();
+    functions.push(Box::new(PulseLog));
+    functions.push(Box::new(PulseIncCounter));
+
+    let mut compile_config = CompileConfig::default();
+    compile_config.set_custom(dynamic_state);
+
+    let result = compile_with_external(program, &functions, &external, compile_config)
+      .map_err(|e| anyhow!("VRL compile error: {}", Formatter::new(program, e)))?;
     if !result.warnings.is_empty() {
       bail!(Formatter::new(program, result.warnings).to_string(),);
     }