Add support for *_proxy env vars

This includes `http_proxy`, `https_proxy` and `all_proxy` semi-standard env vars.

This supports both capital and lower-case env vars because these variables are confusing to
everyone. Curl may have introduced them, and they use lower-case for `http_proxy` and
screaming-snake case for all others[1]. Stack Exchange answers around these questions have multiple
conflicting recommendations (although this one is good[2]), and I've personally seen folks think
that only one or the other (lower, upper) or mixed-use are required for all projects.

Since folks are going to be coming to us with some previous experience and confusion around these,
just supporting both seems like the thing that will cause the fewest number of support requests.

Some downsides of this commit:

* It unfortunately re-introduces time v0.1 into our dependecy tree, but I have a PR up to
  remove it.
* PubNub sources don't respect the headers, the client library hard-codes their hyper::Client type
  parameters, and `hyper::Client<impl Connect + Clone>` in general is hard to type-erase.

[1]: https://curl.se/docs/manual.html#environment-variables
[2]: https://superuser.com/a/1166790

Author: quodlibetor

Cargo.lock

@@ -13,6 +13,7 @@ members = [
     "src/dataflow-types",
     "src/dataflow",
     "src/expr",
+    "src/http-util",
     "src/interchange",
     "src/kafka-util",
     "src/materialized",

Cargo.toml

@@ -24,6 +24,11 @@ skip = [
 
     # Waiting on parse_duration.
     { name = "num-bigint", version = "0.2.6" },
+
+    # waiting on https://github.com/hyperium/headers/pull/83
+    { name = "time", version = "0.1.44" },
+    # this is already merged into tokio-postgres, just waiting for a release
+    { name = "socket2", version = "0.3.19" },
 ]
 deny = [
     # Strum has suspect code quality and includes many unneeded features. Use

deny.toml

@@ -8,11 +8,12 @@ publish = false
 [dependencies]
 anyhow = "1.0.40"
 log = "0.4.13"
+http-util = { path = "../http-util" }
 rusoto_core = { git = "https://github.com/rusoto/rusoto.git" }
 rusoto_credential = { git = "https://github.com/rusoto/rusoto.git" }
 rusoto_kinesis = { git = "https://github.com/rusoto/rusoto.git" }
 rusoto_s3 = { git = "https://github.com/rusoto/rusoto.git" }
-rusoto_sts = { git = "https://github.com/rusoto/rusoto.git" }
 rusoto_sqs = { git = "https://github.com/rusoto/rusoto.git" }
+rusoto_sts = { git = "https://github.com/rusoto/rusoto.git" }
 serde = { version = "1.0.124", features = ["derive"] }
 tokio = "1.4.0"

src/aws-util/Cargo.toml

@@ -10,7 +10,7 @@
 //! Utility functions for AWS.
 
 use anyhow::{anyhow, Context};
-use rusoto_core::{HttpClient, Region};
+use rusoto_core::Region;
 use rusoto_credential::{
     AutoRefreshingProvider, AwsCredentials, ChainProvider, ProvideAwsCredentials, StaticProvider,
 };
@@ -81,7 +81,8 @@ pub async fn account(
     region: Region,
     timeout: Duration,
 ) -> Result<String, anyhow::Error> {
-    let dispatcher = HttpClient::new().context("creating HTTP for AWS STS Account verification")?;
+    let dispatcher =
+        crate::client::http().context("creating HTTP client for AWS STS Account verification")?;
     let sts_client = StsClient::new_with(dispatcher, provider, region);
     let get_identity = sts_client.get_caller_identity(GetCallerIdentityRequest {});
     let account = time::timeout(timeout, get_identity)

src/aws-util/src/aws.rs

@@ -14,7 +14,7 @@
 
 use std::time::Duration;
 
-use anyhow::Context;
+use anyhow::{anyhow, Context};
 use log::info;
 use rusoto_core::HttpClient;
 use rusoto_credential::{AutoRefreshingProvider, AwsCredentials, ChainProvider, StaticProvider};
@@ -24,6 +24,13 @@ use rusoto_sqs::SqsClient;
 
 use crate::aws::ConnectInfo;
 
+/// Get an [`HttpClient`]  that respects the `http_proxy` environment variables
+pub(crate) fn http() -> Result<HttpClient<http_util::ProxiedConnector>, anyhow::Error> {
+    Ok(HttpClient::from_connector(
+        http_util::connector().map_err(|e| anyhow!(e))?,
+    ))
+}
+
 /// Create a function that calls <Client>::new() with credentials sources
 ///
 /// Unfortunately there is no trait that we can rely on in Rusoto to allow us
@@ -51,22 +58,24 @@ The [`AutoRefreshingProvider`] caches the underlying provider's AWS credentials,
 automatically fetching updated credentials if they've expired.
 "]
 pub async fn $name(conn_info: ConnectInfo) -> Result<$client, anyhow::Error> {
-    let request_dispatcher = HttpClient::new().context("creating HTTP client for S3 client")?;
+    let request_dispatcher = http().context(
+        concat!("creating HTTP client for ", $client_name))?;
     let the_client = if let Some(creds) = conn_info.credentials {
-        info!(concat!("Creating a new", stringify!(client), " from provided access_key and secret_access_key"));
+        info!(concat!("Creating a new", $client_name, " from provided access_key and secret_access_key"));
         let provider = StaticProvider::from(AwsCredentials::from(creds));
 
         $client::new_with(request_dispatcher, provider, conn_info.region)
     } else {
         info!(
             concat!("AWS access_key_id and secret_access_key not provided, \
-               creating a new ", stringify!($client), " using a chain provider.")
+               creating a new ", $client_name, " using a chain provider.")
         );
         let mut provider = ChainProvider::new();
 
         provider.set_timeout(Duration::from_secs(10));
         let provider =
-            AutoRefreshingProvider::new(provider).context(concat!("generating AWS credentials refreshing provider for ", stringify!($client)))?;
+            AutoRefreshingProvider::new(provider).context(
+                concat!("generating AWS credentials refreshing provider for ", $client_name))?;
 
         $client::new_with(request_dispatcher, provider, conn_info.region)
     };

src/aws-util/src/client.rs

@@ -6,7 +6,9 @@ edition = "2018"
 publish = false
 
 [dependencies]
+anyhow = "1.0.40"
 futures = "0.3.13"
+http-util = { path = "../http-util" }
 native-tls = "0.2.7"
 openssl = { version = "0.10.33", features = ["vendored"] }
 reqwest = { version = "0.11.2", features = ["blocking", "json", "native-tls-vendored"] }
@@ -15,7 +17,7 @@ serde_json = "1.0.64"
 url = { version = "2.2.1", features = ["serde"] }
 
 [dev-dependencies]
-anyhow = "1.0.40"
-hyper = "0.14.0"
+tracing = { version = "0.1.0", features = ["log"] }
+hyper = { version = "0.14.5", features = ["server"] }
 lazy_static = "1.4.0"
 tokio = { version = "1.4.0", features = ["macros"] }

src/ccsr/Cargo.toml

@@ -7,6 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use anyhow::Context;
 use reqwest::Url;
 use serde::{Deserialize, Serialize};
 
@@ -62,8 +63,9 @@ impl ClientConfig {
     }
 
     /// Builds the [`Client`].
-    pub fn build(self) -> Client {
-        let mut builder = reqwest::Client::builder();
+    pub fn build(self) -> Result<Client, anyhow::Error> {
+        let mut builder = http_util::reqwest_client_builder()
+            .context("Creating HTTP client for schema registry")?;
 
         for root_cert in self.root_certs {
             builder = builder.add_root_certificate(root_cert.into());
@@ -78,6 +80,6 @@ impl ClientConfig {
             .build()
             .unwrap();
 
-        Client::new(inner, self.url, self.auth)
+        Ok(Client::new(inner, self.url, self.auth))
     }
 }

src/ccsr/src/config.rs

@@ -23,7 +23,7 @@
 //! use ccsr::ClientConfig;
 //!
 //! let url = "http://localhost:8080".parse()?;
-//! let client = ClientConfig::new(url).build();
+//! let client = ClientConfig::new(url).build()?;
 //! let subjects = client.list_subjects().await?;
 //! for subject in subjects {
 //!     let schema = client.get_schema_by_subject(&subject).await?;