From e1c39e65de92c214037397ec4c1ad73c1c7de3e9 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 7 Sep 2020 14:12:40 -0400 Subject: [PATCH] Robust proxying with retry Also includes upgraded hyper-reverse-proxy, hyper, and futures. This is looking and feeling way more ergonomic. --- Cargo.lock | 590 +++++++++++++++++++++------------------------ Cargo.toml | 9 +- src/api.rs | 102 ++++++++ src/main.rs | 136 +++++++---- src/speak_proxy.rs | 168 +++++++------ 5 files changed, 556 insertions(+), 449 deletions(-) create mode 100644 src/api.rs diff --git a/Cargo.lock b/Cargo.lock index aeefe27..dcd869e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -15,6 +15,12 @@ version = "1.0.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "85bb70cc08ec97ca5450e6eba421deeea5f172c0fc61f78b5357b2a8e8be195f" +[[package]] +name = "arc-swap" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d25d88fd6b8041580a654f9d0c581a047baee2b3efee13275f2fc392fc75034" + [[package]] name = "atty" version = "0.2.14" @@ -38,22 +44,11 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693" -[[package]] -name = "byteorder" -version = "1.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08c48aae112d48ed9f069b33538ea9e3e90aa263cfa3d1c24309612b1f7472de" - [[package]] name = "bytes" -version = "0.4.12" +version = "0.5.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "206fdffcfa2df7cbe15601ef46c813fce0965eb3286db6b56c583b814b51c81c" -dependencies = [ - "byteorder", - "either", - "iovec", -] +checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38" [[package]] name = "cfg-if" @@ -61,69 +56,6 @@ version = "0.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822" -[[package]] -name = "cloudabi" -version = "0.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddfc5b9aa5d4507acaf872de71051dfd0e309860e88966e1051e462a077aac4f" -dependencies = [ - "bitflags", -] - -[[package]] -name = "crossbeam-deque" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f02af974daeee82218205558e51ec8768b48cf524bd01d550abe5573a608285" -dependencies = [ - "crossbeam-epoch", - "crossbeam-utils", - "maybe-uninit", -] - -[[package]] -name = "crossbeam-epoch" -version = "0.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "058ed274caafc1f60c4997b5fc07bf7dc7cca454af7c6e81edffe5f33f70dace" -dependencies = [ - "autocfg", - "cfg-if", - "crossbeam-utils", - "lazy_static", - "maybe-uninit", - "memoffset", - "scopeguard", -] - -[[package]] -name = "crossbeam-queue" -version = "0.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "774ba60a54c213d409d5353bda12d49cd68d14e45036a285234c8d6f91f92570" -dependencies = [ - "cfg-if", - "crossbeam-utils", - "maybe-uninit", -] - -[[package]] -name = "crossbeam-utils" -version = "0.7.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3c7c73a2d1e9fc0886a08b93e98eb643461230d5f1925e4036204d5f2e261a8" -dependencies = [ - "autocfg", - "cfg-if", - "lazy_static", -] - -[[package]] -name = "either" -version = "1.5.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb1f6b1ce1c140482ea30ddd3335fc0024ac7ee112895426e0a629a6c20adfe3" - [[package]] name = "env_logger" version = "0.7.1" @@ -161,18 +93,97 @@ checksum = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" [[package]] name = "futures" -version = "0.1.29" +version = "0.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b980f2816d6ee8673b6517b52cb0e808a180efc92e5c19d02cdda79066703ef" +checksum = "1e05b85ec287aac0dc34db7d4a569323df697f9c55b99b15d6b4ef8cde49f613" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] [[package]] -name = "futures-cpupool" -version = "0.1.8" +name = "futures-channel" +version = "0.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ab90cde24b3319636588d0c35fe03b1333857621051837ed769faefb4c2162e4" +checksum = "f366ad74c28cca6ba456d95e6422883cfb4b252a83bed929c83abfdbbf2967d5" dependencies = [ - "futures", - "num_cpus", + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59f5fff90fd5d971f936ad674802482ba441b6f09ba5e15fd8b39145582ca399" + +[[package]] +name = "futures-executor" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10d6bb888be1153d3abeb9006b11b02cf5e9b209fda28693c31ae1e4e012e314" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de27142b013a8e869c14957e6d2edeef89e97c289e69d042ee3a49acd8b51789" + +[[package]] +name = "futures-macro" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0b5a30a4328ab5473878237c447333c093297bded83a4983d10f4deea240d39" +dependencies = [ + "proc-macro-hack", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f2032893cb734c7a05d85ce0cc8b8c4075278e93b24b66f9de99d6eb0fa8acc" + +[[package]] +name = "futures-task" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bdb66b5f09e22019b1ab0830f7785bcea8e7a42148683f99214f73f8ec21a626" +dependencies = [ + "once_cell", +] + +[[package]] +name = "futures-util" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8764574ff08b701a084482c3c7031349104b07ac897393010494beaa18ce32c6" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project", + "pin-utils", + "proc-macro-hack", + "proc-macro-nested", + "slab", ] [[package]] @@ -188,20 +199,21 @@ dependencies = [ [[package]] name = "h2" -version = "0.1.26" +version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5b34c246847f938a410a03c5458c7fee2274436675e76d8b903c08efc29c462" +checksum = "993f9e0baeed60001cf565546b0d3dbe6a6ad23f2bd31644a133c641eccf6d53" dependencies = [ - "byteorder", "bytes", "fnv", - "futures", + "futures-core", + "futures-sink", + "futures-util", "http", "indexmap", - "log", "slab", - "string", - "tokio-io", + "tokio", + "tokio-util", + "tracing", ] [[package]] @@ -215,9 +227,9 @@ dependencies = [ [[package]] name = "http" -version = "0.1.21" +version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6ccf5ede3a895d8856620237b2f02972c1bbc78d2965ad7fe8838d4a0ed41f0" +checksum = "28d569972648b2c512421b5f2a405ad6ac9666547189d0c5477a3f200f3e02f9" dependencies = [ "bytes", "fnv", @@ -226,14 +238,12 @@ dependencies = [ [[package]] name = "http-body" -version = "0.1.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6741c859c1b2463a423a1dbce98d418e6c3c3fc720fb0d45528657320920292d" +checksum = "13d5ff830006f7646652e057693569bfe0d51760c0085a071769d142a205111b" dependencies = [ "bytes", - "futures", "http", - "tokio-buf", ] [[package]] @@ -253,41 +263,33 @@ dependencies = [ [[package]] name = "hyper" -version = "0.12.35" +version = "0.13.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dbe6ed1438e1f8ad955a4701e9a944938e9519f6888d12d8558b645e247d5f6" +checksum = "3e68a8dd9716185d9e64ea473ea6ef63529252e3e27623295a0378a19665d5eb" dependencies = [ "bytes", - "futures", - "futures-cpupool", + "futures-channel", + "futures-core", + "futures-util", "h2", "http", "http-body", "httparse", - "iovec", "itoa", - "log", - "net2", - "rustc_version", + "pin-project", + "socket2", "time", "tokio", - "tokio-buf", - "tokio-executor", - "tokio-io", - "tokio-reactor", - "tokio-tcp", - "tokio-threadpool", - "tokio-timer", + "tower-service", + "tracing", "want", ] [[package]] name = "hyper-reverse-proxy" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "474f19c624c13005d914fc4a73aa2a2921b0371f5f7b748146f5130685492ce8" +version = "0.5.0" +source = "git+https://github.com/felipenoris/hyper-reverse-proxy.git#d109fcbd629c62eeb5c34bec1074d737e7b58529" dependencies = [ - "futures", "hyper", "lazy_static", "unicase", @@ -339,15 +341,6 @@ version = "0.2.71" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9457b06509d27052635f90d6466700c65095fdf75409b3fbdd903e988b886f49" -[[package]] -name = "lock_api" -version = "0.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4da24a77a3d8a6d4862d95f72e6fdb9c09a643ecdb402d754004a557f2bec75" -dependencies = [ - "scopeguard", -] - [[package]] name = "log" version = "0.4.8" @@ -357,27 +350,12 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "maybe-uninit" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "60302e4db3a61da70c0cb7991976248362f30319e88850c487b9b95bbf059e00" - [[package]] name = "memchr" version = "2.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3728d817d99e5ac407411fa471ff9800a778d88a24685968b36824eaf4bee400" -[[package]] -name = "memoffset" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4fc2c02a7e374099d4ee95a193111f72d2110197fe200272371758f6c3643d8" -dependencies = [ - "autocfg", -] - [[package]] name = "mio" version = "0.6.22" @@ -391,12 +369,35 @@ dependencies = [ "kernel32-sys", "libc", "log", - "miow", + "miow 0.2.1", "net2", "slab", "winapi 0.2.8", ] +[[package]] +name = "mio-named-pipes" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0840c1c50fd55e521b247f949c241c9997709f23bd7f023b9762cd561e935656" +dependencies = [ + "log", + "mio", + "miow 0.3.5", + "winapi 0.3.9", +] + +[[package]] +name = "mio-uds" +version = "0.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "afcb699eb26d4332647cc848492bbc15eafb26f08d0304550d5aa1f612e066f0" +dependencies = [ + "iovec", + "libc", + "mio", +] + [[package]] name = "miow" version = "0.2.1" @@ -409,6 +410,16 @@ dependencies = [ "ws2_32-sys", ] +[[package]] +name = "miow" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07b88fb9795d4d36d62a012dfbf49a8f5cf12751f36d31a9dbe66d528e58979e" +dependencies = [ + "socket2", + "winapi 0.3.9", +] + [[package]] name = "net2" version = "0.2.34" @@ -447,37 +458,61 @@ dependencies = [ ] [[package]] -name = "parking_lot" -version = "0.9.0" +name = "once_cell" +version = "1.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f842b1982eb6c2fe34036a4fbfb06dd185a3f5c8edfaacdf7d1ea10b07de6252" +checksum = "260e51e7efe62b592207e9e13a68e43692a7a279171d6ba57abd208bf23645ad" + +[[package]] +name = "pin-project" +version = "0.4.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca4433fff2ae79342e497d9f8ee990d174071408f28f726d6d83af93e58e48aa" dependencies = [ - "lock_api", - "parking_lot_core", - "rustc_version", + "pin-project-internal", ] [[package]] -name = "parking_lot_core" -version = "0.6.2" +name = "pin-project-internal" +version = "0.4.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b876b1b9e7ac6e1a74a6da34d25c42e17e8862aa409cbbbdcfc8d86c6f3bc62b" +checksum = "2c0e815c3ee9a031fdf5af21c10aa17c573c9c6a566328d99e3936c34e36461f" dependencies = [ - "cfg-if", - "cloudabi", - "libc", - "redox_syscall", - "rustc_version", - "smallvec", - "winapi 0.3.9", + "proc-macro2", + "quote", + "syn", ] +[[package]] +name = "pin-project-lite" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282adbf10f2698a7a77f8e983a74b2d18176c19a7fd32a45446139ae7b02b715" + +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "ppv-lite86" version = "0.2.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "237a5ed80e274dbc66f86bd59c1e25edc039660be53194b5fe0a482e0f2612ea" +[[package]] +name = "proc-macro-hack" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "99c605b9a0adc77b7211c6b1f722dcb613d68d66859a44f3d485a6da332b0598" + +[[package]] +name = "proc-macro-nested" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eba180dafb9038b050a4c280019bbedf9f2467b61e5d892dcad585bb57aadc5a" + [[package]] name = "proc-macro2" version = "1.0.18" @@ -567,42 +602,12 @@ version = "0.6.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26412eb97c6b088a6997e05f69403a802a92d520de2f8e63c2b65f9e0f47c4e8" -[[package]] -name = "rustc_version" -version = "0.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "138e3e0acb6c9fb258b19b67cb8abd63c00679d2851805ea151465464fe9030a" -dependencies = [ - "semver", -] - [[package]] name = "ryu" version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "71d301d4193d031abdd79ff7e3dd721168a9572ef3fe51a1517aba235bd8f86e" -[[package]] -name = "scopeguard" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd" - -[[package]] -name = "semver" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403" -dependencies = [ - "semver-parser", -] - -[[package]] -name = "semver-parser" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3" - [[package]] name = "serde" version = "1.0.114" @@ -631,6 +636,16 @@ dependencies = [ "serde", ] +[[package]] +name = "signal-hook-registry" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3e12110bc539e657a646068aaf5eb5b63af9d0c1f7b29c97113fad80e15f035" +dependencies = [ + "arc-swap", + "libc", +] + [[package]] name = "slab" version = "0.4.2" @@ -638,21 +653,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c111b5bd5695e56cffe5129854aa230b39c93a305372fdbb2668ca2394eea9f8" [[package]] -name = "smallvec" -version = "0.6.13" +name = "socket2" +version = "0.3.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7b0758c52e15a8b5e3691eae6cc559f08eee9406e548a4477ba4e67770a82b6" +checksum = "03088793f677dce356f3ccc2edb1b314ad191ab702a5de3faf49304f7e104918" dependencies = [ - "maybe-uninit", -] - -[[package]] -name = "string" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d24114bfcceb867ca7f71a0d3fe45d45619ec47a6fbfa98cb14e14250bfa5d6d" -dependencies = [ - "bytes", + "cfg-if", + "libc", + "redox_syscall", + "winapi 0.3.9", ] [[package]] @@ -696,134 +705,51 @@ dependencies = [ [[package]] name = "tokio" -version = "0.1.22" +version = "0.2.22" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a09c0b5bb588872ab2f09afa13ee6e9dac11e10a0ec9e8e3ba39a5a5d530af6" +checksum = "5d34ca54d84bf2b5b4d7d31e901a8464f7b60ac145a284fba25ceb801f2ddccd" dependencies = [ "bytes", - "futures", - "mio", - "num_cpus", - "tokio-current-thread", - "tokio-executor", - "tokio-io", - "tokio-reactor", - "tokio-threadpool", - "tokio-timer", -] - -[[package]] -name = "tokio-buf" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fb220f46c53859a4b7ec083e41dec9778ff0b1851c0942b211edb89e0ccdc46" -dependencies = [ - "bytes", - "either", - "futures", -] - -[[package]] -name = "tokio-current-thread" -version = "0.1.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1de0e32a83f131e002238d7ccde18211c0a5397f60cbfffcb112868c2e0e20e" -dependencies = [ - "futures", - "tokio-executor", -] - -[[package]] -name = "tokio-executor" -version = "0.1.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb2d1b8f4548dbf5e1f7818512e9c406860678f29c300cdf0ebac72d1a3a1671" -dependencies = [ - "crossbeam-utils", - "futures", -] - -[[package]] -name = "tokio-io" -version = "0.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57fc868aae093479e3131e3d165c93b1c7474109d13c90ec0dda2a1bbfff0674" -dependencies = [ - "bytes", - "futures", - "log", -] - -[[package]] -name = "tokio-reactor" -version = "0.1.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09bc590ec4ba8ba87652da2068d150dcada2cfa2e07faae270a5e0409aa51351" -dependencies = [ - "crossbeam-utils", - "futures", - "lazy_static", - "log", - "mio", - "num_cpus", - "parking_lot", - "slab", - "tokio-executor", - "tokio-io", - "tokio-sync", -] - -[[package]] -name = "tokio-sync" -version = "0.1.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "edfe50152bc8164fcc456dab7891fa9bf8beaf01c5ee7e1dd43a397c3cf87dee" -dependencies = [ "fnv", - "futures", + "futures-core", + "iovec", + "lazy_static", + "libc", + "memchr", + "mio", + "mio-named-pipes", + "mio-uds", + "num_cpus", + "pin-project-lite", + "signal-hook-registry", + "slab", + "tokio-macros", + "winapi 0.3.9", ] [[package]] -name = "tokio-tcp" -version = "0.1.4" +name = "tokio-macros" +version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "98df18ed66e3b72e742f185882a9e201892407957e45fbff8da17ae7a7c51f72" +checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499" dependencies = [ "bytes", - "futures", - "iovec", - "mio", - "tokio-io", - "tokio-reactor", -] - -[[package]] -name = "tokio-threadpool" -version = "0.1.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df720b6581784c118f0eb4310796b12b1d242a7eb95f716a8367855325c25f89" -dependencies = [ - "crossbeam-deque", - "crossbeam-queue", - "crossbeam-utils", - "futures", - "lazy_static", + "futures-core", + "futures-sink", "log", - "num_cpus", - "slab", - "tokio-executor", -] - -[[package]] -name = "tokio-timer" -version = "0.2.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93044f2d313c95ff1cb7809ce9a7a05735b012288a888b62d4434fd58c94f296" -dependencies = [ - "crossbeam-utils", - "futures", - "slab", - "tokio-executor", + "pin-project-lite", + "tokio", ] [[package]] @@ -835,6 +761,32 @@ dependencies = [ "serde", ] +[[package]] +name = "tower-service" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e987b6bf443f4b5b3b6f38704195592cca41c5bb7aedd3c3693c7081f8289860" + +[[package]] +name = "tracing" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d79ca061b032d6ce30c660fded31189ca0b9922bf483cd70759f13a2d86786c" +dependencies = [ + "cfg-if", + "log", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f0e00789804e99b20f12bc7003ca416309d28a6f495d6af58d1e2c2842461b5" +dependencies = [ + "lazy_static", +] + [[package]] name = "try-lock" version = "0.2.2" @@ -856,6 +808,7 @@ dependencies = [ "serde", "serde_derive", "serde_json", + "tokio", "toml", ] @@ -882,11 +835,10 @@ checksum = "b5a972e5669d67ba988ce3dc826706fb0a8b01471c088cb0b6110b805cc36aed" [[package]] name = "want" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6395efa4784b027708f7451087e647ec73cc74f5d9bc2e418404248d679a230" +checksum = "1ce8a968cb1cd110d136ff8b819a556d6fb6d919363c61534f6860c7eb172ba0" dependencies = [ - "futures", "log", "try-lock", ] diff --git a/Cargo.toml b/Cargo.toml index 977a05f..68f9542 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,13 +15,14 @@ path = "src/main.rs" [dependencies] anyhow = "1.0.31" env_logger = "0.7.1" -futures = "0.1" -hyper = "0.12" -hyper-reverse-proxy = "0.4" +futures = "0.3" +hyper = "0.13" +hyper-reverse-proxy = { git = "https://github.com/felipenoris/hyper-reverse-proxy.git", commit = "d109fcbd629c62eeb5c34bec1074d737e7b58529" } log = "0.4.8" newrelic = "0.2.2" rand = "0.7.3" serde = "1.0.104" serde_derive = "1.0.104" serde_json = "1.0.55" -toml = "0.5.6" +tokio = { version = "0.2", features = ["full"] } +toml = "0.5.6" \ No newline at end of file diff --git a/src/api.rs b/src/api.rs new file mode 100644 index 0000000..38926a8 --- /dev/null +++ b/src/api.rs @@ -0,0 +1,102 @@ +use anyhow::Result as AnyhowResult; +use futures::TryStreamExt; +use hyper::http::HeaderValue; +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. +#[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, + 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) -> AnyhowResult { + 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::(&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 { + 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 { + data.extend_from_slice(&chunk); + Ok(data) + }).await?; + + Ok(bytes) +} diff --git a/src/main.rs b/src/main.rs index f733e4e..970061b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,32 +1,33 @@ +#![feature(toowned_clone_into)] + #[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, MaybeNewRelicTransaction}; -use crate::speak_proxy::{speak_proxy, speak_proxy_with_retry}; +use crate::speak_proxy::speak_proxy_with_retry; use futures::future::{self, Future}; -use hyper::Method; use hyper::header::HeaderValue; use hyper::http::header::HeaderName; -use hyper::rt::Stream; 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; use std::env; use std::fmt::Debug; use std::fmt::Display; -use std::net::SocketAddr; +use std::net::{SocketAddr, IpAddr}; use std::str::FromStr; use std::sync::Arc; -type BoxFut = Box, Error=hyper::Error> + 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"; @@ -88,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, @@ -120,7 +110,66 @@ impl Router { } } -fn main() -> AnyhowResult<()> { +fn debug_request(req: Request) -> Result, Infallible> { + let body_str = format!("{:?}", req); + 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, + router: Arc, + newrelic_logger: Arc) -> 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; + match result { + 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(error.to_response()), + } + }, + "/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()) + } + } + }, + } +} + +#[tokio::main] +async fn main() { if env::var(ENV_RUST_LOG) .as_ref() .ok() @@ -134,7 +183,8 @@ fn main() -> AnyhowResult<()> { env_logger::init(); let default_route = get_env_string(ENV_DEFAULT_ROUTE, DEFAULT_DEFAULT_ROUTE); - let service_port = get_env_num::(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT)?; + let service_port = get_env_num::(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT) + .expect("should have env var"); let newrelic_api_key = get_env_string(ENV_NEWRELIC_API_KEY, DEFAULT_NEWRELIC_API_KEY); info!("Default route: {}", default_route); @@ -142,7 +192,8 @@ fn main() -> AnyhowResult<()> { let proxy_configs_file = get_env_string(ENV_PROXY_CONFIG_FILE, DEFAULT_PROXY_CONFIG_FILE); info!("Proxy config file: {}", proxy_configs_file); - let proxy_configs = ProxyConfigs::load_from_file(&proxy_configs_file)?; + let proxy_configs = ProxyConfigs::load_from_file(&proxy_configs_file) + .expect("should have configs"); info!("Proxy configs: {:?}", proxy_configs); let mut route_map = HashMap::new(); @@ -165,49 +216,30 @@ fn main() -> AnyhowResult<()> { 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 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(); - service_fn(move |req: Request| { + async move { 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_with_retry(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) - } - } - }) + Ok::<_, Infallible>(service_fn(move |req| { + let router4 = router3.clone(); + let newrelic_logger4 = newrelic_logger3.clone(); + handle(remote_addr, req, router4, newrelic_logger4) + })) + } }); let addr = ([0, 0, 0, 0], service_port).into(); info!("Running server on {:?}", addr); - let server = Server::bind(&addr) - .serve(make_service) - .map_err(|e| eprintln!("server error: {}", e)); + let server = Server::bind(&addr).serve(make_service); - hyper::rt::run(server); - - Ok(()) + if let Err(e) = server.await { + eprintln!("server error: {}", e); + } } diff --git a/src/speak_proxy.rs b/src/speak_proxy.rs index fbe199f..75fa09a 100644 --- a/src/speak_proxy.rs +++ b/src/speak_proxy.rs @@ -1,72 +1,45 @@ use anyhow::Result as AnyhowResult; -use crate::{BoxFut, Router}; +use crate::Router; +use crate::api::{ExternalSpeakRequest, ExternalHttpRequest, InternalSpeakRequest, ErrorResponse, ErrorType}; use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction}; +use futures::TryStreamExt; use futures::future::{self, Future}; -use hyper::Method; use hyper::header::HeaderValue; use hyper::http::header::HeaderName; -use hyper::rt::Stream; use hyper::server::conn::AddrStream; use hyper::service::{service_fn, make_service_fn}; -use hyper::{Body, Request, Response, Server}; +use hyper::{Body, Request, Response, Server, HeaderMap}; +use hyper::{Method, StatusCode}; +use hyper_reverse_proxy::ProxyError; use rand::seq::IteratorRandom; use std::collections::HashMap; -use std::env; +use std::convert::Infallible; use std::fmt::Debug; use std::fmt::Display; -use std::net::SocketAddr; +use std::net::{SocketAddr, IpAddr}; use std::str::FromStr; use std::sync::Arc; -use futures::IntoFuture; +use std::time::Duration; +use std::{env, thread}; -/// 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(Serialize, Deserialize, Debug)] -pub struct SpeakRequest { - /// Slug for the speaker - speaker: String, - /// Raw text to be spoken - text: String, -} - -struct RequestDetails { - pub request_bytes: Vec, - pub speaker: String, -} - -pub fn speak_proxy_with_retry( - req: Request, - remote_addr: SocketAddr, +pub async fn speak_proxy_with_retry( + ip_addr: IpAddr, + request: Request, router: Arc, - transaction: MaybeNewRelicTransaction, - endpoint: &'static str) -> BoxFut + _transaction: MaybeNewRelicTransaction, + endpoint: &'static str) -> Result, ErrorResponse> { - speak_proxy(req, remote_addr, router, transaction, endpoint) - .then(|result| { - match result { - Ok(t) => future::ok(t), - Err(t) => future::err(t), - } - }) - .boxed() -} + let mut attempt : u8 = 0; -pub fn speak_proxy( - req: Request, - remote_addr: SocketAddr, - router: Arc, - transaction: MaybeNewRelicTransaction, - endpoint: &'static str) -> BoxFut -{ - let mut headers = req.headers().clone(); + 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(&remote_addr.ip().to_string()) + 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 = req.headers().get(HeaderName::from_static("x-forwarded-for")) + let upstream_client_ip = external_http_request.headers.get(HeaderName::from_static("x-forwarded-for")) .map(|ip| ip.to_str().unwrap_or("")) .unwrap_or(""); @@ -74,42 +47,89 @@ pub fn speak_proxy( .ok() .unwrap_or(HeaderValue::from_static("")); - headers.insert(HeaderName::from_static("forwarded"), upstream_client_ip.clone()); - headers.insert(HeaderName::from_static("x-forwarded-for"), upstream_client_ip.clone()); + 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.insert(HeaderName::from_static("x-voder-proxy-for"), upstream_client_ip.clone()); - headers.insert(HeaderName::from_static("x-gateway-ip"), gateway_ip); + 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); - Box::new(req.into_body().concat2().map(move |b| { // Builds a BoxedFut to return - let request_bytes = b.as_ref(); + // 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")); - let request = serde_json::from_slice::(request_bytes) - .unwrap(); + // 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); - RequestDetails { - request_bytes: request_bytes.to_vec(), - speaker: request.speaker, - } - }).and_then(move |request_details: RequestDetails| { - let proxy_host = router.get_speaker_host(&request_details.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, + }; - info!("Routing {} for {} to {}", endpoint, &request_details.speaker, &proxy_host); + let result = speak_proxy( + ip_addr, + internal_speak_request, + headers_to_proxy.clone(), + &load_balancer_hostname, + endpoint).await; - transaction.add_attribute("speaker", &request_details.speaker); - - let mut request_builder = Request::builder(); - request_builder.method(Method::POST) - .uri(endpoint); - - for (k, v) in headers.iter() { - request_builder.header(k, v); + match result { + Ok(good_response) => return Ok(good_response), + Err(_) => {}, } - let new_req = request_builder - .body(Body::from(request_details.request_bytes)) - .unwrap(); + attempt += 1; - hyper_reverse_proxy::call(remote_addr.ip(), &proxy_host, new_req) - })) + 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(), + }); + } + } +} + +#[derive(Serialize, Debug)] +pub struct ResponseError { + pub reason: String, +} + +async fn speak_proxy( + ip_addr: IpAddr, + internal_speak_request: InternalSpeakRequest, + headers: HeaderMap, + load_balancer_hostname: &str, + endpoint: &'static str) -> Result, ()> +{ + 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? + } }