Merge and move voder-proxy code into old_code/

old_code/voder-proxy/src/api.rs

@@ -0,0 +1,101 @@
+use anyhow::Result as AnyhowResult;
+use futures::TryStreamExt;
+use hyper::http::HeaderValue;
+use hyper::{Body, Request, HeaderMap, Response, StatusCode};
+
+/// NB: This much match the shape of SpeakRequest in the 'voder/tts_service' code.
+/// This is used for both /speak and /speak_spectrogram requests.
+#[derive(Deserialize, Debug)]
+pub struct ExternalSpeakRequest {
+  /// Slug for the speaker
+  pub speaker: String,
+
+  /// Raw text to be spoken
+  pub text: String,
+}
+
+/// The full request
+pub struct ExternalHttpRequest {
+  pub headers: HeaderMap<HeaderValue>,
+  pub speak_request: ExternalSpeakRequest,
+}
+
+#[derive(Serialize, Debug)]
+pub struct InternalSpeakRequest {
+  /// Slug for the speaker
+  pub speaker: String,
+
+  /// Raw text to be spoken
+  pub text: String,
+
+  /// Retry attempt.
+  /// The first request is "0", and all subsequent requests are incremented.
+  /// Subsequent requests do not trigger the rate limiter.
+  pub retry_attempt_number: u8,
+
+  /// Admin/debug requests get to skip the rate limiter.
+  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 {
+
+  pub async fn decode_request(request: Request<Body>) -> AnyhowResult<Self> {
+    let headers = request.headers().clone();
+    let (_parts, body) = request.into_parts();
+
+    let body_bytes = body_to_bytes(body).await?;
+    let speak_request = serde_json::from_slice::<ExternalSpeakRequest>(&body_bytes)?;
+
+    Ok(ExternalHttpRequest {
+      headers: headers,
+      speak_request:  speak_request,
+    })
+  }
+}
+
+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 {
+    data.extend_from_slice(&chunk);
+    Ok(data)
+  }).await?;
+
+  Ok(bytes)
+}

old_code/voder-proxy/src/main.rs

