netty: implement UdsNameResolver and UdsNettyChannelProvider (#9113)

* netty: implement UdsNameResolver and UdsNettyChannelProvider
When the scheme is "unix:" we get the UdsNettyChannelProvider to
create a NettyChannelBuilder with DomainSocketAddress type and
other related params needed for UDS sockets

api/src/main/java/io/grpc/ManagedChannelRegistry.java

@@ -145,6 +145,11 @@ static List<Class<?>> getHardCodedClasses() {
     } catch (ClassNotFoundException e) {
       logger.log(Level.FINE, "Unable to find NettyChannelProvider", e);
     }
+    try {
+      list.add(Class.forName("io.grpc.netty.UdsNettyChannelProvider"));
+    } catch (ClassNotFoundException e) {
+      logger.log(Level.FINE, "Unable to find UdsNettyChannelProvider", e);
+    }
     return Collections.unmodifiableList(list);
   }
 

build.gradle

@@ -176,6 +176,8 @@ subprojects {
 
             netty: "io.netty:netty-codec-http2:[${nettyVersion}]",
             netty_epoll: "io.netty:netty-transport-native-epoll:${nettyVersion}:linux-x86_64",
+            netty_epoll_common: "io.netty:netty-transport-native-epoll:${nettyVersion}",
+            netty_unix_common: "io.netty:netty-transport-native-unix-common:${nettyVersion}",
             netty_epoll_arm64: "io.netty:netty-transport-native-epoll:${nettyVersion}:linux-aarch_64",
             netty_proxy_handler: "io.netty:netty-handler-proxy:${nettyVersion}",
 

netty/BUILD.bazel

@@ -20,6 +20,7 @@ java_library(
         "@io_netty_netty_codec_http2//jar",
         "@io_netty_netty_codec_socks//jar",
         "@io_netty_netty_common//jar",
+        "@io_netty_netty_transport_native_unix_common//jar",
         "@io_netty_netty_handler//jar",
         "@io_netty_netty_handler_proxy//jar",
         "@io_netty_netty_resolver//jar",

netty/build.gradle

@@ -21,13 +21,15 @@ dependencies {
     implementation libraries.netty_proxy_handler,
             libraries.guava,
             libraries.errorprone,
-            libraries.perfmark
+            libraries.perfmark,
+            libraries.netty_unix_common
 
     // Tests depend on base class defined by core module.
     testImplementation project(':grpc-core').sourceSets.test.output,
             project(':grpc-api').sourceSets.test.output,
             project(':grpc-testing'),
-            project(':grpc-testing-proto')
+            project(':grpc-testing-proto'),
+            libraries.netty_epoll_common
     testRuntimeOnly libraries.netty_tcnative,
             libraries.conscrypt,
             libraries.netty_epoll

netty/src/main/java/io/grpc/netty/UdsNameResolver.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2022 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.netty;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.base.Preconditions;
+import io.grpc.EquivalentAddressGroup;
+import io.grpc.NameResolver;
+import io.netty.channel.unix.DomainSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+final class UdsNameResolver extends NameResolver {
+  private NameResolver.Listener2 listener;
+  private final String authority;
+
+  UdsNameResolver(String authority, String targetPath) {
+    checkArgument(authority == null, "non-null authority not supported");
+    this.authority = targetPath;
+  }
+
+  @Override
+  public String getServiceAuthority() {
+    return this.authority;
+  }
+
+  @Override
+  public void start(Listener2 listener) {
+    Preconditions.checkState(this.listener == null, "already started");
+    this.listener = checkNotNull(listener, "listener");
+    resolve();
+  }
+
+  @Override
+  public void refresh() {
+    resolve();
+  }
+
+  private void resolve() {
+    ResolutionResult.Builder resolutionResultBuilder = ResolutionResult.newBuilder();
+    List<EquivalentAddressGroup> servers = new ArrayList<>(1);
+    servers.add(new EquivalentAddressGroup(new DomainSocketAddress(authority)));
+    resolutionResultBuilder.setAddresses(Collections.unmodifiableList(servers));
+    listener.onResult(resolutionResultBuilder.build());
+  }
+
+  @Override
+  public void shutdown() {}
+}

netty/src/main/java/io/grpc/netty/UdsNameResolverProvider.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2022 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.netty;
+
+import com.google.common.base.Preconditions;
+import io.grpc.Internal;
+import io.grpc.NameResolver;
+import io.grpc.NameResolverProvider;
+import io.netty.channel.unix.DomainSocketAddress;
+import java.net.SocketAddress;
+import java.net.URI;
+import java.util.Collection;
+import java.util.Collections;
+
+@Internal
+public final class UdsNameResolverProvider extends NameResolverProvider {
+
+  private static final String SCHEME = "unix";
+
+  @Override
+  public UdsNameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
+    if (SCHEME.equals(targetUri.getScheme())) {
+      return new UdsNameResolver(targetUri.getAuthority(), getTargetPathFromUri(targetUri));
+    } else {
+      return null;
+    }
+  }
+
+  static String getTargetPathFromUri(URI targetUri) {
+    Preconditions.checkArgument(SCHEME.equals(targetUri.getScheme()), "scheme must be " + SCHEME);
+    String targetPath = targetUri.getPath();
+    if (targetPath == null) {
+      targetPath = Preconditions.checkNotNull(targetUri.getSchemeSpecificPart(), "targetPath");
+    }
+    return targetPath;
+  }
+
+  @Override
+  public String getDefaultScheme() {
+    return SCHEME;
+  }
+
+  @Override
+  protected boolean isAvailable() {
+    return true;
+  }
+
+  @Override
+  protected int priority() {
+    return 3;
+  }
+
+  @Override
+  protected Collection<Class<? extends SocketAddress>> getProducedSocketAddressTypes() {
+    return Collections.singleton(DomainSocketAddress.class);
+  }
+}

netty/src/main/java/io/grpc/netty/UdsNettyChannelProvider.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2015 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.netty;
+
+import io.grpc.CallCredentials;
+import io.grpc.ChannelCredentials;
+import io.grpc.InsecureChannelCredentials;
+import io.grpc.Internal;
+import io.grpc.ManagedChannelProvider;
+import io.grpc.internal.SharedResourcePool;
+import io.netty.channel.unix.DomainSocketAddress;
+import java.net.SocketAddress;
+import java.net.URI;
+import java.util.Collection;
+import java.util.Collections;
+
+/** Provider for {@link NettyChannelBuilder} instances for UDS channels. */
+@Internal
+public final class UdsNettyChannelProvider extends ManagedChannelProvider {
+
+  @Override
+  public boolean isAvailable() {
+    return (Utils.EPOLL_DOMAIN_CLIENT_CHANNEL_TYPE != null);
+  }
+
+  @Override
+  public int priority() {
+    return 3;
+  }
+
+  @Override
+  public NettyChannelBuilder builderForAddress(String name, int port) {
+    throw new UnsupportedOperationException("host:port not supported");
+  }
+
+  @Override
+  public NettyChannelBuilder builderForTarget(String target) {
+    ChannelCredentials creds = InsecureChannelCredentials.create();
+    ProtocolNegotiators.FromChannelCredentialsResult result = ProtocolNegotiators.from(creds);
+    if (result.error != null) {
+      throw new RuntimeException(result.error);
+    }
+    return getNettyChannelBuilder(target, creds, null, result.negotiator);
+  }
+
+  @Override
+  public NewChannelBuilderResult newChannelBuilder(String target, ChannelCredentials creds) {
+    ProtocolNegotiators.FromChannelCredentialsResult result = ProtocolNegotiators.from(creds);
+    if (result.error != null) {
+      return NewChannelBuilderResult.error(result.error);
+    }
+    return NewChannelBuilderResult.channelBuilder(
+        getNettyChannelBuilder(target, creds, result.callCredentials, result.negotiator));
+  }
+
+  private static NettyChannelBuilder getNettyChannelBuilder(
+      String target,
+      ChannelCredentials creds,
+      CallCredentials callCredentials,
+      ProtocolNegotiator.ClientFactory negotiator) {
+    if (Utils.EPOLL_DOMAIN_CLIENT_CHANNEL_TYPE == null) {
+      throw new IllegalStateException("Epoll is not available");
+    }
+    String targetPath = UdsNameResolverProvider.getTargetPathFromUri(URI.create(target));
+    NettyChannelBuilder builder =
+        new NettyChannelBuilder(
+            new DomainSocketAddress(targetPath), creds, callCredentials, negotiator);
+    builder =
+        builder
+            .eventLoopGroupPool(
+                SharedResourcePool.forResource(Utils.DEFAULT_WORKER_EVENT_LOOP_GROUP))
+            .channelType(Utils.EPOLL_DOMAIN_CLIENT_CHANNEL_TYPE);
+    return builder;
+  }
+
+  @Override
+  protected Collection<Class<? extends SocketAddress>> getSupportedSocketAddressTypes() {
+    return Collections.singleton(DomainSocketAddress.class);
+  }
+}

netty/src/main/java/io/grpc/netty/Utils.java

@@ -89,6 +89,7 @@ class Utils {
       = new DefaultEventLoopGroupResource(1, "grpc-nio-boss-ELG", EventLoopGroupType.NIO);
   public static final Resource<EventLoopGroup> NIO_WORKER_EVENT_LOOP_GROUP
       = new DefaultEventLoopGroupResource(0, "grpc-nio-worker-ELG", EventLoopGroupType.NIO);
+
   public static final Resource<EventLoopGroup> DEFAULT_BOSS_EVENT_LOOP_GROUP;
   public static final Resource<EventLoopGroup> DEFAULT_WORKER_EVENT_LOOP_GROUP;
 
@@ -104,6 +105,7 @@ private static final class ByteBufAllocatorPreferHeapHolder {
 
   public static final ChannelFactory<? extends ServerChannel> DEFAULT_SERVER_CHANNEL_FACTORY;
   public static final Class<? extends Channel> DEFAULT_CLIENT_CHANNEL_TYPE;
+  public static final Class<? extends Channel> EPOLL_DOMAIN_CLIENT_CHANNEL_TYPE;
 
   @Nullable
   private static final Constructor<? extends EventLoopGroup> EPOLL_EVENT_LOOP_GROUP_CONSTRUCTOR;
@@ -112,6 +114,7 @@ private static final class ByteBufAllocatorPreferHeapHolder {
     // Decide default channel types and EventLoopGroup based on Epoll availability
     if (isEpollAvailable()) {
       DEFAULT_CLIENT_CHANNEL_TYPE = epollChannelType();
+      EPOLL_DOMAIN_CLIENT_CHANNEL_TYPE = epollDomainSocketChannelType();
       DEFAULT_SERVER_CHANNEL_FACTORY = new ReflectiveChannelFactory<>(epollServerChannelType());
       EPOLL_EVENT_LOOP_GROUP_CONSTRUCTOR = epollEventLoopGroupConstructor();
       DEFAULT_BOSS_EVENT_LOOP_GROUP
@@ -122,6 +125,7 @@ private static final class ByteBufAllocatorPreferHeapHolder {
       logger.log(Level.FINE, "Epoll is not available, using Nio.", getEpollUnavailabilityCause());
       DEFAULT_SERVER_CHANNEL_FACTORY = nioServerChannelFactory();
       DEFAULT_CLIENT_CHANNEL_TYPE = NioSocketChannel.class;
+      EPOLL_DOMAIN_CLIENT_CHANNEL_TYPE = null;
       DEFAULT_BOSS_EVENT_LOOP_GROUP = NIO_BOSS_EVENT_LOOP_GROUP;
       DEFAULT_WORKER_EVENT_LOOP_GROUP = NIO_WORKER_EVENT_LOOP_GROUP;
       EPOLL_EVENT_LOOP_GROUP_CONSTRUCTOR = null;
@@ -326,6 +330,17 @@ private static Class<? extends Channel> epollChannelType() {
     }
   }
 
+  // Must call when epoll is available
+  private static Class<? extends Channel> epollDomainSocketChannelType() {
+    try {
+      Class<? extends Channel> channelType = Class
+          .forName("io.netty.channel.epoll.EpollDomainSocketChannel").asSubclass(Channel.class);
+      return channelType;
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException("Cannot load EpollDomainSocketChannel", e);
+    }
+  }
+
   // Must call when epoll is available
   private static Constructor<? extends EventLoopGroup> epollEventLoopGroupConstructor() {
     try {

netty/src/main/resources/META-INF/services/io.grpc.ManagedChannelProvider

@@ -1 +1,2 @@
 io.grpc.netty.NettyChannelProvider
+io.grpc.netty.UdsNettyChannelProvider

netty/src/main/resources/META-INF/services/io.grpc.NameResolverProvider

@@ -0,0 +1 @@
+io.grpc.netty.UdsNameResolverProvider