@ancient-wolf-19325 how to override the image of pod spec with user input?

You can override the container image of a Flyte task at runtime using the

.with_overrides(container_image=...)

method when invoking the task in a workflow. This allows you to set the image dynamically, including from user input, as long as you pass the desired image string to the override. Note that this works reliably if both your Flyte backend and flytekit are version 1.11.0 or newer; otherwise, the override may not be respected at execution time (Flyte 1.15.0 release notes, Task node overrides not handled in flyteidl are broken). Example usage:

@task
def my_task():
    ...

@workflow
def wf(user_image: str):
    my_task().with_overrides(container_image=user_image)

This will run

my_task

using the image specified by the

user_image

input at runtime. Would you like more detail or a more advanced example? Sources: - Task node overrides not handled in flyteidl are broken - Flyte 1.15.0 release notes 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

is it possible to provide container_image as optional parameter? @ancient-wolf-19325

Yes, you can make the container_image parameter optional by setting a default value (such as None) in your workflow or task function, and only applying the override if a value is provided. If container_image is not specified, Flyte will use the default image configured for the task (flytekit task decorator docs, Multiple images in a workflow). Example:

@task
def my_task():
    ...

@workflow
def wf(container_image: str = None):
    if container_image:
        my_task().with_overrides(container_image=container_image)
    else:
        my_task()

Would you like more detail or a more complex example? Sources: - flytekit task decorator docs - Multiple images in a workflow 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@task
def my_task():
    ...

@workflow
def wf(user_image: str):
    my_task().with_overrides(container_image=user_image)

provide this fully in podspec format @ancient-wolf-19325

To override the image in a Kubernetes PodSpec for a Flyte task, use the

pod_template

parameter with a custom

PodTemplate

object, and set the image in the

V1Container

. You can make the image dynamic by constructing the

PodTemplate

at runtime using the user input (Pod templates documentation). Example:

from kubernetes.client import V1Container, V1PodSpec
import flyte

@flyte.task
async def my_task():
    ...

@flyte.workflow
async def wf(user_image: str):
    pod_template = flyte.PodTemplate(
        primary_container_name="primary",
        pod_spec=V1PodSpec(
            containers=[V1Container(name="primary", image=user_image)]
        ),
    )
    env = flyte.TaskEnvironment(name="my_env", pod_template=pod_template)
    await my_task.override(env=env)()

This sets the primary container's image in the pod spec to the value of

user_image

. Would you like a step-by-step explanation or more advanced usage? Sources: - Pod templates documentation 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

can I place a default image if the user not provided image as a input? @ancient-wolf-19325

Yes, you can set a default image in the PodSpec by using a conditional: if the user does not provide an image, use your default image string. This ensures the pod will always have an image, either user-supplied or default (Pod templates documentation). Example:

from kubernetes.client import V1Container, V1PodSpec
import flyte

@flyte.task
async def my_task():
    ...

@flyte.workflow
async def wf(user_image: str = None):
    image = user_image or "docker.io/myorg/default-image:latest"
    pod_template = flyte.PodTemplate(
        primary_container_name="primary",
        pod_spec=V1PodSpec(
            containers=[V1Container(name="primary", image=image)]
        ),
    )
    env = flyte.TaskEnvironment(name="my_env", pod_template=pod_template)
    await my_task.override(env=env)()

Would you like more detail or a breakdown of this pattern? Sources: - https://github.com/flyteorg/flyte/issues/6387 - https://github.com/flyteorg/flytekit/blob/master/flytekit/core/task.py 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

