Change phase to queue on job submit for webapi plugins

[pingsutw]

Tracking issue

#3936

Why are the changes needed?

log links are not shown up on FlyteConsole when the task status is running.
It will also fix other webAPI plugins.

propeller only sends the event when the current event is not equal to the previous event, so if ResourceCache sets the phase to running here without info, the log link will never show up on flyteconsole when the status is running.

What changes were proposed in this pull request?

The phase should be PhaseQueue when there is no resource in the cacheItem.

The async ResourceCache is empty if the Get request is not yet sent to the agent.

How was this patch tested?

Local

Setup process

Screenshots

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

Related PRs

NA

Docs link

NA

[codecov]

Codecov Report

Attention: Patch coverage is 0% with 1 lines in your changes are missing coverage. Please review.

Project coverage is 59.08%. Comparing base (44c701e) to head (6bedcf1).

Files	Patch %	Lines
...o/tasks/pluginmachinery/internal/webapi/monitor.go	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #5188      +/-   ##
==========================================
+ Coverage   59.07%   59.08%   +0.01%     
==========================================
  Files         646      646              
  Lines       55739    55739              
==========================================
+ Hits        32928    32935       +7     
+ Misses      20215    20208       -7     
  Partials     2596     2596              

Flag	Coverage Δ
unittests	59.08% <0.00%> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[Future-Outlier]

It works, and it can also show the message now.

return core.PhaseInfoRunningWithReason(core.DefaultPhaseVersion, resource.Message, taskInfo), nil

(I'm not sure what's happened to the error in the screenshot.)

However, this is still great, I've tested it in the sensor agent.

return Resource(phase=TaskExecution.RUNNING, outputs=None, log_links=[TaskLog(uri="sensor_test_uri", name="Sensor Console").to_flyte_idl()])

[Future-Outlier]

This is my git diff to enhance this PR.

diff --git a/flyteplugins/go/tasks/pluginmachinery/core/phase.go b/flyteplugins/go/tasks/pluginmachinery/core/phase.go
index 6c80cc4d2..97ea98d41 100644
--- a/flyteplugins/go/tasks/pluginmachinery/core/phase.go
+++ b/flyteplugins/go/tasks/pluginmachinery/core/phase.go
@@ -256,6 +256,12 @@ func PhaseInfoRunning(version uint32, info *TaskInfo) PhaseInfo {
        return phaseInfo(PhaseRunning, version, nil, info, false)
 }
 
+func PhaseInfoRunningWithReason(version uint32, reason string, info *TaskInfo) PhaseInfo {
+       pi := phaseInfo(PhaseRunning, version, nil, info, false)
+       pi.reason = reason
+       return pi
+}
+
 func PhaseInfoSuccess(info *TaskInfo) PhaseInfo {
        return phaseInfo(PhaseSuccess, DefaultPhaseVersion, nil, info, false)
 }
diff --git a/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go b/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go
index cc7f15bd8..a8ba8513f 100644
--- a/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go
+++ b/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go
@@ -249,7 +249,7 @@ func (p Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phase
        case flyteIdl.TaskExecution_INITIALIZING:
                return core.PhaseInfoInitializing(time.Now(), core.DefaultPhaseVersion, resource.Message, taskInfo), nil
        case flyteIdl.TaskExecution_RUNNING:
-               return core.PhaseInfoRunning(core.DefaultPhaseVersion, taskInfo), nil
+               return core.PhaseInfoRunningWithReason(core.DefaultPhaseVersion, resource.Message, taskInfo), nil
        case flyteIdl.TaskExecution_SUCCEEDED:
                err = writeOutput(ctx, taskCtx, resource.Outputs)
                if err != nil {
:...skipping...
diff --git a/flyteplugins/go/tasks/pluginmachinery/core/phase.go b/flyteplugins/go/tasks/pluginmachinery/core/phase.go
index 6c80cc4d2..97ea98d41 100644
--- a/flyteplugins/go/tasks/pluginmachinery/core/phase.go
+++ b/flyteplugins/go/tasks/pluginmachinery/core/phase.go
@@ -256,6 +256,12 @@ func PhaseInfoRunning(version uint32, info *TaskInfo) PhaseInfo {
        return phaseInfo(PhaseRunning, version, nil, info, false)
 }
 
+func PhaseInfoRunningWithReason(version uint32, reason string, info *TaskInfo) PhaseInfo {
+       pi := phaseInfo(PhaseRunning, version, nil, info, false)
+       pi.reason = reason
+       return pi
+}
+
 func PhaseInfoSuccess(info *TaskInfo) PhaseInfo {
        return phaseInfo(PhaseSuccess, DefaultPhaseVersion, nil, info, false)
 }
diff --git a/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go b/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go
index cc7f15bd8..a8ba8513f 100644
--- a/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go
+++ b/flyteplugins/go/tasks/plugins/webapi/agent/plugin.go
@@ -249,7 +249,7 @@ func (p Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phase
        case flyteIdl.TaskExecution_INITIALIZING:
                return core.PhaseInfoInitializing(time.Now(), core.DefaultPhaseVersion, resource.Message, taskInfo), nil
        case flyteIdl.TaskExecution_RUNNING:
-               return core.PhaseInfoRunning(core.DefaultPhaseVersion, taskInfo), nil
+               return core.PhaseInfoRunningWithReason(core.DefaultPhaseVersion, resource.Message, taskInfo), nil
        case flyteIdl.TaskExecution_SUCCEEDED:
                err = writeOutput(ctx, taskCtx, resource.Outputs)
                if err != nil {
~
~
~
~
~
~

[Future-Outlier]

I'm also wondering that cases for databricks agent,
it will return phase=TaskExecution.INITIALIZING with loglinks when running agents.
If we implement this change, it will not show loglinks when phase=TaskExecution.INITIALIZING.

[Future-Outlier]

I'm also wondering that cases for databricks agent, it will return phase=TaskExecution.INITIALIZING with loglinks when running agents. If we implement this change, it will not show loglinks when phase=TaskExecution.INITIALIZING.

One possible solution is that we change the code in flytekit, and this will solve the problem.

[pingsutw]

I changed the initial phase to queue; it will fix the Databricks issue as well.