Merge pull request #9383 from markdroth/http_connect_channel_arg

Trigger HTTP CONNECT handshaker via channel args.

7 files changed, 199 insertions, 88 deletions

diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c
index 2f25fef9a7..865e91a2b4 100644
--- a/src/core/ext/client_channel/client_channel.c
+++ b/src/core/ext/client_channel/client_channel.c
@@ -43,6 +43,8 @@
 #include <grpc/support/sync.h>
 #include <grpc/support/useful.h>
 
+#include "src/core/ext/client_channel/http_connect_handshaker.h"
+#include "src/core/ext/client_channel/http_proxy.h"
 #include "src/core/ext/client_channel/lb_policy_registry.h"
 #include "src/core/ext/client_channel/resolver_registry.h"
 #include "src/core/ext/client_channel/subchannel.h"
@@ -150,6 +152,10 @@ static void *method_parameters_create_from_json(const grpc_json *json) {
  */
 
 typedef struct client_channel_channel_data {
+  /** server name */
+  char *server_name;
+  /** HTTP CONNECT proxy to use, if any */
+  char *proxy_name;
   /** resolver for this channel */
   grpc_resolver *resolver;
   /** have we started resolving this channel */
@@ -310,6 +316,17 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
     // Use pick_first if nothing was specified and we didn't select grpclb
     // above.
     if (lb_policy_name == NULL) lb_policy_name = "pick_first";
+    // If using a proxy, add channel arg for server in HTTP CONNECT request.
+    if (chand->proxy_name != NULL) {
+      grpc_arg new_arg;
+      new_arg.key = GRPC_ARG_HTTP_CONNECT_SERVER;
+      new_arg.type = GRPC_ARG_STRING;
+      new_arg.value.string = chand->server_name;
+      grpc_channel_args *tmp_args = chand->resolver_result;
+      chand->resolver_result =
+          grpc_channel_args_copy_and_add(chand->resolver_result, &new_arg, 1);
+      grpc_channel_args_destroy(exec_ctx, tmp_args);
+    }
     // Instantiate LB policy.
     grpc_lb_policy_args lb_policy_args;
     lb_policy_args.args = chand->resolver_result;
@@ -528,9 +545,12 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
   arg = grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVER_URI);
   GPR_ASSERT(arg != NULL);
   GPR_ASSERT(arg->type == GRPC_ARG_STRING);
-  chand->resolver =
-      grpc_resolver_create(exec_ctx, arg->value.string, args->channel_args,
-                           chand->interested_parties);
+  chand->server_name = gpr_strdup(arg->value.string);
+  chand->proxy_name = grpc_get_http_proxy_server();
+  char *name_to_resolve =
+      chand->proxy_name == NULL ? chand->server_name : chand->proxy_name;
+  chand->resolver = grpc_resolver_create(
+      exec_ctx, name_to_resolve, args->channel_args, chand->interested_parties);
   if (chand->resolver == NULL) {
     return GRPC_ERROR_CREATE("resolver creation failed");
   }
@@ -541,7 +561,8 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
 static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
                                     grpc_channel_element *elem) {
   channel_data *chand = elem->channel_data;
-
+  gpr_free(chand->server_name);
+  gpr_free(chand->proxy_name);
   if (chand->resolver != NULL) {
     grpc_resolver_shutdown(exec_ctx, chand->resolver);
     GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel");
diff --git a/src/core/ext/client_channel/http_connect_handshaker.c b/src/core/ext/client_channel/http_connect_handshaker.c
index fba32561ac..622d236320 100644
--- a/src/core/ext/client_channel/http_connect_handshaker.c
+++ b/src/core/ext/client_channel/http_connect_handshaker.c
@@ -49,15 +49,12 @@
 #include "src/core/lib/http/parser.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/support/env.h"
+#include "src/core/lib/support/string.h"
 
 typedef struct http_connect_handshaker {
   // Base class.  Must be first.
   grpc_handshaker base;
 
-  char* proxy_server;
-  grpc_http_header* headers;
-  size_t num_headers;
-
   gpr_refcount refcount;
   gpr_mu mu;
 
@@ -91,12 +88,6 @@ static void http_connect_handshaker_unref(grpc_exec_ctx* exec_ctx,
                                          handshaker->read_buffer_to_destroy);
       gpr_free(handshaker->read_buffer_to_destroy);
     }
-    gpr_free(handshaker->proxy_server);
-    for (size_t i = 0; i < handshaker->num_headers; ++i) {
-      gpr_free(handshaker->headers[i].key);
-      gpr_free(handshaker->headers[i].value);
-    }
-    gpr_free(handshaker->headers);
     grpc_slice_buffer_destroy_internal(exec_ctx, &handshaker->write_buffer);
     grpc_http_parser_destroy(&handshaker->http_parser);
     grpc_http_response_destroy(&handshaker->http_response);
@@ -276,64 +267,88 @@ static void http_connect_handshaker_do_handshake(
     grpc_tcp_server_acceptor* acceptor, grpc_closure* on_handshake_done,
     grpc_handshaker_args* args) {
   http_connect_handshaker* handshaker = (http_connect_handshaker*)handshaker_in;
-  // Get server name from channel args.
-  const grpc_arg* arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI);
-  GPR_ASSERT(arg != NULL);
+  // Check for HTTP CONNECT channel arg.
+  // If not found, invoke on_handshake_done without doing anything.
+  const grpc_arg* arg =
+      grpc_channel_args_find(args->args, GRPC_ARG_HTTP_CONNECT_SERVER);
+  if (arg == NULL) {
+    // Set shutdown to true so that subsequent calls to
+    // http_connect_handshaker_shutdown() do nothing.
+    gpr_mu_lock(&handshaker->mu);
+    handshaker->shutdown = true;
+    gpr_mu_unlock(&handshaker->mu);
+    grpc_closure_sched(exec_ctx, on_handshake_done, GRPC_ERROR_NONE);
+    return;
+  }
   GPR_ASSERT(arg->type == GRPC_ARG_STRING);
-  char* canonical_uri =
-      grpc_resolver_factory_add_default_prefix_if_needed(arg->value.string);
-  grpc_uri* uri = grpc_uri_parse(canonical_uri, 1);
-  char* server_name = uri->path;
-  if (server_name[0] == '/') ++server_name;
+  char* server_name = arg->value.string;
+  // Get headers from channel args.
+  arg = grpc_channel_args_find(args->args, GRPC_ARG_HTTP_CONNECT_HEADERS);
+  grpc_http_header* headers = NULL;
+  size_t num_headers = 0;
+  char** header_strings = NULL;
+  size_t num_header_strings = 0;
+  if (arg != NULL) {
+    GPR_ASSERT(arg->type == GRPC_ARG_STRING);
+    gpr_string_split(arg->value.string, "\n", &header_strings,
+                     &num_header_strings);
+    headers = gpr_malloc(sizeof(grpc_http_header) * num_header_strings);
+    for (size_t i = 0; i < num_header_strings; ++i) {
+      char* sep = strchr(header_strings[i], ':');
+      if (sep == NULL) {
+        gpr_log(GPR_ERROR, "skipping unparseable HTTP CONNECT header: %s",
+                header_strings[i]);
+        continue;
+      }
+      *sep = '\0';
+      headers[num_headers].key = header_strings[i];
+      headers[num_headers].value = sep + 1;
+      ++num_headers;
+    }
+  }
   // Save state in the handshaker object.
   gpr_mu_lock(&handshaker->mu);
   handshaker->args = args;
   handshaker->on_handshake_done = on_handshake_done;
-  // Send HTTP CONNECT request.
+  // Log connection via proxy.
+  char* proxy_name = grpc_endpoint_get_peer(args->endpoint);
   gpr_log(GPR_INFO, "Connecting to server %s via HTTP proxy %s", server_name,
-          handshaker->proxy_server);
+          proxy_name);
+  gpr_free(proxy_name);
+  // Construct HTTP CONNECT request.
   grpc_httpcli_request request;
   memset(&request, 0, sizeof(request));
   request.host = server_name;
   request.http.method = "CONNECT";
   request.http.path = server_name;
-  request.http.hdrs = handshaker->headers;
-  request.http.hdr_count = handshaker->num_headers;
+  request.http.hdrs = headers;
+  request.http.hdr_count = num_headers;
   request.handshaker = &grpc_httpcli_plaintext;
   grpc_slice request_slice = grpc_httpcli_format_connect_request(&request);
   grpc_slice_buffer_add(&handshaker->write_buffer, request_slice);
+  // Clean up.
+  gpr_free(headers);
+  for (size_t i = 0; i < num_header_strings; ++i) {
+    gpr_free(header_strings[i]);
+  }
+  gpr_free(header_strings);
   // Take a new ref to be held by the write callback.
   gpr_ref(&handshaker->refcount);
   grpc_endpoint_write(exec_ctx, args->endpoint, &handshaker->write_buffer,
                       &handshaker->request_done_closure);
   gpr_mu_unlock(&handshaker->mu);
-  // Clean up.
-  gpr_free(canonical_uri);
-  grpc_uri_destroy(uri);
 }
 
 static const grpc_handshaker_vtable http_connect_handshaker_vtable = {
     http_connect_handshaker_destroy, http_connect_handshaker_shutdown,
     http_connect_handshaker_do_handshake};
 
-grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
-                                                     grpc_http_header* headers,
-                                                     size_t num_headers) {
-  GPR_ASSERT(proxy_server != NULL);
+static grpc_handshaker* grpc_http_connect_handshaker_create() {
   http_connect_handshaker* handshaker = gpr_malloc(sizeof(*handshaker));
   memset(handshaker, 0, sizeof(*handshaker));
   grpc_handshaker_init(&http_connect_handshaker_vtable, &handshaker->base);
   gpr_mu_init(&handshaker->mu);
   gpr_ref_init(&handshaker->refcount, 1);
-  handshaker->proxy_server = gpr_strdup(proxy_server);
-  if (num_headers > 0) {
-    handshaker->headers = gpr_malloc(sizeof(grpc_http_header) * num_headers);
-    for (size_t i = 0; i < num_headers; ++i) {
-      handshaker->headers[i].key = gpr_strdup(headers[i].key);
-      handshaker->headers[i].value = gpr_strdup(headers[i].value);
-    }
-    handshaker->num_headers = num_headers;
-  }
   grpc_slice_buffer_init(&handshaker->write_buffer);
   grpc_closure_init(&handshaker->request_done_closure, on_write_done,
                     handshaker, grpc_schedule_on_exec_ctx);
@@ -344,30 +359,6 @@ grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
   return &handshaker->base;
 }
 
-char* grpc_get_http_proxy_server() {
-  char* uri_str = gpr_getenv("http_proxy");
-  if (uri_str == NULL) return NULL;
-  grpc_uri* uri = grpc_uri_parse(uri_str, false /* suppress_errors */);
-  char* proxy_name = NULL;
-  if (uri == NULL || uri->authority == NULL) {
-    gpr_log(GPR_ERROR, "cannot parse value of 'http_proxy' env var");
-    goto done;
-  }
-  if (strcmp(uri->scheme, "http") != 0) {
-    gpr_log(GPR_ERROR, "'%s' scheme not supported in proxy URI", uri->scheme);
-    goto done;
-  }
-  if (strchr(uri->authority, '@') != NULL) {
-    gpr_log(GPR_ERROR, "userinfo not supported in proxy URI");
-    goto done;
-  }
-  proxy_name = gpr_strdup(uri->authority);
-done:
-  gpr_free(uri_str);
-  grpc_uri_destroy(uri);
-  return proxy_name;
-}
-
 //
 // handshaker factory
 //
@@ -375,13 +366,8 @@ done:
 static void handshaker_factory_add_handshakers(
     grpc_exec_ctx* exec_ctx, grpc_handshaker_factory* factory,
     const grpc_channel_args* args, grpc_handshake_manager* handshake_mgr) {
-  char* proxy_name = grpc_get_http_proxy_server();
-  if (proxy_name != NULL) {
-    grpc_handshake_manager_add(
-        handshake_mgr,
-        grpc_http_connect_handshaker_create(proxy_name, NULL, 0));
-    gpr_free(proxy_name);
-  }
+  grpc_handshake_manager_add(handshake_mgr,
+                             grpc_http_connect_handshaker_create());
 }
 
 static void handshaker_factory_destroy(grpc_exec_ctx* exec_ctx,
diff --git a/src/core/ext/client_channel/http_connect_handshaker.h b/src/core/ext/client_channel/http_connect_handshaker.h
index c2e68de716..3059d551e3 100644
--- a/src/core/ext/client_channel/http_connect_handshaker.h
+++ b/src/core/ext/client_channel/http_connect_handshaker.h
@@ -34,17 +34,14 @@
 #ifndef GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_CONNECT_HANDSHAKER_H
 #define GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_CONNECT_HANDSHAKER_H
 
-#include "src/core/lib/channel/handshaker.h"
-#include "src/core/lib/http/parser.h"
+/// Channel arg indicating the server in HTTP CONNECT request (string).
+/// The presence of this arg triggers the use of HTTP CONNECT.
+#define GRPC_ARG_HTTP_CONNECT_SERVER "grpc.http_connect_server"
 
-/// Creates a new HTTP CONNECT handshaker.
-grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
-                                                     grpc_http_header* headers,
-                                                     size_t num_headers);
-
-/// Returns the name of the proxy to use, or NULL if no proxy is configured.
-/// Caller takes ownership of result.
-char* grpc_get_http_proxy_server();
+/// Channel arg indicating HTTP CONNECT headers (string).
+/// Multiple headers are separated by newlines.  Key/value pairs are
+/// seperated by colons.
+#define GRPC_ARG_HTTP_CONNECT_HEADERS "grpc.http_connect_headers"
 
 /// Registers handshaker factory.
 void grpc_http_connect_register_handshaker_factory();
diff --git a/src/core/ext/client_channel/http_proxy.c b/src/core/ext/client_channel/http_proxy.c
new file mode 100644
index 0000000000..9a6c818c4e
--- /dev/null
+++ b/src/core/ext/client_channel/http_proxy.c
@@ -0,0 +1,68 @@
+/*
+ *
+ * Copyright 2016, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "src/core/ext/client_channel/http_proxy.h"
+
+#include <stdbool.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+
+#include "src/core/ext/client_channel/uri_parser.h"
+#include "src/core/lib/support/env.h"
+
+char* grpc_get_http_proxy_server() {
+  char* uri_str = gpr_getenv("http_proxy");
+  if (uri_str == NULL) return NULL;
+  grpc_uri* uri = grpc_uri_parse(uri_str, false /* suppress_errors */);
+  char* proxy_name = NULL;
+  if (uri == NULL || uri->authority == NULL) {
+    gpr_log(GPR_ERROR, "cannot parse value of 'http_proxy' env var");
+    goto done;
+  }
+  if (strcmp(uri->scheme, "http") != 0) {
+    gpr_log(GPR_ERROR, "'%s' scheme not supported in proxy URI", uri->scheme);
+    goto done;
+  }
+  if (strchr(uri->authority, '@') != NULL) {
+    gpr_log(GPR_ERROR, "userinfo not supported in proxy URI");
+    goto done;
+  }
+  proxy_name = gpr_strdup(uri->authority);
+done:
+  gpr_free(uri_str);
+  grpc_uri_destroy(uri);
+  return proxy_name;
+}
diff --git a/src/core/ext/client_channel/http_proxy.h b/src/core/ext/client_channel/http_proxy.h
new file mode 100644
index 0000000000..0d77ae253b
--- /dev/null
+++ b/src/core/ext/client_channel/http_proxy.h
@@ -0,0 +1,41 @@
+/*
+ *
+ * Copyright 2016, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_PROXY_H
+#define GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_PROXY_H
+
+/// Returns the name of the proxy to use, or NULL if no proxy is configured.
+/// Caller takes ownership of result.
+char* grpc_get_http_proxy_server();
+
+#endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_HTTP_PROXY_H */
diff --git a/src/core/ext/resolver/dns/native/dns_resolver.c b/src/core/ext/resolver/dns/native/dns_resolver.c
index 655d9dc586..bf2f4e5ee4 100644
--- a/src/core/ext/resolver/dns/native/dns_resolver.c
+++ b/src/core/ext/resolver/dns/native/dns_resolver.c
@@ -37,7 +37,6 @@
 #include <grpc/support/host_port.h>
 #include <grpc/support/string_util.h>
 
