WebAPI plugins optimization

[pingsutw]

Tracking issue

NA

Why are the changes needed?

There are some issues in the webapi plugin:

• Push duplicate items into the workqueue since we add obj into the workqueue instead of itemID
• It doesn't correctly handle stale CRD. Propeller transitions the status to Queue after the task succeeds since the plugin returns PhaseInfoQueued when an item is being deleted from the cache.
• It pushed the item to the workqueue after the task was completed because it only marked the item as complete when evaluating the node. Therefore, in the below case, the propeller will continue to send the Get request to the external system.

Sync Cycle	Async Cache	Propeller
1		Add Item
2	plugin.get()
3	plugin.get()
4	plugin.get()
5	plugin.get()
6		Update the phase in the item
7	Remove the item

What changes were proposed in this pull request?

• Add processing set to prevent handling the same item from different workers
• Add IsTerminal to the webapi.Resource for early termination of the processed item.
• Add PhaseVersion to cache.state. If the cache is missed, propeller will return the previous phase and version and wait for the item to be processed by the async cache.
• Fixes cold start issue in the AutoRefreshCache by pushing the item to the workqueue when it's being added to the lru cache.

How was this patch tested?

Local/remote cluster

Setup process

from flytekit import workflow, task, LaunchPlan
from flytekitplugins.mock_agent import Sleep


@task(task_config=Sleep(duration=2))
def sleep_task() -> str:
    return "Hello World!"


@workflow()
def sleep_wf():
    for i in range(100):
        sleep_task()


@workflow
def load_test_wf():
    sleep_lp = LaunchPlan.get_or_create(name="fixed_inputs", workflow=sleep_wf, max_parallelism=25)
    for i in range(20):
        sleep_lp()


if __name__ == "__main__":
    sleep_task()

Screenshots

Before:
~5 mins / per workflow

~30s for a single task

After:
~3 mins / per workflow

~15s for a single task

Check all the applicable boxes

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

Related PRs

NA

Docs link

NA

[EngHabu]

nit:

Suggested change

	Processing: newSyncSet(),
	processing: newSyncSet(),

[EngHabu]

Add TIMED_OUT too?

[pingsutw]

there is no timeout.

flyte/flyteidl/protos/flyteidl/core/execution.proto

Lines 44 to 56 in 97a79c0

	message TaskExecution{
	enum Phase {
	UNDEFINED = 0;
	QUEUED = 1;
	RUNNING = 2;
	SUCCEEDED = 3;
	ABORTED = 4;
	FAILED = 5;
	// To indicate cases where task is initializing, like: ErrImagePull, ContainerCreating, PodInitializing
	INITIALIZING = 6;
	// To address cases, where underlying resource is not available: Backoff error, Resource quota exceeded
	WAITING_FOR_RESOURCES = 7;
	}

[EngHabu]

Why 2?

[pingsutw]

It was 2.

flyte/flyteplugins/go/tasks/pluginmachinery/internal/webapi/launcher.go

Line 52 in e8588f3

return state, core.PhaseInfoQueued(time.Now(), 2, "launched"), nil

because we set it to 0 when we allocate a token.

flyte/flyteplugins/go/tasks/pluginmachinery/internal/webapi/allocation_token.go

Line 54 in e8588f3

}, core.PhaseInfoQueued(a.clock.Now(), 0, "Allocation token required"), nil