This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-management.git
The following commit(s) were added to refs/heads/main by this push:
new 126bcb83 improve lease code (#911)
126bcb83 is described below
commit 126bcb839c124065156acfde86b64bfa73744acb
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Aug 26 09:47:32 2026 +0100
improve lease code (#911)
* improve lease code
* revert m8
* Update LeaseActorSpec.scala
* add heartbeatMaxRetries
* deal with FIXME deal with failure from releasing the the lock
* Create operation-in-progress.backwards.excludes
* Update operation-in-progress.backwards.excludes
* Update operation-in-progress.backwards.excludes
* Update operation-in-progress.backwards.excludes
* Bound acquire conflict retries and pace release retries off the operation
timeout
Motivation:
Two retry loops in LeaseActor could run for longer than the caller waits.
The Granting conflict retry was unbounded and had no deadline, so a lease
version that kept moving on made the actor issue update requests in a tight
loop. The caller's ask expired at lease-operation-timeout while the actor
stayed in Granting, failing every later acquire/release with InvalidRequest.
If the loop eventually won, the lease was granted after the caller had
already
been told it was not. The retry also stored the original version rather than
the one just attempted, which defeated the require that guards against a
conflict response repeating a version already tried.
The Releasing retry was spaced at heartbeat-interval / (retries + 1). With
the
default 12s heartbeat interval that is 3s per retry, roughly 9s in total,
against a 5s lease-operation-timeout ask. The caller therefore saw an ask
timeout and the eventual reply went to dead letters. The retries only
appeared
to work because LeaseActorSpec configures a 25ms heartbeat interval.
Modification:
Give the Granting conflict retry a deadline: once the elapsed time since the
operation started exceeds lease-operation-timeout, reply with a
LeaseTimeoutException and return to Idle instead of retrying. Record the
version being attempted so a subsequent conflict is compared against it.
Pace release retries at lease-operation-timeout / (retries + 1) and stop
once
the remaining budget cannot fit another delay, so the caller is always told
the outcome before its ask expires.
Document in reference.conf that heartbeat-max-retries also bounds release
retries and that the two are spaced against different timeouts.
Two existing tests had the mock return the same resourceVersion for a
successful PUT, which a real API server never does; they now return a fresh
version, which the tightened conflict tracking requires.
Result:
Neither retry loop outlives the caller, and a lease can no longer be granted
after the acquire has been reported as failed.
Tests:
- sbt "lease-kubernetes/test" headerCheckAll - 108 tests pass
- sbt scalafmtCheckAll "+lease-kubernetes/mimaReportBinaryIssues" - pass on
2.13.18, 3.3.8 and 3.8.4
References:
Refs #911
* Separate release retries from heartbeat retries and queue concurrent
acquires
Motivation:
heartbeat-max-retries governed both the heartbeat retry loop and the release
retry loop, even though its name and documentation only describe the former
and the two are now spaced against different timeouts. Nothing rejected a
negative value, and the default of 3 was stated both in reference.conf and
as
a default argument on LeaseActor.props.
An Acquire arriving while an acquire was already in flight was answered with
InvalidRequest. That is better than the silent drop it replaced, but the
common case is the same owner asking for the same lease twice, and both
callers want the same outcome. The spec has carried a TODO about replying to
all such callers since the code was first written.
Modification:
Add a release-max-retries setting, defaulting to 3, and use it for the
Releasing retry loop. heartbeat-max-retries now only bounds heartbeat
retries.
Both are validated as non-negative in KubernetesSettings, and reference.conf
documents what each one spaces its retries against. Drop the duplicated
default from LeaseActor.props so reference.conf is the only source of truth.
Queue an Acquire that arrives during PendingRead or Granting instead of
rejecting it: the extra sender is recorded in the FSM data and every queued
caller gets the same LeaseAcquired, LeaseTaken or failure as the original.
The most recently supplied lease lost callback wins, matching the existing
re-acquire behaviour in Granted. An Acquire during Releasing is
contradictory
rather than duplicate, so it is still answered with InvalidRequest.
LeaseActorSpec now overrides the retry counts rather than the actor itself,
which also stops each no-retry test creating a second, orphaned LeaseActor.
Result:
The two retry loops are configured independently, and concurrent acquires of
the same lease no longer fail one of the callers.
Tests:
- sbt "lease-kubernetes/test" headerCheckAll - 118 tests pass
- sbt scalafmtCheckAll "+lease-kubernetes/mimaReportBinaryIssues" - pass on
2.13.18, 3.3.8 and 3.8.4
References:
Refs #911
* Tidy LeaseActorSpec retry tests and cover repeated acquire conflicts
Motivation:
The retry tests hardcoded the attempt count as a literal 4, which is only
correct while both retry settings default to 3, and repeated the same
"initial attempt + 3 retries" loop and comment in six places. The version
offset used to simulate a conflict was a bare 6 with no explanation. The
test
for the acquire conflict retry sat in the block for a disabled retry setting
that has no bearing on it. Nothing covered a conflict that recurs, which is
the case the version tracking in the retry exists for, or a retry that finds
the lease taken by then.
Modification:
Add failAllHeartbeatAttempts and failAllReleaseAttempts helpers to the Test
trait that derive the number of attempts from the configured retry counts,
and
use them for every such loop. heartBeatFailureNoRetry is now redundant and
is
removed, since the shared helper honours the overridden retry count.
Name the conflict version offset otherClientUpdates and say what it stands
for.
Move the conflict retry test into the acquire conflict retry block, and
assert
that nothing is reported to the caller until the retry has been answered.
Add
a test for a conflict that recurs, asserting the retry uses the version from
the most recent conflict rather than the original, and a test for a retry
that
finds another owner holding the lease.
Result:
The retry tests follow the configured retry counts instead of restating
them,
and the acquire conflict retry is covered beyond the single conflict case.
Tests:
- sbt "lease-kubernetes/test" headerCheckAll - 120 tests pass
- sbt scalafmtCheckAll "+lease-kubernetes/mimaReportBinaryIssues" - pass on
2.13.18, 3.3.8 and 3.8.4
References:
Refs #911
---
.../operation-in-progress.backwards.excludes | 36 +++
lease-kubernetes/src/main/resources/reference.conf | 12 +
.../lease/kubernetes/AbstractKubernetesLease.scala | 3 +-
.../lease/kubernetes/KubernetesSettings.scala | 14 +-
.../coordination/lease/kubernetes/LeaseActor.scala | 215 +++++++++---
.../internal/NativeKubernetesApiImpl.scala | 5 +-
.../lease/kubernetes/KubernetesSettingsSpec.scala | 22 ++
.../lease/kubernetes/LeaseActorSpec.scala | 359 +++++++++++++++++++--
8 files changed, 589 insertions(+), 77 deletions(-)
diff --git
a/lease-kubernetes/src/main/mima-filters/2.0.x.backwards.excludes/operation-in-progress.backwards.excludes
b/lease-kubernetes/src/main/mima-filters/2.0.x.backwards.excludes/operation-in-progress.backwards.excludes
new file mode 100644
index 00000000..94406d10
--- /dev/null
+++
b/lease-kubernetes/src/main/mima-filters/2.0.x.backwards.excludes/operation-in-progress.backwards.excludes
@@ -0,0 +1,36 @@
+# 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.
+
+# OperationInProgress changes
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#GrantedVersion.unapply")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#GrantedVersion.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#GrantedVersion.this")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#GrantedVersion.apply")
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor$GrantedVersion$")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#OperationInProgress.unapply")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#OperationInProgress.apply")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#OperationInProgress.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#OperationInProgress.this")
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor$OperationInProgress$")
+
+# PendingReadData changes
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#PendingReadData.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#PendingReadData.this")
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor$PendingReadData$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#PendingReadData.apply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#PendingReadData.unapply")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.coordination.lease.kubernetes.LeaseActor#ReplyRequired.alsoReplyTo")
diff --git a/lease-kubernetes/src/main/resources/reference.conf
b/lease-kubernetes/src/main/resources/reference.conf
index dc88efbf..36af9267 100644
--- a/lease-kubernetes/src/main/resources/reference.conf
+++ b/lease-kubernetes/src/main/resources/reference.conf
@@ -37,6 +37,18 @@ pekko.coordination.lease.kubernetes {
# having been updated
heartbeat-timeout = 120s
+ # Number of times to retry a failed heartbeat before giving up the lease.
Retries are spaced
+ # evenly within the heartbeat-interval, so raising this shortens the gap
between attempts
+ # rather than extending how long the lease is held without a successful
heartbeat.
+ # Set to 0 to give up the lease on the first failed heartbeat.
+ heartbeat-max-retries = 3
+
+ # Number of times to retry a failed lease release before reporting the
failure to the caller.
+ # Retries are spaced evenly within the lease-operation-timeout so that the
caller is told the
+ # outcome before its own timeout expires.
+ # Set to 0 to report the failure on the first failed release.
+ release-max-retries = 3
+
# The individual timeout for each HTTP request. Defaults to 2/5 of the
lease-operation-timeout
# Can't be greater than then lease-operation-timeout
api-server-request-timeout = ""
diff --git
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/AbstractKubernetesLease.scala
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/AbstractKubernetesLease.scala
index 419aaae1..2486b5c1 100644
---
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/AbstractKubernetesLease.scala
+++
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/AbstractKubernetesLease.scala
@@ -140,7 +140,8 @@ abstract class AbstractKubernetesLease(system:
ExtendedActorSystem, leaseTaken:
k8sSettings.leaseLabelMaxLength,
k8sSettings.onTruncateAddHashLength)
private val leaseActor = system.systemActorOf(
- LeaseActor.props(k8sApi, settings, leaseName, leaseTaken),
+ LeaseActor.props(k8sApi, settings, leaseName, leaseTaken,
k8sSettings.heartbeatMaxRetries,
+ k8sSettings.releaseMaxRetries),
s"kubernetesLease${AbstractKubernetesLease.leaseCounter.incrementAndGet}")
if (leaseName != settings.leaseName) {
logger.info(
diff --git
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettings.scala
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettings.scala
index b4d23221..351bce90 100644
---
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettings.scala
+++
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettings.scala
@@ -53,6 +53,12 @@ private[pekko] object KubernetesSettings {
apiServerRequestTimeout < leaseTimeoutSettings.operationTimeout,
"'api-server-request-timeout can not be less than
'lease-operation-timeout'")
+ val heartbeatMaxRetries = config.getInt("heartbeat-max-retries")
+ require(heartbeatMaxRetries >= 0, "'heartbeat-max-retries' must not be
negative")
+
+ val releaseMaxRetries = config.getInt("release-max-retries")
+ require(releaseMaxRetries >= 0, "'release-max-retries' must not be
negative")
+
val retryConfPath = "token-rotation-retry"
val tokenRetrySettings = new TokenRetrySettings(
@@ -75,7 +81,9 @@ private[pekko] object KubernetesSettings {
bodyReadTimeout = apiServerRequestTimeout / 2,
tokenRetrySettings = tokenRetrySettings,
leaseLabelMaxLength = config.getInt("lease-name-max-length"),
- onTruncateAddHashLength = config.getInt("on-truncate-add-hash-length"))
+ onTruncateAddHashLength = config.getInt("on-truncate-add-hash-length"),
+ heartbeatMaxRetries = heartbeatMaxRetries,
+ releaseMaxRetries = releaseMaxRetries)
}
}
@@ -111,4 +119,6 @@ private[pekko] class KubernetesSettings(
0.3
),
val leaseLabelMaxLength: Int = 63,
- val onTruncateAddHashLength: Int = 8)
+ val onTruncateAddHashLength: Int = 8,
+ val heartbeatMaxRetries: Int = 3,
+ val releaseMaxRetries: Int = 3)
diff --git
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActor.scala
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActor.scala
index d50a32b7..1b60251f 100644
---
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActor.scala
+++
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActor.scala
@@ -48,19 +48,36 @@ private[pekko] object LeaseActor {
sealed trait ReplyRequired {
def replyTo: ActorRef
+
+ /**
+ * Callers that asked to acquire the same lease while this operation was
already in flight.
+ * They get the same response as [[replyTo]] once the operation completes.
+ */
+ def alsoReplyTo: Set[ActorRef]
+
+ def allReplyTo: Set[ActorRef] = alsoReplyTo + replyTo
}
// Awaiting a read to try and get the lease
- case class PendingReadData(replyTo: ActorRef, leaseLostCallback:
Option[Throwable] => Unit)
+ case class PendingReadData(
+ replyTo: ActorRef,
+ leaseLostCallback: Option[Throwable] => Unit,
+ alsoReplyTo: Set[ActorRef] = Set.empty)
extends Data
with ReplyRequired
case class OperationInProgress(
replyTo: ActorRef,
version: String,
leaseLostCallback: Option[Throwable] => Unit,
- operationStartTime: Long = System.nanoTime())
+ operationStartTime: Long = System.nanoTime(),
+ releaseRetries: Int = 0,
+ alsoReplyTo: Set[ActorRef] = Set.empty)
extends Data
with ReplyRequired
- case class GrantedVersion(version: String, leaseLostCallback:
Option[Throwable] => Unit) extends Data
+ case class GrantedVersion(
+ version: String,
+ leaseLostCallback: Option[Throwable] => Unit,
+ heartbeatFailures: Int = 0)
+ extends Data
sealed trait Command
case class Acquire(leaseLostCallback: Option[Throwable] => Unit =
ConstantFun.scalaAnyToUnit) extends Command
@@ -70,6 +87,8 @@ private[pekko] object LeaseActor {
private case class ReadResponse(response: LeaseResource) extends Command
private case class WriteResponse(response: Either[LeaseResource,
LeaseResource]) extends Command
private case object Heartbeat extends Command
+ private case object HeartbeatRetry extends Command
+ private case object ReleaseRetry extends Command
sealed trait Response
case object LeaseAcquired extends Response
@@ -77,8 +96,14 @@ private[pekko] object LeaseActor {
case object LeaseReleased extends Response with DeadLetterSuppression
case class InvalidRequest(reason: String) extends Response with
DeadLetterSuppression
- def props(k8sApi: KubernetesApi, settings: LeaseSettings, leaseName: String,
granted: AtomicBoolean): Props = {
- Props(new LeaseActor(k8sApi, settings, leaseName, granted))
+ def props(
+ k8sApi: KubernetesApi,
+ settings: LeaseSettings,
+ leaseName: String,
+ granted: AtomicBoolean,
+ heartbeatMaxRetries: Int,
+ releaseMaxRetries: Int): Props = {
+ Props(new LeaseActor(k8sApi, settings, leaseName, granted,
heartbeatMaxRetries, releaseMaxRetries))
}
}
@@ -87,8 +112,13 @@ private[pekko] object LeaseActor {
* INTERNAL API
*/
@InternalApi
-private[pekko] class LeaseActor(k8sApi: KubernetesApi, settings:
LeaseSettings, leaseName: String,
- granted: AtomicBoolean)
+private[pekko] class LeaseActor(
+ k8sApi: KubernetesApi,
+ settings: LeaseSettings,
+ leaseName: String,
+ granted: AtomicBoolean,
+ heartbeatMaxRetries: Int,
+ releaseMaxRetries: Int)
extends LoggingFSM[LeaseActor.State, LeaseActor.Data] {
import pekko.pattern.pipe
@@ -113,10 +143,11 @@ private[pekko] class LeaseActor(k8sApi: KubernetesApi,
settings: LeaseSettings,
when(PendingRead) {
// Lock not taken
- case Event(ReadResponse(LeaseResource(None, version, _)),
PendingReadData(who, leaseLost)) =>
- tryGetLease(version, who, leaseLost)
- case Event(ReadResponse(LeaseResource(Some(currentOwner), version, time)),
PendingReadData(who, leaseLost))
- if currentOwner == ownerName =>
+ case Event(ReadResponse(LeaseResource(None, version, _)), prd @
PendingReadData(who, leaseLost, _)) =>
+ tryGetLease(version, who, leaseLost, prd.alsoReplyTo)
+ case Event(
+ ReadResponse(LeaseResource(Some(currentOwner), version, time)),
+ prd @ PendingReadData(who, leaseLost, _)) if currentOwner ==
ownerName =>
// We have the lock from a different incarnation
if (hasLeaseTimedOut(time)) {
log.warning(
@@ -125,26 +156,27 @@ private[pekko] class LeaseActor(k8sApi: KubernetesApi,
settings: LeaseSettings,
leaseName,
ownerName,
time)
- tryGetLease(version, who, leaseLost)
+ tryGetLease(version, who, leaseLost, prd.alsoReplyTo)
} else {
log.warning(
"Lease {} requested by client {} is already owned by client.
Previous lease was not released due to ungraceful shutdown. " +
"Lease is still within timeout so granting immediately",
leaseName,
ownerName)
- who ! LeaseAcquired
+ replyToAll(prd, LeaseAcquired)
goto(Granted).using(GrantedVersion(version, leaseLost))
}
- case Event(ReadResponse(LeaseResource(Some(currentOwner), version, time)),
PendingReadData(who, leaseLost)) =>
+ case Event(ReadResponse(LeaseResource(Some(currentOwner), version, time)),
+ prd @ PendingReadData(who, leaseLost, _)) =>
if (hasLeaseTimedOut(time)) {
log.warning(
"Lease {} has reached TTL. Owner {} has failed to heartbeat, have
they crashed?. Allowing {} to try and take lease",
leaseName,
currentOwner,
ownerName)
- tryGetLease(version, who, leaseLost)
+ tryGetLease(version, who, leaseLost, prd.alsoReplyTo)
} else {
- who ! LeaseTaken
+ replyToAll(prd, LeaseTaken)
// Even though we have a version there is no benefit to storing it as
we can't update a lease that has a client
goto(Idle).using(ReadRequired)
}
@@ -153,58 +185,92 @@ private[pekko] class LeaseActor(k8sApi: KubernetesApi,
settings: LeaseSettings,
when(Granting) {
case Event(
WriteResponse(Right(response)),
- cc @ OperationInProgress(who, oldVersion, leaseLost,
operationStartTime)) =>
+ cc @ OperationInProgress(_, oldVersion, leaseLost,
operationStartTime, _, _)) =>
require(
oldVersion != response.version,
s"Update response from Kubernetes API should not return the same
version: Response: $response. Client: $cc")
val operationDuration = System.nanoTime() - operationStartTime
if (operationDuration >
(settings.timeoutSettings.heartbeatTimeout.toNanos / 2)) {
log.warning("API server took too long to respond to update: {}. ",
operationDuration.nanos.pretty)
- who ! Failure(
- new LeaseTimeoutException(s"API server took too long to respond:
${operationDuration.nanos.pretty}"))
+ replyToAll(cc,
+ Failure(new LeaseTimeoutException(s"API server took too long to
respond: ${operationDuration.nanos.pretty}")))
goto(Idle).using(ReadRequired)
} else {
granted.set(true)
- who ! LeaseAcquired
+ replyToAll(cc, LeaseAcquired)
goto(Granted).using(GrantedVersion(response.version, leaseLost))
}
- case Event(WriteResponse(Left(LeaseResource(None, version, _))),
OperationInProgress(who, oldVersion, _, _)) =>
+ case Event(WriteResponse(Left(LeaseResource(None, version, _))),
+ op @ OperationInProgress(_, oldVersion, _, startTime, _, _)) =>
require(oldVersion != version)
- who ! LeaseAcquired
- // Try again as lock version has moved on but is not taken
- pipe(k8sApi.updateLeaseResource(leaseName, ownerName, version).map(r =>
WriteResponse(r))).to(self)
- stay()
- case Event(WriteResponse(Left(LeaseResource(Some(_), _, _))),
OperationInProgress(who, _, _, _)) =>
+ val operationDuration = (System.nanoTime() - startTime).nanos
+ if (operationDuration > settings.timeoutSettings.operationTimeout) {
+ // The lease version keeps moving on. Give up rather than retrying for
longer than the
+ // caller is prepared to wait, otherwise the lease could be granted
after the caller has
+ // already been told the acquire failed.
+ log.warning(
+ "Failed to acquire lease {} for owner {} after {}: lease version
kept moving on.",
+ leaseName,
+ ownerName,
+ operationDuration.pretty)
+ replyToAll(op,
+ Failure(new LeaseTimeoutException(
+ s"Timed out trying to acquire lease [$leaseName, $ownerName] after
${operationDuration.pretty}")))
+ goto(Idle).using(ReadRequired)
+ } else {
+ // Try again as lock version has moved on but is not taken.
+ // Do not reply yet — wait for the retry to succeed. Record the
version being attempted so
+ // that a subsequent conflict is compared against it rather than
against the original version.
+ pipe(k8sApi.updateLeaseResource(leaseName, ownerName, version).map(r
=> WriteResponse(r))).to(self)
+ stay().using(op.copy(version = version))
+ }
+ case Event(WriteResponse(Left(LeaseResource(Some(_), _, _))), op:
OperationInProgress) =>
// The audacity, someone else has taken the lease :(
- who ! LeaseTaken
+ replyToAll(op, LeaseTaken)
goto(Idle).using(ReadRequired) // can't use version as another owner has
the lock
}
when(Granted) {
- case Event(Heartbeat, GrantedVersion(version, _)) =>
+ case Event(Heartbeat, GrantedVersion(version, _, _)) =>
log.debug("Heartbeat: updating lease time. Version {}", version)
pipe(k8sApi.updateLeaseResource(leaseName, ownerName,
version).map(WriteResponse.apply)).to(self)
stay()
+ case Event(HeartbeatRetry, GrantedVersion(version, _, _)) =>
+ log.debug("Heartbeat retry: updating lease time. Version {}", version)
+ pipe(k8sApi.updateLeaseResource(leaseName, ownerName,
version).map(WriteResponse.apply)).to(self)
+ stay()
case Event(WriteResponse(Right(resource)), gv: GrantedVersion) =>
require(
resource.owner.contains(ownerName),
"response from API server has different owner for success: " +
resource)
log.debug("Heartbeat: lease time updated: Version {}", resource.version)
startSingleTimer("heartbeat", Heartbeat,
settings.timeoutSettings.heartbeatInterval)
- stay().using(gv.copy(version = resource.version))
- case Event(WriteResponse(Left(lr @ _)), GrantedVersion(_, leaseLost)) =>
+ stay().using(gv.copy(version = resource.version, heartbeatFailures = 0))
+ case Event(WriteResponse(Left(lr @ _)), GrantedVersion(_, leaseLost, _)) =>
log.warning("Conflict during heartbeat to lease {}. Lease assumed to be
released.", lr)
granted.set(false)
executeLeaseLockCallback(leaseLost, None)
goto(Idle).using(ReadRequired)
- case Event(Failure(t), GrantedVersion(_, leaseLost)) =>
- // FIXME, retry if timeout far enough off:
https://github.com/lightbend/akka-commercial-addons/issues/501
- log.warning("Failure during heartbeat to lease: [{}]. Lease assumed to
be released.", t.getMessage)
- granted.set(false)
- executeLeaseLockCallback(leaseLost, Some(t))
- goto(Idle).using(ReadRequired)
- case Event(Release(), GrantedVersion(version, leaseLost)) =>
+ case Event(Failure(t), gv @ GrantedVersion(_, leaseLost, failures)) =>
+ if (failures < heartbeatMaxRetries) {
+ log.warning(
+ "Failure during heartbeat to lease: [{}]. Retrying (attempt {}/{}).",
+ t.getMessage,
+ failures + 1,
+ heartbeatMaxRetries)
+ val retryDelay = settings.timeoutSettings.heartbeatInterval /
(heartbeatMaxRetries + 1)
+ startSingleTimer("heartbeat-retry", HeartbeatRetry, retryDelay)
+ stay().using(gv.copy(heartbeatFailures = failures + 1))
+ } else {
+ log.warning(
+ "Failure during heartbeat to lease: [{}]. Retries exhausted. Lease
assumed to be released.",
+ t.getMessage)
+ granted.set(false)
+ executeLeaseLockCallback(leaseLost, Some(t))
+ goto(Idle).using(ReadRequired)
+ }
+ case Event(Release(), GrantedVersion(version, leaseLost, _)) =>
pipe(k8sApi.updateLeaseResource(leaseName, "",
version).map(WriteResponse.apply)).to(self)
goto(Releasing).using(OperationInProgress(sender(), version, leaseLost))
case Event(Acquire(leaseLostCallback), gv: GrantedVersion) =>
@@ -221,33 +287,69 @@ private[pekko] class LeaseActor(k8sApi: KubernetesApi,
settings: LeaseSettings,
}
when(Releasing) {
- // FIXME deal with failure from releasing the the lock, currently handled
in whenUnhandled but could retry to remove:
https://github.com/lightbend/akka-commercial-addons/issues/502
- case Event(WriteResponse(Right(lr)), OperationInProgress(who, _, _, _)) =>
+ case Event(WriteResponse(Right(lr)), OperationInProgress(who, _, _, _, _,
_)) =>
require(lr.owner.isEmpty, "Released lease has unexpected owner: " + lr)
who ! LeaseReleased
goto(Idle).using(LeaseCleared(lr.version))
- case Event(WriteResponse(Left(lr @ LeaseResource(None, _, _))),
OperationInProgress(who, _, _, _)) =>
+ case Event(WriteResponse(Left(lr @ LeaseResource(None, _, _))),
OperationInProgress(who, _, _, _, _, _)) =>
log.warning(
"Release conflict and owner has been removed: {}. Lease will continue
to work but TTL must have been reached to allow another node to remove lease.",
lr)
who ! LeaseReleased
goto(Idle).using(ReadRequired)
- case Event(WriteResponse(Left(lr @ LeaseResource(Some(_), _, _))),
OperationInProgress(who, _, _, _)) =>
+ case Event(WriteResponse(Left(lr @ LeaseResource(Some(_), _, _))),
OperationInProgress(who, _, _, _, _, _)) =>
log.warning(
"Release conflict and owner has changed: {}. Lease will continue to
work but TTL must have been reached to allow another node to update the lease.",
lr)
who ! LeaseReleased
goto(Idle).using(ReadRequired)
+ case Event(Failure(t), op @ OperationInProgress(who, _, _, startTime,
retries, _)) =>
+ // Pace release retries off the lease operation timeout rather than the
heartbeat interval:
+ // the caller is waiting on an ask that uses the operation timeout, so
retries that run past
+ // it would leave the caller with an ask timeout and the reply in dead
letters.
+ val operationTimeout = settings.timeoutSettings.operationTimeout
+ val retryDelay = operationTimeout / (releaseMaxRetries + 1)
+ val elapsed = (System.nanoTime() - startTime).nanos
+ if (retries < releaseMaxRetries && (elapsed + retryDelay) <
operationTimeout) {
+ log.warning(
+ "Failure releasing lease: [{}]. Retrying in {} (attempt {}/{}).",
+ t.getMessage,
+ retryDelay.pretty,
+ retries + 1,
+ releaseMaxRetries)
+ startSingleTimer("release-retry", ReleaseRetry, retryDelay)
+ stay().using(op.copy(releaseRetries = retries + 1))
+ } else {
+ log.warning("Failure releasing lease: [{}]. Retries exhausted.",
t.getMessage)
+ who ! Failure(t)
+ goto(Idle).using(ReadRequired)
+ }
+ case Event(ReleaseRetry, OperationInProgress(_, version, _, _, _, _)) =>
+ log.debug("Release retry: releasing lease. Version {}", version)
+ pipe(k8sApi.updateLeaseResource(leaseName, "",
version).map(WriteResponse.apply)).to(self)
+ stay()
+ case Event(Acquire(_), _) =>
+ // Acquiring while a release of the same lease is in flight is
contradictory, so unlike an
+ // acquire during an in-flight acquire this is rejected rather than
queued.
+ log.info(
+ "Acquire request for owner {} lease {} while a release is in
progress.",
+ ownerName,
+ leaseName)
+ sender() ! InvalidRequest("Tried to acquire a lease while a release is
in progress")
+ stay()
}
whenUnhandled {
- case Event(Acquire(_), data @ _) =>
+ case Event(Acquire(leaseLostCallback), data: ReplyRequired) =>
+ // An acquire for the same lease is already in flight. Queue this caller
rather than
+ // rejecting it: they all want the same outcome and will get the same
response.
log.info(
- "Acquire request for owner {} lease {} while previous acquire/release
still in progress. Current state: {}",
+ "Acquire request for owner {} lease {} while a previous acquire is
still in progress, " +
+ "the caller will get the result of that acquire. Current state: {}",
ownerName,
leaseName,
stateName)
- stay().using(data)
+ stay().using(addAcquirer(data, sender(), leaseLostCallback))
case Event(Release(), data @ _) =>
log.info(
"Release request for owner {} lease {} while previous acquire/release
still in progress. Current state: {}",
@@ -263,24 +365,43 @@ private[pekko] class LeaseActor(k8sApi: KubernetesApi,
settings: LeaseSettings,
leaseName,
t.getMessage,
stateName)
- replyRequired.replyTo ! Failure(t)
+ replyToAll(replyRequired, Failure(t))
goto(Idle).using(ReadRequired)
}
+ private def replyToAll(data: ReplyRequired, response: Any): Unit =
+ data.allReplyTo.foreach(_ ! response)
+
+ /**
+ * Add a caller that asked to acquire the lease while an acquire was already
in flight. The most
+ * recently supplied lease lost callback wins, matching the re-acquire
behaviour in `Granted`.
+ */
+ private def addAcquirer(data: ReplyRequired, who: ActorRef, leaseLost:
Option[Throwable] => Unit): Data =
+ data match {
+ case prd: PendingReadData =>
+ prd.copy(leaseLostCallback = leaseLost, alsoReplyTo = prd.alsoReplyTo
+ who)
+ case op: OperationInProgress =>
+ op.copy(leaseLostCallback = leaseLost, alsoReplyTo = op.alsoReplyTo +
who)
+ }
+
onTransition {
case _ -> Granted =>
startSingleTimer("heartbeat", Heartbeat,
settings.timeoutSettings.heartbeatInterval)
case Granted -> _ =>
cancelTimer("heartbeat")
+ cancelTimer("heartbeat-retry")
granted.set(false)
+ case Releasing -> _ =>
+ cancelTimer("release-retry")
}
private def tryGetLease(
version: String,
reply: ActorRef,
- leaseLost: Option[Throwable] => Unit): FSM.State[LeaseActor.State, Data]
= {
+ leaseLost: Option[Throwable] => Unit,
+ alsoReplyTo: Set[ActorRef]): FSM.State[LeaseActor.State, Data] = {
pipe(k8sApi.updateLeaseResource(leaseName, ownerName, version).map(r =>
WriteResponse(r))).to(self)
- goto(Granting).using(OperationInProgress(reply, version, leaseLost))
+ goto(Granting).using(OperationInProgress(reply, version, leaseLost,
alsoReplyTo = alsoReplyTo))
}
private def hasLeaseTimedOut(leaseTime: Long): Boolean = {
diff --git
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/NativeKubernetesApiImpl.scala
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/NativeKubernetesApiImpl.scala
index 40a47bf7..d6e2a2f7 100644
---
a/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/NativeKubernetesApiImpl.scala
+++
b/lease-kubernetes/src/main/scala/org/apache/pekko/coordination/lease/kubernetes/internal/NativeKubernetesApiImpl.scala
@@ -39,6 +39,7 @@ object NativeKubernetesApiImpl {
new DateTimeFormatterBuilder().parseDefaulting(ChronoField.OFFSET_SECONDS,
0).append(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss")).optionalStart.appendFraction(
ChronoField.NANO_OF_SECOND, 6, 6,
true).optionalEnd.appendLiteral("Z").toFormatter
+ private val UTC_FORMATTER: DateTimeFormatter =
RFC3339MICRO_FORMATTER.withZone(ZoneId.of("UTC"))
}
/**
@@ -182,7 +183,7 @@ object NativeKubernetesApiImpl {
}
private def currentTimeRFC3339: String = {
- RFC3339MICRO_FORMATTER.withZone(ZoneId.of("UTC")).format(Instant.now())
+ NativeKubernetesApiImpl.UTC_FORMATTER.format(Instant.now())
}
private def toLeaseResource(lcr: NativeLeaseResource) = {
@@ -195,7 +196,7 @@ object NativeKubernetesApiImpl {
case other => Some(other)
}
LeaseResource(owner, lcr.metadata.resourceVersion.get,
- LocalDateTime.parse(lcr.spec.acquireTime, RFC3339MICRO_FORMATTER)
+ LocalDateTime.parse(lcr.spec.acquireTime,
NativeKubernetesApiImpl.RFC3339MICRO_FORMATTER)
.atZone(ZoneId.of("UTC"))
.toInstant
.toEpochMilli)
diff --git
a/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettingsSpec.scala
b/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettingsSpec.scala
index dbbf9c2f..ae1e585f 100644
---
a/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettingsSpec.scala
+++
b/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/KubernetesSettingsSpec.scala
@@ -59,6 +59,28 @@ class KubernetesSettingsSpec extends AnyWordSpec with
Matchers {
"support on-truncate-add-hash-length override" in {
conf("on-truncate-add-hash-length=12").onTruncateAddHashLength
shouldEqual 12
}
+ "default heartbeat-max-retries to 3" in {
+ conf("").heartbeatMaxRetries shouldEqual 3
+ }
+ "support heartbeat-max-retries override" in {
+ conf("heartbeat-max-retries=5").heartbeatMaxRetries shouldEqual 5
+ }
+ "default release-max-retries to 3" in {
+ conf("").releaseMaxRetries shouldEqual 3
+ }
+ "support release-max-retries override" in {
+ conf("release-max-retries=1").releaseMaxRetries shouldEqual 1
+ }
+ "not allow a negative heartbeat-max-retries" in {
+ intercept[IllegalArgumentException] {
+ conf("heartbeat-max-retries=-1")
+ }.getMessage shouldEqual "requirement failed: 'heartbeat-max-retries'
must not be negative"
+ }
+ "not allow a negative release-max-retries" in {
+ intercept[IllegalArgumentException] {
+ conf("release-max-retries=-1")
+ }.getMessage shouldEqual "requirement failed: 'release-max-retries' must
not be negative"
+ }
"not allow server request timeout greater than operation timeout" in {
intercept[IllegalArgumentException] {
conf("""
diff --git
a/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActorSpec.scala
b/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActorSpec.scala
index 02037f74..14e816cc 100644
---
a/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActorSpec.scala
+++
b/lease-kubernetes/src/test/scala/org/apache/pekko/coordination/lease/kubernetes/LeaseActorSpec.scala
@@ -19,9 +19,9 @@ import org.apache.pekko
import pekko.actor.Status.Failure
import pekko.actor.{ ActorRef, ActorSystem }
import pekko.coordination.lease.kubernetes.LeaseActor._
-import pekko.coordination.lease.{ LeaseException, LeaseSettings,
TimeoutSettings }
+import pekko.coordination.lease.{ LeaseException, LeaseSettings,
LeaseTimeoutException, TimeoutSettings }
import pekko.pattern.ask
-import pekko.testkit.{ TestKit, TestProbe }
+import pekko.testkit.{ TestDuration, TestKit, TestProbe }
import pekko.util.{ ConstantFun, Timeout }
import com.typesafe.config.ConfigFactory
import org.scalatest.BeforeAndAfterAll
@@ -68,6 +68,10 @@ class LeaseActorSpec
val leaseName = "sbr"
+ // How far the lease version is moved on by other clients when simulating a
conflict. Any value
+ // greater than zero works: the test only needs a version ahead of the one
the actor sent.
+ val otherClientUpdates = 6
+
"LeaseActor" should {
// TODO what if the same client asks for the lease when granting? respond
to both or ignore?
@@ -174,9 +178,7 @@ class LeaseActorSpec
val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
acquireLease()
underTest ! Release()
- updateProbe.expectMsg(("", currentVersion))
- incrementVersion()
- updateProbe.reply(Failure(k8sApiFailure))
+ failAllReleaseAttempts(k8sApiFailure)
senderProbe.expectMsg(Failure(k8sApiFailure))
}
@@ -214,12 +216,13 @@ class LeaseActorSpec
// Version from the previous lock so can skip the read of the resource
unless the CAS fails
underTest ! LeaseActor.Acquire()
updateProbe.expectMsg((ownerName, currentVersion))
- // Fail due to cas, version has moved on by 6 but no one owns the lock
- val failedVersion = currentVersionCount + 6
+ // Fail due to cas, the version has moved on but no one owns the lock
+ val failedVersion = currentVersionCount + otherClientUpdates
updateProbe.reply(Left(LeaseResource(None, failedVersion.toString,
System.currentTimeMillis())))
- // Try again
+ // Try again, a successful update moves the version on again
updateProbe.expectMsg((ownerName, failedVersion.toString))
- updateProbe.reply(Right(LeaseResource(Some(ownerName),
failedVersion.toString, System.currentTimeMillis())))
+ currentVersionCount = failedVersion + 1
+ updateProbe.reply(Right(LeaseResource(Some(ownerName), currentVersion,
System.currentTimeMillis())))
senderProbe.expectMsg(LeaseAcquired)
}
@@ -250,30 +253,26 @@ class LeaseActorSpec
}
}
- "heartbeat fail should set granted to false" in new Test {
+ "heartbeat fail should set granted to false after retries exhausted" in
new Test {
val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
acquireLease()
expectHeartBeat()
granted.get() shouldEqual true
- updateProbe.expectMsg((ownerName, currentVersion))
- incrementVersion()
- updateProbe.reply(Failure(k8sApiFailure))
+ failAllHeartbeatAttempts(k8sApiFailure)
awaitAssert {
granted.get() shouldEqual false
}
}
- "heartbeat fail should call lease lost callback" in new Test {
+ "heartbeat fail should call lease lost callback after retries exhausted"
in new Test {
val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
@volatile var callbackCalled: Option[Throwable] = None
acquireLease(reason => callbackCalled = reason)
expectHeartBeat()
granted.get() shouldEqual true
- updateProbe.expectMsg((ownerName, currentVersion))
- incrementVersion()
- updateProbe.reply(Failure(k8sApiFailure))
+ failAllHeartbeatAttempts(k8sApiFailure)
awaitAssert {
callbackCalled shouldEqual Some(k8sApiFailure)
}
@@ -293,10 +292,80 @@ class LeaseActorSpec
acquireLease()
}
- "lease acquire in reading state" in new Test {
- // TODO this could accumulate senders and reply to all, atm it'll log
saying
- // previous action hasn't finished
- pending
+ "reply LeaseAcquired to both callers when acquire arrives while read is
pending" in new Test {
+ val secondSender = TestProbe()
+ underTest.tell(LeaseActor.Acquire(), senderProbe.ref)
+ leaseProbe.expectMsg(leaseName)
+
+ // second acquire while first is still pending, only one read/update is
issued for both
+ underTest.tell(LeaseActor.Acquire(), secondSender.ref)
+ leaseProbe.expectNoMessage(100.millis)
+
+ leaseProbe.reply(LeaseResource(None, currentVersion,
System.currentTimeMillis()))
+ updateProbe.expectMsg((ownerName, currentVersion))
+ incrementVersion()
+ updateProbe.reply(Right(LeaseResource(Some(ownerName), currentVersion,
System.currentTimeMillis())))
+
+ senderProbe.expectMsg(LeaseAcquired)
+ secondSender.expectMsg(LeaseAcquired)
+ }
+
+ "reply LeaseAcquired to both callers when acquire arrives while grant is
in progress" in new Test {
+ val secondSender = TestProbe()
+ underTest.tell(LeaseActor.Acquire(), senderProbe.ref)
+ leaseProbe.expectMsg(leaseName)
+ leaseProbe.reply(LeaseResource(None, currentVersion,
System.currentTimeMillis()))
+ updateProbe.expectMsg((ownerName, currentVersion))
+
+ // second acquire while granting, no extra update is issued
+ underTest.tell(LeaseActor.Acquire(), secondSender.ref)
+ updateProbe.expectNoMessage(100.millis)
+
+ incrementVersion()
+ updateProbe.reply(Right(LeaseResource(Some(ownerName), currentVersion,
System.currentTimeMillis())))
+
+ senderProbe.expectMsg(LeaseAcquired)
+ secondSender.expectMsg(LeaseAcquired)
+ }
+
+ "reply LeaseTaken to both callers when the lease turns out to be taken" in
new Test {
+ val secondSender = TestProbe()
+ underTest.tell(LeaseActor.Acquire(), senderProbe.ref)
+ leaseProbe.expectMsg(leaseName)
+ underTest.tell(LeaseActor.Acquire(), secondSender.ref)
+
+ leaseProbe.reply(LeaseResource(Some("someone else"), currentVersion,
System.currentTimeMillis()))
+
+ senderProbe.expectMsg(LeaseTaken)
+ secondSender.expectMsg(LeaseTaken)
+ }
+
+ "reply the failure to both callers when the in progress acquire fails" in
new Test {
+ val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
+ val secondSender = TestProbe()
+ underTest.tell(LeaseActor.Acquire(), senderProbe.ref)
+ leaseProbe.expectMsg(leaseName)
+ underTest.tell(LeaseActor.Acquire(), secondSender.ref)
+
+ leaseProbe.reply(Failure(k8sApiFailure))
+
+ senderProbe.expectMsg(Failure(k8sApiFailure))
+ secondSender.expectMsg(Failure(k8sApiFailure))
+ }
+
+ "reply InvalidRequest when acquire arrives while a release is in progress"
in new Test {
+ val secondSender = TestProbe()
+ acquireLease()
+ underTest ! Release()
+ updateProbe.expectMsg(("", currentVersion))
+
+ underTest.tell(LeaseActor.Acquire(), secondSender.ref)
+ secondSender.expectMsg(InvalidRequest("Tried to acquire a lease while a
release is in progress"))
+
+ // the release itself is unaffected
+ incrementVersion()
+ updateProbe.reply(Right(LeaseResource(None, currentVersion,
System.currentTimeMillis())))
+ senderProbe.expectMsg(LeaseReleased)
}
"return lease taken if conflict when updating lease" in new Test {
@@ -384,10 +453,11 @@ class LeaseActorSpec
trait Test {
val ownerName = "owner1"
+ def timeoutSettings: TimeoutSettings = new TimeoutSettings(25.millis,
250.millis, 1.second)
val leaseSettings: LeaseSettings = new LeaseSettings(
leaseName,
ownerName,
- new TimeoutSettings(25.millis, 250.millis, 1.second),
+ timeoutSettings,
ConfigFactory.empty())
var currentVersionCount = 1
@@ -397,7 +467,11 @@ class LeaseActorSpec
val updateProbe = TestProbe()
val mockKubernetesApi = new MockKubernetesApi(system, leaseProbe.ref,
updateProbe.ref)
val granted = new AtomicBoolean(false)
- val underTest = system.actorOf(LeaseActor.props(mockKubernetesApi,
leaseSettings, leaseSettings.leaseName, granted))
+ def heartbeatMaxRetries: Int = 3
+ def releaseMaxRetries: Int = 3
+ val underTest = system.actorOf(
+ LeaseActor.props(mockKubernetesApi, leaseSettings,
leaseSettings.leaseName, granted, heartbeatMaxRetries,
+ releaseMaxRetries))
val senderProbe = TestProbe()
implicit val sender: ActorRef = senderProbe.ref
@@ -476,13 +550,248 @@ class LeaseActorSpec
}
}
+ /**
+ * Fail the initial heartbeat and every retry of it. A failed heartbeat
does not move the lease
+ * version on, so the same version is expected for each attempt.
+ */
+ def failAllHeartbeatAttempts(failure: Throwable): Unit =
+ for (_ <- 0 to heartbeatMaxRetries) {
+ updateProbe.expectMsg((ownerName, currentVersion))
+ updateProbe.reply(Failure(failure))
+ }
+
+ /** Fail the initial release and every retry of it. */
+ def failAllReleaseAttempts(failure: Throwable): Unit =
+ for (_ <- 0 to releaseMaxRetries) {
+ updateProbe.expectMsg(("", currentVersion))
+ updateProbe.reply(Failure(failure))
+ }
+
def heartBeatFailure(): Unit = {
+ failAllHeartbeatAttempts(new LeaseException("Failed to communicate with
API server"))
+ awaitAssert {
+ granted.get() shouldEqual false
+ }
+ }
+
+ }
+
+ trait NoRetryTest extends Test {
+ override def heartbeatMaxRetries: Int = 0
+ override def releaseMaxRetries: Int = 0
+ }
+
+ "LeaseActor with retries disabled" should {
+
+ "immediately release lease on heartbeat failure" in new NoRetryTest {
+ acquireLease()
+ expectHeartBeat()
+ granted.get() shouldEqual true
+
+ heartBeatFailure()
+ }
+
+ "call lease lost callback immediately on heartbeat failure" in new
NoRetryTest {
+ @volatile var callbackCalled: Option[Throwable] = None
+ acquireLease(reason => callbackCalled = reason)
+ expectHeartBeat()
+ granted.get() shouldEqual true
+
+ failAllHeartbeatAttempts(new LeaseException("Failed to communicate with
API server"))
+ awaitAssert {
+ callbackCalled shouldBe defined
+ }
+ }
+
+ "allow re-acquire after immediate heartbeat failure" in new NoRetryTest {
+ acquireLease()
+ expectHeartBeat()
+ heartBeatFailure()
+ acquireLease()
+ }
+
+ "immediately report release failure with no retries" in new NoRetryTest {
+ val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
+ acquireLease()
+ underTest ! Release()
+ updateProbe.expectMsg(("", currentVersion))
+ updateProbe.reply(Failure(k8sApiFailure))
+ senderProbe.expectMsg(Failure(k8sApiFailure))
+ }
+
+ "allow re-acquire after immediate release failure" in new NoRetryTest {
+ acquireLease()
+ underTest ! Release()
+ updateProbe.expectMsg(("", currentVersion))
+ updateProbe.reply(Failure(new LeaseException("Failed")))
+ senderProbe.expectMsgType[Failure]
+ acquireLease()
+ }
+
+ }
+
+ trait ShortOperationTimeoutTest extends Test {
+ // an operation timeout that is already spent by the time the first
conflict is handled
+ override def timeoutSettings: TimeoutSettings = new
TimeoutSettings(25.millis, 250.millis, 1.nano)
+ }
+
+ trait ReleaseRetryTimingTest extends Test {
+ // heartbeat-interval is deliberately far larger than the operation
timeout: if release retries
+ // were paced off the heartbeat interval they would be 5s apart and the
probe would never see them
+ override def timeoutSettings: TimeoutSettings = new
TimeoutSettings(20.seconds, 60.seconds, 1.second)
+ }
+
+ trait IndependentRetryCountsTest extends Test {
+ override def heartbeatMaxRetries: Int = 0
+ override def releaseMaxRetries: Int = 2
+ }
+
+ "LeaseActor retry settings" should {
+
+ "apply heartbeat-max-retries and release-max-retries independently" in new
IndependentRetryCountsTest {
+ val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
+ acquireLease()
+ expectHeartBeat()
+
+ // heartbeat-max-retries is 0, so the lease is given up on the first
failed heartbeat
updateProbe.expectMsg((ownerName, currentVersion))
- incrementVersion()
- updateProbe.reply(Failure(new LeaseException("Failed to communicate with
API server")))
+ updateProbe.reply(Failure(k8sApiFailure))
awaitAssert {
granted.get() shouldEqual false
}
+
+ // release-max-retries is 2, so the release is attempted three times
before the caller is told
+ acquireLease()
+ underTest ! Release()
+ failAllReleaseAttempts(k8sApiFailure)
+ senderProbe.expectMsg(Failure(k8sApiFailure))
+ }
+
+ }
+
+ "LeaseActor acquire conflict retry" should {
+
+ "reply LeaseAcquired only once the retry succeeds" in new Test {
+ acquireLease()
+ releaseLease()
+
+ // Start acquire, will hit a conflict and retry
+ underTest ! LeaseActor.Acquire()
+ updateProbe.expectMsg((ownerName, currentVersion))
+ // Conflict: version moved on, no owner
+ val conflictVersion = currentVersionCount + otherClientUpdates
+ updateProbe.reply(Left(LeaseResource(None, conflictVersion.toString,
System.currentTimeMillis())))
+ // Nothing is reported to the caller until the retry has been answered
+ senderProbe.expectNoMessage(100.millis)
+ // Retry uses the version from the conflict response, and success moves
the version on again
+ updateProbe.expectMsg((ownerName, conflictVersion.toString))
+ currentVersionCount = conflictVersion + 1
+ updateProbe.reply(Right(LeaseResource(Some(ownerName), currentVersion,
System.currentTimeMillis())))
+ senderProbe.expectMsg(LeaseAcquired)
+ granted.get() shouldEqual true
+ }
+
+ "keep retrying while the version keeps moving on" in new Test {
+ acquireLease()
+ releaseLease()
+
+ underTest ! LeaseActor.Acquire()
+ updateProbe.expectMsg((ownerName, currentVersion))
+ // the version moves on again between the first conflict and the retry
landing
+ val firstConflict = currentVersionCount + otherClientUpdates
+ updateProbe.reply(Left(LeaseResource(None, firstConflict.toString,
System.currentTimeMillis())))
+ updateProbe.expectMsg((ownerName, firstConflict.toString))
+ val secondConflict = firstConflict + otherClientUpdates
+ updateProbe.reply(Left(LeaseResource(None, secondConflict.toString,
System.currentTimeMillis())))
+
+ // the third attempt uses the version from the second conflict, not the
original one
+ updateProbe.expectMsg((ownerName, secondConflict.toString))
+ currentVersionCount = secondConflict + 1
+ updateProbe.reply(Right(LeaseResource(Some(ownerName), currentVersion,
System.currentTimeMillis())))
+ senderProbe.expectMsg(LeaseAcquired)
+ granted.get() shouldEqual true
+ }
+
+ "reply LeaseTaken if another owner has the lease by the time the retry
lands" in new Test {
+ acquireLease()
+ releaseLease()
+
+ underTest ! LeaseActor.Acquire()
+ updateProbe.expectMsg((ownerName, currentVersion))
+ val conflictVersion = currentVersionCount + otherClientUpdates
+ updateProbe.reply(Left(LeaseResource(None, conflictVersion.toString,
System.currentTimeMillis())))
+
+ updateProbe.expectMsg((ownerName, conflictVersion.toString))
+ updateProbe.reply(
+ Left(LeaseResource(Some("i got there first"), (conflictVersion +
1).toString, System.currentTimeMillis())))
+ senderProbe.expectMsg(LeaseTaken)
+ granted.get() shouldEqual false
+ }
+
+ "give up and fail the caller once the lease operation timeout is spent" in
new ShortOperationTimeoutTest {
+ underTest ! LeaseActor.Acquire()
+ leaseProbe.expectMsg(leaseName)
+ leaseProbe.reply(LeaseResource(None, currentVersion,
System.currentTimeMillis()))
+ updateProbe.expectMsg((ownerName, currentVersion))
+ incrementVersion()
+ // version has moved on but the lease is not taken, so this would
normally be retried
+ updateProbe.reply(Left(LeaseResource(None, currentVersion,
System.currentTimeMillis())))
+
+ senderProbe.expectMsgType[Failure].cause shouldBe
a[LeaseTimeoutException]
+ // no further retry is issued, the lease is not granted behind the
caller's back
+ updateProbe.expectNoMessage(200.millis)
+ granted.get() shouldEqual false
+ }
+
+ "be able to acquire again after giving up on conflict retries" in new
ShortOperationTimeoutTest {
+ underTest ! LeaseActor.Acquire()
+ leaseProbe.expectMsg(leaseName)
+ leaseProbe.reply(LeaseResource(None, currentVersion,
System.currentTimeMillis()))
+ updateProbe.expectMsg((ownerName, currentVersion))
+ incrementVersion()
+ updateProbe.reply(Left(LeaseResource(None, currentVersion,
System.currentTimeMillis())))
+ senderProbe.expectMsgType[Failure].cause shouldBe
a[LeaseTimeoutException]
+
+ acquireLease()
+ }
+
+ }
+
+ "LeaseActor release retry" should {
+
+ "pace retries off the lease operation timeout, not the heartbeat interval"
in new ReleaseRetryTimingTest {
+ val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
+ acquireLease()
+ val operationTimeout = leaseSettings.timeoutSettings.operationTimeout
+ val start = System.nanoTime()
+ // Paced off the operation timeout these attempts are 250ms apart; paced
off the heartbeat
+ // interval they would be 5s apart and the probe would time out waiting
for them.
+ underTest ! Release()
+ failAllReleaseAttempts(k8sApiFailure)
+ senderProbe.expectMsg(Failure(k8sApiFailure))
+ // the caller is told the outcome before the ask it is waiting on would
have timed out
+ (System.nanoTime() - start).nanos should be < operationTimeout.dilated
+ }
+
+ "retry release on failure and succeed" in new Test {
+ acquireLease()
+ underTest ! Release()
+ // First attempt fails
+ updateProbe.expectMsg(("", currentVersion))
+ updateProbe.reply(Failure(new LeaseException("transient error")))
+ // Retry succeeds
+ updateProbe.expectMsg(("", currentVersion))
+ incrementVersion()
+ updateProbe.reply(Right(LeaseResource(None, currentVersion,
System.currentTimeMillis())))
+ senderProbe.expectMsg(LeaseReleased)
+ }
+
+ "report release failure after retries exhausted" in new Test {
+ val k8sApiFailure = new LeaseException("Failed to communicate with API
server")
+ acquireLease()
+ underTest ! Release()
+ failAllReleaseAttempts(k8sApiFailure)
+ senderProbe.expectMsg(Failure(k8sApiFailure))
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
