Skip to content

Commit

Permalink
Factor out RpcHealth module
Browse files Browse the repository at this point in the history
  • Loading branch information
mvines committed Jun 2, 2020
1 parent 189aa79 commit 9dbf3d5
Show file tree
Hide file tree
Showing 3 changed files with 131 additions and 100 deletions.
1 change: 1 addition & 0 deletions core/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ pub mod retransmit_stage;
pub mod rewards_recorder_service;
pub mod rpc;
pub mod rpc_error;
pub mod rpc_health;
pub mod rpc_pubsub;
pub mod rpc_pubsub_service;
pub mod rpc_service;
Expand Down
91 changes: 91 additions & 0 deletions core/src/rpc_health.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
use crate::cluster_info::ClusterInfo;
use solana_sdk::pubkey::Pubkey;
use std::{
collections::HashSet,
sync::atomic::{AtomicBool, Ordering},
sync::Arc,
};

pub enum RpcHealthStatus {
Ok,
Behind, // Validator is behind its trusted validators
}

pub struct RpcHealth {
cluster_info: Arc<ClusterInfo>,
trusted_validators: Option<HashSet<Pubkey>>,
health_check_slot_distance: u64,
override_health_check: Arc<AtomicBool>,
}

impl RpcHealth {
pub fn new(
cluster_info: Arc<ClusterInfo>,
trusted_validators: Option<HashSet<Pubkey>>,
health_check_slot_distance: u64,
override_health_check: Arc<AtomicBool>,
) -> Self {
Self {
cluster_info,
trusted_validators,
health_check_slot_distance,
override_health_check,
}
}

pub fn check(&self) -> RpcHealthStatus {
if self.override_health_check.load(Ordering::Relaxed) {
RpcHealthStatus::Ok
} else if let Some(trusted_validators) = &self.trusted_validators {
let (latest_account_hash_slot, latest_trusted_validator_account_hash_slot) = {
(
self.cluster_info
.get_accounts_hash_for_node(&self.cluster_info.id(), |hashes| {
hashes
.iter()
.max_by(|a, b| a.0.cmp(&b.0))
.map(|slot_hash| slot_hash.0)
})
.flatten()
.unwrap_or(0),
trusted_validators
.iter()
.map(|trusted_validator| {
self.cluster_info
.get_accounts_hash_for_node(&trusted_validator, |hashes| {
hashes
.iter()
.max_by(|a, b| a.0.cmp(&b.0))
.map(|slot_hash| slot_hash.0)
})
.flatten()
.unwrap_or(0)
})
.max()
.unwrap_or(0),
)
};

// This validator is considered healthy if its latest account hash slot is within
// `health_check_slot_distance` of the latest trusted validator's account hash slot
if latest_account_hash_slot > 0
&& latest_trusted_validator_account_hash_slot > 0
&& latest_account_hash_slot
> latest_trusted_validator_account_hash_slot
.saturating_sub(self.health_check_slot_distance)
{
RpcHealthStatus::Ok
} else {
warn!(
"health check: me={}, latest trusted_validator={}",
latest_account_hash_slot, latest_trusted_validator_account_hash_slot
);
RpcHealthStatus::Behind
}
} else {
// No trusted validator point of reference available, so this validator is healthy
// because it's running
RpcHealthStatus::Ok
}
}
}
139 changes: 39 additions & 100 deletions core/src/rpc_service.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
//! The `rpc_service` module implements the Solana JSON RPC service.
use crate::{
cluster_info::ClusterInfo, commitment::BlockCommitmentCache, rpc::*, validator::ValidatorExit,
cluster_info::ClusterInfo, commitment::BlockCommitmentCache, rpc::*, rpc_health::*,
validator::ValidatorExit,
};
use jsonrpc_core::MetaIoHandler;
use jsonrpc_http_server::{
Expand All @@ -19,7 +20,7 @@ use std::{
collections::HashSet,
net::SocketAddr,
path::{Path, PathBuf},
sync::atomic::{AtomicBool, Ordering},
sync::atomic::AtomicBool,
sync::{mpsc::channel, Arc, RwLock},
thread::{self, Builder, JoinHandle},
};
Expand All @@ -38,22 +39,16 @@ struct RpcRequestMiddleware {
ledger_path: PathBuf,
snapshot_archive_path_regex: Regex,
snapshot_config: Option<SnapshotConfig>,
cluster_info: Arc<ClusterInfo>,
trusted_validators: Option<HashSet<Pubkey>>,
bank_forks: Arc<RwLock<BankForks>>,
health_check_slot_distance: u64,
override_health_check: Arc<AtomicBool>,
health: Arc<RpcHealth>,
}

impl RpcRequestMiddleware {
pub fn new(
ledger_path: PathBuf,
snapshot_config: Option<SnapshotConfig>,
cluster_info: Arc<ClusterInfo>,
trusted_validators: Option<HashSet<Pubkey>>,
bank_forks: Arc<RwLock<BankForks>>,
health_check_slot_distance: u64,
override_health_check: Arc<AtomicBool>,
health: Arc<RpcHealth>,
) -> Self {
Self {
ledger_path,
Expand All @@ -62,11 +57,8 @@ impl RpcRequestMiddleware {
)
.unwrap(),
snapshot_config,
cluster_info,
trusted_validators,
bank_forks,
health_check_slot_distance,
override_health_check,
health,
}
}

Expand Down Expand Up @@ -137,60 +129,10 @@ impl RpcRequestMiddleware {
}

fn health_check(&self) -> &'static str {
let response = if self.override_health_check.load(Ordering::Relaxed) {
"ok"
} else if let Some(trusted_validators) = &self.trusted_validators {
let (latest_account_hash_slot, latest_trusted_validator_account_hash_slot) = {
(
self.cluster_info
.get_accounts_hash_for_node(&self.cluster_info.id(), |hashes| {
hashes
.iter()
.max_by(|a, b| a.0.cmp(&b.0))
.map(|slot_hash| slot_hash.0)
})
.flatten()
.unwrap_or(0),
trusted_validators
.iter()
.map(|trusted_validator| {
self.cluster_info
.get_accounts_hash_for_node(&trusted_validator, |hashes| {
hashes
.iter()
.max_by(|a, b| a.0.cmp(&b.0))
.map(|slot_hash| slot_hash.0)
})
.flatten()
.unwrap_or(0)
})
.max()
.unwrap_or(0),
)
};

// This validator is considered healthy if its latest account hash slot is within
// `health_check_slot_distance` of the latest trusted validator's account hash slot
if latest_account_hash_slot > 0
&& latest_trusted_validator_account_hash_slot > 0
&& latest_account_hash_slot
> latest_trusted_validator_account_hash_slot
.saturating_sub(self.health_check_slot_distance)
{
"ok"
} else {
warn!(
"health check: me={}, latest trusted_validator={}",
latest_account_hash_slot, latest_trusted_validator_account_hash_slot
);
"behind"
}
} else {
// No trusted validator point of reference available, so this validator is healthy
// because it's running
"ok"
let response = match self.health.check() {
RpcHealthStatus::Ok => "ok",
RpcHealthStatus::Behind => "behind",
};

info!("health check: {}", response);
response
}
Expand Down Expand Up @@ -299,7 +241,14 @@ impl JsonRpcService {
) -> Self {
info!("rpc bound to {:?}", rpc_addr);
info!("rpc configuration: {:?}", config);
let health_check_slot_distance = config.health_check_slot_distance;

let health = Arc::new(RpcHealth::new(
cluster_info.clone(),
trusted_validators,
config.health_check_slot_distance,
override_health_check,
));

let request_processor = Arc::new(RwLock::new(JsonRpcRequestProcessor::new(
config,
bank_forks.clone(),
Expand All @@ -324,11 +273,8 @@ impl JsonRpcService {
let request_middleware = RpcRequestMiddleware::new(
ledger_path,
snapshot_config,
cluster_info.clone(),
trusted_validators,
bank_forks.clone(),
health_check_slot_distance,
override_health_check,
health.clone(),
);
let server = ServerBuilder::with_meta_extractor(
io,
Expand Down Expand Up @@ -403,7 +349,10 @@ mod tests {
};
use solana_runtime::bank::Bank;
use solana_sdk::signature::Signer;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::{
net::{IpAddr, Ipv4Addr, SocketAddr},
sync::atomic::Ordering,
};

#[test]
fn test_rpc_new() {
Expand Down Expand Up @@ -481,18 +430,16 @@ mod tests {

#[test]
fn test_is_file_get_path() {
let cluster_info = Arc::new(ClusterInfo::new_with_invalid_keypair(ContactInfo::default()));
let bank_forks = create_bank_forks();

let rrm = RpcRequestMiddleware::new(
PathBuf::from("/"),
None,
cluster_info.clone(),
let health = Arc::new(RpcHealth::new(
Arc::new(ClusterInfo::new_with_invalid_keypair(ContactInfo::default())),
None,
bank_forks.clone(),
42,
Arc::new(AtomicBool::new(false)),
);
));

let rrm =
RpcRequestMiddleware::new(PathBuf::from("/"), None, bank_forks.clone(), health.clone());
let rrm_with_snapshot_config = RpcRequestMiddleware::new(
PathBuf::from("/"),
Some(SnapshotConfig {
Expand All @@ -501,11 +448,8 @@ mod tests {
snapshot_path: PathBuf::from("/"),
compression: CompressionType::Bzip2,
}),
cluster_info,
None,
bank_forks,
42,
Arc::new(AtomicBool::new(false)),
health,
);

assert!(rrm.is_file_get_path("/genesis.tar.bz2"));
Expand All @@ -531,37 +475,32 @@ mod tests {

#[test]
fn test_health_check_with_no_trusted_validators() {
let cluster_info = Arc::new(ClusterInfo::new_with_invalid_keypair(ContactInfo::default()));

let rm = RpcRequestMiddleware::new(
PathBuf::from("/"),
None,
cluster_info,
let health = Arc::new(RpcHealth::new(
Arc::new(ClusterInfo::new_with_invalid_keypair(ContactInfo::default())),
None,
create_bank_forks(),
42,
Arc::new(AtomicBool::new(false)),
);
));

let rm = RpcRequestMiddleware::new(PathBuf::from("/"), None, create_bank_forks(), health);
assert_eq!(rm.health_check(), "ok");
}

#[test]
fn test_health_check_with_trusted_validators() {
let cluster_info = Arc::new(ClusterInfo::new_with_invalid_keypair(ContactInfo::default()));

let health_check_slot_distance = 123;

let override_health_check = Arc::new(AtomicBool::new(false));
let trusted_validators = vec![Pubkey::new_rand(), Pubkey::new_rand(), Pubkey::new_rand()];
let rm = RpcRequestMiddleware::new(
PathBuf::from("/"),
None,

let health = Arc::new(RpcHealth::new(
cluster_info.clone(),
Some(trusted_validators.clone().into_iter().collect()),
create_bank_forks(),
health_check_slot_distance,
override_health_check.clone(),
);
));

let rm = RpcRequestMiddleware::new(PathBuf::from("/"), None, create_bank_forks(), health);

// No account hashes for this node or any trusted validators == "behind"
assert_eq!(rm.health_check(), "behind");
Expand Down

0 comments on commit 9dbf3d5

Please sign in to comment.