[flytepropeller][flyteadmin] Streaming Decks V2 #6053
Conversation
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Signed-off-by: Future-Outlier <eric901201@gmail.com> Co-authored-by: Yi Cheng <luyc58576@gmail.com> Co-authored-by: pingsutw <pingsutw@apache.org>
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #6053 +/- ##
==========================================
+ Coverage 37.05% 37.07% +0.01%
==========================================
Files 1318 1318
Lines 132638 132693 +55
==========================================
+ Hits 49151 49191 +40
- Misses 79237 79246 +9
- Partials 4250 4256 +6
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
| switch pluginTrns.pInfo.Phase() { | ||
| case pluginCore.PhaseSuccess: | ||
| // This is to prevent the console from potentially checking the deck URI that does not exist if in final phase(PhaseSuccess). | ||
| err = pluginTrns.RemoveNonexistentDeckURI(ctx, tCtx) |
There was a problem hiding this comment.
Does this do a head call on the deck URI for every task that succeeds? Two thoughts here:
(1) does the flyteadmin merge algorithm then remove the deckURI from the execution metadata?
(2) this is incurring a 20-30ms performance degredation to every task execution
There was a problem hiding this comment.
will take a look tmr, thank you!!!
There was a problem hiding this comment.
Does this do a head call on the deck URI for every task that succeeds?
yes it will do a head call by RemoteFileOutputReader
There was a problem hiding this comment.
how do you know the performance degradation?
did you use grafana or other performance tools?
There was a problem hiding this comment.
does the flyteadmin merge algorithm then remove the deckURI from the execution metadata?
flyteadmin will set the deckURI in the execution metadata to nil if the propeller removes it.
How to test it?
flytectl demo start --image futureoutlier/sandbox:deck-1205-1138 --forcecd flytekit
gh pr checkout 2779
from flytekit import ImageSpec, task, workflow
from flytekit.deck import Deck
flytekit_hash = "473ae1119af6f86c26c0790dee0affa3eb29be64"
flytekit = f"git+https://github.com/flyteorg/flytekit.git@{flytekit_hash}"
# Define custom image for the task
custom_image = ImageSpec(packages=[flytekit],
apt_packages=["git"],
registry="localhost:30000",
env={"FLYTE_SDK_LOGGING_LEVEL": 10},
)
@task(enable_deck=True, container_image=custom_image)
def t_deck():
import time
"""
1st deck only show timeline deck
2nd will show
"""
for i in range(5):
Deck.publish()
# # raise Exception("This is an exception")
time.sleep(3)
@workflow
def wf():
t_deck()
if __name__ == "__main__":
from flytekit.clis.sdk_in_container import pyflyte
from click.testing import CliRunner
import os
runner = CliRunner()
path = os.path.realpath(__file__)
# result = runner.invoke(pyflyte.main,
# ["run", path, "wf"])
# print("Local Execution: ", result.output)
result = runner.invoke(pyflyte.main,
["run", "--remote", path,"wf"])
# "--remote"
print("Remote Execution: ", result.output) |
|
Mind adding screenshots for the rendered deck and refresh to the PR description? |
Yes no problem |
its provided! |
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Code Review Agent Run #f3ef5eActionable Suggestions - 1
Additional Suggestions - 1
Review Details
AI Code Review powered by |
Changelist by BitoThis pull request implements the following key changes.
|
| // - We relied on a HEAD request to check if the deck file exists, then added the URI to the event. | ||
| // | ||
| // After (new behavior): | ||
| // - If `FLYTE_ENABLE_DECK = true` is set in the task template config (requires Flytekit > 1.14.0), |
There was a problem hiding this comment.
this comment is no longer correct right?
There was a problem hiding this comment.
yes super nice catch
| @@ -380,6 +430,27 @@ func (t Handler) fetchPluginTaskMetrics(pluginID, taskType string) (*taskMetrics | |||
| return t.taskMetricsMap[metricNameKey], nil | |||
| } | |||
|
|
|||
| func GetDeckStatus(ctx context.Context, tCtx *taskExecutionContext) (DeckStatus, error) { | |||
| // FLYTE_ENABLE_DECK is used when flytekit > 1.14.0 | |||
|
|
||
| metadata := template.GetMetadata() | ||
| if metadata == nil { | ||
| return DeckUnknown, nil |
There was a problem hiding this comment.
is this correct in older versions of flytekit? didn't tasks in the past also have this field? this means that this function will always return Disabled right for older versions of flytekit. meaning the condition on line 567 won't get triggered cuz it'll be disabled instead of unknown.
There was a problem hiding this comment.
oh yes, you are right, thinking solution.
There was a problem hiding this comment.
older versions of flytekit will return DeckUnknown, which ends up calling AddDeckURIIfDeckExists (which is the old code path where we would hit the blob store to check for the deck file before adding it to the events).
Code Review Agent Run Status
|
| // GetDeckStatus determines whether a task generates a deck based on its execution context. | ||
| // | ||
| // This function ensures backward compatibility with older Flytekit versions using the following logic: | ||
| // 1. For Flytekit > 1.14.3, the task template's metadata includes the `generates_deck` flag: | ||
| // - If `generates_deck` is set to true, it indicates that the task generates a deck, and DeckEnabled is returned. | ||
| // 2. If `generates_deck` is set to false or is not set (likely from older Flytekit versions): | ||
| // - DeckUnknown is returned as a placeholder status. | ||
| // - In terminal states, a HEAD request can be made to check if the deck file exists. | ||
| // | ||
| // In future implementations, a `DeckDisabled` status could be introduced for better performance optimization: | ||
| // - This would eliminate the need for a HEAD request in the final phase. | ||
| // - However, the tradeoff is that a new field would need to be added to FlyteIDL to support this behavior. | ||
|
|
||
| template, err := tCtx.tr.Read(ctx) |
There was a problem hiding this comment.
Better comments!
cc @wild-endeavor
|
Streaming Decks #!/usr/bin/env bash
set -ex
ARCH="$(uname -m)"
case ${ARCH} in
x86_64|amd64)
IMAGE_ARCH=amd64
;;
aarch64|arm64)
IMAGE_ARCH=arm64
;;
*)
>&2 echo "ERROR: Unsupported architecture: ${ARCH}"
exit 1
;;
esac
FLYTECONSOLE_IMAGE="localhost:30000/flyteconsole:1216-2134"
IMAGE_DIGEST="$(docker manifest inspect --verbose localhost:30000/flyteconsole:1216-2134 | \
jq --arg IMAGE_ARCH "${IMAGE_ARCH}" --raw-output \
'.[] | select(.Descriptor.platform.architecture == $IMAGE_ARCH) | .Descriptor.digest')"
# Short circuit if we already have the correct distribution
[ -f cmd/single/dist/.digest ] && grep -Fxq ${IMAGE_DIGEST} cmd/single/dist/.digest && exit 0
# Create container from desired image
CONTAINER_ID=$(docker create localhost:30000/flyteconsole:1216-2134)
trap 'docker rm -f ${CONTAINER_ID}' EXIT
# Copy distribution
rm -rf cmd/single/dist
docker cp ${CONTAINER_ID}:/app cmd/single/dist
printf '%q' ${IMAGE_DIGEST} > cmd/single/dist/.digest
|
Future-Outlier
force-pushed
the
streaming-deck-v2
branch
from
04f7fbc to
1204c0e
Compare
Code Review Agent Run Status
|
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Code Review Agent Run Status
|
|
/review |
Code Review Agent Run #ecb1b4Actionable Suggestions - 2
Review Details
AI Code Review powered by |
| func (p *pluginRequestedTransition) ObserveSuccess(outputPath storage.DataReference, taskMetadata *event.TaskNodeMetadata) { | ||
| if p.execInfo.OutputInfo == nil { | ||
| p.execInfo.OutputInfo = &handler.OutputInfo{ | ||
| OutputURI: outputPath, | ||
| } | ||
| } else { | ||
| p.execInfo.OutputInfo.OutputURI = outputPath |
There was a problem hiding this comment.
Consider checking if the OutputURI is empty before updating it. The current implementation may overwrite an existing valid output URI with an empty one.
Code suggestion
Check the AI-generated fix before applying
- if p.execInfo.OutputInfo == nil {
+ if p.execInfo.OutputInfo == nil && len(outputPath) > 0 {
p.execInfo.OutputInfo = &handler.OutputInfo{
OutputURI: outputPath,
}
} else {
- p.execInfo.OutputInfo.OutputURI = outputPath
+ if len(outputPath) > 0 {
+ p.execInfo.OutputInfo.OutputURI = outputPath
+ }
}
Code Review Run #ecb1b4
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| } | ||
| } | ||
| pluginTrns.ObserveSuccess(tCtx.ow.GetOutputPath(), deckURI, | ||
| pluginTrns.ObserveSuccess(tCtx.ow.GetOutputPath(), |
There was a problem hiding this comment.
The ObserveSuccess method signature has been modified to remove the deckURI parameter, but the deck URI is now being added through separate methods. Consider if this change maintains backward compatibility with existing code.
Code suggestion
Check the AI-generated fix before applying
| pluginTrns.ObserveSuccess(tCtx.ow.GetOutputPath(), | |
| // TODO: Deprecated - deckURI parameter will be removed in future versions | |
| pluginTrns.ObserveSuccess(tCtx.ow.GetOutputPath(), nil, |
Code Review Run #ecb1b4
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
|
|
||
| metadata := template.GetMetadata() | ||
| if metadata == nil { | ||
| return DeckUnknown, nil |
There was a problem hiding this comment.
older versions of flytekit will return DeckUnknown, which ends up calling AddDeckURIIfDeckExists (which is the old code path where we would hit the blob store to check for the deck file before adding it to the events).
| @@ -88,6 +89,8 @@ message RuntimeMetadata { | |||
|
|
|||
| // Task Metadata | |||
| message TaskMetadata { | |||
| // Remove generates_deck from here. | |||
There was a problem hiding this comment.
Can you leave a comment explaining that we're reusing the name generates_deck for the field number 15, but since we're changing its type we're reserving the old field number?
| @@ -51,6 +51,7 @@ var childExecutionID = &core.WorkflowExecutionIdentifier{ | |||
| const dynamicWorkflowClosureRef = "s3://bucket/admin/metadata/workflow" | |||
|
|
|||
| const testInputURI = "fake://bucket/inputs.pb" | |||
| const DeckURI = "fake://bucket/deck.html" | |||
There was a problem hiding this comment.
Can you add a test here showing that the deck uri also shows up in the case of failed executions?
There was a problem hiding this comment.
| // This is for backward compatibility with older Flytekit versions. | ||
| // Older Flytekit versions did not set the `generates_deck` flag in the task template's metadata. | ||
| // So, we need to add deck URI to the event if it exists. | ||
| err = pluginTrns.AddDeckURIIfDeckExists(ctx, tCtx) |
There was a problem hiding this comment.
Shouldn't we always check if the file exists in the terminal state? if flytekit fails to generate a deck for some reasons, we should not add deck_uri to the output info, right
There was a problem hiding this comment.
What are the concerns with having a deck_uri set in the event? flyteconsole will still make the call to ensure that the file exists before showing the final deck, no?
There was a problem hiding this comment.
Hi, @eapolinario
Yes, but FlyteConsole currently needs to make an additional call to check if the task is in a terminal phase.
I think it's better to handle all the logic in Propeller, as this would make maintenance easier. It would also simplify FlyteConsole's implementation.
In summary:
We should keep as much backend logic in the backend as possible. This approach reduces the maintenance burden on FlyteConsole and improves the readability of the backend code.
Co-authored-by: Kevin Su <pingsutw@apache.org> Signed-off-by: Han-Ru Chen (Future-Outlier) <eric901201@gmail.com>
Co-authored-by: Kevin Su <pingsutw@apache.org> Signed-off-by: Han-Ru Chen (Future-Outlier) <eric901201@gmail.com>
Code Review Agent Run #9d753eActionable Suggestions - 2
Review Details
AI Code Review powered by |
| return DeckUnknown, regErrors.Wrapf(err, "failed to read task template") | ||
| } | ||
|
|
||
| deckValue := template.GetMetadata().GetGeneratesDeck() |
There was a problem hiding this comment.
Consider adding error handling for the case when GetMetadata() returns nil. Currently, if template.GetMetadata() returns nil, the code will panic when calling GetGeneratesDeck() on a nil pointer.
Code suggestion
Check the AI-generated fix before applying
| deckValue := template.GetMetadata().GetGeneratesDeck() | |
| metadata := template.GetMetadata() | |
| if metadata == nil { | |
| return DeckUnknown, nil | |
| } | |
| deckValue := metadata.GetGeneratesDeck() |
Code Review Run #9d753e
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| tags: _containers.ScalarMap[str, str] | ||
| pod_template_name: str | ||
| cache_ignore_input_vars: _containers.RepeatedScalarFieldContainer[str] | ||
| is_eager: bool | ||
| def __init__(self, discoverable: bool = ..., runtime: _Optional[_Union[RuntimeMetadata, _Mapping]] = ..., timeout: _Optional[_Union[_duration_pb2.Duration, _Mapping]] = ..., retries: _Optional[_Union[_literals_pb2.RetryStrategy, _Mapping]] = ..., discovery_version: _Optional[str] = ..., deprecated_error_message: _Optional[str] = ..., interruptible: bool = ..., cache_serializable: bool = ..., generates_deck: bool = ..., tags: _Optional[_Mapping[str, str]] = ..., pod_template_name: _Optional[str] = ..., cache_ignore_input_vars: _Optional[_Iterable[str]] = ..., is_eager: bool = ...) -> None: ... | ||
| generates_deck: _wrappers_pb2.BoolValue |
There was a problem hiding this comment.
Consider updating the type hint for generates_deck to be consistent between the class attribute and constructor parameter. The class attribute uses _wrappers_pb2.BoolValue while the constructor parameter uses _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]].
Code suggestion
Check the AI-generated fix before applying
| generates_deck: _wrappers_pb2.BoolValue | |
| generates_deck: _Optional[_Union[_wrappers_pb2.BoolValue, _Mapping]] |
Code Review Run #9d753e
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Signed-off-by: Eduardo Apolinario <eapolinario@users.noreply.github.com>
Code Review Agent Run #4a1c3eActionable Suggestions - 0Additional Suggestions - 1
Review Details
AI Code Review powered by |
Signed-off-by: Future-Outlier <eric901201@gmail.com>
Code Review Agent Run #daa4f1Actionable Suggestions - 0Review Details
AI Code Review powered by |
|
Hi, @eapolinario @pingsutw @wild-endeavor Now all of these cases will be considered.
|
Code Review Agent Run #dc455eActionable Suggestions - 5
Review Details
AI Code Review powered by |
| if (deckStatus == DeckUnknown || deckStatus == DeckEnabled) && pluginTrns.pInfo.Phase().IsTerminal() { | ||
| if err := pluginTrns.RemoveDeckURIIfDeckNotExists(ctx, tCtx); err != nil { | ||
| logger.Errorf(ctx, "Failed to remove deck URI if deck does not exist. Error: %v", err) | ||
| } | ||
| } | ||
| }() |
There was a problem hiding this comment.
Consider checking for errors from RemoveDeckURIIfDeckNotExists before proceeding with the task completion. The current implementation only logs the error but continues execution which could lead to inconsistent state.
Code suggestion
Check the AI-generated fix before applying
| if (deckStatus == DeckUnknown || deckStatus == DeckEnabled) && pluginTrns.pInfo.Phase().IsTerminal() { | |
| if err := pluginTrns.RemoveDeckURIIfDeckNotExists(ctx, tCtx); err != nil { | |
| logger.Errorf(ctx, "Failed to remove deck URI if deck does not exist. Error: %v", err) | |
| } | |
| } | |
| }() | |
| if (deckStatus == DeckUnknown || deckStatus == DeckEnabled) && pluginTrns.pInfo.Phase().IsTerminal() { | |
| if err := pluginTrns.RemoveDeckURIIfDeckNotExists(ctx, tCtx); err != nil { | |
| logger.Errorf(ctx, "Failed to remove deck URI if deck does not exist. Error: %v", err) | |
| // Return error to allow proper handling at higher levels | |
| return pluginTrns, err | |
| } | |
| } | |
| } | |
| }() |
Code Review Run #dc455e
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| defer func() { | ||
| if (deckStatus == DeckUnknown || deckStatus == DeckEnabled) && pluginTrns.pInfo.Phase().IsTerminal() { | ||
| if err := pluginTrns.RemoveDeckURIIfDeckNotExists(ctx, tCtx); err != nil { | ||
| logger.Errorf(ctx, "Failed to remove deck URI if deck does not exist. Error: %v", err) | ||
| } | ||
| } | ||
| }() |
There was a problem hiding this comment.
Consider moving the deck URI cleanup logic to a separate function for better code organization and reusability. The deferred function could be simplified by extracting the logic into a named function.
Code suggestion
Check the AI-generated fix before applying
| defer func() { | |
| if (deckStatus == DeckUnknown || deckStatus == DeckEnabled) && pluginTrns.pInfo.Phase().IsTerminal() { | |
| if err := pluginTrns.RemoveDeckURIIfDeckNotExists(ctx, tCtx); err != nil { | |
| logger.Errorf(ctx, "Failed to remove deck URI if deck does not exist. Error: %v", err) | |
| } | |
| } | |
| }() | |
| defer cleanupDeckURI(ctx, tCtx, deckStatus, pluginTrns) | |
| func cleanupDeckURI(ctx context.Context, tCtx *taskExecutionContext, deckStatus DeckStatus, pluginTrns *pluginRequestedTransition) { | |
| if (deckStatus == DeckUnknown || deckStatus == DeckEnabled) && pluginTrns.pInfo.Phase().IsTerminal() { | |
| if err := pluginTrns.RemoveDeckURIIfDeckNotExists(ctx, tCtx); err != nil { | |
| logger.Errorf(ctx, "Failed to remove deck URI if deck does not exist. Error: %v", err) | |
| } | |
| } | |
| } |
Code Review Run #dc455e
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| p.execInfo.OutputInfo.DeckURI = deckURI | ||
| } | ||
|
|
||
| func (p *pluginRequestedTransition) RemoveDeckURIIfDeckNotExists(ctx context.Context, tCtx *taskExecutionContext) error { |
There was a problem hiding this comment.
The function name RemoveDeckURIIfDeckNotExists seems to have a double negative and may be confusing. Consider renaming to ClearDeckURIIfMissing or similar for better clarity.
Code suggestion
Check the AI-generated fix before applying
| func (p *pluginRequestedTransition) RemoveDeckURIIfDeckNotExists(ctx context.Context, tCtx *taskExecutionContext) error { | |
| func (p *pluginRequestedTransition) ClearDeckURIIfMissing(ctx context.Context, tCtx *taskExecutionContext) error { |
Code Review Run #dc455e
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| exists, err := reader.DeckExists(ctx) | ||
| if err != nil { | ||
| logger.Errorf(ctx, "Failed to check deck file existence. Error: %v", err) | ||
| p.execInfo.OutputInfo.DeckURI = nil |
There was a problem hiding this comment.
Consider initializing OutputInfo if it's nil before attempting to set DeckURI to avoid potential nil pointer dereference in RemoveDeckURIIfDeckNotExists
Code suggestion
Check the AI-generated fix before applying
@@ -102,2 +102,5 @@
logger.Errorf(ctx, "Failed to check deck file existence. Error: %v", err)
+ if p.execInfo.OutputInfo == nil {
+ p.execInfo.OutputInfo = &handler.OutputInfo{}
+ }
p.execInfo.OutputInfo.DeckURI = nil
Code Review Run #dc455e
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
| error := &core.ExecutionError{ | ||
| Code: "foo", | ||
| } | ||
| request := admin.NodeExecutionEventRequest{ | ||
| Event: &event.NodeExecutionEvent{ | ||
| Phase: core.NodeExecution_FAILED, | ||
| OutputResult: &event.NodeExecutionEvent_Error{ | ||
| Error: error, |
There was a problem hiding this comment.
Consider using error as a variable name may shadow the built-in error type. Consider renaming to execError or similar.
Code suggestion
Check the AI-generated fix before applying
- error := &core.ExecutionError{
+ execError := &core.ExecutionError{
Code: "foo",
}
- Error: error,
+ Error: execError,
Code Review Run #dc455e
Is this a valid issue, or was it incorrectly flagged by the Agent?
- it was incorrectly flagged
Tracking issue
#5574
Why are the changes needed?
To enhance user visibility into Flyte Decks at different stages of workflow execution (running, failing, and succeeding), enabling better debugging and analysis.
Summary
What changes were proposed in this pull request?
2025/01/14 update
BoolValue, it can return 3 values, which can represent 3 statusnil -> DeckUnknown
true -> DeckEnabled
false -> DeckDisabled
Concept:
NodeExecutionEvent, and send it to admin.flyte/flytepropeller/pkg/controller/nodes/executor.go
Lines 1251 to 1261 in b3330ba
Life Cycle:
use new flytekit > 1.14.0
summary:
HEADrequest to be called. (save resource)details:
DeckURIwhen the task is running ifFLYTE_ENABLE_DECK=truein the task template.DeckURIto node info, and turn it toNodeExecutionEventto flyte admin.DeckURItoClosureflyte/flyteadmin/dataproxy/service.go
Lines 175 to 189 in b3330ba
node Closure, it will not show theFlyte Deckbutton.old flytekit <= 1.14.0
summary:
details:
HEADrequest to know if the Deck URI exists or not.if exist, then put it to the node info.
How was this patch tested?
python code:
Setup process
single binary.
flyte: this branch
flytekit: flyteorg/flytekit#2779
flyteconsole: flyteorg/flyteconsole#890
Screenshots
flytekit branch:
flyteorg/flytekit#2779
NEW FLYTEKIT, NO DECK, RUNNING With Deck, SUCCEED, and FAILED
OSS-STREAMING-DECK-small.mov
OLD FLYTEKIT, NO DECK, RUNNING With Deck, SUCCEED, and FAILED
OSS-STREAMING-DECK-OLD-FLYTEKIT-small.mov
Check all the applicable boxes
Related PRs
follow up questions
Abortphase for the streaming deck?should we support
EPhaseAbortin this file?https://github.com/flyteorg/flyte/blob/b3330ba4430538f91ae9fc7d868a29a2e96db8bd/flytepropeller/pkg/controller/nodes/handler/transition_info.go
Summary by Bito
This PR enhances Flyte Decks V2 streaming functionality with tri-state support through BoolValue and improved protocol buffer definitions. The changes implement refined deck URI handling with task condition-based status determination and enhanced terminal state management. Updates include renaming environment variable fields and improving secret handling mechanisms across TypeScript, Go, JavaScript, Python, and Rust implementations. The implementation maintains backward compatibility while optimizing workflow visibility and cluster assignment capabilities.Unit tests added: True
Estimated effort to review (1-5, lower is better): 5