Long-term ledger storage with BigTable (bp #11222) (#11392)

* ledger-storage-bigtable boilerplate

(cherry picked from commit 9d2293bb32)

* $ wget https://pki.goog/roots.pem -O pki-goog-roots.pem

(cherry picked from commit 1617a025ce)

* Add access_token module

(cherry picked from commit 59d266a111)

* Add root_ca_certificate

(cherry picked from commit faa016e4b7)

* Add build-proto

(cherry picked from commit c31e1f5bf0)

* UiTransactionEncoding is now copy

(cherry picked from commit 494968be66)

* Increase timeout

(cherry picked from commit 57dfebc5ba)

* Add build-proto/build.sh output

(cherry picked from commit 54dae6ba2c)

* Supress doctest errors

(cherry picked from commit 019c75797d)

* Add compression

(cherry picked from commit 243e05d59f)

* Add bigtable

(cherry picked from commit 6e0353965a)

* Add configuration info

(cherry picked from commit 98cca1e774)

* Add ledger-tool bigtable subcommands

(cherry picked from commit f9049d6ee4)

# Conflicts:
#	ledger-tool/Cargo.toml

* Make room for tokio 0.2

(cherry picked from commit b876fb84ba)

# Conflicts:
#	core/Cargo.toml

* Setup a tokio 0.2 runtime for RPC usage

(cherry picked from commit 0e02740565)

# Conflicts:
#	core/Cargo.toml

* Plumb Bigtable ledger storage into the RPC subsystem

(cherry picked from commit dfae9a9864)

# Conflicts:
#	core/Cargo.toml

* Add RPC transaction history design

(cherry picked from commit e56ea138c7)

* Simplify access token refreshing

(cherry picked from commit 1f7af14386)

* Report block status more frequently

(cherry picked from commit 22c46ebf96)

* after -> before

(cherry picked from commit 227ea934ff)

* Rebase

* Cargo.lock

Co-authored-by: Michael Vines <mvines@gmail.com>

core/src/rpc_service.rs

@@ -23,7 +23,7 @@ use std::{
     sync::{mpsc::channel, Arc, RwLock},
     thread::{self, Builder, JoinHandle},
 };
-use tokio::prelude::Future;
+use tokio::runtime;
 
 pub struct JsonRpcService {
     thread_hdl: JoinHandle<()>,
@@ -32,6 +32,7 @@ pub struct JsonRpcService {
     pub request_processor: JsonRpcRequestProcessor, // Used only by test_rpc_new()...
 
     close_handle: Option<CloseHandle>,
+    runtime: runtime::Runtime,
 }
 
 struct RpcRequestMiddleware {
@@ -97,6 +98,9 @@ impl RpcRequestMiddleware {
     }
 
     fn process_file_get(&self, path: &str) -> RequestMiddlewareAction {
+        // Stuck on tokio 0.1 until the jsonrpc-http-server crate upgrades to tokio 0.2
+        use tokio_01::prelude::*;
+
         let stem = path.split_at(1).1; // Drop leading '/' from path
         let filename = {
             match path {
@@ -115,10 +119,10 @@ impl RpcRequestMiddleware {
         RequestMiddlewareAction::Respond {
             should_validate_hosts: true,
             response: Box::new(
-                tokio_fs::file::File::open(filename)
+                tokio_fs_01::file::File::open(filename)
                     .and_then(|file| {
                         let buf: Vec<u8> = Vec::new();
-                        tokio_io::io::read_to_end(file, buf)
+                        tokio_io_01::io::read_to_end(file, buf)
                             .and_then(|item| Ok(hyper::Response::new(item.1.into())))
                             .or_else(|_| Ok(RpcRequestMiddleware::internal_server_error()))
                     })
@@ -249,6 +253,28 @@ impl JsonRpcService {
         ));
 
         let tpu_address = cluster_info.my_contact_info().tpu;
+        let mut runtime = runtime::Builder::new()
+            .threaded_scheduler()
+            .thread_name("rpc-runtime")
+            .enable_all()
+            .build()
+            .expect("Runtime");
+
+        let bigtable_ledger_storage = if config.enable_bigtable_ledger_storage {
+            runtime
+                .block_on(solana_storage_bigtable::LedgerStorage::new(false))
+                .map(|x| {
+                    info!("BigTable ledger storage initialized");
+                    Some(x)
+                })
+                .unwrap_or_else(|err| {
+                    error!("Failed to initialize BigTable ledger storage: {:?}", err);
+                    None
+                })
+        } else {
+            None
+        };
+
         let (request_processor, receiver) = JsonRpcRequestProcessor::new(
             config,
             bank_forks.clone(),
@@ -258,6 +284,8 @@ impl JsonRpcService {
             health.clone(),
             cluster_info,
             genesis_hash,
+            &runtime,
+            bigtable_ledger_storage,
         );
 
         let exit_send_transaction_service = Arc::new(AtomicBool::new(false));
@@ -325,6 +353,7 @@ impl JsonRpcService {
             .register_exit(Box::new(move || close_handle_.close()));
         Self {
             thread_hdl,
+            runtime,
             #[cfg(test)]
             request_processor: test_request_processor,
             close_handle: Some(close_handle),
@@ -338,6 +367,7 @@ impl JsonRpcService {
     }
 
     pub fn join(self) -> thread::Result<()> {
+        self.runtime.shutdown_background();
         self.thread_hdl.join()
     }
 }