New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Use latest code if action's revision is mismatched #4954
Merged
Merged
Changes from all commits
Commits
Show all changes
2 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -164,6 +164,61 @@ class InvokerReactive( | |
private val pool = | ||
actorSystem.actorOf(ContainerPool.props(childFactory, poolConfig, activationFeed, prewarmingConfigs)) | ||
|
||
def handleActivationMessage(msg: ActivationMessage)(implicit transid: TransactionId): Future[Unit] = { | ||
val namespace = msg.action.path | ||
val name = msg.action.name | ||
val actionid = FullyQualifiedEntityName(namespace, name).toDocId.asDocInfo(msg.revision) | ||
val subject = msg.user.subject | ||
|
||
logging.debug(this, s"${actionid.id} $subject ${msg.activationId}") | ||
|
||
// caching is enabled since actions have revision id and an updated | ||
// action will not hit in the cache due to change in the revision id; | ||
// if the doc revision is missing, then bypass cache | ||
if (actionid.rev == DocRevision.empty) logging.warn(this, s"revision was not provided for ${actionid.id}") | ||
|
||
WhiskAction | ||
.get(entityStore, actionid.id, actionid.rev, fromCache = actionid.rev != DocRevision.empty) | ||
.flatMap(action => { | ||
action.toExecutableWhiskAction match { | ||
case Some(executable) => | ||
pool ! Run(executable, msg) | ||
Future.successful(()) | ||
case None => | ||
logging.error(this, s"non-executable action reached the invoker ${action.fullyQualifiedName(false)}") | ||
Future.failed(new IllegalStateException("non-executable action reached the invoker")) | ||
} | ||
}) | ||
.recoverWith { | ||
case DocumentRevisionMismatchException(_) => | ||
// if revision is mismatched, the action may have been updated, | ||
// so try again with the latest code | ||
handleActivationMessage(msg.copy(revision = DocRevision.empty)) | ||
case t => | ||
val response = t match { | ||
case _: NoDocumentException => | ||
ActivationResponse.applicationError(Messages.actionRemovedWhileInvoking) | ||
case _: DocumentTypeMismatchException | _: DocumentUnreadable => | ||
ActivationResponse.whiskError(Messages.actionMismatchWhileInvoking) | ||
case _ => | ||
ActivationResponse.whiskError(Messages.actionFetchErrorWhileInvoking) | ||
} | ||
activationFeed ! MessageFeed.Processed | ||
|
||
val activation = generateFallbackActivation(msg, response) | ||
ack( | ||
msg.transid, | ||
activation, | ||
msg.blocking, | ||
msg.rootControllerIndex, | ||
msg.user.namespace.uuid, | ||
CombinedCompletionAndResultMessage(transid, activation, instance)) | ||
|
||
store(msg.transid, activation, msg.blocking, UserContext(msg.user)) | ||
Future.successful(()) | ||
} | ||
} | ||
|
||
/** Is called when an ActivationMessage is read from Kafka */ | ||
def processActivationMessage(bytes: Array[Byte]): Future[Unit] = { | ||
Future(ActivationMessage.parse(new String(bytes, StandardCharsets.UTF_8))) | ||
|
@@ -179,58 +234,7 @@ class InvokerReactive( | |
|
||
if (!namespaceBlacklist.isBlacklisted(msg.user)) { | ||
val start = transid.started(this, LoggingMarkers.INVOKER_ACTIVATION, logLevel = InfoLevel) | ||
val namespace = msg.action.path | ||
val name = msg.action.name | ||
val actionid = FullyQualifiedEntityName(namespace, name).toDocId.asDocInfo(msg.revision) | ||
val subject = msg.user.subject | ||
|
||
logging.debug(this, s"${actionid.id} $subject ${msg.activationId}") | ||
|
||
// caching is enabled since actions have revision id and an updated | ||
// action will not hit in the cache due to change in the revision id; | ||
// if the doc revision is missing, then bypass cache | ||
if (actionid.rev == DocRevision.empty) logging.warn(this, s"revision was not provided for ${actionid.id}") | ||
|
||
WhiskAction | ||
.get(entityStore, actionid.id, actionid.rev, fromCache = actionid.rev != DocRevision.empty) | ||
.flatMap { action => | ||
action.toExecutableWhiskAction match { | ||
case Some(executable) => | ||
pool ! Run(executable, msg) | ||
Future.successful(()) | ||
case None => | ||
logging.error(this, s"non-executable action reached the invoker ${action.fullyQualifiedName(false)}") | ||
Future.failed(new IllegalStateException("non-executable action reached the invoker")) | ||
} | ||
} | ||
.recoverWith { | ||
case t => | ||
// If the action cannot be found, the user has concurrently deleted it, | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This comment is still valid - why was it removed? |
||
// making this an application error. All other errors are considered system | ||
// errors and should cause the invoker to be considered unhealthy. | ||
val response = t match { | ||
case _: NoDocumentException => | ||
ActivationResponse.applicationError(Messages.actionRemovedWhileInvoking) | ||
case _: DocumentTypeMismatchException | _: DocumentUnreadable => | ||
ActivationResponse.whiskError(Messages.actionMismatchWhileInvoking) | ||
case _ => | ||
ActivationResponse.whiskError(Messages.actionFetchErrorWhileInvoking) | ||
} | ||
|
||
activationFeed ! MessageFeed.Processed | ||
|
||
val activation = generateFallbackActivation(msg, response) | ||
ack( | ||
msg.transid, | ||
activation, | ||
msg.blocking, | ||
msg.rootControllerIndex, | ||
msg.user.namespace.uuid, | ||
CombinedCompletionAndResultMessage(transid, activation, instance)) | ||
|
||
store(msg.transid, activation, msg.blocking, UserContext(msg.user)) | ||
Future.successful(()) | ||
} | ||
handleActivationMessage(msg) | ||
} else { | ||
// Iff the current namespace is blacklisted, an active-ack is only produced to keep the loadbalancer protocol | ||
// Due to the protective nature of the blacklist, a database entry is not written. | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -501,6 +501,46 @@ class WskSequenceTests extends TestHelpers with WskTestHelpers with StreamLoggin | |
checkEchoSeqRuleResult(newRun, seqName, JsObject(newPayload)) | ||
} | ||
|
||
it should "run a sub-action even if it is updated while the sequence action is running" in withAssetCleaner(wskprops) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I've added a test case. And this test currently does not pass on the master branch. |
||
(wp, assetHelper) => | ||
val seqName = "sequence" | ||
val sleep = "sleep" | ||
val echo = "echo" | ||
val slowInvokeDuration = 5.seconds | ||
|
||
// create echo action | ||
val echoFile = TestUtils.getTestActionFilename(s"$echo.js") | ||
assetHelper.withCleaner(wsk.action, echo) { (action, actionName) => | ||
action.create(name = actionName, artifact = Some(echoFile), timeout = Some(allowedActionDuration)) | ||
} | ||
// create sleep action | ||
val sleepFile = TestUtils.getTestActionFilename(s"$sleep.js") | ||
assetHelper.withCleaner(wsk.action, sleep) { (action, actionName) => | ||
action.create( | ||
name = sleep, | ||
artifact = Some(sleepFile), | ||
parameters = Map("sleepTimeInMs" -> slowInvokeDuration.toMillis.toJson), | ||
timeout = Some(allowedActionDuration)) | ||
} | ||
|
||
// create sequence | ||
assetHelper.withCleaner(wsk.action, seqName) { (action, seqName) => | ||
action.create(seqName, Some(s"$sleep,$echo"), kind = Some("sequence")) | ||
} | ||
val run = wsk.action.invoke(seqName) | ||
|
||
// update the sub-action before the sequence action invokes it | ||
wsk.action.create(name = echo, artifact = None, annotations = Map("a" -> JsString("A")), update = true) | ||
wsk.action.invoke(echo) | ||
|
||
wsk.action.create(name = echo, artifact = None, annotations = Map("b" -> JsString("B")), update = true) | ||
wsk.action.invoke(echo) | ||
|
||
withActivation(wsk.activation, run, totalWait = 2 * allowedActionDuration) { activation => | ||
activation.response.status shouldBe "success" | ||
} | ||
} | ||
|
||
/** | ||
* checks the result of an echo sequence connected to a trigger through a rule | ||
* @param triggerFireRun the run result of firing the trigger | ||
|
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This can be controversial but I think the system should take care of this.
This is to handle the case where the underlying actions are updated while a sequence action is being invoked.
If we consider the sequence as just a coordinator for underlying actions, it would be fine for it to always invoke the latest code. Once a sequence action is defined with some actions, we don't need to update the sequence action whenever the underlying actions are updated. It means the sequence action itself does not care about the version of actions in it but just focuses on the relation and the execution order of them.
So I think it is reasonable to invoke the latest codes all the time.
And regarding the implementation, while this is great, can we differentiate the sequence case with the others?
I feel like there can be some side effects.
For example, if any activation sent to Kafka arrives at the invoker side late, it could happen.
In such a case the activation is intended for the old codes but the latest code will be invoked with this change.
I did not look into code deeply yet, but can we make the controller setup subsequent activations with the latest codes while invoking a sequence action?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thank you for your review and I understand your concern.
Currently, Openwhisk doesn't have a versioning feature. So I don't think there's any problem with always serving the latest code if the DB can't fetch the older version, because action developers are responsible for compatibility between old and new action codes.
HDYT?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Not quite sure about this.
It would be great to listen to other reviewers' opinions as well.
From some point of view, it could be a semantic change as activations that are supposed to be rejected would be invoked successfully.
(With an assumption that the codes are backward compatible.)
How about sending an email to the dev list to get more people to attend this PR?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we leave at least a warning log to describe the system takes a fallback to the latest codes because of the revision mismatch?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Since you're also working on versioning - I think this change should be considered in that broader context.
If the sequence or composition references actions specifically by version, then it should be an error to invoke an alternate version. If the sequence uses "latest" then this change is acceptable.
Was the intent for this change strictly to address compositions?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes.
That needs to be considered in the action versioning feature I think.
CC: @jiangpengcheng