@@ -0,0 +1,292 @@
+#[macro_use] extern crate anyhow;
+#[macro_use] extern crate log;
+#[macro_use] extern crate serde_derive;
+
+pub mod api;
+pub mod newrelic_logger;
+pub mod speak_proxy;
+
+use anyhow::Result as AnyhowResult;
+use crate::newrelic_logger::NewRelicLogger;
+use crate::speak_proxy::speak_proxy_with_retry;
+use hyper::server::conn::AddrStream;
+use hyper::service::{service_fn, make_service_fn};
+use hyper::{Body, Request, Response, Server};
+use hyper::{StatusCode, Method};
+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::IpAddr;
+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;
+
+#[derive(Deserialize, Debug, Clone)]
+struct ProxyConfigs {
+  pub backends: Vec<ProxyConfig>,
+}
+
+#[derive(Deserialize, Debug, Clone)]
+struct ProxyConfig {
+  pub voice : String,
+  pub ip: String,
+  pub port: String,
+}
+
+impl ProxyConfigs {
+  pub fn load_from_file(filename: &str) -> AnyhowResult<Self> {
+    let contents = std::fs::read_to_string(filename)?;
+    let configs = toml::from_str(&contents)?;
+    Ok(configs)
+  }
+}
+
+fn get_env_string(env_name: &str, default: &str) -> String {
+  match env::var(env_name).as_ref().ok() {
+    Some(s) => s.to_string(),
+    None => {
+      warn!("Env var '{}' not supplied. Using default '{}'.", env_name, default);
+      default.to_string()
+    },
+  }
+}
+
+// the trait `std::error::Error` is not implemented for `<T as std::str::FromStr>::Err`
+fn get_env_num<T>(env_name: &str, default: T) -> AnyhowResult<T>
+  where T: FromStr + Display,
+        T::Err: Debug
+{
+  match env::var(env_name).as_ref().ok() {
+    None => {
+      warn!("Env var '{}' not supplied. Using default '{}'.", env_name, default);
+      Ok(default)
+    },
+    Some(val) => {
+      match val.parse::<T>() {
+        Err(e) => bail!("Can't parse value '{:?}'. Error: {:?}", val, e),
+        Ok(val) => Ok(val),
+      }
+    },
+  }
+}
+
+pub struct Router {
+  pub routes: HashMap<String, String>,
+  pub default_route: String,
+}
+
+impl Router {
+  fn get_random_host(&self) -> String {
+    let mut rng = rand::thread_rng();
+    self.routes.values()
+      .choose(&mut rng)
+      .map(|url| url.to_string())
+      .unwrap_or(self.default_route.clone())
+  }
+
+  fn get_speaker_host(&self, voice: &str) -> String {
+    self.routes.get(voice)
+      .map(|url| url.to_string())
+      .unwrap_or(self.default_route.clone())
+  }
+}
+
+// fn debug_request(req: Request<Body>) -> Result<Response<Body>, Infallible>  {
+//   let body_str = format!("{:?}", req);
+//   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>,
+  router: Arc<Router>,
+  newrelic_logger: Arc<NewRelicLogger>,
+  server_params: ServerParams) -> Result<Response<Body>, Infallible>
+{
+  match (req.method(), req.uri().path()) {
+    (&Method::POST, "/speak") => {
+      let nr_transaction = newrelic_logger.web_transaction("/speak");
+      let result = speak_proxy_with_retry(
+        client_ip,
+        req,
+        router,
+        nr_transaction,
+        server_params.max_retries,
+        server_params.retry_wait_ms,
+        &server_params.server_hostname,
+        "/speak").await;
+      match result {
+        Ok(response) => Ok(response),
+        Err(error) => Ok(error.to_response()),
+      }
+    },
+    (&Method::POST, "/speak_spectrogram") => {
+      let nr_transaction = newrelic_logger.web_transaction("/speak_spectrogram");
+      let result = speak_proxy_with_retry(
+        client_ip,
+        req,
+        router,
+        nr_transaction,
+        server_params.max_retries,
+        server_params.retry_wait_ms,
+        &server_params.server_hostname,
+        "/speak_spectrogram").await;
+      match result {
+        Ok(response) => Ok(response),
+        Err(error) => Ok(error.to_response()),
+      }
+    },
+    (&Method::GET, "/proxy_health") => {
+      let _droppable_transaction = newrelic_logger.web_transaction("/proxy_health");
+      health_check_response(req)
+    },
+    _ => {
+      let _droppable_transaction = newrelic_logger
+        .web_transaction("unmatched endpoint");
+      let forward = router.get_random_host();
+
+      info!("Forwarding to `{}` random host: {}", &forward, req.uri());
+
+      match hyper_reverse_proxy::call(client_ip, &forward, req).await {
+        Ok(response) => {
+          Ok(response)
+        },
+        Err(_) => {
+          Ok(Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(Body::from("unknown proxy error"))
+            .unwrap())
+        }
+      }
+    },
+  }
+}
+
+#[derive(Debug, Clone)]
+pub struct ServerParams {
+  pub max_retries: u8,
+  pub retry_wait_ms: u64,
+  pub server_hostname: String,
+}
+
+#[tokio::main]
+async fn main() {
+  if env::var(ENV_RUST_LOG)
+    .as_ref()
+    .ok()
+    .is_none()
+  {
+    println!("Setting default logging level to \"{}\", override with env var {}.",
+             DEFAULT_RUST_LOG, ENV_RUST_LOG);
+    std::env::set_var(ENV_RUST_LOG, DEFAULT_RUST_LOG);
+  }
+
+  env_logger::init();
+
+  let default_route = get_env_string(ENV_DEFAULT_ROUTE, DEFAULT_DEFAULT_ROUTE);
+  let service_port = get_env_num::<u16>(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT)
+    .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);
+
+  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::<u8>(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::<u64>(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_hostname = hostname::get()
+    .ok()
+    .and_then(|h| h.into_string().ok())
+    .unwrap_or("proxy-unknown".to_string());
+
+  let server_params = ServerParams {
+    max_retries,
+    retry_wait_ms,
+    server_hostname,
+  };
+
+  let mut route_map = HashMap::new();
+
+  for backend in proxy_configs.backends.iter() {
+    let host = format!("http://{}:{}", backend.ip, backend.port);
+    route_map.insert(backend.voice.clone(), host);
+  }
+
+  let router = Arc::new(Router {
+    routes: route_map,
+    default_route: default_route,
+  });
+
+  let newrelic_logger = if newrelic_api_key.is_empty() {
+    NewRelicLogger::null_instance()
+  } else {
+    NewRelicLogger::try_new_or_null("voder-proxy", &newrelic_api_key)
+  };
+
+  let newrelic_logger = Arc::new(newrelic_logger);
+
+  let make_service = make_service_fn(move |conn: &AddrStream| {
+    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, server_params.clone())
+      }))
+    }
+  });
+
+  let addr = ([0, 0, 0, 0], service_port).into();
+
+  info!("Running server on {:?}", addr);
+
+  let server = Server::bind(&addr).serve(make_service);
+
+  if let Err(e) = server.await {
+    eprintln!("server error: {}", e);
+  }
+}

