diff --git a/src/request.rs b/src/api.rs similarity index 62% rename from src/request.rs rename to src/api.rs index 7e3d628..38926a8 100644 --- a/src/request.rs +++ b/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 { + 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> { // Also: let body_bytes = hyper::body::to_bytes(body) let bytes = body.try_fold(Vec::new(), |mut data, chunk| async move { diff --git a/src/main.rs b/src/main.rs index 09f71bb..970061b 100644 --- a/src/main.rs +++ b/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; -type BoxFut = Box> + 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(env_name: &str, default: T) -> AnyhowResult } } -/*fn _debug_request(req: Request) -> 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) -> BoxFut { - let response = Response::new(Body::from("healthy")); - Box::new(future::ok(response)) -}*/ - pub struct Router { pub routes: HashMap, pub default_route: String, @@ -129,6 +115,10 @@ fn debug_request(req: Request) -> Result, Infallible> { Ok(Response::new(Body::from(body_str))) } +fn health_check_response(_req: Request) -> Result, Infallible> { + Ok(Response::new(Body::from("healthy"))) +} + pub async fn handle( client_ip: IpAddr, req: Request, @@ -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| { - 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(); diff --git a/src/speak_proxy.rs b/src/speak_proxy.rs index b277f6f..75fa09a 100644 --- a/src/speak_proxy.rs +++ b/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, router: Arc, _transaction: MaybeNewRelicTransaction, - endpoint: &'static str) -> Result, ResponseError> + endpoint: &'static str) -> Result, 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, load_balancer_hostname: &str, - endpoint: &'static str) -> Result, ResponseError> + endpoint: &'static str) -> Result, ()> { - 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? } }