✅ Integration Testing

.env.example

@@ -8,4 +8,6 @@ MAINNET_RPC_URL=
 # Config
 NETWORK=testnet
 ORACLE_ADDRESS=0x
-PAIRS=BTC/USD,ETH/USD
+PAIRS=BTC/USD,ETH/USD
+IGNORE_SOURCES=BITSTAMP,DEFILLAMA
+IGNORE_PUBLISHERS=BINANCE

Cargo.toml

@@ -43,3 +43,6 @@ strum = { version = "0.25.0", features = ["derive"] }
 tokio = { version = "1", features = ["full"] }
 url = "2.5.0"
 uuid = { version = "1.4", features = ["fast-rng", "v4", "serde"] }
+
+[dev-dependencies]
+rstest = "0.18.2"

prometheus/alertmanager.yml

@@ -1,18 +1,16 @@
 global:
-  smtp_smarthost: "${SMTP_HOST}"
-  smtp_from: "${SMTP_FROM}"
-  smtp_auth_username: "${SMTP_AUTH_USERNAME}"
-  smtp_auth_password: "${SMTP_AUTH_PASSWORD}"
-  smtp_require_tls: true
+  resolve_timeout: 5m
 
 route:
   group_by: ["instance", "severity"]
   group_wait: 30s
   group_interval: 5m
   repeat_interval: 4h
-  receiver: "email_configs"
+  receiver: "main-telegram"
+
 receivers:
-  - name: "email_configs"
-    email_configs:
-      - to: "${EMAIL_TO}"
-        send_resolved: true
+  - name: "main-telegram"
+    telegram_configs:
+      - bot_token: "${TELEGRAM_BOT_TOKEN}"
+        chat_id: -1001904637278
+        parse_mode: ""

prometheus/alerts.rules.yml

@@ -1,11 +1,35 @@
 groups:
-  - name: example
+  - name: Oracle
     rules:
       - alert: TimeSinceLastUpdateTooHigh
         expr: time_since_last_update_seconds > 1200
         for: 5m
         labels:
-          severity: critical
+          severity: warning
         annotations:
           summary: "Time since the last update is too high"
-          description: "The time since the last update of {{ $labels.pair }} from {{ $labels.source }} has exceeded 1200 seconds."
+          description: "The time since the last update from {{ $labels.publisher }} has exceeded 1200 seconds."
+      - alert: WrongPrice
+        expr: price_deviation > 0.02
+        for: 5m
+        labels:
+          severity: critical
+        annotations:
+          summary: "Price deviation is too high"
+          description: "The price deviation of {{ $labels.pair }} from {{ $labels.source }} has exceeded 5%."
+      - alert: TooFewSources
+        expr: num_sources < 5
+        for: 5m
+        labels:
+          severity: critical
+        annotations:
+          summary: "Too few sources"
+          description: "The number of sources for {{ $labels.pair }} has fallen below 5."
+      - alert: SourceDeviation
+        expr: price_deviation_source > 0.02
+        for: 5m
+        labels:
+          severity: critical
+        annotations:
+          summary: "Source deviation is too high"
+          description: "The source deviation of {{ $labels.pair }} from {{ $labels.source }} has exceeded 5%."

prometheus/prometheus.yml

@@ -6,10 +6,12 @@ scrape_configs:
   - job_name: "prometheus_monitoring"
     static_configs:
       - targets: ["host.docker.internal:8081"]
+
 rule_files:
   - "alerts.rules.yml"
+
 alerting:
   alertmanagers:
     - static_configs:
         - targets:
-            - localhost:9093
+            - host.docker.internal:9093

scripts/db.sh

@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+set -ex
+cd "$(dirname "$0")"
+
+docker rm -f oracle_monitoring
+docker run -d --name=oracle_monitoring -p 5432:5432 -e POSTGRES_PASSWORD=password postgres
+sleep 5
+
+DATABASE_URL=postgresql://postgres:password@localhost:5432/test cargo test

scripts/prom.sh

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+set -e
+cd "$(dirname "$0")"
+
+echo "Starting prometheus on: http://localhost:9090"
+
+docker rm prom
+docker run \
+    -p 9090:9090 \
+    -v $(pwd)/../prometheus:/etc/prometheus \
+    --name prom \
+    prom/prometheus

src/config.rs

@@ -21,6 +21,8 @@ pub enum NetworkName {
     Mainnet,
     #[strum(ascii_case_insensitive)]
     Testnet,
+    #[strum(ascii_case_insensitive)]
+    Katana,
 }
 
 #[derive(Debug, Clone)]
@@ -93,6 +95,26 @@ impl Config {
                     },
                 }
             }
