Channels
#ask-the-community
Title
# ask-the-community
v
Victor Gustavo da Silva Oliveira
05/02/2023, 2:38 PMHello 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 PMAre you setting resources in the
@task(task_config=...)d
Dan Rammer (hamersaw)
05/02/2023, 2:41 PMHey 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 PMThanks, 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 PMAh, 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 PMMay 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...
d
Dan Rammer (hamersaw)
05/02/2023, 3:02 PMThese 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 PMCool. I'm talking with Jaine to send you the code snippet ASAP
Copy code
@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:d
Dan Rammer (hamersaw)
05/02/2023, 3:09 PMthanks! 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_resourcesv
Victor Gustavo da Silva Oliveira
05/02/2023, 4:06 PMYup
We have an extended_task decorator, with gcloud integration
d
Dan Rammer (hamersaw)
05/02/2023, 4:13 PMCan you say a little more? Extending the flytekit
@taskv
Victor Gustavo da Silva Oliveira
05/02/2023, 4:22 PMYup
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
Copy code
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 Copy code
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:
...d
Dan Rammer (hamersaw)
05/02/2023, 6:56 PMit 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 PMNope
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 PMthe
...n3-0-0n3-0-1skusn0modelsn1preparedn2v
Victor Gustavo da Silva Oliveira
05/02/2023, 7:24 PMAll 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 PMAs 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.
Copy code
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_successesv
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 PMI'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 PMsounds great, thanks! just want to make sure we address this.
a
Ariel Kaspit
07/20/2023, 2:49 PM@Nizar Hattab @Eyal Abbas
3 Views