Support address rewriting proxy mapper

[ejona86]

gRFC A1 describes HTTP CONNECT support. grpc-java has supported "Use case 1" to one level or another since v1.0.3. There's been some ways to do "use case 2" (and 3), but these needed custom name resolvers and Netty ProtocolNegotiators for other reasons, and they could do what they needed without any direct support in grpc-java.

We're seeing some new cases for "use case 2." The important part of that case is the client is aware of backends so can load balance between them. Name resolution is generally not DNS, and may actually use a "use case 1" approach to get addresses.

We need to add an API, at least to Netty, to choose to use a proxy on a per-address basis. We can potentially use the ProxyDetector API for this, but we'd call it with resolved SocketAddresses instead of the normal unresolved addresses. I don't know if that's a good idea or not, also because ProxyDetector can block and we probably can't block in the usage here for Netty.

The implementation for this can be a Netty protocol negotiator that wraps the real negotiator if the builder was configured with one of these per-address mappers. It probably can reuse large portions of ProtocolNegotiators.httpProxy().

[ejona86]

Looking at this more, I don't think we need a new protocol negotiatior. This can be done cross-transport as ClientTransportOptions already encodes whether a proxy should be used. Worst-case, the proxy mapping could be run in InternalSubchannel, but we could instead put the logic in ManagedChannelImpl or ManagedChannelImplBuilder by using a wrapping ClientTransportFactory.

[ejona86]

With the current API, you can hack something like the proxy mapper with:

LoadBalancerRegistry registry = LoadBalancerRegistry.getDefaultRegistry();
// Some LB policy you are using
registry.register(new ProxyingLoadBalancerProvider(registry.getProvider("round_robin")));

// rewriteAddress() is the main interesting method below
final class ProxyingLoadBalancerProvider extends LoadBalancerProvider {
  private final LoadBalancerProvider delegate;

  public ProxyingLoadBalancerProvider(LoadBalancerProvider delegate) {
    if (delegate == null) {
      throw new NullPointerException("delegate");
    }
    this.delegate = delegate;
  }

  @Override public boolean isAvailable() { return delegate.isAvailable(); }
  @Override public int getPriority() { return delegate.getPriority() + 1; }
  @Override public String getPolicyName() { return delegate.getPolicyName(); }
  @Override public NameResolver.ConfigOrError parseLoadBalancingPolicyConfig(Map<String, ?> raw) {
    return delegate.parseLoadBalancingPolicyConfig(raw);
  }

  @Override
  public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
    return delegate.newLoadBalancer(new ProxyingHelper(helper));
  }

  private List<EquivalentAddressGroup> rewriteAddresses(List<EquivalentAddressGroup> eags) {
    List<EquivalentAddressGroup> newEags = new ArrayList<>(eags.size());
    for (EquivalentAddressGroup eag : eags) {
      List<SocketAddress> addrs = new ArrayList<>(eag.getAddresses().size());
      for (SocketAddress addr : eag.getAddresses()) {
        addrs.add(rewriteAddress(addr));
      }
      newEags.add(new EquivalentAddressGroup(addrs, eag.getAttributes()));
    }
    return newEags;
  }

  /** THIS IS ESSENTIALLY THE PROXY MAPPER */
  private SocketAddress rewriteAddress(SocketAddress addr) {
    if (!(addr instanceof InetSocketAddress)) {
      return addr;
    }
    InetSocketAddress inet = (InetSocketAddress) addr;
    InetAddress newAddr;
    try {
      // Strip any hostname
      newAddr = InetAddress.getByAddress(inet.getAddress().getAddress());
    } catch (UnknownHostException ex) {
      throw new AssertionError(ex); // Not possible
    }
    addr = HttpConnectProxiedSocketAddress.newBuilder()
        .setTargetAddress(new InetSocketAddress(newAddr, inet.getPort()))
        .setProxyAddress(new InetSocketAddress("127.0.0.1", 1234)) // CHANGE ME
        .build();
    return addr;
  }

  private final class ProxyingHelper extends io.grpc.util.ForwardingLoadBalancerHelper {
    private final LoadBalancer.Helper delegate;

    public ProxyingHelper(LoadBalancer.Helper delegate) {
      this.delegate = delegate;
    }

    @Override protected LoadBalancer.Helper delegate() { return delegate; }

    @Override
    public LoadBalancer.Subchannel createSubchannel(LoadBalancer.CreateSubchannelArgs args) {
      return new ProxyingSubchannel(
          delegate.createSubchannel(
              args.toBuilder().setAddresses(rewriteAddresses(args.getAddresses())).build()),
          args.getAddresses());
    }
  }

  private final class ProxyingSubchannel extends io.grpc.util.ForwardingSubchannel {
    private final LoadBalancer.Subchannel delegate;
    private List<EquivalentAddressGroup> addresses;

    public ProxyingSubchannel(
        LoadBalancer.Subchannel delegate, List<EquivalentAddressGroup> addresses) {
      this.delegate = delegate;
      this.addresses = addresses;
    }

    @Override protected LoadBalancer.Subchannel delegate() { return delegate; }
    @Override public List<EquivalentAddressGroup> getAllAddresses() { return addresses; }

    @Override
    public void updateAddresses(List<EquivalentAddressGroup> addrs) {
      super.updateAddresses(rewriteAddresses(addrs));
      addresses = new ArrayList<>(addrs);
    }
  }
}