feat: Periodic client discovery refresh #1152

project/Dependencies.scala

@@ -17,7 +17,7 @@ object Dependencies {
     // We don't force Akka updates because downstream projects can upgrade
     // themselves. For more information see
     // https://doc.akka.io//docs/akka/current/project/downstream-upgrade-strategy.html
-    val akka = "2.9.0"
+    val akka = "2.9.1-M1+11-809c4920-SNAPSHOT"
     val akkaBinary = "2.9"
     val akkaHttp = "10.6.0"
     val akkaHttpBinary = "10.6"

runtime/src/main/resources/reference.conf

@@ -17,6 +17,14 @@ akka.grpc.client."*" {
 
     # timeout for service discovery resolving
     resolve-timeout = 1s
+
+    # Set this to a duration to trigger periodic refresh of the resolved endpoints, evicting cached entries
+    # if the discovery mechanism supports that. Expected use is for client side load-balancing, to detect new services
+    # to load balance across. The default value "off" disables periodic refresh and instead only does refresh when
+    # the client implementation decides to.
+    #
+    # Currently only supported by the Netty client backend.
+    refresh-interval = off
   }
 
   # port to use if service-discovery-mechism is static or service discovery does not return a port

runtime/src/main/scala/akka/grpc/GrpcClientSettings.scala

@@ -15,10 +15,12 @@ import com.typesafe.config.{ Config, ConfigValueFactory }
 import io.grpc.CallCredentials
 import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslProvider
-import javax.net.ssl.{ SSLContext, TrustManager }
 
+import java.util.Optional
+import javax.net.ssl.{ SSLContext, TrustManager }
 import scala.collection.immutable
 import scala.concurrent.duration.{ Duration, _ }
+import scala.jdk.OptionConverters.RichOptional
 
 object GrpcClientSettings {
 
@@ -148,7 +150,9 @@ object GrpcClientSettings {
       getOptionalString(clientConfiguration, "user-agent"),
       clientConfiguration.getBoolean("use-tls"),
       getOptionalString(clientConfiguration, "load-balancing-policy"),
-      clientConfiguration.getString("backend"))
+      clientConfiguration.getString("backend"),
+      identity,
+      getOptionalDuration(clientConfiguration, "discovery.refresh-interval"))
 
   private def getOptionalString(config: Config, path: String): Option[String] =
     config.getString(path) match {
@@ -162,6 +166,12 @@ object GrpcClientSettings {
       case other => Some(other)
     }
 
+  private def getOptionalDuration(config: Config, path: String): Option[FiniteDuration] =
+    config.getString(path) match {
+      case "off" => None
+      case _     => Some(config.getDuration(path).asScala)
+    }
+
   private def getPotentiallyInfiniteDuration(underlying: Config, path: String): Duration =
     Helpers.toRootLowerCase(underlying.getString(path)) match {
       case "infinite" => Duration.Inf
@@ -196,7 +206,8 @@ final class GrpcClientSettings private (
     val useTls: Boolean,
     val loadBalancingPolicy: Option[String],
     val backend: String,
-    val channelBuilderOverrides: NettyChannelBuilder => NettyChannelBuilder = identity) {
+    val channelBuilderOverrides: NettyChannelBuilder => NettyChannelBuilder = identity,
+    val discoveryRefreshInterval: Option[FiniteDuration]) {
   require(
     sslContext.isEmpty || trustManager.isEmpty,
     "Configuring the sslContext or the trustManager is mutually exclusive")
@@ -284,6 +295,28 @@ final class GrpcClientSettings private (
   def withBackend(value: String): GrpcClientSettings =
     copy(backend = value)
 
+  /**
+   * Scala API: Set this to a duration to trigger periodic refresh of the resolved endpoints, evicting cached entries
+   * if the discovery mechanism supports that. The default value `None` disables periodic refresh and instead
+   * only does refresh when the client implementation decides to.
+   *
+   * Currently only supported by the Netty client backend.
+   */
+  @ApiMayChange
+  def withDiscoveryRefreshInterval(refreshInterval: Option[FiniteDuration]): GrpcClientSettings =
+    copy(discoveryRefreshInterval = refreshInterval)
+
+  /**
+   * Java API: Set this to a duration to trigger periodic refresh of the resolved endpoints, evicting cached entries
+   * if the discovery mechanism supports that. The default value `None` disables periodic refresh and instead
+   * only does refresh when the client implementation decides to.
+   *
+   * Currently only supported by the Netty client backend.
+   */
+  @ApiMayChange
+  def withDiscoveryRefreshInterval(refreshInterval: Optional[java.time.Duration]): GrpcClientSettings =
+    copy(discoveryRefreshInterval = refreshInterval.map(_.asScala).toScala)
+
   private def copy(
       serviceName: String = serviceName,
       servicePortName: Option[String] = servicePortName,
@@ -301,8 +334,8 @@ final class GrpcClientSettings private (
       connectionAttempts: Option[Int] = connectionAttempts,
       loadBalancingPolicy: Option[String] = loadBalancingPolicy,
       backend: String = backend,
-      channelBuilderOverrides: NettyChannelBuilder => NettyChannelBuilder = channelBuilderOverrides)
-      : GrpcClientSettings =
+      channelBuilderOverrides: NettyChannelBuilder => NettyChannelBuilder = channelBuilderOverrides,
+      discoveryRefreshInterval: Option[FiniteDuration] = discoveryRefreshInterval): GrpcClientSettings =
     new GrpcClientSettings(
       callCredentials = callCredentials,
       serviceDiscovery = serviceDiscovery,
@@ -321,5 +354,6 @@ final class GrpcClientSettings private (
       connectionAttempts = connectionAttempts,
       loadBalancingPolicy = loadBalancingPolicy,
       backend = backend,
-      channelBuilderOverrides = channelBuilderOverrides)
+      channelBuilderOverrides = channelBuilderOverrides,
+      discoveryRefreshInterval = discoveryRefreshInterval)
 }

runtime/src/main/scala/akka/grpc/internal/AkkaDiscoveryNameResolver.scala

@@ -4,55 +4,101 @@
 
 package akka.grpc.internal
 
-import java.net.{ InetAddress, InetSocketAddress, UnknownHostException }
+import akka.actor.ActorSystem
+import akka.actor.Cancellable
+import akka.annotation.InternalApi
 
+import java.net.{ InetAddress, InetSocketAddress, UnknownHostException }
 import akka.discovery.ServiceDiscovery.ResolvedTarget
 import akka.discovery.{ Lookup, ServiceDiscovery }
+import akka.event.Logging
 import akka.grpc.GrpcClientSettings
 import io.grpc.{ Attributes, EquivalentAddressGroup, NameResolver, Status }
 import io.grpc.NameResolver.Listener
 
+import java.util.concurrent.atomic.AtomicReference
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{ ExecutionContext, Promise }
 import scala.util.{ Failure, Success }
 
-class AkkaDiscoveryNameResolver(
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] final class AkkaDiscoveryNameResolver(
     discovery: ServiceDiscovery,
     defaultPort: Int,
     serviceName: String,
     portName: Option[String],
     protocol: Option[String],
-    resolveTimeout: FiniteDuration)(implicit val ec: ExecutionContext)
+    resolveTimeout: FiniteDuration,
+    refreshInterval: Option[FiniteDuration])(implicit val ec: ExecutionContext, system: ActorSystem)
     extends NameResolver {
+
+  private final val log = Logging(system, getClass)
+
   override def getServiceAuthority: String = serviceName
 
-  val listener: Promise[Listener] = Promise()
+  private val listener: Promise[Listener] = Promise()
+
+  // initialized after first resolve if needed
+  private val refreshTask = new AtomicReference[Cancellable]
 
   override def start(l: Listener): Unit = {
+    log.debug("Name resolver for {} started", serviceName)
     listener.trySuccess(l)
-    lookup(l)
+    lookup(l, evict = false)
   }
 
-  override def refresh(): Unit =
+  override def refresh(): Unit = refresh(false)
+
+  private def refresh(evict: Boolean): Unit =
     listener.future.onComplete {
-      case Success(l) => lookup(l)
+      case Success(l) =>
+        log.debug("Name resolver for {} refreshing", serviceName)
+        lookup(l, evict)
       case Failure(_) => // We never fail this promise
     }
 
-  def lookup(listener: Listener): Unit = {
-    discovery.lookup(Lookup(serviceName, portName, protocol), resolveTimeout).onComplete {
+  def lookup(listener: Listener, evict: Boolean): Unit = {
+    val request = {
+      val l = Lookup(serviceName, portName, protocol)
+      if (evict) l.withDiscardCache
+      else l
+    }
+    val result = discovery.lookup(request, resolveTimeout)
+
+    result.onComplete {
       case Success(result) =>
         try {
+          if (log.isDebugEnabled)
+            log.debug(
+              "Successful service discovery for service {}, found addresses: {}",
+              serviceName,
+              result.addresses.mkString(", "))
           listener.onAddresses(addresses(result.addresses), Attributes.EMPTY)
         } catch {
           case e: UnknownHostException =>
-            // TODO at least log
+            log.warning(e, s"Unknown host for service $serviceName")
             listener.onError(Status.UNKNOWN.withDescription(e.getMessage))
         }
       case Failure(e) =>
-        // TODO at least log
+        log.warning(e, s"Service discovery failed for service $serviceName")
         listener.onError(Status.UNKNOWN.withDescription(e.getMessage))
     }
+
+    // initialize refresh timer after first lookup, if configured
+    if (refreshInterval.isDefined && refreshTask.get() == null) {
+      result.onComplete { _ =>
+        refreshInterval.foreach { interval =>
+          val cancellable = system.scheduler.scheduleWithFixedDelay(interval, interval)(() => refresh(evict = true))
+          if (!refreshTask.compareAndSet(null, cancellable)) {
+            // concurrent update beat us to it, there already is a scheduled task
+            cancellable.cancel()
+          }
+        }
+      }
+    }
   }
 
   @throws[UnknownHostException]
@@ -67,16 +113,25 @@ class AkkaDiscoveryNameResolver(
       .asJava
   }
 
-  override def shutdown(): Unit = ()
+  override def shutdown(): Unit = {
+    val refreshCancellable = refreshTask.get()
+    if (refreshCancellable ne null) refreshCancellable.cancel()
+  }
 }
 
-object AkkaDiscoveryNameResolver {
-  def apply(settings: GrpcClientSettings)(implicit ec: ExecutionContext): AkkaDiscoveryNameResolver =
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] object AkkaDiscoveryNameResolver {
+  def apply(
+      settings: GrpcClientSettings)(implicit ec: ExecutionContext, system: ActorSystem): AkkaDiscoveryNameResolver =
     new AkkaDiscoveryNameResolver(
       settings.serviceDiscovery,
       settings.defaultPort,
       settings.serviceName,
       settings.servicePortName,
       settings.serviceProtocol,
-      settings.resolveTimeout)
+      settings.resolveTimeout,
+      settings.discoveryRefreshInterval)
 }

runtime/src/main/scala/akka/grpc/internal/AkkaDiscoveryNameResolverProvider.scala

@@ -4,20 +4,27 @@
 
 package akka.grpc.internal
 
-import java.net.URI
+import akka.actor.ActorSystem
+import akka.annotation.InternalApi
 
+import java.net.URI
 import akka.discovery.ServiceDiscovery
 import io.grpc.{ NameResolver, NameResolverProvider }
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration.FiniteDuration
 
-class AkkaDiscoveryNameResolverProvider(
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] final class AkkaDiscoveryNameResolverProvider(
     discovery: ServiceDiscovery,
     defaultPort: Int,
     portName: Option[String],
     protocol: Option[String],
-    resolveTimeout: FiniteDuration)(implicit ec: ExecutionContext)
+    resolveTimeout: FiniteDuration,
+    refreshInterval: Option[FiniteDuration])(implicit ec: ExecutionContext, system: ActorSystem)
     extends NameResolverProvider {
   override def isAvailable: Boolean = true
 
@@ -27,6 +34,13 @@ class AkkaDiscoveryNameResolverProvider(
 
   override def newNameResolver(targetUri: URI, args: NameResolver.Args): AkkaDiscoveryNameResolver = {
     require(targetUri.getAuthority != null, s"target uri should not have null authority, got [$targetUri]")
-    new AkkaDiscoveryNameResolver(discovery, defaultPort, targetUri.getAuthority, portName, protocol, resolveTimeout)
+    new AkkaDiscoveryNameResolver(
+      discovery,
+      defaultPort,
+      targetUri.getAuthority,
+      portName,
+      protocol,
+      resolveTimeout,
+      refreshInterval)
   }
 }

runtime/src/main/scala/akka/grpc/internal/ChannelUtils.scala

@@ -38,7 +38,7 @@ object ChannelUtils {
       implicit sys: ClassicActorSystemProvider): InternalChannel = {
     settings.backend match {
       case "netty" =>
-        NettyClientUtils.createChannel(settings, log)(sys.classicSystem.dispatcher)
+        NettyClientUtils.createChannel(settings, log)(sys.classicSystem.dispatcher, sys.classicSystem)
       case "akka-http" =>
         AkkaHttpClientUtils.createChannel(settings, log)
       case _ => throw new IllegalArgumentException(s"Unexpected backend [${settings.backend}]")

runtime/src/main/scala/akka/grpc/internal/NettyClientUtils.scala

@@ -4,6 +4,8 @@
 
 package akka.grpc.internal
 
+import akka.actor.ActorSystem
+
 import java.util.concurrent.TimeUnit
 import javax.net.ssl.SSLContext
 import akka.{ Done, NotUsed }
@@ -43,7 +45,8 @@ object NettyClientUtils {
    */
   @InternalApi
   def createChannel(settings: GrpcClientSettings, log: LoggingAdapter)(
-      implicit ec: ExecutionContext): InternalChannel = {
+      implicit ec: ExecutionContext,
+      system: ActorSystem): InternalChannel = {
 
     @nowarn("msg=deprecated")
     var builder =
@@ -59,7 +62,8 @@ object NettyClientUtils {
             settings.defaultPort,
             settings.servicePortName,
             settings.serviceProtocol,
-            settings.resolveTimeout))
+            settings.resolveTimeout,
+            settings.discoveryRefreshInterval))
 
     if (!settings.useTls)
       builder = builder.usePlaintext()