cleanup

src/api.rs

@@ -1,7 +1,8 @@
 use anyhow::Result as AnyhowResult;
 use futures::TryStreamExt;
 use hyper::http::HeaderValue;
-use hyper::{Body, Request, HeaderMap};
+use hyper::{Body, Request, HeaderMap, Response, StatusCode};
+use hyper::body::Bytes;
 
 /// NB: This much match the shape of SpeakRequest in the 'voder/tts_service' code.
 /// This is used for both /speak and /speak_spectrogram requests.
@@ -37,6 +38,21 @@ pub struct InternalSpeakRequest {
   pub skip_rate_limiter: bool,
 }
 
+pub enum ErrorType {
+  /// The proxy couldn't find a healthy host (or it died mid-connection)
+  ProxyError,
+  // The backend server reported a 500
+  ServerError,
+  /// Client is rate limited
+  RateLimitError,
+}
+
+pub struct ErrorResponse {
+  /// The type of error
+  pub error_type: ErrorType,
+  /// Description of the error
+  pub error_description: String,
+}
 
 impl ExternalHttpRequest {
 
@@ -54,6 +70,27 @@ impl ExternalHttpRequest {
   }
 }
 
+impl ErrorType {
+  pub fn status_code(&self) -> StatusCode {
+    match *self {
+      ErrorType::ProxyError =>StatusCode::INTERNAL_SERVER_ERROR,
+      ErrorType::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
+      ErrorType::RateLimitError => StatusCode::TOO_MANY_REQUESTS,
+    }
+  }
+}
+
+impl ErrorResponse {
+  pub fn to_response(&self) -> Response<Body> {
+    let status_code = self.error_type.status_code();
+    let body = Body::from(self.error_description.clone());
+    Response::builder()
+      .status(status_code)
+      .body(body)
+      .unwrap()
+  }
+}
+
 async fn body_to_bytes(body: Body) -> AnyhowResult<Vec<u8>> {
   // Also: let body_bytes = hyper::body::to_bytes(body)
   let bytes = body.try_fold(Vec::new(), |mut data, chunk| async move {

src/main.rs

@@ -4,20 +4,20 @@
 #[macro_use] extern crate log;
 #[macro_use] extern crate serde_derive;
 
+pub mod api;
 pub mod newrelic_logger;
-pub mod request;
 pub mod speak_proxy;
 
 use anyhow::Result as AnyhowResult;
 use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction};
-use futures::future::{self, Future};
 use crate::speak_proxy::speak_proxy_with_retry;
-use hyper::{Method, StatusCode};
+use futures::future::{self, Future};
 use hyper::header::HeaderValue;
 use hyper::http::header::HeaderName;
 use hyper::server::conn::AddrStream;
 use hyper::service::{service_fn, make_service_fn};
 use hyper::{Body, Request, Response, Server};
+use hyper::{Method, StatusCode};
 use rand::seq::IteratorRandom;
 use std::collections::HashMap;
 use std::convert::Infallible;
@@ -28,9 +28,6 @@ use std::net::{SocketAddr, IpAddr};
 use std::str::FromStr;
 use std::sync::Arc;
 
-//type BoxFut = Box<dyn Future + Send>;
-type BoxFut = Box<dyn Future<Output=Response<Body>> + Send>;
-
 const ENV_DEFAULT_ROUTE : &'static str = "DEFAULT_ROUTE";
 const ENV_NEWRELIC_API_KEY : &'static str = "NEWRELIC_API_KEY";
 const ENV_PROXY_CONFIG_FILE : &'static str = "PROXY_CONFIG_FILE";
@@ -92,17 +89,6 @@ fn get_env_num<T>(env_name: &str, default: T) -> AnyhowResult<T>
   }
 }
 
