feat(uptime): add assertion checks to check_executor (#437)

Author: klochek

Cargo.lock

@@ -65,7 +65,8 @@ iprange = "0.6.7"
 http = "1.2.0"
 relay-pattern = { git = "https://github.com/getsentry/relay", rev = "7ba03b5f6ffe95c2429ba315f661d2b8ee3c0efe" }
 pest = "2.7.15"
-axum = "0.8.6"
+axum = { version = "0.8.6", features = ["macros"] }
+associative-cache = "2.0.0"
 
 [patch.crates-io]
 rdkafka = { git = "https://github.com/fede1024/rust-rdkafka" }
@@ -82,3 +83,5 @@ redis-test-macro = { path = "./redis-test-macro" }
 ntest = "0.9.3"
 socket-server-mocker = "0.5.0"
 httpmock = "0.7.0-rc.1"
+tower = { version = "0.5.2", features = ["util"] }
+http-body-util = "0.1.0"

Cargo.toml

@@ -0,0 +1,76 @@
+use crate::assertions::{self, compiled};
+use associative_cache::{Capacity8192, HashDirectMapped, LruReplacement, LruTimestamp};
+use std::{
+    fmt,
+    sync::{Arc, RwLock},
+    time::Instant,
+};
+
+pub struct CacheEntry {
+    assertion: Arc<compiled::Assertion>,
+    timestamp: RwLock<Instant>,
+}
+
+impl LruTimestamp for CacheEntry {
+    type Timestamp<'a> = Instant;
+
+    fn get_timestamp(&self) -> Self::Timestamp<'_> {
+        *self.timestamp.read().expect("not poisoned")
+    }
+
+    fn update_timestamp(&self) {
+        *self.timestamp.write().expect("not poisoned") = Instant::now();
+    }
+}
+
+pub struct Cache {
+    compiled: Arc<RwLock<AssocCache>>,
+}
+
+impl fmt::Debug for Cache {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Cache").finish()
+    }
+}
+
+type AssocCache = associative_cache::AssociativeCache<
+    assertions::Assertion,
+    CacheEntry,
+    Capacity8192,
+    HashDirectMapped,
+    LruReplacement,
+>;
+
+impl Cache {
+    pub fn new() -> Self {
+        Self {
+            compiled: Arc::new(RwLock::new(AssocCache::default())),
+        }
+    }
+    pub fn get_or_compile(
+        &self,
+        key: &assertions::Assertion,
+    ) -> Result<Arc<compiled::Assertion>, compiled::Error> {
+        if let Some(entry) = self.compiled.read().expect("not poisoned").get(key) {
+            return Ok(entry.assertion.clone());
+        }
+
+        let comp = compiled::compile(key);
+        match comp {
+            Ok(comp) => {
+                let mut wl = self.compiled.write().expect("not poisoned");
+                let comp = Arc::new(comp);
+                wl.insert(
+                    key.clone(),
+                    CacheEntry {
+                        assertion: comp.clone(),
+                        timestamp: RwLock::new(Instant::now()),
+                    },
+                );
+
+                Ok(comp)
+            }
+            Err(err) => Err(err),
+        }
+    }
+}

src/assertions/cache.rs

@@ -3,6 +3,7 @@ use jsonpath_rust::{
     parser::{errors::JsonPathError, model::JpQuery, parse_json_path},
     query::js_path_process,
 };