def dynamic_pod_spec_recompute(claim_name: str, execution_id: str) -> V1PodSpec: """Dynamic_pod_spec_recompute function is used to set the dynamic pvc claim_name using with_overrides function.""" pod_spec = V1PodSpec( image_pull_secrets=[V1LocalObjectReference(name="te13535-artifactory-cred")], node_selector=RecomputeConfig.NODE_SELECTOR, host_ipc=True, restart_policy="OnFailure", active_deadline_seconds=RecomputeConfig.GKE_POD_ACTIVE_DEADLINE_SECONDS, service_account=RecomputeConfig.SERVICE_ACCOUNT, service_account_name=RecomputeConfig.SERVICE_ACCOUNT, containers=[ V1Container( name="primary", image_pull_policy="Always", ), V1Container( name="simple1", image="simple:v1", security_context=V1SecurityContext(privileged=True), env=[ V1EnvVar(name="PYTHONUNBUFFERED", value="1"), V1EnvVar( name="TEST_SUITES", value_from=V1EnvVarSource( config_map_key_ref=V1ConfigMapKeySelector( name="recompute-config-v4", key="TEST_SUITES" ) ) ), ], image_pull_policy="Always", command=["/bin/bash"], args=[ "-c", """ ls """ ], volume_mounts=[ V1VolumeMount( name="gcs-volume", mount_path="/home/julia/shared_memory_app/logs", sub_path="logs" ) ] ), can you update this accordingly @ancient-wolf-19325

It seems likely you want to update your

dynamic_pod_spec_recompute

function so that the primary container’s image can be set dynamically (e.g., via user input or with_overrides), and a default image is used if none is provided. You can add an

image

parameter (defaulting to your preferred image) and set it in the

V1Container

for "primary". The rest of your pod spec can remain unchanged (Pod templates documentation). Example:

from kubernetes.client import (
    V1PodSpec, V1Container, V1LocalObjectReference, V1SecurityContext,
    V1EnvVar, V1EnvVarSource, V1ConfigMapKeySelector, V1VolumeMount
)

def dynamic_pod_spec_recompute(
    claim_name: str,
    execution_id: str,
    image: str = "docker.io/myorg/default-image:latest"
) -> V1PodSpec:
    pod_spec = V1PodSpec(
        image_pull_secrets=[V1LocalObjectReference(name="te13535-artifactory-cred")],
        node_selector=RecomputeConfig.NODE_SELECTOR,
        host_ipc=True,
        restart_policy="OnFailure",
        active_deadline_seconds=RecomputeConfig.GKE_POD_ACTIVE_DEADLINE_SECONDS,
        service_account=RecomputeConfig.SERVICE_ACCOUNT,
        service_account_name=RecomputeConfig.SERVICE_ACCOUNT,
        containers=[
            V1Container(
                name="primary",
                image=image,
                image_pull_policy="Always",
            ),
            V1Container(
                name="simple1",
                image="simple:v1",
                security_context=V1SecurityContext(privileged=True),
                env=[
                    V1EnvVar(name="PYTHONUNBUFFERED", value="1"),
                    V1EnvVar(
                        name="TEST_SUITES",
                        value_from=V1EnvVarSource(
                            config_map_key_ref=V1ConfigMapKeySelector(
                                name="recompute-config-v4",
                                key="TEST_SUITES"
                            )
                        )
                    ),
                ],
                image_pull_policy="Always",
                command=["/bin/bash"],
                args=[
                    "-c",
                    "ls"
                ],
                volume_mounts=[
                    V1VolumeMount(
                        name="gcs-volume",
                        mount_path="/home/julia/shared_memory_app/logs",
                        sub_path="logs"
                    )
                ]
            ),
        ]
    )
    return pod_spec

You can now pass a user-supplied image, or let it default. Would you like more detail or an example of using this with Flyte’s

with_overrides

? Sources: - https://github.com/flyteorg/flyte/blob/master/CHANGELOG/CHANGELOG-v1.15.0.md - https://github.com/flyteorg/flyte/issues/6463 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

I want to update with this image image="simple:v1", I want some default image if user provides any image it should take new image provided by user other wise take it as default image is it possible? @ancient-wolf-19325

Yes, this is possible. You can add an image parameter to your function, set its default to your desired default image (e.g., "simple:v1"), and use the user-provided image if given. In your pod spec, set the primary container's image to this parameter. If the user does not provide an image, the default will be used (Pod templates documentation). Example:

def dynamic_pod_spec_recompute(
    claim_name: str,
    execution_id: str,
    image: str = "simple:v1"
) -> V1PodSpec:
    pod_spec = V1PodSpec(
        # ... other fields ...
        containers=[
            V1Container(
                name="primary",
                image=image,
                image_pull_policy="Always",
            ),
            # ... other containers ...
        ]
    )
    return pod_spec

Now, if the user provides an image, it will be used; otherwise, "simple:v1" is the default. Would you like more detail or an example of how to pass the image dynamically? Sources: - Pod templates documentation 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

not for primary i want to provide it for simple1 container @ancient-wolf-19325

Yes, you can make the image for the "simple1" container dynamic by adding an image parameter (defaulting to "simple:v1") and setting it in the "simple1" container's image field. If the user provides an image, it will be used; otherwise, the default is used (Pod templates documentation). Example:

def dynamic_pod_spec_recompute(
    claim_name: str,
    execution_id: str,
    simple1_image: str = "simple:v1"
) -> V1PodSpec:
    pod_spec = V1PodSpec(
        # ... other fields ...
        containers=[
            V1Container(
                name="primary",
                image_pull_policy="Always",
            ),
            V1Container(
                name="simple1",
                image=simple1_image,
                image_pull_policy="Always",
                # ... other fields ...
            ),
        ]
    )
    return pod_spec

Now, pass

simple1_image

as a parameter to override the image for "simple1". Would you like a full example with all fields included? Sources: - Pod templates documentation 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

yes, provide full example @ancient-wolf-19325

It could be possible that you want a full example where the "simple1" container image is set dynamically based on user input, defaulting to "simple:v1" if not provided. Though I'm not entirely certain, it might be that you want to see how this fits into a Flyte workflow using a PodTemplate and TaskEnvironment. Here’s a complete example based on the Pod templates documentation:

from kubernetes.client import (
    V1PodSpec, V1Container, V1LocalObjectReference, V1SecurityContext,
    V1EnvVar, V1EnvVarSource, V1ConfigMapKeySelector, V1VolumeMount
)
import flyte

def dynamic_pod_spec_recompute(
    claim_name: str,
    execution_id: str,
    simple1_image: str = "simple:v1"
) -> V1PodSpec:
    pod_spec = V1PodSpec(
        image_pull_secrets=[V1LocalObjectReference(name="te13535-artifactory-cred")],
        node_selector={"example": "selector"},
        host_ipc=True,
        restart_policy="OnFailure",
        active_deadline_seconds=3600,
        service_account="default",
        service_account_name="default",
        containers=[
            V1Container(
                name="primary",
                image_pull_policy="Always",
            ),
            V1Container(
                name="simple1",
                image=simple1_image,
                security_context=V1SecurityContext(privileged=True),
                env=[
                    V1EnvVar(name="PYTHONUNBUFFERED", value="1"),
                    V1EnvVar(
                        name="TEST_SUITES",
                        value_from=V1EnvVarSource(
                            config_map_key_ref=V1ConfigMapKeySelector(
                                name="recompute-config-v4",
                                key="TEST_SUITES"
                            )
                        )
                    ),
                ],
                image_pull_policy="Always",
                command=["/bin/bash"],
                args=[
                    "-c",
                    "ls"
                ],
                volume_mounts=[
                    V1VolumeMount(
                        name="gcs-volume",
                        mount_path="/home/julia/shared_memory_app/logs",
                        sub_path="logs"
                    )
                ]
            ),
        ]
    )
    return pod_spec

@flyte.task
async def my_task():
    ...

@flyte.workflow
async def wf(user_image: str = None):
    pod_spec = dynamic_pod_spec_recompute(
        claim_name="my-claim",
        execution_id="my-exec",
        simple1_image=user_image or "simple:v1"
    )
    pod_template = flyte.PodTemplate(
        primary_container_name="primary",
        pod_spec=pod_spec,
    )
    env = flyte.TaskEnvironment(name="my_env", pod_template=pod_template)
    await my_task.override(env=env)()