📞 Support the gRPC flag for dynamic backends (#308)

Enabling this flags adds **experimental** support for gRPC origins for compute services.

* Add a grpc flag for backends.
* Add some ALPN machinery.
* A simple test case for the http2 connection.
* Comment out the test until the feature is in the released SDK.

cli/tests/integration/common/backends.rs

@@ -73,6 +73,7 @@ impl TestBackends {
                 override_host: backend.override_host.clone(),
                 cert_host: backend.cert_host.clone(),
                 use_sni: backend.use_sni,
+                grpc: false,
                 client_cert: None,
             };
             backends.insert(name.to_string(), Arc::new(backend_config));

cli/tests/integration/grpc.rs

@@ -0,0 +1,39 @@
+//use crate::common::{Test, TestResult};
+//use hyper::http::response;
+//use hyper::server::conn::AddrIncoming;
+//use hyper::service::{make_service_fn, service_fn};
+//use hyper::{Request, Server, StatusCode};
+//use std::net::SocketAddr;
+//
+//#[tokio::test(flavor = "multi_thread")]
+//async fn grpc_creates_h2_connection() -> TestResult {
+//    let test = Test::using_fixture("grpc.wasm");
+//    let server_addr: SocketAddr = "127.0.0.1:0".parse().expect("localhost parses");
+//    let incoming = AddrIncoming::bind(&server_addr).expect("bind");
+//    let bound_port = incoming.local_addr().port();
+//
+//    let service = make_service_fn(|_| async move {
+//        Ok::<_, std::io::Error>(service_fn(move |_req| async {
+//            response::Builder::new()
+//                .status(200)
+//                .body("Hello!".to_string())
+//        }))
+//    });
+//
+//    let server = Server::builder(incoming).http2_only(true).serve(service);
+//    tokio::spawn(server);
+//
+//    let resp = test
+//        .against(
+//            Request::post("/")
+//                .header("port", bound_port)
+//                .body("Hello, Viceroy!")
+//                .unwrap(),
+//        )
+//        .await;
+//    assert_eq!(resp.status(), StatusCode::OK);
+//    assert_eq!(resp.into_body().read_into_string().await?, "Hello!");
+//
+//    Ok(())
+//}
+//

cli/tests/integration/main.rs

@@ -6,6 +6,7 @@ mod dictionary_lookup;
 mod downstream_req;
 mod env_vars;
 mod geolocation_lookup;
+mod grpc;
 mod http_semantics;
 mod kv_store;
 mod logging;

lib/src/config/backends.rs

@@ -14,6 +14,7 @@ pub struct Backend {
     pub override_host: Option<HeaderValue>,
     pub cert_host: Option<String>,
     pub use_sni: bool,
+    pub grpc: bool,
     pub client_cert: Option<ClientCertInfo>,
 }
 
@@ -126,13 +127,26 @@ mod deserialization {
                 .transpose()?
                 .unwrap_or(true);
 
+            let grpc = toml
+                .remove("grpc")
+                .map(|grpc| {
+                    if let Value::Boolean(grpc) = grpc {
+                        Ok(grpc)
+                    } else {
+                        Err(BackendConfigError::InvalidGrpcEntry)
+                    }
+                })
+                .transpose()?
+                .unwrap_or(true);
+
             check_for_unrecognized_keys(&toml)?;
 
             Ok(Self {
                 uri,
                 override_host,
                 cert_host,
                 use_sni,
+                grpc,
                 // NOTE: Update when we support client certs in static backends
                 client_cert: None,
             })

lib/src/error.rs

@@ -138,6 +138,9 @@ pub enum Error {
 
     #[error("invalid client certificate")]
     InvalidClientCert(#[from] crate::config::ClientCertError),
+
+    #[error("Invalid response to ALPN request; wanted '{0}', got '{1}'")]
+    InvalidAlpnRepsonse(&'static str, String),
 }
 
 impl Error {
@@ -202,7 +205,8 @@ impl Error {
             | Error::ObjectStoreKeyValidationError(_)
             | Error::UnfinishedStreamingBody
             | Error::SharedMemory
-            | Error::ToStr(_) => FastlyStatus::Error,
+            | Error::ToStr(_)
+            | Error::InvalidAlpnRepsonse(_, _) => FastlyStatus::Error,
         }
     }
 
@@ -393,6 +397,9 @@ pub enum BackendConfigError {
     #[error("'use_sni' field was not a boolean")]
     InvalidUseSniEntry,
 
+    #[error("'grpc' field was not a boolean")]
+    InvalidGrpcEntry,
+
     #[error("invalid url: {0}")]
     InvalidUrl(#[from] http::uri::InvalidUri),
 

lib/src/upstream.rs

@@ -128,6 +128,9 @@ impl hyper::service::Service<Uri> for BackendConnector {
                     config.partial_config.with_no_client_auth()
                 };
                 config.enable_sni = backend.use_sni;
+                if backend.grpc {
+                    config.alpn_protocols = vec![b"h2".to_vec()];
+                }
                 let connector = TlsConnector::from(Arc::new(config));
 
                 let cert_host = backend
@@ -138,6 +141,29 @@ impl hyper::service::Service<Uri> for BackendConnector {
                 let dnsname = ServerName::try_from(cert_host).map_err(Box::new)?;
 
                 let tls = connector.connect(dnsname, tcp).await.map_err(Box::new)?;
+
+                if backend.grpc {
+                    let (_, tls_state) = tls.get_ref();
+
+                    match tls_state.alpn_protocol() {
+                        None => {
+                            tracing::warn!(
+                                "Unexpected; request h2 for grpc, but got nothing back from ALPN"
+                            );
+                        }
+
+                        Some(b"h2") => {}
+
+                        Some(other_value) => {
+                            return Err(Error::InvalidAlpnRepsonse(
+                                "h2",
+                                String::from_utf8_lossy(other_value).to_string(),
+                            )
+                            .into())
+                        }
+                    }
+                }
+
                 Ok(Connection::Https(Box::new(tls)))
             } else {
                 Ok(Connection::Http(tcp))
@@ -237,9 +263,11 @@ pub fn send_request(
     req.headers_mut().insert(hyper::header::HOST, host);
     *req.uri_mut() = uri;
 
+    let h2only = backend.grpc;
     async move {
         let basic_response = Client::builder()
             .set_host(false)
+            .http2_only(h2only)
             .build(connector)
             .request(req)
             .await

lib/src/wiggle_abi/req_impl.rs

@@ -375,6 +375,8 @@ impl FastlyHttpReq for Session {
             None
         };
 
+        let grpc = backend_info_mask.contains(BackendConfigOptions::GRPC);
+
         let new_backend = Backend {
             uri: Uri::builder()
                 .scheme(scheme)
@@ -384,6 +386,7 @@ impl FastlyHttpReq for Session {
             override_host,
             cert_host,
             use_sni,
+            grpc,
             client_cert,
         };
 

test-fixtures/src/bin/grpc.rs

@@ -0,0 +1,27 @@
+use fastly::{Backend, Error, Request};
+//use fastly::experimental::GrpcBackend;
+use std::str::FromStr;
+
+/// Pass everything from the downstream request through to the backend, then pass everything back
+/// from the upstream request to the downstream response.
+fn main() -> Result<(), Error> {
+    let client_req = Request::from_client();
+    let Some(port_str) = client_req.get_header_str("Port") else {
+            panic!("Couldn't find out what port to use!");
+    };
+    let port = u16::from_str(port_str).unwrap();
+
+    let backend = Backend::builder("grpc-backend", format!("localhost:{}", port))
+//        .for_grpc(true)
+        .finish()
+        .expect("can build backend");
+
+    Request::get("http://localhost/")
+        .with_header("header", "is-a-thing")
+        .with_body("hello")
+        .send(backend)
+        .unwrap()
+        .send_to_client();
+
+    Ok(())
+}