+use serde::Serialize;
 use std::{
     borrow::{Borrow, BorrowMut},
     str::FromStr,
@@ -11,7 +12,7 @@ use std::{
 const GLOB_COMPLEXITY_LIMIT: u64 = 20;
 const ASSERTION_MAX_GAS: u32 = 100;
 
-#[derive(thiserror::Error, Debug)]
+#[derive(thiserror::Error, Debug, Serialize)]
 pub enum Error {
     #[error("Invalid glob: {0}")]
     InvalidGlob(String),
@@ -58,7 +59,7 @@ impl BorrowMut<u32> for Gas {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct Assertion {
     root: Op,
 }
@@ -75,7 +76,7 @@ impl Assertion {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq)]
 enum Comparison {
     LessThan,
     GreaterThan,
@@ -94,7 +95,7 @@ impl From<&super::Comparison> for Comparison {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq)]
 enum HeaderOperand {
     Literal { value: Value },
     Glob { value: relay_pattern::Pattern },
@@ -123,7 +124,7 @@ impl TryFrom<&super::HeaderOperand> for HeaderOperand {
     }
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, PartialEq)]
 enum Value {
     I64(i64),
     F64(f64),
@@ -132,7 +133,7 @@ enum Value {
 
 impl Eq for Value {}
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq)]
 enum HeaderComparison {
     Always,
     Never,
@@ -261,7 +262,7 @@ impl HeaderComparison {
     }
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, PartialEq)]
 enum Op {
     And {
         children: Vec<Op>,

src/assertions/compiled.rs

@@ -1,38 +1,39 @@
+pub mod cache;
 pub mod compiled;
 
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
 pub struct Assertion {
-    root: Op,
+    pub(crate) root: Op,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
 #[serde(tag = "cmp", rename_all = "snake_case")]
-enum Comparison {
+pub(crate) enum Comparison {
     LessThan,
     GreaterThan,
     Equal,
     NotEqual,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
 #[serde(rename_all = "snake_case")]
-struct GlobPattern {
-    value: String,
+pub(crate) struct GlobPattern {
+    pub(crate) value: String,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
 #[serde(tag = "header_op", rename_all = "snake_case")]
-enum HeaderOperand {
+pub(crate) enum HeaderOperand {
     Literal { value: String },
     Glob { pattern: GlobPattern },
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
 #[serde(tag = "header_cmp", rename_all = "snake_case")]
-enum HeaderComparison {
+pub(crate) enum HeaderComparison {
     Always,
     Never,
     Equals { test_value: HeaderOperand },
@@ -41,9 +42,9 @@ enum HeaderComparison {
     GreaterThan { test_value: String },
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
 #[serde(tag = "op", rename_all = "snake_case")]
-enum Op {
+pub(crate) enum Op {
     And {
         children: Vec<Op>,
     },

src/assertions/mod.rs

@@ -168,7 +168,7 @@ impl RedisConfigProvider {
                     .get_config_store()
                     .write()
                     .unwrap()
-                    .add_config(Arc::new(config));
+                    .add_config(config);
             }
             let partition_loading_time = partition_start_loading.elapsed().as_secs_f64();
             metrics::histogram!(
@@ -267,7 +267,7 @@ impl RedisConfigProvider {
                         .get_config_store()
                         .write()
                         .unwrap()
-                        .add_config(Arc::new(config));
+                        .add_config(config);
                 }
                 let partition_update_duration = partition_update_start.elapsed().as_secs_f64();
                 metrics::histogram!(

src/check_config_provider/redis_config_provider.rs

@@ -16,7 +16,7 @@ use crate::checker::HttpChecker;
 use crate::config_store::Tick;
 use crate::producer::ResultsProducer;
 use crate::types::check_config::CheckConfig;
-use crate::types::result::{CheckResult, CheckStatus, CheckStatusReasonType};
+use crate::types::result::{CheckResult, CheckStatus};
 
 const SLOW_POLL_THRESHOLD: Duration = Duration::from_millis(100);
 const LONG_DELAY_THRESHOLD: Duration = Duration::from_millis(100);
@@ -38,6 +38,21 @@ pub struct ScheduledCheck {
 }
 
 impl ScheduledCheck {
+    pub fn new(
+        kind: CheckKind,
+        tick: Tick,
+        config: Arc<CheckConfig>,
+        resolve_tx: Sender<Option<CheckResult>>,
+    ) -> ScheduledCheck {
+        ScheduledCheck {
+            kind,
+            tick,
+            config,
+            resolve_tx,
+            retry_count: 0,
+        }
+    }
+
     #[cfg(test)]
     pub fn new_for_test(tick: Tick, config: CheckConfig) -> Self {
         let (resolve_tx, _) = tokio::sync::oneshot::channel();
@@ -107,13 +122,7 @@ impl CheckSender {
     ) -> anyhow::Result<Receiver<Option<CheckResult>>> {
         let (resolve_tx, resolve_rx) = oneshot::channel();
 
-        let scheduled_check = ScheduledCheck {
-            tick,
-            config,
-            resolve_tx,
-            retry_count: 0,
-            kind: check_kind,
-        };
+        let scheduled_check = ScheduledCheck::new(check_kind, tick, config, resolve_tx);
 
         self.queue_size.fetch_add(1, Ordering::Relaxed);
 
@@ -284,7 +293,7 @@ async fn executor_loop(
 }
 
 #[allow(clippy::too_many_arguments)]
-async fn do_check(
+pub(crate) async fn do_check(
     failure_retries: u16,
     scheduled_check: ScheduledCheck,
     job_checker: Arc<HttpChecker>,
@@ -367,15 +376,7 @@ fn record_result_metrics(result: &CheckResult, is_retry: bool, will_retry: bool)
         CheckStatus::MissedWindow => "missed_window",
         CheckStatus::DisallowedByRobots => "disallowed_by_robots",
     };
-    let failure_reason = match status_reason.as_ref().map(|r| r.status_type) {
-        Some(CheckStatusReasonType::Failure) => Some("failure"),
-        Some(CheckStatusReasonType::DnsError) => Some("dns_error"),
-        Some(CheckStatusReasonType::Timeout) => Some("timeout"),
-        Some(CheckStatusReasonType::TlsError) => Some("tls_error"),
-        Some(CheckStatusReasonType::ConnectionError) => Some("connection_error"),
-        Some(CheckStatusReasonType::RedirectError) => Some("redirect_error"),
-        None => None,
-    };
+    let failure_reason = status_reason.as_ref().map(|r| r.status_type.as_str());
     let status_code = match request_info.as_ref().and_then(|a| a.http_status_code) {
         None => "none".to_string(),
         Some(status) => status.to_string(),