Add getClusterNodes/getSlotLeader JSON RPC API (#3940)

* Minor cleanup

* Include _this_ node in the contact info trace

* Add getClusterNodes/getSlotLeader RPC API

book/src/jsonrpc-api.md

@@ -24,8 +24,10 @@ Methods
 * [confirmTransaction](#confirmtransaction)
 * [getAccountInfo](#getaccountinfo)
 * [getBalance](#getbalance)
+* [getClusterNodes](#getclusternodes)
 * [getRecentBlockhash](#getrecentblockhash)
 * [getSignatureStatus](#getsignaturestatus)
+* [getSlotLeader](#getslotleader)
 * [getNumBlocksSinceSignatureConfirmation](#getnumblockssincesignatureconfirmation)
 * [getTransactionCount](#gettransactioncount)
 * [requestAirdrop](#requestairdrop)
@@ -114,6 +116,30 @@ curl -X POST -H "Content-Type: application/json" -d '{"jsonrpc":"2.0", "id":1, "
 
 ---
 
+### getClusterNodes
+Returns information about all the nodes participating in the cluster
+
+##### Parameters:
+None
+
+##### Results:
+The result field will be an array of JSON objects, each with the following sub fields:
+* `id` - Node identifier, as base-58 encoded string
+* `gossip` - Gossip network address for the node
+* `tpu` - TPU network address for the node
+* `rpc` - JSON RPC network address for the node, or `null` if the JSON RPC service is not enabled
+
+##### Example:
+```bash
+// Request
+curl -X POST -H "Content-Type: application/json" -d '{"jsonrpc":"2.0", "id":1, "method":"getClusterNodes"}' http://localhost:8899
+
+// Result
+{"jsonrpc":"2.0","result":[{"gossip":"10.239.6.48:8001","id":"9QzsJf7LPLj8GkXbYT3LFDKqsj2hHG7TA3xinJHu8epQ","rpc":"10.239.6.48:8899","tpu":"10.239.6.48:8856"}],"id":1}
+```
+
+---
+
 ### getAccountInfo
 Returns all information associated with the account of provided Pubkey
 
@@ -183,7 +209,27 @@ curl -X POST -H "Content-Type: application/json" -d '{"jsonrpc":"2.0", "id":1, "
 {"jsonrpc":"2.0","result":"SignatureNotFound","id":1}
 ```
 
----
+-----
+
+### getSlotLeader
+Returns the current slot leader
+
+##### Parameters:
+None
+
+##### Results:
+* `string` - Node Id as base-58 encoded string
+
+##### Example:
+```bash
+// Request
+curl -X POST -H "Content-Type: application/json" -d '{"jsonrpc":"2.0","id":1, "method":"getSlotLeader"}' http://localhost:8899
+
+// Result
+{"jsonrpc":"2.0","result":"ENvAW7JScgYq6o4zKZwewtkzzJgDzuJAFxYasvmEQdpS","id":1}
+```
+
+-----
 
 ### getNumBlocksSinceSignatureConfirmation
 Returns the current number of blocks since signature has been confirmed.

core/src/cluster_info.rs

@@ -245,6 +245,7 @@ impl ClusterInfo {
     pub fn contact_info_trace(&self) -> String {
         let now = timestamp();
         let mut spy_nodes = 0;
+        let my_id = self.my_data().id;
         let nodes: Vec<_> = self
             .all_peers()
             .into_iter()
@@ -261,12 +262,13 @@ impl ClusterInfo {
                 }
 
                 format!(
-                    "- gossip: {:20} | {:5}ms | {}\n  \
+                    "- gossip: {:20} | {:5}ms | {} {}\n  \
                      tpu:    {:20} |         |\n  \
                      rpc:    {:20} |         |\n",
                     addr_to_string(&node.gossip),
                     now.saturating_sub(node.wallclock),
                     node.id,
+                    if node.id == my_id { "(me)" } else { "" }.to_string(),
                     addr_to_string(&node.tpu),
                     addr_to_string(&node.rpc),
                 )
@@ -346,14 +348,12 @@ impl ClusterInfo {
     }
 
     // All nodes in gossip, including spy nodes
-    fn all_peers(&self) -> Vec<ContactInfo> {
+    pub(crate) fn all_peers(&self) -> Vec<ContactInfo> {
-        let me = self.my_data().id;
         self.gossip
             .crds
             .table
             .values()
             .filter_map(|x| x.value.contact_info())
-            .filter(|x| x.id != me)
             .cloned()
             .collect()
     }

core/src/rpc.rs

@@ -2,6 +2,7 @@
 
 use crate::bank_forks::BankForks;
 use crate::cluster_info::ClusterInfo;
+use crate::contact_info::ContactInfo;
 use crate::packet::PACKET_DATA_SIZE;
 use crate::storage_stage::StorageState;
 use bincode::{deserialize, serialize};
@@ -171,6 +172,18 @@ pub struct Meta {
 }
 impl Metadata for Meta {}
 
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct RpcContactInfo {
+    /// Base58 id
+    pub id: String,
+    /// Gossip port
+    pub gossip: Option<SocketAddr>,
+    /// Tpu port
+    pub tpu: Option<SocketAddr>,
+    /// JSON RPC port
+    pub rpc: Option<SocketAddr>,
+}
+
 #[rpc(server)]
 pub trait RpcSol {
     type Metadata;
@@ -184,6 +197,9 @@ pub trait RpcSol {
     #[rpc(meta, name = "getBalance")]
     fn get_balance(&self, _: Self::Metadata, _: String) -> Result<u64>;
 
+    #[rpc(meta, name = "getClusterNodes")]
+    fn get_cluster_nodes(&self, _: Self::Metadata) -> Result<Vec<RpcContactInfo>>;
+
     #[rpc(meta, name = "getRecentBlockhash")]
     fn get_recent_blockhash(&self, _: Self::Metadata) -> Result<String>;
 
@@ -203,6 +219,9 @@ pub trait RpcSol {
     #[rpc(meta, name = "sendTransaction")]
     fn send_transaction(&self, _: Self::Metadata, _: Vec<u8>) -> Result<String>;
 
+    #[rpc(meta, name = "getSlotLeader")]
+    fn get_slot_leader(&self, _: Self::Metadata) -> Result<String>;
+
     #[rpc(meta, name = "getStorageBlockhash")]
     fn get_storage_blockhash(&self, _: Self::Metadata) -> Result<String>;
 
@@ -263,6 +282,33 @@ impl RpcSol for RpcSolImpl {
         Ok(meta.request_processor.read().unwrap().get_balance(&pubkey))
     }
 
+    fn get_cluster_nodes(&self, meta: Self::Metadata) -> Result<Vec<RpcContactInfo>> {
+        let cluster_info = meta.cluster_info.read().unwrap();
+        fn valid_address_or_none(addr: &SocketAddr) -> Option<SocketAddr> {
+            if ContactInfo::is_valid_address(addr) {
+                Some(*addr)
+            } else {
+                None
+            }
+        }
+        Ok(cluster_info
+            .all_peers()
+            .iter()
+            .filter_map(|contact_info| {
+                if ContactInfo::is_valid_address(&contact_info.gossip) {
+                    Some(RpcContactInfo {
+                        id: contact_info.id.to_string(),
+                        gossip: Some(contact_info.gossip),
+                        tpu: valid_address_or_none(&contact_info.tpu),
+                        rpc: valid_address_or_none(&contact_info.rpc),
+                    })
+                } else {
+                    None // Exclude spy nodes
+                }
+            })
+            .collect())
+    }
+
     fn get_recent_blockhash(&self, meta: Self::Metadata) -> Result<String> {
         debug!("get_recent_blockhash rpc request received");
         Ok(meta
@@ -402,6 +448,15 @@ impl RpcSol for RpcSolImpl {
         Ok(signature)
     }
 
+    fn get_slot_leader(&self, meta: Self::Metadata) -> Result<String> {
+        let cluster_info = meta.cluster_info.read().unwrap();
+        let leader_data_option = cluster_info.leader_data();
+        Ok(leader_data_option
+            .and_then(|leader_data| Some(leader_data.id))
+            .unwrap_or_default()
+            .to_string())
+    }
+
     fn get_storage_blockhash(&self, meta: Self::Metadata) -> Result<String> {
         meta.request_processor
             .read()
@@ -445,7 +500,9 @@ mod tests {
     use solana_sdk::transaction::TransactionError;
     use std::thread;
 
-    fn start_rpc_handler_with_tx(pubkey: &Pubkey) -> (MetaIoHandler<Meta>, Meta, Hash, Keypair) {
+    fn start_rpc_handler_with_tx(
+        pubkey: &Pubkey,
+    ) -> (MetaIoHandler<Meta>, Meta, Hash, Keypair, Pubkey) {
         let (bank_forks, alice) = new_bank_forks();
         let bank = bank_forks.read().unwrap().working_bank();
         let exit = Arc::new(AtomicBool::new(false));
@@ -477,7 +534,7 @@ mod tests {
             request_processor,
             cluster_info,
         };
-        (io, meta, blockhash, alice)
+        (io, meta, blockhash, alice, leader.id)
     }
 
     #[test]
@@ -505,7 +562,7 @@ mod tests {
     #[test]
     fn test_rpc_get_balance() {
         let bob_pubkey = Pubkey::new_rand();
-        let (io, meta, _blockhash, _alice) = start_rpc_handler_with_tx(&bob_pubkey);
+        let (io, meta, _blockhash, _alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
 
         let req = format!(
             r#"{{"jsonrpc":"2.0","id":1,"method":"getBalance","params":["{}"]}}"#,
@@ -520,10 +577,46 @@ mod tests {
         assert_eq!(expected, result);
     }
 
+    #[test]
+    fn test_rpc_get_cluster_nodes() {
+        let bob_pubkey = Pubkey::new_rand();
+        let (io, meta, _blockhash, _alice, leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
+
+        let req = format!(r#"{{"jsonrpc":"2.0","id":1,"method":"getClusterNodes"}}"#);
+        let res = io.handle_request_sync(&req, meta);
+        let result: Response = serde_json::from_str(&res.expect("actual response"))
+            .expect("actual response deserialization");
+
+        let expected = format!(
+            r#"{{"jsonrpc":"2.0","result":[{{"id": "{}", "gossip": "127.0.0.1:1235", "tpu": "127.0.0.1:1234", "rpc": "127.0.0.1:8899"}}],"id":1}}"#,
+            leader_id,
+        );
+
+        let expected: Response =
+            serde_json::from_str(&expected).expect("expected response deserialization");
+        assert_eq!(expected, result);
+    }
+
+    #[test]
+    fn test_rpc_get_slot_leader() {
+        let bob_pubkey = Pubkey::new_rand();
+        let (io, meta, _blockhash, _alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
+
+        let req = format!(r#"{{"jsonrpc":"2.0","id":1,"method":"getSlotLeader"}}"#);
+        let res = io.handle_request_sync(&req, meta);
+        let expected =
+            format!(r#"{{"jsonrpc":"2.0","result":"11111111111111111111111111111111","id":1}}"#);
+        let expected: Response =
+            serde_json::from_str(&expected).expect("expected response deserialization");
+        let result: Response = serde_json::from_str(&res.expect("actual response"))
+            .expect("actual response deserialization");
+        assert_eq!(expected, result);
+    }
+
     #[test]
     fn test_rpc_get_tx_count() {
         let bob_pubkey = Pubkey::new_rand();
-        let (io, meta, _blockhash, _alice) = start_rpc_handler_with_tx(&bob_pubkey);
+        let (io, meta, _blockhash, _alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
 
         let req = format!(r#"{{"jsonrpc":"2.0","id":1,"method":"getTransactionCount"}}"#);
         let res = io.handle_request_sync(&req, meta);
@@ -538,7 +631,7 @@ mod tests {
     #[test]
     fn test_rpc_get_account_info() {
         let bob_pubkey = Pubkey::new_rand();
-        let (io, meta, _blockhash, _alice) = start_rpc_handler_with_tx(&bob_pubkey);
+        let (io, meta, _blockhash, _alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
 
         let req = format!(
             r#"{{"jsonrpc":"2.0","id":1,"method":"getAccountInfo","params":["{}"]}}"#,
@@ -565,7 +658,7 @@ mod tests {
     #[test]
     fn test_rpc_confirm_tx() {
         let bob_pubkey = Pubkey::new_rand();
-        let (io, meta, blockhash, alice) = start_rpc_handler_with_tx(&bob_pubkey);
+        let (io, meta, blockhash, alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
         let tx = system_transaction::transfer(&alice, &bob_pubkey, 20, blockhash, 0);
 
         let req = format!(
@@ -584,7 +677,7 @@ mod tests {
     #[test]
     fn test_rpc_get_signature_status() {
         let bob_pubkey = Pubkey::new_rand();
-        let (io, meta, blockhash, alice) = start_rpc_handler_with_tx(&bob_pubkey);
+        let (io, meta, blockhash, alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
         let tx = system_transaction::transfer(&alice, &bob_pubkey, 20, blockhash, 0);
 
         let req = format!(
@@ -648,7 +741,7 @@ mod tests {
     #[test]
     fn test_rpc_get_recent_blockhash() {
         let bob_pubkey = Pubkey::new_rand();
-        let (io, meta, blockhash, _alice) = start_rpc_handler_with_tx(&bob_pubkey);
+        let (io, meta, blockhash, _alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
 
         let req = format!(r#"{{"jsonrpc":"2.0","id":1,"method":"getRecentBlockhash"}}"#);
         let res = io.handle_request_sync(&req, meta);
@@ -663,7 +756,7 @@ mod tests {
     #[test]
     fn test_rpc_fail_request_airdrop() {
         let bob_pubkey = Pubkey::new_rand();
-        let (io, meta, _blockhash, _alice) = start_rpc_handler_with_tx(&bob_pubkey);
+        let (io, meta, _blockhash, _alice, _leader_id) = start_rpc_handler_with_tx(&bob_pubkey);
 
         // Expect internal error because no drone is available
         let req = format!(

core/src/rpc_service.rs

@@ -15,7 +15,9 @@ use std::time::Duration;
 
 pub struct JsonRpcService {
     thread_hdl: JoinHandle<()>,
-    pub request_processor: Arc<RwLock<JsonRpcRequestProcessor>>, // Used only by tests...
+
+    #[cfg(test)]
+    pub request_processor: Arc<RwLock<JsonRpcRequestProcessor>>, // Used only by test_rpc_new()...
 }
 
 impl JsonRpcService {
@@ -37,7 +39,7 @@ impl JsonRpcService {
         )));
         let request_processor_ = request_processor.clone();
 
-        let info = cluster_info.clone();
+        let cluster_info = cluster_info.clone();
         let exit_ = exit.clone();
 
         let thread_hdl = Builder::new()
@@ -50,7 +52,7 @@ impl JsonRpcService {
                 let server =
                     ServerBuilder::with_meta_extractor(io, move |_req: &hyper::Request<hyper::Body>| Meta {
                         request_processor: request_processor_.clone(),
-                        cluster_info: info.clone(),
+                        cluster_info: cluster_info.clone(),
                     }).threads(4)
                         .cors(DomainsValidation::AllowOnly(vec![
                             AccessControlAllowOrigin::Any,
@@ -68,6 +70,7 @@ impl JsonRpcService {
             .unwrap();
         Self {
             thread_hdl,
+            #[cfg(test)]
             request_processor,
         }
     }