-#include "src/core/ext/client_channel/http_connect_handshaker.h"
 #include "src/core/ext/client_channel/lb_policy_registry.h"
 #include "src/core/ext/client_channel/resolver_registry.h"
 #include "src/core/lib/channel/channel_args.h"
@@ -261,16 +260,14 @@ static grpc_resolver *dns_create(grpc_exec_ctx *exec_ctx,
     return NULL;
   }
   // Get name from args.
-  const char *path = args->uri->path;
+  char *path = args->uri->path;
   if (path[0] == '/') ++path;
-  // Get proxy name, if any.
-  char *proxy_name = grpc_get_http_proxy_server();
   // Create resolver.
   dns_resolver *r = gpr_malloc(sizeof(dns_resolver));
   memset(r, 0, sizeof(*r));
   gpr_mu_init(&r->mu);
   grpc_resolver_init(&r->base, &dns_resolver_vtable);
-  r->name_to_resolve = proxy_name == NULL ? gpr_strdup(path) : proxy_name;
+  r->name_to_resolve = gpr_strdup(path);
   r->default_port = gpr_strdup(default_port);
   r->channel_args = grpc_channel_args_copy(args->args);
   r->interested_parties = grpc_pollset_set_create();
diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py
index ba64174863..ed80c5c747 100644
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@@ -248,6 +248,7 @@ CORE_SOURCE_FILES = [
   'src/core/ext/client_channel/connector.c',
   'src/core/ext/client_channel/default_initial_connect_string.c',
   'src/core/ext/client_channel/http_connect_handshaker.c',
+  'src/core/ext/client_channel/http_proxy.c',
   'src/core/ext/client_channel/initial_connect_string.c',
   'src/core/ext/client_channel/lb_policy.c',
   'src/core/ext/client_channel/lb_policy_factory.c',