Merge pull request #1356 from murgatroid99/grpc-js_proxy_support_take_2

grpc-js: Interact with proxies properly

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "0.8.0",
+  "version": "0.8.1",
   "description": "gRPC Library for Node - pure JS implementation",
   "homepage": "https://grpc.io/",
   "repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js",

packages/grpc-js/src/channel.ts

@@ -38,6 +38,7 @@ import { ServiceConfig, validateServiceConfig } from './service-config';
 import { trace, log } from './logging';
 import { SubchannelAddress } from './subchannel';
 import { MaxMessageSizeFilterFactory } from './max-message-size-filter';
+import { mapProxyName } from './http_proxy';
 
 export enum ConnectivityState {
   CONNECTING,
@@ -163,6 +164,14 @@ export class ChannelImplementation implements Channel {
         );
       }
     }
+    if (this.options['grpc.default_authority']) {
+      this.defaultAuthority = this.options['grpc.default_authority'] as string;
+    } else {
+      this.defaultAuthority = getDefaultAuthority(target);
+    }
+    const proxyMapResult = mapProxyName(target, options);
+    this.target = proxyMapResult.target;
+    this.options = Object.assign({}, this.options, proxyMapResult.extraOptions);
     /* The global boolean parameter to getSubchannelPool has the inverse meaning to what
      * the grpc.use_local_subchannel_pool channel option means. */
     this.subchannelPool = getSubchannelPool(
@@ -207,7 +216,7 @@ export class ChannelImplementation implements Channel {
       );
     }
     this.resolvingLoadBalancer = new ResolvingLoadBalancer(
-      target,
+      this.target,
       channelControlHelper,
       defaultServiceConfig
     );
@@ -217,12 +226,6 @@ export class ChannelImplementation implements Channel {
       new MaxMessageSizeFilterFactory(this.options),
       new CompressionFilterFactory(this),
     ]);
-    // TODO(murgatroid99): Add more centralized handling of channel options
-    if (this.options['grpc.default_authority']) {
-      this.defaultAuthority = this.options['grpc.default_authority'] as string;
-    } else {
-      this.defaultAuthority = getDefaultAuthority(target);
-    }
   }
 
   /**

packages/grpc-js/src/http_proxy.ts

@@ -22,7 +22,12 @@ import { parseTarget } from './resolver-dns';
 import { Socket } from 'net';
 import * as http from 'http';
 import * as logging from './logging';
-import { SubchannelAddress, isTcpSubchannelAddress } from './subchannel';
+import {
+  SubchannelAddress,
+  isTcpSubchannelAddress,
+  subchannelAddressToString,
+} from './subchannel';
+import { ChannelOptions } from './channel-options';
 
 const TRACER_NAME = 'proxy';
 
@@ -89,8 +94,6 @@ function getProxyInfo(): ProxyInfo {
   return result;
 }
 
-const PROXY_INFO = getProxyInfo();
-
 function getNoProxyHostList(): string[] {
   /* Prefer using 'no_grpc_proxy'. Fallback on 'no_proxy' if it is not set. */
   let noProxyStr: string | undefined = process.env.no_grpc_proxy;
@@ -107,81 +110,109 @@ function getNoProxyHostList(): string[] {
   }
 }
 
-const NO_PROXY_HOSTS = getNoProxyHostList();
+export interface ProxyMapResult {
+  target: string;
+  extraOptions: ChannelOptions;
+}
 
