diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala index a48340712..9e76240c9 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala @@ -126,14 +126,15 @@ final class ClusterBootstrap(implicit system: ExtendedActorSystem) extends Exten * * We give the required selfContactPoint some time to be set asynchronously, or else log an error. */ - @InternalApi private[bootstrap] def ensureSelfContactPoint(): Unit = system.scheduler.scheduleOnce(10.seconds) { - if (!selfContactPoint.isCompleted) { - _selfContactPointUri.failure(new TimeoutException("Awaiting Bootstrap.selfContactPoint timed out.")) - log.error( - "'Bootstrap.selfContactPoint' was NOT set, but is required for the bootstrap to work " + - "if binding bootstrap routes manually and not via pekko-management.") + @InternalApi private[bootstrap] def ensureSelfContactPoint(): Unit = + system.scheduler.scheduleOnce(ClusterBootstrap.SelfContactPointTimeout) { + if (!selfContactPoint.isCompleted) { + _selfContactPointUri.failure(new TimeoutException("Awaiting Bootstrap.selfContactPoint timed out.")) + log.error( + "'Bootstrap.selfContactPoint' was NOT set, but is required for the bootstrap to work " + + "if binding bootstrap routes manually and not via pekko-management.") + } } - } /** * INTERNAL API @@ -153,6 +154,15 @@ final class ClusterBootstrap(implicit system: ExtendedActorSystem) extends Exten object ClusterBootstrap extends ExtensionId[ClusterBootstrap] with ExtensionIdProvider { + /** + * INTERNAL API + * + * How long `selfContactPoint` is given to be set before the promise behind it is failed. Anything + * waiting on that promise has to outlast this, so it is shared rather than restated - see + * `SelfAwareJoinDecider.selfContactPointTimeout`. + */ + @InternalApi private[bootstrap] val SelfContactPointTimeout: FiniteDuration = 10.seconds + override def lookup: ClusterBootstrap.type = ClusterBootstrap override def get(system: ActorSystem): ClusterBootstrap = super.get(system) diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala index 4ae8d4dea..7d79ef969 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala @@ -13,11 +13,15 @@ package org.apache.pekko.management.cluster.bootstrap +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.{ AtomicBoolean, AtomicReference } + import org.apache.pekko import pekko.actor.ActorSystem import pekko.annotation.InternalApi import pekko.discovery.ServiceDiscovery.ResolvedTarget import pekko.event.{ LogSource, Logging } +import pekko.http.scaladsl.model.Uri import scala.concurrent.Await import scala.concurrent.duration._ @@ -42,18 +46,59 @@ import scala.concurrent.duration._ protected def contactPointString(contactPoint: ResolvedTarget): String = s"${contactPoint.host}:${contactPoint.port.getOrElse("0")}" + /** + * How long to block waiting for the self contact point to be set. `ClusterBootstrap` fails the + * promise itself after `ClusterBootstrap.SelfContactPointTimeout`, so this only has to outlast + * that; it is derived from it rather than restated so that the two cannot drift apart. Overridable + * for tests, which cannot afford to wait this long. + */ + protected def selfContactPointTimeout: FiniteDuration = ClusterBootstrap.SelfContactPointTimeout * 3 + + // `decide` is called from the BootstrapCoordinator actor and guarded by its `decisionInProgress` + // flag, so in-tree callers are already serialised. `JoinDecider` is a user-pluggable interface + // though, and a custom one is free to resolve the contact point inside the Future it returns, so + // these stay atomic rather than plain vars. + // + // `cachedSelfContactPoint` holds the resolved value, `null` until there is one. Failures are not + // cached, so a contact point set later is still picked up. + private val cachedSelfContactPoint = new AtomicReference[(String, Int)]() + // Set by whichever caller takes the one permitted blocking wait; everyone else fails fast. + private val hasBlockedForSelfContactPoint = new AtomicBoolean(false) + /** * The value `ClusterBootstrap(system).selfContactPoints` is set prior * to HTTP binding, during [[pekko.management.scaladsl.PekkoManagement.start()]], hence we * accept blocking on this initialization. If no value is received, the future will fail with * a `TimeoutException` and ClusterBootstrap will log an explanatory error to the user. + * + * A resolved value is cached. A failure is not, so that a contact point set later is still picked + * up, but the blocking wait is only ever paid once: once `ClusterBootstrap.start()` has run, the + * promise is always completed within `ClusterBootstrap.SelfContactPointTimeout`, so reaching + * the timeout at all means it never ran and no amount of further waiting will help. */ - private[bootstrap] def selfContactPoint: (String, Int) = - Await.result( - ClusterBootstrap(system).selfContactPoint - .map(uri => (uri.authority.host.toString, uri.authority.port))(system.dispatcher), - Duration.Inf // the future has a timeout - ) + private[bootstrap] def selfContactPoint: (String, Int) = { + val cached = cachedSelfContactPoint.get() + if (cached ne null) cached + else { + val pending = ClusterBootstrap(system).selfContactPoint + val uri = pending.value match { + case Some(completed) => completed.get + case None if hasBlockedForSelfContactPoint.compareAndSet(false, true) => + Await.result(pending, selfContactPointTimeout) + case None => + throw new TimeoutException( + "'Bootstrap.selfContactPoint' is still not set after waiting " + + s"[$selfContactPointTimeout] for it once. It is required for the bootstrap to work " + + "if binding bootstrap routes manually and not via pekko-management.") + } + val resolved = toContactPoint(uri) + cachedSelfContactPoint.compareAndSet(null, resolved) + resolved + } + } + + private def toContactPoint(uri: Uri): (String, Int) = + (uri.authority.host.toString, uri.authority.port) /** * Determines whether it has the need and ability to join self and create a new cluster. @@ -66,7 +111,7 @@ import scala.concurrent.duration._ log.warning( BootstrapLogMarker.inProgress(info.contactPoints.map(contactPointString), info.allSeedNodes), "Self contact point [{}] not found in targets {}", - contactPointString(selfContactPoint), + contactPointString(self), info.contactPoints.mkString(", ")) } false diff --git a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala index dba790253..8cece9d68 100644 --- a/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala +++ b/management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala @@ -16,7 +16,7 @@ package org.apache.pekko.management.cluster.bootstrap.internal import java.time.LocalDateTime import java.util.concurrent.ThreadLocalRandom import scala.collection.immutable -import scala.concurrent.ExecutionContext +import scala.concurrent.{ ExecutionContext, Future } import org.apache.pekko import pekko.actor.Actor import pekko.actor.ActorRef @@ -34,6 +34,7 @@ import pekko.pattern.pipe import scala.concurrent.duration._ import scala.util.Try +import scala.util.control.NonFatal import pekko.event.Logging import pekko.management.cluster.bootstrap.{ BootstrapLogMarker, @@ -392,8 +393,16 @@ private[pekko] class BootstrapCoordinator( decisionInProgress = true - joinDecider - .decide(info) + // `decide` may throw synchronously before it ever builds a Future (it resolves the self + // contact point eagerly), and `recover` would not see that. Letting it escape would leave + // `decisionInProgress` stuck true and stop this coordinator deciding ever again. + val decision = + try joinDecider.decide(info) + catch { + case NonFatal(e) => Future.failed(e) + } + + decision .recover { case e => log.error(e, "Join decision failed: {}", e) diff --git a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala new file mode 100644 index 000000000..f14ada4c1 --- /dev/null +++ b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.pekko.management.cluster.bootstrap + +import java.util.concurrent.TimeoutException + +import com.typesafe.config.ConfigFactory +import org.apache.pekko +import pekko.actor.ActorSystem +import pekko.event.NoLogging +import pekko.testkit.TestKit +import org.scalatest.BeforeAndAfterAll +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import scala.concurrent.duration._ + +object SelfContactPointResolutionSpec { + + /** Waits a fraction of the real timeout, so that the blocking path can be exercised in a test. */ + class ShortTimeoutJoinDecider(system: ActorSystem, settings: ClusterBootstrapSettings) + extends LowestAddressJoinDecider(system, settings) { + override protected def selfContactPointTimeout: FiniteDuration = 500.millis + } + + /** Reads back the production timeout, which is otherwise protected. */ + class TimeoutReadingJoinDecider(system: ActorSystem, settings: ClusterBootstrapSettings) + extends LowestAddressJoinDecider(system, settings) { + def configuredTimeout: FiniteDuration = selfContactPointTimeout + } + +} + +class SelfContactPointResolutionSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll { + + import SelfContactPointResolutionSpec._ + + private val config = ConfigFactory.parseString(""" + pekko { + loglevel = INFO + remote.artery.canonical.port = 0 + management.http { + hostname = "10.0.0.2" + base-path = "test" + } + } + """).withFallback(ConfigFactory.load()) + + // a system per test, because setSelfContactPoint completes a promise that cannot be reset + private var systems: List[ActorSystem] = Nil + + private def newDecider(name: String): (ActorSystem, ShortTimeoutJoinDecider) = { + val system = ActorSystem(name, config) + systems = system :: systems + val settings = ClusterBootstrapSettings(system.settings.config, NoLogging) + (system, new ShortTimeoutJoinDecider(system, settings)) + } + + private def elapsed(body: => Unit): FiniteDuration = { + val started = System.nanoTime() + body + (System.nanoTime() - started).nanos + } + + "SelfAwareJoinDecider.selfContactPoint" should { + + "resolve the contact point once it has been set" in { + val (system, decider) = newDecider("self-contact-point-resolves") + ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test") + + decider.selfContactPoint should ===(("10.0.0.2", 8558)) + } + + "keep returning the resolved contact point on later calls" in { + val (system, decider) = newDecider("self-contact-point-caches") + ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test") + + val first = decider.selfContactPoint + val second = decider.selfContactPoint + + first should ===(("10.0.0.2", 8558)) + second should ===(first) + } + + "time out instead of blocking forever when the contact point is never set" in { + val (_, decider) = newDecider("self-contact-point-times-out") + + a[TimeoutException] should be thrownBy decider.selfContactPoint + } + + "block for the timeout only once, then fail fast" in { + val (_, decider) = newDecider("self-contact-point-fails-fast") + + val firstCall = elapsed(a[TimeoutException] should be thrownBy decider.selfContactPoint) + val secondCall = elapsed(a[TimeoutException] should be thrownBy decider.selfContactPoint) + + firstCall should be >= 500.millis + // without this the bootstrap coordinator would park a dispatcher thread for the full + // timeout on every probe, for as long as the contact point stays unset + secondCall should be < 250.millis + } + + "still pick up a contact point that is set after a timeout" in { + val (system, decider) = newDecider("self-contact-point-set-late") + + a[TimeoutException] should be thrownBy decider.selfContactPoint + + ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test") + + decider.selfContactPoint should ===(("10.0.0.2", 8558)) + } + + } + + "ClusterBootstrap.SelfContactPointTimeout" should { + + "be outlasted by the decider's own wait, so that the promise fails first" in { + val (system, _) = newDecider("self-contact-point-timeout-ordering") + val settings = ClusterBootstrapSettings(system.settings.config, NoLogging) + val decider = new TimeoutReadingJoinDecider(system, settings) + + decider.configuredTimeout should be > ClusterBootstrap.SelfContactPointTimeout + } + + } + + override def afterAll(): Unit = + systems.foreach(TestKit.shutdownActorSystem(_, 5.seconds)) + +} diff --git a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala index 9995694cb..7591ed363 100644 --- a/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala +++ b/management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala @@ -13,7 +13,8 @@ package org.apache.pekko.management.cluster.bootstrap.internal -import java.util.concurrent.atomic.AtomicReference +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.{ AtomicInteger, AtomicReference } import org.apache.pekko import pekko.actor.{ ActorRef, ActorSystem, Props } @@ -21,7 +22,12 @@ import pekko.discovery.ServiceDiscovery.{ Resolved, ResolvedTarget } import pekko.discovery.{ Lookup, MockDiscovery } import pekko.http.scaladsl.model.Uri import pekko.management.cluster.bootstrap.internal.BootstrapCoordinator.Protocol.InitiateBootstrapping -import pekko.management.cluster.bootstrap.{ ClusterBootstrapSettings, LowestAddressJoinDecider } +import pekko.management.cluster.bootstrap.{ + ClusterBootstrapSettings, + JoinDecision, + LowestAddressJoinDecider, + SeedNodesInformation +} import com.typesafe.config.ConfigFactory import org.scalatest.BeforeAndAfterAll import org.scalatest.concurrent.Eventually @@ -177,6 +183,48 @@ class BootstrapCoordinatorSpec extends AnyWordSpec with Matchers with BeforeAndA } } + "The bootstrap coordinator, when a JoinDecider throws synchronously" should { + + // SelfAwareJoinDecider.selfContactPoint resolves eagerly and can throw before `decide` ever + // builds a Future, so the failure never reaches `recover`. If that escapes, `decisionInProgress` + // is left true and the coordinator stops deciding for the rest of its life. + "keep making decisions instead of wedging" in { + val decideServiceName = "bootstrap-coordinator-throwing-decider" + val decideSettings = ClusterBootstrapSettings( + ConfigFactory.parseString(s""" + |pekko.management.cluster.bootstrap { + | contact-point-discovery.service-name = $decideServiceName + | contact-point-discovery.required-contact-point-nr = 1 + | contact-point-discovery.contact-with-all-contact-points = false + |} + """.stripMargin).withFallback(system.settings.config), + system.log) + + MockDiscovery.set( + Lookup(decideServiceName, portName = None, protocol = Some("tcp")), + () => Future.successful(Resolved(decideServiceName, List(ResolvedTarget("host1", Some(7626), None))))) + + val decideCalls = new AtomicInteger(0) + val throwingDecider = new LowestAddressJoinDecider(system, decideSettings) { + override def decide(info: SeedNodesInformation): Future[JoinDecision] = { + decideCalls.incrementAndGet() + throw new TimeoutException("'Bootstrap.selfContactPoint' was never set") + } + } + + val coordinator = system.actorOf(Props(new BootstrapCoordinator(discovery, throwingDecider, decideSettings) { + override def ensureProbing(selfContactPointScheme: String, contactPoint: ResolvedTarget): Option[ActorRef] = + None + })) + coordinator ! InitiateBootstrapping(selfUri) + + // more than one call proves decisionInProgress was reset after the throw + eventually { + decideCalls.get should be >= 2 + } + } + } + override def afterAll(): Unit = { Await.result(system.terminate(), 10.seconds) super.afterAll()