send data to pingthing

bench/Cargo.toml

@@ -23,6 +23,7 @@ lazy_static = "1.4.0"
 bincode = { workspace = true }
 itertools = "0.10.5"
 spl-memo = "4.0.0"
+reqwest = "0.11"
 
 [dev-dependencies]
 bincode = { workspace = true }

bench/src/bin/pingthing-tester.rs

@@ -0,0 +1,61 @@
+use std::collections::HashMap;
+use std::error::Error;
+use std::str::FromStr;
+use std::sync::Arc;
+use std::time::Duration;
+use clap::Parser;
+use log::info;
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use solana_rpc_client::rpc_client::RpcClient;
+use solana_sdk::clock::Slot;
+use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
+use solana_sdk::genesis_config::ClusterType;
+use solana_sdk::pubkey::Pubkey;
+use solana_sdk::signature::Signature;
+use bench::pingthing::PingThing;
+
+/// Simple program to greet a person
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Name of the person to greet
+    #[arg(long)]
+    va_api_key: String,
+}
+
+
+/// https://www.validators.app/ping-thing
+/// see https://github.com/Block-Logic/ping-thing-client/blob/main/ping-thing-client.mjs#L161C10-L181C17
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tracing_subscriber::fmt::init();
+
+    let args = Args::parse();
+    let va_api_key = args.va_api_key;
+
+    let rpc_client = Arc::new(RpcClient::new_with_commitment(
+        "http://api.mainnet-beta.solana.com/",
+        CommitmentConfig::confirmed(),
+    ));
+
+    let current_slot = rpc_client.get_slot().unwrap();
+    info!("Current slot: {}", current_slot);
+
+    // some random transaction
+    let hardcoded_example = Signature::from_str("3yKgzowsEUnXXv7TdbLcHFRqrFvn4CtaNKMELzfqrvokp8Pgw4LGFVAZqvnSLp92B9oY4HGhSEZhSuYqLzkT9sC8").unwrap();
+
+    let tx_success = true;
+    let target_slot = current_slot + 0;
+    info!("Target slot: {}", current_slot);
+
+
+    let pingthing = PingThing {
+        va_api_key,
+    };
+    pingthing.submit_stats(Duration::from_millis(5555), hardcoded_example, tx_success, target_slot, target_slot).await??;
+
+    Ok(())
+}
+
+

bench/src/cli.rs

@@ -23,4 +23,8 @@ pub struct Args {
     // choose between small (179 bytes) and large (1186 bytes) transactions
     #[arg(short = 'L', long, default_value_t = false)]
     pub large_transactions: bool,
+    #[arg(long, default_value_t = false)]
+    pub pingthing_enable: bool,
+    #[arg(long,)]
+    pub pingthing_va_api_key: Option<String>,
 }

bench/src/lib.rs

@@ -1,3 +1,4 @@
 pub mod cli;
 pub mod helpers;
 pub mod metrics;
+pub mod pingthing;

bench/src/main.rs

@@ -1,12 +1,8 @@
-use bench::{
-    cli::Args,
-    helpers::BenchHelper,
-    metrics::{AvgMetric, Metric, TxMetricData},
-};
+use bench::{cli::Args, helpers::BenchHelper, metrics::{AvgMetric, Metric, TxMetricData}, pingthing};
 use clap::Parser;
 use dashmap::DashMap;
 use futures::future::join_all;
-use log::{error, info, warn};
+use log::{debug, error, info, warn};
 use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::signature::Signature;
 use solana_sdk::{
@@ -21,6 +17,7 @@ use tokio::{
     sync::{mpsc::UnboundedSender, RwLock},
     time::{Duration, Instant},
 };
+use bench::pingthing::PingThing;
 
 #[tokio::main(flavor = "multi_thread", worker_threads = 16)]
 async fn main() {
@@ -34,6 +31,8 @@ async fn main() {
         lite_rpc_addr,
         transaction_save_file,
         large_transactions,
+        pingthing_enable,
+        pingthing_va_api_key,
     } = Args::parse();
 
     let mut run_interval_ms = tokio::time::interval(Duration::from_millis(run_interval_ms));
@@ -44,6 +43,15 @@ async fn main() {
         TransactionSize::Small
     };
 
+    let pingthing_config = Arc::new(if pingthing_enable {
+        Some(PingThing {
+            va_api_key: pingthing_va_api_key.expect("va_api_key must be provided - see https://github.com/Block-Logic/ping-thing-client/tree/main#install-notes")
+        })
+    } else {
+        None
+    });
+
+
     info!("Connecting to LiteRPC using {lite_rpc_addr}");
 
     let mut avg_metric = AvgMetric::default();
@@ -113,6 +121,7 @@ async fn main() {
             tx_log_sx.clone(),
             log_transactions,
             transaction_size,
+            pingthing_config.clone(),
         )));
         // wait for an interval
         run_interval_ms.tick().await;