+            NetworkName::Katana => {
+                let url = Url::parse("http://localhost:5050").expect("Invalid JSON RPC URL");
+                let rpc_client = JsonRpcClient::new(HttpTransport::new(url)); // Katana URL
+
+                let (decimals, sources, publishers, publisher_registry_address) =
+                    init_oracle_config(&rpc_client, oracle_address, pairs.clone()).await;
+
+                Self {
+                    pairs,
+                    sources,
+                    publishers,
+                    decimals,
+                    network: Network {
+                        name: "katana".to_string(),
+                        provider: Arc::new(rpc_client),
+                        oracle_address,
+                        publisher_registry_address,
+                    },
+                }
+            }
         }
     }
 }
@@ -140,9 +162,27 @@ async fn init_oracle_config(
 
     let publishers = publishers[1..].to_vec();
 
+    // Exclude publishers that are not supported by the monitoring service
+    let excluded_publishers = std::env::var("IGNORE_PUBLISHERS")
+        .unwrap_or("".to_string())
+        .split(',')
+        .map(|publisher| publisher.to_string())
+        .collect::<Vec<String>>();
+
+    let publishers = publishers
+        .into_iter()
+        .filter(|publisher| !excluded_publishers.contains(publisher))
+        .collect::<Vec<String>>();
+
     let mut sources: HashMap<String, Vec<String>> = HashMap::new();
     let mut decimals: HashMap<String, u32> = HashMap::new();
 
+    let excluded_sources = std::env::var("IGNORE_SOURCES")
+        .unwrap_or("".to_string())
+        .split(',')
+        .map(|source| source.to_string())
+        .collect::<Vec<String>>();
+
     for pair in &pairs {
         let field_pair = cairo_short_string_to_felt(pair).unwrap();
 
@@ -212,7 +252,7 @@ async fn init_oracle_config(
 
         for source in spot_pair_sources {
             let source = parse_cairo_short_string(&source).unwrap();
-            if !pair_sources.contains(&source) {
+            if !pair_sources.contains(&source) && !excluded_sources.contains(&source) {
                 pair_sources.push(source);
             }
         }

src/constants.rs

@@ -1,4 +1,6 @@
+use lazy_static::lazy_static;
 use phf::phf_map;
+use prometheus::{opts, register_gauge_vec, register_int_gauge_vec, GaugeVec, IntGaugeVec};
 
 pub(crate) static COINGECKO_IDS: phf::Map<&'static str, &'static str> = phf_map! {
     "BTC/USD" => "bitcoin",
@@ -11,3 +13,51 @@ pub(crate) static COINGECKO_IDS: phf::Map<&'static str, &'static str> = phf_map!
     "WSTETH/USD" => "wrapped-steth",
     "LORDS/USD" => "lords",
 };
+
+lazy_static! {
+    pub static ref TIME_SINCE_LAST_UPDATE_PUBLISHER: GaugeVec = register_gauge_vec!(
+        opts!(
+            "time_since_last_update_seconds",
+            "Time since the last update in seconds."
+        ),
+        &["publisher"]
+    )
+    .unwrap();
+    pub static ref PAIR_PRICE: GaugeVec = register_gauge_vec!(
+        opts!("pair_price", "Price of the pair from the source."),
+        &["pair", "source"]
+    )
+    .unwrap();
+    pub static ref TIME_SINCE_LAST_UPDATE_PAIR_ID: GaugeVec = register_gauge_vec!(
+        opts!(
+            "time_since_last_update_pair_id",
+            "Time since the last update in seconds."
+        ),
+        &["pair"]
+    )
+    .unwrap();
+    pub static ref PRICE_DEVIATION: GaugeVec = register_gauge_vec!(
+        opts!(
+            "price_deviation",
+            "Price deviation from the reference price."
+        ),
+        &["pair", "source"]
+    )
+    .unwrap();
+    pub static ref PRICE_DEVIATION_SOURCE: GaugeVec = register_gauge_vec!(
+        opts!(
+            "price_deviation_source",
+            "Price deviation from the reference price."
+        ),
+        &["pair", "source"]
+    )
+    .unwrap();
+    pub static ref NUM_SOURCES: IntGaugeVec = register_int_gauge_vec!(
+        opts!(
+            "num_sources",
+            "Number of sources that have published data for a pair."
+        ),
+        &["pair"]
+    )
+    .unwrap();
+}

src/error.rs

@@ -1,5 +1,7 @@
 use std::{error::Error as StdError, fmt};
 
+use starknet::providers::ProviderError;
+
 #[derive(Debug)]
 pub enum MonitoringError {
     Price(String),
@@ -8,6 +10,7 @@ pub enum MonitoringError {
     Api(String),
     Conversion(String),
     OnChain(String),
+    Provider(ProviderError),
 }
 
 impl StdError for MonitoringError {}
@@ -21,6 +24,7 @@ impl fmt::Display for MonitoringError {
             MonitoringError::Api(e) => write!(f, "API Error: {}", e),
             MonitoringError::Conversion(e) => write!(f, "Conversion Error: {}", e),
             MonitoringError::OnChain(e) => write!(f, "OnChain Error: {}", e),
+            MonitoringError::Provider(e) => write!(f, "Provider Error: {}", e),
         }
     }
 }