https://flyte.org logo
Powered by
Title
v

Victor Gustavo da Silva Oliveira

05/02/2023, 2:38 PM
Hello guys We are trying to use a map task in a workflow, but no matter how we increase the pod memory, it keeps giving OOMKilled with error 137... By looking at the pods creation, in a successfull execution (with less inputs), we can verify that the resources setted on the task are not the same that are requested in GKE... The image shows the pods resources, when we setted the task with 6Gi memory and 2CPU...
n

Nicholas Roberson

05/02/2023, 2:40 PM
Are you setting resources in the 
@task(task_config=...)
?
d

Dan Rammer (hamersaw)

05/02/2023, 2:41 PM
Hey Victor, what version of Flyte are you running? Do you have a minimal repro for code here? Would be useful to test locally. It seems like this could be a bug.
v

Victor Gustavo da Silva Oliveira

05/02/2023, 2:45 PM
@Nicholas Roberson not in task_config, but in limits @Dan Rammer (hamersaw) yes, @Jaine Rannow Budke can you help with this?
d

Dan Rammer (hamersaw)

05/02/2023, 2:50 PM
Thanks, this is a high priority bug - once we can repro I will have a fix out in the next day or so.
v

Victor Gustavo da Silva Oliveira

05/02/2023, 2:54 PM
Ah, about the version, we are using flyte 1.3.0
k

Ketan (kumare3)

05/02/2023, 2:56 PM
@Dan Rammer (hamersaw) is this similar to what @Nicholas LoFaso is seeing that pods land on the same node
d

Dan Rammer (hamersaw)

05/02/2023, 2:57 PM
@Ketan (kumare3) that's my concern - this is now my top priority today
k

Ketan (kumare3)

05/02/2023, 2:57 PM
Ya
v

Victor Gustavo da Silva Oliveira

05/02/2023, 2:57 PM
May be this could be related to the problem I've mentioned before about dynamic workflows?
Somehow, since it's about pods on the same node...
https://flyte-org.slack.com/archives/CP2HDHKE1/p1682535159838099 This one
d

Dan Rammer (hamersaw)

05/02/2023, 3:02 PM
These could be related, but seem to be separate issues. I'm going to start diving into maptasks and see what I can find.
v

Victor Gustavo da Silva Oliveira

05/02/2023, 3:04 PM
Cool. I'm talking with Jaine to send you the code snippet ASAP
@workflow
def image_workflow(inputs..) -> None:

    skus = get_sku_list(...)

    models = load_model()

    prepared = prepare_map_inputs(...)

    result = map_task(
        process_images,
        metadata=TaskMetadata(retries=1),
        concurrency=2
    )(input=prepared)


@task(requests=Resources(mem='2Gi'))
def prepare_map_inputs(
    skus: List[str],
    ...
    ) -> List[MapInput]:
    
    
@task(requests=Resources(cpu='2', mem='6Gi'))
def process_images(
    input: MapInput
    ) -> bool:
This is a snippet, we had to ommit some info...
d

Dan Rammer (hamersaw)

05/02/2023, 3:09 PM
thanks! an no problem on ommiting, it was expected 😄
I'm having difficulty reproducing this in 1.3.0. Do you have any unique configuration in flyteadmin? For example the 
task_resources
configuration option (sets default resource request) or any matchable attributes for example task level or workflow level, etc
v

Victor Gustavo da Silva Oliveira

05/02/2023, 4:06 PM
Yup We have an extended_task decorator, with gcloud integration
d

Dan Rammer (hamersaw)

05/02/2023, 4:13 PM
Can you say a little more? Extending the flytekit 
@task
?
v

Victor Gustavo da Silva Oliveira

05/02/2023, 4:22 PM
Yup This extension just creates a new gcloud credentials integration, with the authentication needed for the execution
k

Kevin Su

05/02/2023, 6:49 PM
@Victor Gustavo da Silva Oliveira just to confirm, did you pass the task resource to the parent class. like 
class GcloudFunctionTask(PythonFunctionTask[gcloud]):    
    def __init__(self, task_config: gcloud, task_function: Callable, **kwargs):
        super(PysparkFunctionTask, self).__init__(
            task_config=task_config,
            task_type="gcloud",
            task_function=task_function,
            **kwargs,  # resource limit is in it
        )
v

Victor Gustavo da Silva Oliveira