@@ -169,6 +178,7 @@ async fn bench(
     tx_metric_sx: UnboundedSender<TxMetricData>,
     log_txs: bool,
     transaction_size: TransactionSize,
+    pingthing_config: Arc<Option<PingThing>>,
 ) -> Metric {
     let map_of_txs: Arc<DashMap<Signature, TxSendData>> = Arc::new(DashMap::new());
     // transaction sender task
@@ -254,6 +264,14 @@ async fn bench(
                             time_to_confirm_in_millis: time_to_confirm.as_millis() as u64,
                         });
                     }
+
+                    if let Some(pingthing) = pingthing_config.as_ref() {
+                        pingthing.submit_stats(
+                            time_to_confirm, signature.clone(), true, tx_data.sent_slot, current_slot.load(Ordering::Relaxed));
+                        // TODO check
+
+                    }
+
                     drop(tx_data);
                     map_of_txs.remove(signature);
                     confirmed_count += 1;

bench/src/pingthing.rs

@@ -0,0 +1,77 @@
+use std::collections::HashMap;
+use std::error::Error;
+use std::str::FromStr;
+use std::sync::Arc;
+use std::time::Duration;
+use futures::FutureExt;
+use log::{debug, info};
+use reqwest::{Response, StatusCode};
+use serde::{Deserialize, Serialize};
+use solana_rpc_client::rpc_client::RpcClient;
+use solana_sdk::clock::Slot;
+use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
+use solana_sdk::genesis_config::ClusterType;
+use solana_sdk::pubkey::Pubkey;
+use solana_sdk::signature::Signature;
+use tokio::task::JoinHandle;
+
+
+pub struct PingThing {
+    pub va_api_key: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Request {
+    time: u128,
+    signature: String, // Tx sig
+    transaction_type: String, // 'transfer',
+    success: bool, // txSuccess
+    application: String, // e.g. 'web3'
+    commitment_level: String, // e.g. 'confirmed'
+    slot_sent: Slot,
+    slot_landed: Slot,
+}
+
+
+impl PingThing {
+    pub fn submit_stats(&self, tx_elapsed: Duration, tx_sig: Signature, tx_success: bool, slot_sent: Slot, slot_landed: Slot) -> JoinHandle<anyhow::Result<()>> {
+        let jh = tokio::spawn(submit_stats_to_ping_thing(
+            self.va_api_key.clone(),
+            tx_elapsed, tx_sig, tx_success, slot_sent, slot_landed));
+        jh
+    }
+
+}
+
+// subit to https://www.validators.app/ping-thing?network=mainnet
+async fn submit_stats_to_ping_thing(va_api_key: String, tx_elapsed: Duration, tx_sig: Signature, tx_success: bool, slot_sent: Slot, slot_landed: Slot)
+    -> anyhow::Result<()> {
+    // TODO only works on mainnet - skip for all others
+
+    let foo = Request {
+        time: tx_elapsed.as_millis(),
+        signature: tx_sig.to_string(),
+        transaction_type: "transfer".to_string(),
+        success: tx_success,
+        application: "LiteRPC.bench".to_string(),
+        commitment_level: "confirmed".to_string(),
+        slot_sent,
+        slot_landed,
+    };
+
+    let client = reqwest::Client::new();
+    let response = client.post("https://www.validators.app/api/v1/ping-thing/mainnet")
+        .header("Content-Type", "application/json")
+        .header("Token", va_api_key)
+        .json(&foo)
+        .send()
+        .await?
+        .error_for_status()?;
+
+    assert_eq!(response.status(), StatusCode::CREATED);
+
+    debug!("Sent data for tx {} to ping-thing server", tx_sig);
+    Ok(())
+}
+
+