refactor: switch validator metadata reader to JSON API

Update HttpValidatorMetadataReader to GET /api/v1/validators-metadata
and parse a JSON list of ValidatorMetadataItem instead of POSTing a
SQL query. Rename DataApiValidatorMetadataReader to
HttpValidatorMetadataReader and DEFAULT_DATA_API_URL to
DEFAULT_VALIDATOR_METADATA_URL.

Author: snormore

controlplane/doublezero-admin/src/cli/sentinel.rs

@@ -9,7 +9,7 @@ use doublezero_sentinel::{
     multicast_find::{apply_filters, FindFilters},
     output::{print_table, OutputOptions},
     validator_metadata_reader::{
-        DataApiValidatorMetadataReader, ValidatorMetadataReader, DEFAULT_DATA_API_URL,
+        HttpValidatorMetadataReader, ValidatorMetadataReader, DEFAULT_VALIDATOR_METADATA_URL,
     },
 };
 use doublezero_serviceability::pda::get_tenant_pda;
@@ -114,7 +114,7 @@ pub struct FindValidatorMulticastPublishersCommand {
     summary: bool,
 
     /// Data API URL for validator metadata.
-    #[arg(long, value_name = "URL", default_value = DEFAULT_DATA_API_URL)]
+    #[arg(long, value_name = "URL", default_value = DEFAULT_VALIDATOR_METADATA_URL)]
     data_api_url: String,
 
     #[command(flatten)]
@@ -128,7 +128,7 @@ impl FindValidatorMulticastPublishersCommand {
 
         let codes = dz_ledger_reader::fetch_device_codes(rpc_client, &program_id).ok();
 
-        let validator_metadata = DataApiValidatorMetadataReader {
+        let validator_metadata = HttpValidatorMetadataReader {
             api_url: self.data_api_url.clone(),
         };
         let dz_ledger = RpcDzLedgerReader::new(
@@ -405,7 +405,7 @@ pub struct CreateValidatorMulticastPublishersCommand {
     client: Option<String>,
 
     /// Data API URL for validator metadata.
-    #[arg(long, value_name = "URL", default_value = DEFAULT_DATA_API_URL)]
+    #[arg(long, value_name = "URL", default_value = DEFAULT_VALIDATOR_METADATA_URL)]
     data_api_url: String,
 
     /// Simulate transactions without sending.
@@ -424,7 +424,7 @@ impl CreateValidatorMulticastPublishersCommand {
 
         let codes = dz_ledger_reader::fetch_device_codes(rpc_client, &program_id).ok();
 
-        let validator_metadata = DataApiValidatorMetadataReader {
+        let validator_metadata = HttpValidatorMetadataReader {
             api_url: self.data_api_url.clone(),
         };
         let dz_ledger = RpcDzLedgerReader::new(

crates/sentinel/src/validator_metadata_reader.rs

@@ -32,68 +32,57 @@ pub trait ValidatorMetadataReader: Send + Sync {
 // HTTP implementation
 // ---------------------------------------------------------------------------
 
-pub const DEFAULT_DATA_API_URL: &str = "https://data.malbeclabs.com/api/sql/query";
+pub const DEFAULT_VALIDATOR_METADATA_URL: &str =
+    "https://data.malbeclabs.com/api/v1/validators-metadata";
 
-pub struct DataApiValidatorMetadataReader {
+pub struct HttpValidatorMetadataReader {
     pub api_url: String,
 }
 
 #[derive(serde::Deserialize)]
-struct SqlResponse {
-    rows: Vec<Vec<serde_json::Value>>,
+struct ValidatorMetadataItem {
+    ip: String,
+    active_stake: i64,
+    vote_account: String,
+    software_client: String,
+    software_version: String,
 }
 
 #[async_trait::async_trait]
-impl ValidatorMetadataReader for DataApiValidatorMetadataReader {
+impl ValidatorMetadataReader for HttpValidatorMetadataReader {
     async fn fetch_validators(&self) -> Result<HashMap<Ipv4Addr, ValidatorRecord>> {
-        let query = r#"
-            SELECT
-                v.vote_account,
-                v.software_client,
-                v.software_version,
-                v.active_stake,
-                v.ip
-            FROM validatorsapp_validators_current v
-            WHERE v.is_active = 1
-        "#;
-
         let client = reqwest::Client::new();
         let resp = client
-            .post(&self.api_url)
-            .json(&serde_json::json!({ "query": query }))
+            .get(&self.api_url)
             .send()
             .await
-            .context("failed to query data API for validators")?;
+            .context("failed to fetch validator metadata")?;
 
         let status = resp.status();
         if !status.is_success() {
             let body = resp.text().await.unwrap_or_default();
-            bail!("data API returned {status}: {body}");
+            bail!("validator metadata service returned {status}: {body}");
         }
 
-        let body: SqlResponse = resp
+        let items: Vec<ValidatorMetadataItem> = resp
             .json()
             .await
-            .context("failed to parse data API response")?;
+            .context("failed to parse validator metadata response")?;
 
         let mut map = HashMap::new();
-        for row in &body.rows {
-            if row.len() < 5 {
-                continue;
-            }
-
-            let gossip_ip: Ipv4Addr = match row[4].as_str().unwrap_or_default().parse() {
+        for item in items {
+            let gossip_ip: Ipv4Addr = match item.ip.parse() {
                 Ok(ip) => ip,
                 Err(_) => continue,
             };
 
             map.insert(
                 gossip_ip,
                 ValidatorRecord {
-                    vote_account: row[0].as_str().unwrap_or_default().to_string(),
-                    software_client: row[1].as_str().unwrap_or_default().to_string(),
-                    software_version: row[2].as_str().unwrap_or_default().to_string(),
-                    activated_stake_sol: row[3].as_i64().unwrap_or(0) as f64 / 1_000_000_000.0,
+                    vote_account: item.vote_account,
+                    software_client: item.software_client,
+                    software_version: item.software_version,
+                    activated_stake_sol: item.active_stake as f64 / 1_000_000_000.0,
                     gossip_ip,
                 },
             );