old_code/voder-proxy/src/newrelic_logger.rs

@@ -0,0 +1,56 @@
+use anyhow::Result as AnyhowResult;
+use newrelic::{App, Transaction};
+
+pub struct NewRelicLogger {
+  newrelic_api: Option<App>,
+}
+
+impl NewRelicLogger {
+  pub fn new(app_name: &str, api_key: &str) -> AnyhowResult<Self> {
+    let newrelic_api = App::new(app_name, api_key)?;
+    Ok(Self {
+      newrelic_api: Some(newrelic_api),
+    })
+  }
+
+  pub fn null_instance() -> Self {
+    Self {
+      newrelic_api: None,
+    }
+  }
+
+  pub fn try_new_or_null(app_name: &str, api_key: &str) -> Self {
+    match Self::new(app_name, api_key) {
+      Ok(instance) => instance,
+      Err(_) => Self::null_instance(),
+    }
+  }
+
+  pub fn web_transaction(&self, transaction_name: &str) -> MaybeNewRelicTransaction {
+    if let Some(ref newrelic_api) = self.newrelic_api {
+      info!("Logging request to NewRelic");
+      let maybe_transaction = newrelic_api.web_transaction(transaction_name).ok();
+      return MaybeNewRelicTransaction::from_optional(maybe_transaction);
+    }
+    MaybeNewRelicTransaction::from_optional(None)
+  }
+}
+
+pub struct MaybeNewRelicTransaction {
+  transaction: Option<Transaction>,
+}
+
+impl MaybeNewRelicTransaction {
+  pub fn from_optional(optional: Option<Transaction>) -> Self {
+    Self {
+      transaction: optional,
+    }
+  }
+
+  pub fn add_attribute(&self, name: &str, value: &str) {
+    if let Some(ref transaction) = self.transaction {
+      info!("Logging attribute to NewRelic");
+      let _result = transaction.add_attribute(name, value);
+    }
+  }
+}

old_code/voder-proxy/src/speak_proxy.rs