05/02/2023, 6:55 PM
def extended_task(
    function: Callable = None, *, integrations: List[str] = None, **task_kwargs
) -> Callable:
    if function is None:
        return functools.partial(
            extended_task, integrations=integrations, **task_kwargs
        )

    if integrations is not None and not isinstance(integrations, list):
        raise ValueError("'integrations' parameter must be a list.")

    task_integrations = TaskIntegrations(integrations)

    if task_integrations.has_requests:
        task_kwargs["task_config"] = generate_task_config(task_integrations)

    @task(**task_kwargs)
    @functools.wraps(function)
    def wrapper(*args, **kwargs) -> Any:
        ...
It's like this
d

Dan Rammer (hamersaw)

05/02/2023, 6:56 PM
it looks like the mem is applied but no the cpu right? the tasks got 6G Mem just not 2CPU?
v

Victor Gustavo da Silva Oliveira

05/02/2023, 7:01 PM
Nope Neither is working If you look in the image, it is creating the pods with 2Gi, but in the requests we've setted 6Gi
d

Dan Rammer (hamersaw)

05/02/2023, 7:04 PM
the 
...n3-0-0
and 
n3-0-1
are the maptask Pods right? 
skus
is 
n0
, 
models
is 
n1
, and 
prepared
is 
n2
from your example.
v

Victor Gustavo da Silva Oliveira

05/02/2023, 7:24 PM
All these pods are supposed to be the map task
Sorry, you are right
The n3 pods are the maptask
n

Nicholas LoFaso

05/03/2023, 3:56 PM
@Dan Rammer (hamersaw) Let me know if I can be of assistance here. At a glance this doesn’t appear to be the same issue I was experiencing, but both are happening on GKE with map tasks so maybe they are related
k

Ketan (kumare3)

05/03/2023, 3:58 PM
@Nicholas LoFaso i would love to understand whats happening
i feel something is wrong in the setup
it could be that you are running out of disk space to start container or something
n

Nicholas LoFaso

05/03/2023, 3:59 PM
As far as I can tell the container starts, but then is evicted (preepmted) by k8s to make room for a “calico” pod which is a GKE system pod
We are setting CPU/Mem to have equal request/limit so k8s should know exactly how many to schedule on a given node. Not sure why it is over scheduling when using map_task but not dyanimc. I confirmed with Dan that I was setting concurrency to 25 for both dynamic and map_task so should be equivalent load to k8s API.
k

Ketan (kumare3)

05/03/2023, 9:03 PM
Cc @Dan Rammer (hamersaw)
k

Kevin Su

05/03/2023, 9:16 PM
@Victor Gustavo da Silva Oliveira I’ve tested it with different version of propeller, still can’t reproduce it. could you run a map task that maps over regular python task, and see if it can properly override the resource. just want to make sure the extended_task doesn’t have anything wrong or system config.
you could run this example. 
from typing import List, Optional
from flytekit import map_task, task, workflow, Resources, TaskMetadata


@task(requests=Resources(cpu="1", mem="800Mi"))
def my_task(should_succeed: bool) -> Optional[bool]:
    if should_succeed:
        return should_succeed
    raise ValueError("This is a failure")


@task(requests=Resources(cpu="1", mem="800Mi"))
def get_num_successes_task(my_task_results: List[Optional[bool]]) -> int:
    return len(my_task_results)


@workflow
def wf(inputs: List[bool] = [False, True, False]) -> int:
    my_task_results = map_task(
        my_task,
        metadata=TaskMetadata(retries=1),
        min_success_ratio=0.5,
        concurrency=2,
    )(should_succeed=inputs)
    num_successes = get_num_successes_task(my_task_results=my_task_results)
    return num_successes
v

Victor Gustavo da Silva Oliveira

05/04/2023, 5:11 PM
@Kevin Su ok! When I've done testing I come back here!
d

Dan Rammer (hamersaw)

05/11/2023, 1:27 PM
@Victor Gustavo da Silva Oliveira were you able to get this solved? Is the example above working?
v

Victor Gustavo da Silva Oliveira

05/11/2023, 1:54 PM
I'm trying to update Flyte to 1.5.4, but we having some issues. When we finish this task, I send you a heads up, is that all right?
d

Dan Rammer (hamersaw)

05/11/2023, 1:55 PM
sounds great, thanks! just want to make sure we address this.
2 Views
#ask-the-communityJoin Slack
Powered by