-/*fn _debug_request(req: Request<Body>) -> BoxFut {
-  let body_str = format!("{:?}", req);
-  let response = Response::new(Body::from(body_str));
-  Box::new(future::ok(response))
-}*/
-
-/*fn health_check_response(_req: Request<Body>) -> BoxFut {
-  let response = Response::new(Body::from("healthy"));
-  Box::new(future::ok(response))
-}*/
-
 pub struct Router {
   pub routes: HashMap<String, String>,
   pub default_route: String,
@@ -129,6 +115,10 @@ fn debug_request(req: Request<Body>) -> Result<Response<Body>, Infallible>  {
   Ok(Response::new(Body::from(body_str)))
 }
 
+fn health_check_response(_req: Request<Body>) -> Result<Response<Body>, Infallible> {
+  Ok(Response::new(Body::from("healthy")))
+}
+
 pub async fn handle(
   client_ip: IpAddr,
   req: Request<Body>,
@@ -140,36 +130,21 @@ pub async fn handle(
       let nr_transaction = newrelic_logger.web_transaction("/speak");
       let result = speak_proxy_with_retry(client_ip, req, router, nr_transaction, "/speak").await;
       match result {
-        Ok(response) => {
-          Ok(response)
-        },
-        Err(_error) => {
-          Ok(Response::builder()
-            .status(StatusCode::INTERNAL_SERVER_ERROR)
-            .body(Body::empty())
-            .unwrap())
-        }
+        Ok(response) => Ok(response),
+        Err(error) => Ok(error.to_response()),
       }
     },
     "/speak_spectrogram" => {
       let nr_transaction = newrelic_logger.web_transaction("/speak_spectrogram");
       let result = speak_proxy_with_retry(client_ip, req, router, nr_transaction, "/speak_spectrogram").await;
       match result {
-        Ok(response) => {
-          Ok(response)
-        },
-        Err(_error) => {
-          Ok(Response::builder()
-            .status(StatusCode::INTERNAL_SERVER_ERROR)
-            .body(Body::empty())
-            .unwrap())
-        }
+        Ok(response) => Ok(response),
+        Err(error) => Ok(error.to_response()),
       }
     },
     "/proxy_health" => {
       let _droppable_transaction = newrelic_logger.web_transaction("/proxy_health");
-      //health_check_response(req)
-      debug_request(req)
+      health_check_response(req)
     },
     _ => {
       let _droppable_transaction = newrelic_logger
@@ -182,10 +157,10 @@ pub async fn handle(
         Ok(response) => {
           Ok(response)
         },
-        Err(error) => {
+        Err(_) => {
           Ok(Response::builder()
             .status(StatusCode::INTERNAL_SERVER_ERROR)
-            .body(Body::empty())
+            .body(Body::from("unknown proxy error"))
             .unwrap())
         }
       }
@@ -241,40 +216,6 @@ async fn main() {
 
   let newrelic_logger = Arc::new(newrelic_logger);
 
-  /*let make_service = make_service_fn(move |socket: &AddrStream| {
-    let remote_addr = socket.remote_addr();
-    info!("Got a request from: {:?}", remote_addr);
-
-    let router2 = router.clone();
-    let newrelic_logger2 = newrelic_logger.clone();
-
-    service_fn(move |req: Request<Body>| {
-      let router3 = router2.clone();
-      let newrelic_logger3 = newrelic_logger2.clone();
-
-      match (req.method(), req.uri().path()) {
-        /*(&Method::POST, "/speak") => {
-          let nr_transaction = newrelic_logger3.web_transaction("/speak");
-          speak_proxy(req, remote_addr.clone(), router3, nr_transaction, "/speak")
-        },
-        (&Method::POST, "/speak_spectrogram") => {
-          let nr_transaction = newrelic_logger3.web_transaction("/speak_spectrogram");
-          speak_proxy(req, remote_addr.clone(), router3, nr_transaction, "/speak_spectrogram")
-        },*/
-        /*(&Method::GET, "/proxy_health") => {
-          let _droppable_transaction = newrelic_logger3.web_transaction("/proxy_health");
-          health_check_response(req)
-        },*/
-        _ => {
-          let _droppable_transaction = newrelic_logger3.web_transaction("unmatched endpoint");
-          let forward = router3.get_random_host();
-          info!("Forwarding to `{}` random host: {}", &forward, req.uri());
-          hyper_reverse_proxy::call(remote_addr.ip(), &forward, req)
-        }
-      }
-    })
-  });*/
-
   let make_service = make_service_fn(move |conn: &AddrStream| {
     let remote_addr = conn.remote_addr().ip();
     let router2 = router.clone();

src/speak_proxy.rs

@@ -1,6 +1,7 @@
 use anyhow::Result as AnyhowResult;
+use crate::Router;
+use crate::api::{ExternalSpeakRequest, ExternalHttpRequest, InternalSpeakRequest, ErrorResponse, ErrorType};
 use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction};
-use crate::{BoxFut, Router};
 use futures::TryStreamExt;
 use futures::future::{self, Future};
 use hyper::header::HeaderValue;
@@ -13,20 +14,20 @@ use hyper_reverse_proxy::ProxyError;
 use rand::seq::IteratorRandom;
 use std::collections::HashMap;
 use std::convert::Infallible;
-use std::env;
 use std::fmt::Debug;
 use std::fmt::Display;
 use std::net::{SocketAddr, IpAddr};
 use std::str::FromStr;
 use std::sync::Arc;
-use crate::request::{ExternalSpeakRequest, ExternalHttpRequest, InternalSpeakRequest};
+use std::time::Duration;
+use std::{env, thread};
 
 pub async fn speak_proxy_with_retry(
   ip_addr: IpAddr,
   request: Request<Body>,
   router: Arc<Router>,
   _transaction: MaybeNewRelicTransaction,
-  endpoint: &'static str) -> Result<Response<Body>, ResponseError>
+  endpoint: &'static str) -> Result<Response<Body>, ErrorResponse>
 {
   let mut attempt : u8 = 0;
 
@@ -79,14 +80,20 @@ pub async fn speak_proxy_with_retry(
       &load_balancer_hostname,
       endpoint).await;
 
-    if result.is_ok() {
-      return result;
+    match result {
+      Ok(good_response) => return Ok(good_response),
+      Err(_) => {},
     }
 
     attempt += 1;
 
-    if attempt > 3 {
-      return Err(ResponseError { reason: "max attempts reached".to_string() });
+    thread::sleep(Duration::from_millis(400)); // TODO: Needs backoff + jitter
+
+    if attempt > 2 {
+      return Err(ErrorResponse {
+        error_type: ErrorType::ProxyError,
+        error_description: "max internal retry attempts reached".to_string(),
+      });
     }
   }
 }
@@ -101,9 +108,10 @@ async fn speak_proxy(
   internal_speak_request: InternalSpeakRequest,
   headers: HeaderMap<HeaderValue>,
   load_balancer_hostname: &str,
-  endpoint: &'static str) -> Result<Response<Body>, ResponseError>
+  endpoint: &'static str) -> Result<Response<Body>, ()>
 {
-  info!("Routing {} for {} to {}", endpoint, &internal_speak_request.speaker, &load_balancer_hostname);
+  info!("Routing {} for {} to {} (attempt # {})", endpoint, &internal_speak_request.speaker,
+        &load_balancer_hostname, internal_speak_request.retry_attempt_number);
 
   let mut request_builder = Request::builder()
     .method(Method::POST)
@@ -116,17 +124,12 @@ async fn speak_proxy(
 
   let json_speak_request = serde_json::to_string(&internal_speak_request).unwrap();
 
-  println!("Request: {:?}", internal_speak_request);
-
   let request = request_builder
     .body(Body::from(json_speak_request))
     .unwrap();
 
-  //Box::new(future::ok(Vec::new()))
   match hyper_reverse_proxy::call(ip_addr, &load_balancer_hostname, request).await {
     Ok(result) => Ok(result),
-    Err(_result) => Err(ResponseError {
-      reason: "Uknown failure".to_string()
-    }),
+    Err(_result) => Err(()), // TODO: Surface this?
   }
 }