Merge branch 'mempool' into junderw/use-blocking-spawn-b

src/bin/electrs.rs

@@ -93,7 +93,7 @@ fn run_server(config: Arc<Config>) -> Result<()> {
     ));
 
     // TODO: configuration for which servers to start
-    let rest_server = rest::start(Arc::clone(&config), Arc::clone(&query));
+    let rest_server = rest::start(Arc::clone(&config), Arc::clone(&query), &metrics);
     let electrum_server = ElectrumRPC::start(Arc::clone(&config), Arc::clone(&query), &metrics);
 
     if let Some(ref precache_file) = config.precache_scripts {

src/rest.rs

@@ -1,6 +1,7 @@
 use crate::chain::{address, BlockHash, Network, OutPoint, Script, Transaction, TxIn, TxOut, Txid};
 use crate::config::{Config, VERSION_STRING};
 use crate::errors;
+use crate::metrics::Metrics;
 use crate::new_index::{compute_script_hash, Query, SpendingInput, Utxo};
 use crate::util::{
     create_socket, electrum_merkle, extract_tx_prevouts, full_hash, get_innerscripts, get_tx_fee,
@@ -17,6 +18,7 @@ use bitcoin::hashes::Error as HashError;
 use hex::{self, FromHexError};
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Response, Server, StatusCode};
+use prometheus::{HistogramOpts, HistogramVec};
 use tokio::sync::oneshot;
 
 use hyperlocal::UnixServerExt;
@@ -552,7 +554,12 @@ fn prepare_txs(
 }
 
 #[tokio::main]
-async fn run_server(config: Arc<Config>, query: Arc<Query>, rx: oneshot::Receiver<()>) {
+async fn run_server(
+    config: Arc<Config>,
+    query: Arc<Query>,
+    rx: oneshot::Receiver<()>,
+    metric: HistogramVec,
+) {
     let addr = &config.http_addr;
     let socket_file = &config.http_socket_file;
 
@@ -562,31 +569,49 @@ async fn run_server(config: Arc<Config>, query: Arc<Query>, rx: oneshot::Receive
     let make_service_fn_inn = || {
         let query = Arc::clone(&query);
         let config = Arc::clone(&config);
+        let metric = metric.clone();
 
         async move {
             Ok::<_, hyper::Error>(service_fn(move |req| {
                 let query = Arc::clone(&query);
                 let config = Arc::clone(&config);
+                let timer = metric.with_label_values(&["all_methods"]).start_timer();
 
                 async move {
                     let method = req.method().clone();
                     let uri = req.uri().clone();
                     let body = hyper::body::to_bytes(req.into_body()).await?;
 
-                    let mut resp = handle_request(method, uri, body, &query, &config)
-                        .unwrap_or_else(|err| {
-                            warn!("{:?}", err);
-                            Response::builder()
-                                .status(err.0)
-                                .header("Content-Type", "text/plain")
-                                .header("X-Powered-By", &**VERSION_STRING)
-                                .body(Body::from(err.1))
-                                .unwrap()
-                        });
-                    if let Some(ref origins) = config.cors {
+                    let cors = config
+                        .cors
+                        .as_ref()
+                        .map(|c| c.parse::<hyper::http::HeaderValue>().unwrap());
+
+                    let mut resp = tokio::task::spawn_blocking(move || {
+                        handle_request(method, uri, body, &query, &config)
+                    })
+                    .await
+                    .unwrap_or_else(|err| {
+                        warn!("JoinHandle error: {:?}", err);
+                        Err(HttpError(
+                            StatusCode::INTERNAL_SERVER_ERROR,
+                            String::from("Internal Server Error: JoinHandle"),
+                        ))
+                    })
+                    .unwrap_or_else(|err| {
+                        warn!("{:?}", err);
+                        Response::builder()
+                            .status(err.0)
+                            .header("Content-Type", "text/plain")
+                            .header("X-Powered-By", &**VERSION_STRING)
+                            .body(Body::from(err.1))
+                            .unwrap()
+                    });
+                    if let Some(origins) = cors {
                         resp.headers_mut()
-                            .insert("Access-Control-Allow-Origin", origins.parse().unwrap());
+                            .insert("Access-Control-Allow-Origin", origins);
                     }
+                    timer.observe_duration();
                     Ok::<_, hyper::Error>(resp)
                 }
             }))
@@ -633,13 +658,17 @@ async fn run_server(config: Arc<Config>, query: Arc<Query>, rx: oneshot::Receive
     }
 }
 
-pub fn start(config: Arc<Config>, query: Arc<Query>) -> Handle {
+pub fn start(config: Arc<Config>, query: Arc<Query>, metrics: &Metrics) -> Handle {
     let (tx, rx) = oneshot::channel::<()>();
+    let response_timer = metrics.histogram_vec(
+        HistogramOpts::new("electrs_rest_api", "Electrs REST API response timings"),
+        &["method"],
+    );
 
     Handle {
         tx,
         thread: crate::util::spawn_thread("rest-server", move || {
-            run_server(config, query, rx);
+            run_server(config, query, rx, response_timer);
         }),
     }
 }