Offload trigger inputs at registration

[katrogan]

And read kickoff time input from the run context, using the newly added run_start_time from #1100

Depends on flyteorg/flyte#7482

Testing
[x] regular trigger (TaskTriggerSpec) Ran https://github.com/flyteorg/flyte-sdk/blob/main/examples/triggers/basic.py and verified output included valid kickoff time
[x] deployed trigger (TriggerSpec) Using

import flyte
from flyte.remote import Trigger

CONFIG = ".flyte/config-oss-local.yaml"
TASK_NAME = "cached_example_task.cached_task"

TRIGGER_NAME = "standalone_minutely"  # distinct from the embedded trigger names

flyte.init_from_config(CONFIG)

Trigger.create(
    trigger=flyte.Trigger.minutely("start_time", name=TRIGGER_NAME),
    task_name=TASK_NAME,
)

print(f"Created standalone trigger '{TRIGGER_NAME}' on task '{TASK_NAME}'.")

[x] cached trigger (TaskTriggerSpec) Using this script

import flyte

env = flyte.TaskEnvironment(
    name="cached_example_task",
    env_vars={"_U_USE_ACTIONS": "1"},
)


@env.task(
    cache=flyte.Cache(behavior="auto", ignored_inputs="start_time"),
    triggers=flyte.Trigger.minutely("start_time", name="minutely_cached"),
)
def cached_task(start_time: datetime, x: int = 1) -> str:
    return f"Cached task executed at {start_time.isoformat()} with x={x}"


if __name__ == "__main__":
    flyte.init_from_config()
    flyte.deploy(env)

And verified subsequent runs used cached output from the first run.

[x] run time always injected: Using this script

import flyte

env = flyte.TaskEnvironment(
    name="hello_run_start_time",
)


@env.task()
async def hello_run_start_time() -> str:
    ctx = flyte.ctx()
    assert ctx is not None
    # run_start_time is populated by the backend from the run's start time (for scheduled-trigger
    # runs, the trigger's fire time) via the {{.runStartTime}} container arg. For ad-hoc runs it
    # defaults to the run-creation time.
    run_start_time = ctx.run_start_time
    print(f"Run start time: {run_start_time.isoformat()}")
    return run_start_time.isoformat()


if __name__ == "__main__":
    flyte.init_from_config()

    run = flyte.run(hello_run_start_time)
    print(run.name)
    print(run.url)
    run.wait()

Verified individual runs output the kickoff time

ref eng26-533

[katrogan]

will update to an official release version before merge

[katrogan]

I'll revert all these examples before merge

[pingsutw]

LGTM

[kumare3]

can we not do it here and instead do it at -

flyte-sdk/src/flyte/_internal/runtime/convert.py

Line 252 in e932b24

literals = {named_literal.name: named_literal.value for named_literal in inputs.proto_inputs.literals}

you have it in the ctx already so you can just get it from the ctx internal_ctx or task context

[katrogan]

thanks, done

[kumare3]

remove this part, we dont need to reopen the inputs

[katrogan]

done

[kumare3]

only if we have data?

[katrogan]

done