-export function shouldUseProxy(target: string): boolean {
-  if (!PROXY_INFO.address) {
-    return false;
+export function mapProxyName(
+  target: string,
+  options: ChannelOptions
+): ProxyMapResult {
+  const noProxyResult: ProxyMapResult = {
+    target: target,
+    extraOptions: {},
+  };
+  const proxyInfo = getProxyInfo();
+  if (!proxyInfo.address) {
+    return noProxyResult;
   }
-  let serverHost: string;
   const parsedTarget = parseTarget(target);
-  if (parsedTarget) {
-    serverHost = parsedTarget.host;
-  } else {
-    return false;
+  if (!parsedTarget) {
+    return noProxyResult;
   }
-  for (const host of NO_PROXY_HOSTS) {
+  const serverHost = parsedTarget.host;
+  for (const host of getNoProxyHostList()) {
     if (host === serverHost) {
       trace('Not using proxy for target in no_proxy list: ' + target);
-      return false;
+      return noProxyResult;
     }
   }
-  return true;
+  const extraOptions: ChannelOptions = {
+    'grpc.http_connect_target': target,
+  };
+  if (proxyInfo.creds) {
+    extraOptions['grpc.http_connect_creds'] = proxyInfo.creds;
+  }
+  return {
+    target: `dns:${proxyInfo.address}`,
+    extraOptions: extraOptions,
+  };
+}
+
+export interface ProxyConnectionResult {
+  socket?: Socket;
+  realTarget?: string;
 }
 
 export function getProxiedConnection(
-  target: string,
-  subchannelAddress: SubchannelAddress
-): Promise<Socket> {
-  if (
-    !(
-      PROXY_INFO.address &&
-      shouldUseProxy(target) &&
-      isTcpSubchannelAddress(subchannelAddress)
-    )
-  ) {
-    return Promise.reject<Socket>();
+  address: SubchannelAddress,
+  channelOptions: ChannelOptions
+): Promise<ProxyConnectionResult> {
+  if (!('grpc.http_connect_target' in channelOptions)) {
+    return Promise.resolve<ProxyConnectionResult>({});
   }
-  const subchannelAddressPathString = `${subchannelAddress.host}:${subchannelAddress.port}`;
-  trace(
-    'Using proxy ' +
-      PROXY_INFO.address +
-      ' to connect to ' +
-      target +
-      ' at ' +
-      subchannelAddress
-  );
+  const realTarget = channelOptions['grpc.http_connect_target'] as string;
+  const parsedTarget = parseTarget(realTarget)!;
   const options: http.RequestOptions = {
     method: 'CONNECT',
-    host: PROXY_INFO.address,
-    path: subchannelAddressPathString,
   };
-  if (PROXY_INFO.creds) {
+  // Connect to the subchannel address as a proxy
+  if (isTcpSubchannelAddress(address)) {
+    options.host = address.host;
+    options.port = address.port;
+  } else {
+    options.socketPath = address.path;
+  }
+  if (parsedTarget.port === undefined) {
+    options.path = parsedTarget.host;
+  } else {
+    options.path = `${parsedTarget.host}:${parsedTarget.port}`;
+  }
+  if ('grpc.http_connect_creds' in channelOptions) {
     options.headers = {
       'Proxy-Authorization':
-        'Basic ' + Buffer.from(PROXY_INFO.creds).toString('base64'),
+        'Basic ' +
+        Buffer.from(
+          channelOptions['grpc.http_connect_creds'] as string
+        ).toString('base64'),
     };
   }
-  return new Promise<Socket>((resolve, reject) => {
+  const proxyAddressString = subchannelAddressToString(address);
+  trace('Using proxy ' + proxyAddressString + ' to connect to ' + options.path);
+  return new Promise<ProxyConnectionResult>((resolve, reject) => {
     const request = http.request(options);
     request.once('connect', (res, socket, head) => {
       request.removeAllListeners();
       socket.removeAllListeners();
       if (res.statusCode === 200) {
         trace(
           'Successfully connected to ' +
-            subchannelAddress +
+            options.path +
             ' through proxy ' +
-            PROXY_INFO.address
+            proxyAddressString
         );
-        resolve(socket);
+        resolve({
+          socket,
+          realTarget,
+        });
       } else {
         log(
           LogVerbosity.ERROR,
           'Failed to connect to ' +
-            subchannelAddress +
+            options.path +
             ' through proxy ' +
-            PROXY_INFO.address +
+            proxyAddressString +
             ' with status ' +
             res.statusCode
         );
@@ -193,11 +224,12 @@ export function getProxiedConnection(
       log(
         LogVerbosity.ERROR,
         'Failed to connect to proxy ' +
-          PROXY_INFO.address +
+          proxyAddressString +
           ' with error ' +
           err.message
       );
       reject();
     });
+    request.end();
   });
 }

packages/grpc-js/src/resolver-dns.ts

@@ -171,7 +171,15 @@ class DnsResolver implements Resolver {
       });
       return;
     }
-    if (this.dnsHostname !== null) {
+    if (this.dnsHostname === null) {
+      setImmediate(() => {
+        this.listener.onError({
+          code: Status.UNAVAILABLE,
+          details: `Failed to parse DNS address ${this.target}`,
+          metadata: new Metadata(),
+        });
+      });
+    } else {
       /* We clear out latestLookupResult here to ensure that it contains the
        * latest result since the last time we started resolving. That way, the
        * TXT resolution handler can use it, but only if it finishes second. We

packages/grpc-js/src/subchannel.ts

@@ -26,7 +26,7 @@ import { BackoffTimeout, BackoffOptions } from './backoff-timeout';
 import { getDefaultAuthority } from './resolver';
 import * as logging from './logging';
 import { LogVerbosity } from './constants';
-import { shouldUseProxy, getProxiedConnection } from './http_proxy';
+import { getProxiedConnection, ProxyConnectionResult } from './http_proxy';
 import * as net from 'net';
 
 const clientVersion = require('../../package.json').version;
@@ -278,7 +278,7 @@ export class Subchannel {
     clearTimeout(this.keepaliveTimeoutId);
   }
 
-  private createSession(socket?: net.Socket) {
+  private createSession(proxyConnectionResult: ProxyConnectionResult) {
     let connectionOptions: http2.SecureClientSessionOptions =
       this.credentials._getConnectionOptions() || {};
     let addressScheme = 'http://';
@@ -299,16 +299,16 @@ export class Subchannel {
         };
         connectionOptions.servername = sslTargetNameOverride;
       }
-      if (socket) {
-        connectionOptions.socket = socket;
+      if (proxyConnectionResult.socket) {
+        connectionOptions.socket = proxyConnectionResult.socket;
       }
     } else {
       /* In all but the most recent versions of Node, http2.connect does not use
        * the options when establishing plaintext connections, so we need to
        * establish that connection explicitly. */
       connectionOptions.createConnection = (authority, option) => {
-        if (socket) {
-          return socket;
+        if (proxyConnectionResult.socket) {
+          return proxyConnectionResult.socket;
         } else {
           /* net.NetConnectOpts is declared in a way that is more restrictive
            * than what net.connect will actually accept, so we use the type
@@ -339,7 +339,10 @@ export class Subchannel {
      * determines whether the connection will be established over TLS or not.
      */
     const session = http2.connect(
-      addressScheme + getDefaultAuthority(this.channelTarget),
+      addressScheme +
+        getDefaultAuthority(
+          proxyConnectionResult.realTarget ?? this.channelTarget
+        ),
       connectionOptions
     );
     this.session = session;
@@ -409,21 +412,17 @@ export class Subchannel {
   }
 
   private startConnectingInternal() {
-    if (shouldUseProxy(this.channelTarget)) {
-      getProxiedConnection(this.channelTarget, this.subchannelAddress).then(
-        (socket) => {
-          this.createSession(socket);
-        },
-        (reason) => {
-          this.transitionToState(
-            [ConnectivityState.CONNECTING],
-            ConnectivityState.TRANSIENT_FAILURE
-          );
-        }
-      );
-    } else {
-      this.createSession();
-    }
+    getProxiedConnection(this.subchannelAddress, this.options).then(
+      (result) => {
+        this.createSession(result);
+      },
+      (reason) => {
+        this.transitionToState(
+          [ConnectivityState.CONNECTING],
+          ConnectivityState.TRANSIENT_FAILURE
+        );
+      }
+    );
   }
 
   /**