diff --git a/src/main.rs b/src/main.rs index 970061b..ccd3b28 100644 --- a/src/main.rs +++ b/src/main.rs @@ -29,14 +29,18 @@ use std::str::FromStr; use std::sync::Arc; const ENV_DEFAULT_ROUTE : &'static str = "DEFAULT_ROUTE"; +const ENV_MAX_RETRIES : &'static str = "MAX_RETRIES"; const ENV_NEWRELIC_API_KEY : &'static str = "NEWRELIC_API_KEY"; const ENV_PROXY_CONFIG_FILE : &'static str = "PROXY_CONFIG_FILE"; +const ENV_RETRY_WAIT_MS : &'static str = "RETRY_WAIT_MS"; const ENV_RUST_LOG : &'static str = "RUST_LOG"; const ENV_SERVICE_PORT : &'static str = "SERVICE_PORT"; const DEFAULT_DEFAULT_ROUTE : &'static str = "http://127.0.0.1:12345"; +const DEFAULT_MAX_RETRIES : u8 = 3; const DEFAULT_NEWRELIC_API_KEY : &'static str = ""; const DEFAULT_PROXY_CONFIG_FILE : &'static str = "proxy_configs.toml"; +const DEFAULT_RETRY_WAIT_MS : u64 = 400; const DEFAULT_RUST_LOG : &'static str = "tokio_reactor=warn,hyper=info,debug"; const DEFAULT_SERVICE_PORT : u16 = 5555; @@ -123,12 +127,20 @@ pub async fn handle( client_ip: IpAddr, req: Request, router: Arc, - newrelic_logger: Arc) -> Result, Infallible> + newrelic_logger: Arc, + server_params: ServerParams) -> Result, Infallible> { match req.uri().path() { "/speak" => { let nr_transaction = newrelic_logger.web_transaction("/speak"); - let result = speak_proxy_with_retry(client_ip, req, router, nr_transaction, "/speak").await; + let result = speak_proxy_with_retry( + client_ip, + req, + router, + nr_transaction, + server_params.max_retries, + server_params.retry_wait_ms, + "/speak").await; match result { Ok(response) => Ok(response), Err(error) => Ok(error.to_response()), @@ -136,7 +148,14 @@ pub async fn handle( }, "/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; + let result = speak_proxy_with_retry( + client_ip, + req, + router, + nr_transaction, + server_params.max_retries, + server_params.retry_wait_ms, + "/speak_spectrogram").await; match result { Ok(response) => Ok(response), Err(error) => Ok(error.to_response()), @@ -168,6 +187,12 @@ pub async fn handle( } } +#[derive(Debug, Clone)] +pub struct ServerParams { + pub max_retries: u8, + pub retry_wait_ms: u64, +} + #[tokio::main] async fn main() { if env::var(ENV_RUST_LOG) @@ -184,7 +209,7 @@ async fn main() { let default_route = get_env_string(ENV_DEFAULT_ROUTE, DEFAULT_DEFAULT_ROUTE); let service_port = get_env_num::(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT) - .expect("should have env var"); + .expect("should have service port env var"); let newrelic_api_key = get_env_string(ENV_NEWRELIC_API_KEY, DEFAULT_NEWRELIC_API_KEY); info!("Default route: {}", default_route); @@ -192,10 +217,26 @@ async fn main() { let proxy_configs_file = get_env_string(ENV_PROXY_CONFIG_FILE, DEFAULT_PROXY_CONFIG_FILE); info!("Proxy config file: {}", proxy_configs_file); + let max_retries = get_env_num::(ENV_MAX_RETRIES, DEFAULT_MAX_RETRIES) + .expect("should have max retries env var"); + + info!("Max retries: {}", max_retries); + + let retry_wait_ms = get_env_num::(ENV_RETRY_WAIT_MS, DEFAULT_RETRY_WAIT_MS) + .expect("should have retry wait ms env var"); + + info!("Retry wait: {}", retry_wait_ms); + let proxy_configs = ProxyConfigs::load_from_file(&proxy_configs_file) .expect("should have configs"); + info!("Proxy configs: {:?}", proxy_configs); + let server_params = ServerParams { + max_retries, + retry_wait_ms, + }; + let mut route_map = HashMap::new(); for backend in proxy_configs.backends.iter() { @@ -220,15 +261,17 @@ async fn main() { let remote_addr = conn.remote_addr().ip(); let router2 = router.clone(); let newrelic_logger2 = newrelic_logger.clone(); + let server_params = server_params.clone(); async move { let router3 = router2.clone(); let newrelic_logger3 = newrelic_logger2.clone(); + let server_params = server_params.clone(); Ok::<_, Infallible>(service_fn(move |req| { let router4 = router3.clone(); let newrelic_logger4 = newrelic_logger3.clone(); - handle(remote_addr, req, router4, newrelic_logger4) + handle(remote_addr, req, router4, newrelic_logger4, server_params.clone()) })) } }); diff --git a/src/speak_proxy.rs b/src/speak_proxy.rs index 75fa09a..2824591 100644 --- a/src/speak_proxy.rs +++ b/src/speak_proxy.rs @@ -27,6 +27,8 @@ pub async fn speak_proxy_with_retry( request: Request, router: Arc, _transaction: MaybeNewRelicTransaction, + max_retries: u8, + retry_wait_ms: u64, endpoint: &'static str) -> Result, ErrorResponse> { let mut attempt : u8 = 0; @@ -87,9 +89,9 @@ pub async fn speak_proxy_with_retry( attempt += 1; - thread::sleep(Duration::from_millis(400)); // TODO: Needs backoff + jitter + thread::sleep(Duration::from_millis(retry_wait_ms)); // TODO: Needs backoff + jitter - if attempt > 2 { + if attempt >= max_retries { return Err(ErrorResponse { error_type: ErrorType::ProxyError, error_description: "max internal retry attempts reached".to_string(),