@@ -0,0 +1,149 @@
+use crate::Router;
+use crate::api::{ExternalHttpRequest, InternalSpeakRequest, ErrorResponse, ErrorType};
+use crate::newrelic_logger::{MaybeNewRelicTransaction};
+use hyper::Method;
+use hyper::header::HeaderValue;
+use hyper::http::header::HeaderName;
+use hyper::{Body, Request, Response, HeaderMap};
+use std::fmt::Debug;
+use std::net::IpAddr;
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+pub async fn speak_proxy_with_retry(
+  ip_addr: IpAddr,
+  request: Request<Body>,
+  router: Arc<Router>,
+  _transaction: MaybeNewRelicTransaction,
+  max_retries: u8,
+  retry_wait_ms: u64,
+  proxy_hostname: &str,
+  endpoint: &'static str) -> Result<Response<Body>, ErrorResponse>
+{
+  let mut attempt : u8 = 0;
+
+  let external_http_request = ExternalHttpRequest::decode_request(request).await.unwrap();
+
+  // This is the IP of the Digital Ocean load balancer.
+  let gateway_ip = HeaderValue::from_str(&ip_addr.to_string())
+    .ok()
+    .unwrap_or(HeaderValue::from_static(""));
+
+  // This is the IP of the upstream client browser.
+  let upstream_client_ip = external_http_request.headers.get(HeaderName::from_static("x-forwarded-for"))
+    .map(|ip| ip.to_str().unwrap_or(""))
+    .unwrap_or("");
+
+  let upstream_client_ip = HeaderValue::from_str(upstream_client_ip)
+    .ok()
+    .unwrap_or(HeaderValue::from_static(""));
+
+  let mut headers_to_proxy = external_http_request.headers.clone();
+
+  headers_to_proxy.insert(HeaderName::from_static("forwarded"), upstream_client_ip.clone());
+  headers_to_proxy.insert(HeaderName::from_static("x-forwarded-for"), upstream_client_ip.clone());
+
+  // Unfortunately it looks like this middleware trounces the standard headers, so
+  // here we put it somewhere it won't be overwritten.
+  headers_to_proxy.insert(HeaderName::from_static("x-voder-proxy-for"), upstream_client_ip.clone());
+  headers_to_proxy.insert(HeaderName::from_static("x-gateway-ip"), gateway_ip);
+
+  // We're going to be changing the request body length, so remove the reported value
+  // so Hyper doesn't choke on any mismatch.
+  headers_to_proxy.remove(HeaderName::from_static("content-length"));
+
+  // The proxy host is a Kubernetes load balancer. We're not doing the round robin magic here,
+  // so it makes sense to reuse the same value for each retry.
+  let load_balancer_hostname = router.get_speaker_host(&external_http_request.speak_request.speaker);
+
+  loop {
+    let internal_speak_request = InternalSpeakRequest {
+      speaker: external_http_request.speak_request.speaker.clone(),
+      text: external_http_request.speak_request.text.clone(),
+      retry_attempt_number: attempt,
+      skip_rate_limiter: false,
+    };
+
+    let result = speak_proxy(
+      ip_addr,
+      internal_speak_request,
+      headers_to_proxy.clone(),
+      &load_balancer_hostname,
+      endpoint).await;
+
+    match result {
+      Ok(good_response) => {
+        // All we want to do is add our special header
+        return Ok(modify_response(good_response, proxy_hostname, attempt));
+      },
+      Err(_) => {},
+    }
+
+    attempt += 1;
+
+    thread::sleep(Duration::from_millis(retry_wait_ms)); // TODO: Needs backoff + jitter
+
+    if attempt >= max_retries {
+      return Err(ErrorResponse {
+        error_type: ErrorType::ProxyError,
+        error_description: "max internal retry attempts reached".to_string(),
+      });
+    }
+  }
+}
+
+#[derive(Serialize, Debug)]
+pub struct ResponseError {
+  pub reason: String,
+}
+
+async fn speak_proxy(
+  ip_addr: IpAddr,
+  internal_speak_request: InternalSpeakRequest,
+  headers: HeaderMap<HeaderValue>,
+  load_balancer_hostname: &str,
+  endpoint: &'static str) -> Result<Response<Body>, ()>
+{
+  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)
+    .uri(endpoint);
+
+  for (k, v) in headers.iter() {
+    // NB: Handling move because of builder class
+    request_builder = request_builder.header(k, v);
+  }
+
+  let json_speak_request = serde_json::to_string(&internal_speak_request).unwrap();
+
+  let request = request_builder
+    .body(Body::from(json_speak_request))
+    .unwrap();
+
+  match hyper_reverse_proxy::call(ip_addr, &load_balancer_hostname, request).await {
+    Ok(result) => Ok(result),
+    Err(_result) => Err(()), // TODO: Surface this?
+  }
+}
+
+fn modify_response(mut response: Response<Body>, proxy_hostname: &str, retry_count: u8)
+  -> Response<Body>
+{
+  let headers = response.headers_mut();
+
+  let proxy_hostname = HeaderValue::from_str(proxy_hostname)
+    .ok()
+    .unwrap_or(HeaderValue::from_static("proxy-unknown"));
+
+  let retry_count = HeaderValue::from_str(&retry_count.to_string())
+    .ok()
+    .unwrap_or(HeaderValue::from_static("unknown"));
+
+  headers.append(HeaderName::from_static("x-proxy-hostname"), proxy_hostname);
+  headers.append(HeaderName::from_static("x-retry-count"), retry_count);
+
+  response
+}