Add alternative to Spy Nodes that can fully participate in Gossip (#4087)
automerge
This commit is contained in:
parent
af2e7ea285
commit
9add8d0afc
|
@ -250,7 +250,7 @@ impl ClusterInfo {
|
|||
.all_peers()
|
||||
.into_iter()
|
||||
.map(|(node, last_updated)| {
|
||||
if !ContactInfo::is_valid_address(&node.gossip) {
|
||||
if Self::is_spy_node(&node) {
|
||||
spy_nodes += 1;
|
||||
}
|
||||
fn addr_to_string(addr: &SocketAddr) -> String {
|
||||
|
@ -412,6 +412,12 @@ impl ClusterInfo {
|
|||
.collect()
|
||||
}
|
||||
|
||||
fn is_spy_node(contact_info: &ContactInfo) -> bool {
|
||||
!ContactInfo::is_valid_address(&contact_info.tpu)
|
||||
|| !ContactInfo::is_valid_address(&contact_info.gossip)
|
||||
|| !ContactInfo::is_valid_address(&contact_info.tvu)
|
||||
}
|
||||
|
||||
fn sort_by_stake<S: std::hash::BuildHasher>(
|
||||
peers: &[ContactInfo],
|
||||
stakes: &HashMap<Pubkey, u64, S>,
|
||||
|
@ -1377,6 +1383,26 @@ impl ClusterInfo {
|
|||
.unwrap()
|
||||
}
|
||||
|
||||
/// An alternative to Spy Node that has a valid gossip address and fully participate in Gossip.
|
||||
pub fn gossip_node(id: &Pubkey, gossip_addr: &SocketAddr) -> (ContactInfo, UdpSocket) {
|
||||
let (port, gossip_socket) = Node::get_gossip_port(gossip_addr, FULLNODE_PORT_RANGE);
|
||||
let daddr = socketaddr_any!();
|
||||
|
||||
let node = ContactInfo::new(
|
||||
id,
|
||||
SocketAddr::new(gossip_addr.ip(), port),
|
||||
daddr,
|
||||
daddr,
|
||||
daddr,
|
||||
daddr,
|
||||
daddr,
|
||||
daddr,
|
||||
timestamp(),
|
||||
);
|
||||
(node, gossip_socket)
|
||||
}
|
||||
|
||||
/// A Node with invalid ports to spy on gossip via pull requests
|
||||
pub fn spy_node(id: &Pubkey) -> (ContactInfo, UdpSocket) {
|
||||
let (_, gossip_socket) = bind_in_range(FULLNODE_PORT_RANGE).unwrap();
|
||||
let daddr = socketaddr_any!();
|
||||
|
@ -1638,6 +1664,16 @@ mod tests {
|
|||
use std::net::{IpAddr, Ipv4Addr};
|
||||
use std::sync::{Arc, RwLock};
|
||||
|
||||
#[test]
|
||||
fn test_gossip_node() {
|
||||
//check that a gossip nodes always show up as spies
|
||||
let (node, _) = ClusterInfo::spy_node(&Pubkey::new_rand());
|
||||
assert!(ClusterInfo::is_spy_node(&node));
|
||||
let (node, _) =
|
||||
ClusterInfo::gossip_node(&Pubkey::new_rand(), &"1.1.1.1:1111".parse().unwrap());
|
||||
assert!(ClusterInfo::is_spy_node(&node));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cluster_spy_gossip() {
|
||||
//check that gossip doesn't try to push to invalid addresses
|
||||
|
|
|
@ -53,22 +53,24 @@ impl GossipService {
|
|||
}
|
||||
|
||||
pub fn discover_nodes(
|
||||
gossip_addr: &SocketAddr,
|
||||
entry_point: &SocketAddr,
|
||||
num_nodes: usize,
|
||||
) -> std::io::Result<Vec<ContactInfo>> {
|
||||
discover(gossip_addr, Some(num_nodes), Some(30), None)
|
||||
discover(entry_point, Some(num_nodes), Some(30), None, None)
|
||||
}
|
||||
|
||||
pub fn discover(
|
||||
gossip_addr: &SocketAddr,
|
||||
entry_point: &SocketAddr,
|
||||
num_nodes: Option<usize>,
|
||||
timeout: Option<u64>,
|
||||
find_node: Option<Pubkey>,
|
||||
gossip_addr: Option<&SocketAddr>,
|
||||
) -> std::io::Result<Vec<ContactInfo>> {
|
||||
let exit = Arc::new(AtomicBool::new(false));
|
||||
let (gossip_service, spy_ref) = make_spy_node(gossip_addr, &exit);
|
||||
let (gossip_service, spy_ref) = make_gossip_node(entry_point, &exit, gossip_addr);
|
||||
|
||||
let id = spy_ref.read().unwrap().keypair.pubkey();
|
||||
info!("Gossip entry point: {:?}", gossip_addr);
|
||||
info!("Gossip entry point: {:?}", entry_point);
|
||||
info!("Spy node id: {:?}", id);
|
||||
|
||||
let (met_criteria, secs, tvu_peers) = spy(spy_ref.clone(), num_nodes, timeout, find_node);
|
||||
|
@ -153,14 +155,21 @@ fn spy(
|
|||
(met_criteria, now.elapsed().as_secs(), tvu_peers)
|
||||
}
|
||||
|
||||
fn make_spy_node(
|
||||
gossip_addr: &SocketAddr,
|
||||
/// Makes a spy or gossip node based on whether or not a gossip_addr was passed in
|
||||
/// Pass in a gossip addr to fully participate in gossip instead of relying on just pulls
|
||||
fn make_gossip_node(
|
||||
entry_point: &SocketAddr,
|
||||
exit: &Arc<AtomicBool>,
|
||||
gossip_addr: Option<&SocketAddr>,
|
||||
) -> (GossipService, Arc<RwLock<ClusterInfo>>) {
|
||||
let keypair = Arc::new(Keypair::new());
|
||||
let (node, gossip_socket) = ClusterInfo::spy_node(&keypair.pubkey());
|
||||
let (node, gossip_socket) = if let Some(gossip_addr) = gossip_addr {
|
||||
ClusterInfo::gossip_node(&keypair.pubkey(), gossip_addr)
|
||||
} else {
|
||||
ClusterInfo::spy_node(&keypair.pubkey())
|
||||
};
|
||||
let mut cluster_info = ClusterInfo::new(node, keypair);
|
||||
cluster_info.set_entrypoint(ContactInfo::new_gossip_entry_point(gossip_addr));
|
||||
cluster_info.set_entrypoint(ContactInfo::new_gossip_entry_point(entry_point));
|
||||
let cluster_info = Arc::new(RwLock::new(cluster_info));
|
||||
let gossip_service =
|
||||
GossipService::new(&cluster_info.clone(), None, None, gossip_socket, &exit);
|
||||
|
|
|
@ -1,11 +1,15 @@
|
|||
//! A command-line executable for monitoring a cluster's gossip plane.
|
||||
|
||||
#[macro_use]
|
||||
extern crate solana;
|
||||
|
||||
use clap::{crate_description, crate_name, crate_version, App, AppSettings, Arg, SubCommand};
|
||||
use solana::contact_info::ContactInfo;
|
||||
use solana::gossip_service::discover;
|
||||
use solana_client::rpc_client::RpcClient;
|
||||
use solana_sdk::pubkey::Pubkey;
|
||||
use std::error;
|
||||
use std::net::Ipv4Addr;
|
||||
use std::net::SocketAddr;
|
||||
use std::process::exit;
|
||||
|
||||
|
@ -38,6 +42,18 @@ fn main() -> Result<(), Box<dyn error::Error>> {
|
|||
SubCommand::with_name("spy")
|
||||
.about("Monitor the gossip network")
|
||||
.setting(AppSettings::DisableVersion)
|
||||
.arg(
|
||||
clap::Arg::with_name("public_address")
|
||||
.long("public-address")
|
||||
.takes_value(false)
|
||||
.help("Advertise public machine address in gossip. By default the local machine address is advertised"),
|
||||
)
|
||||
.arg(
|
||||
clap::Arg::with_name("pull_only")
|
||||
.long("pull-only")
|
||||
.takes_value(false)
|
||||
.help("Use a partial gossip node (Pulls only) to spy on the network. By default it will use a full fledged gossip node(Pushes and Pulls). Useful when behind a NAT"),
|
||||
)
|
||||
.arg(
|
||||
Arg::with_name("num_nodes")
|
||||
.short("N")
|
||||
|
@ -111,7 +127,25 @@ fn main() -> Result<(), Box<dyn error::Error>> {
|
|||
.value_of("node_pubkey")
|
||||
.map(|pubkey_str| pubkey_str.parse::<Pubkey>().unwrap());
|
||||
|
||||
let nodes = discover(&network_addr, num_nodes, timeout, pubkey)?;
|
||||
let gossip_addr = if matches.is_present("pull_only") {
|
||||
None
|
||||
} else {
|
||||
let mut addr = socketaddr_any!();
|
||||
if matches.is_present("public_address") {
|
||||
addr.set_ip(solana_netutil::get_public_ip_addr().unwrap());
|
||||
} else {
|
||||
addr.set_ip(solana_netutil::get_ip_addr(false).unwrap());
|
||||
}
|
||||
Some(addr)
|
||||
};
|
||||
|
||||
let nodes = discover(
|
||||
&network_addr,
|
||||
num_nodes,
|
||||
timeout,
|
||||
pubkey,
|
||||
gossip_addr.as_ref(),
|
||||
)?;
|
||||
|
||||
if timeout.is_some() {
|
||||
if let Some(num) = num_nodes {
|
||||
|
@ -146,7 +180,7 @@ fn main() -> Result<(), Box<dyn error::Error>> {
|
|||
.unwrap()
|
||||
.parse::<Pubkey>()
|
||||
.unwrap();
|
||||
let nodes = discover(&network_addr, None, None, Some(pubkey))?;
|
||||
let nodes = discover(&network_addr, None, None, Some(pubkey), None)?;
|
||||
let node = nodes.iter().find(|x| x.id == pubkey).unwrap();
|
||||
|
||||
if !ContactInfo::is_valid_address(&node.rpc) {
|
||||
|
|
Loading…
Reference in New Issue