Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Additional job check #37

Merged
merged 8 commits into from
Dec 5, 2023
Merged

Additional job check #37

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9a3edf2
Add additional waiting
Nov 21, 2023
c38520b
fix mock of kube api
Nov 21, 2023
72275a0
add util method to check child task
Nov 21, 2023
eb30b12
fix exception
Nov 21, 2023
0ab310b
Merge remote-tracking branch 'origin/main' into feature/additional_jo…
Dec 1, 2023
d352a31
Remove uncalled assert
Dec 1, 2023
a71874e
Add type annotation
Dec 4, 2023
76b7c4e
fix
Dec 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 18 additions & 6 deletions kannon/master.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,16 +49,17 @@ def build(self, root_task: gokart.TaskOnKart) -> None:
task_queue = self._create_task_queue(root_task)

# consume task queue
launched_task_ids: Set[str] = set()
maronuu marked this conversation as resolved.
Show resolved Hide resolved
logger.info("Consuming task queue...")
while task_queue:
task = task_queue.popleft()
if task.complete():
logger.info(f"Task {self._gen_task_info(task)} is already completed.")
continue
if task.make_unique_id() in launched_task_ids:
if task.make_unique_id() in self.task_id_to_job_name:
# check if task is still running on child job
assert self._check_child_task_status(task), f"Child task {self._gen_task_info(task)} failed."
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would be better to keep the master node running while the child nodes are running. If we add ownerReference to child node, the running child nodes will be killed by Kubernetes. #38

What do you think?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kitagry
Sorry for late reply!

You mean we should not kill the master node while one or more child nodes are running, right?

In the current implementation, the master node fails right after finding a child node failed, even when other child nodes are still living.
And, in my opinion, when it finds out that a child node fails, it means a failure of the whole pipeline, so I would prefer the current one.

However, as you pointed out, there might be a need for the strategy where we do not stop the master node and all living child nodes keep running even other child nodes fail.
So, I'll create an issue about it, and prefer keeping the current implementation for now.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@maronuu @kitagry I agree with @mamo3gr

I think it's ok to kill the master (and other children with !38 ) for now.

I'm not sure how it's problematic to kill them because they may have gokart caches and easy to retry.

Let's add other option if we confirm any problem.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This assert will not be called, because this method raise error if job is not succeed.

Copy link
Collaborator Author

@maronuu maronuu Dec 4, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, now fixed!

logger.info(f"Task {self._gen_task_info(task)} is still running on child job.")
task_queue.append(task)
task_queue.append(task) # re-enqueue task to check if it is done
continue

# TODO: enable user to specify duration to sleep for each task
Expand All @@ -72,7 +73,6 @@ def build(self, root_task: gokart.TaskOnKart) -> None:
if isinstance(task, TaskOnBullet):
logger.info(f"Trying to run task {self._gen_task_info(task)} on child job...")
self._exec_bullet_task(task)
launched_task_ids.add(task.make_unique_id()) # mark as already launched task
task_queue.append(task) # re-enqueue task to check if it is done
elif isinstance(task, gokart.TaskOnKart):
logger.info(f"Executing task {self._gen_task_info(task)} on master job...")
Expand Down Expand Up @@ -125,8 +125,7 @@ def _exec_bullet_task(self, task: TaskOnBullet) -> None:
)
create_job(self.api_instance, job, self.namespace)
logger.info(f"Created child job {job_name} with task {self._gen_task_info(task)}")
task_unique_id = task.make_unique_id()
self.task_id_to_job_name[task_unique_id] = job_name
self.task_id_to_job_name[task.make_unique_id()] = job_name

def _create_child_job_object(self, job_name: str, task_pkl_path: str) -> client.V1Job:
# TODO: use python -c to avoid dependency to execute_task.py
Expand Down Expand Up @@ -163,6 +162,19 @@ def _gen_task_info(task: gokart.TaskOnKart) -> str:
def _gen_pkl_path(task: gokart.TaskOnKart) -> str:
return os.path.join(task.workspace_directory, 'kannon', f'task_obj_{task.make_unique_id()}.pkl')

def _check_child_task_status(self, task: TaskOnBullet) -> bool:
if task.make_unique_id() not in self.task_id_to_job_name:
raise ValueError(f"Task {self._gen_task_info(task)} is not found in `task_id_to_job_name`")
job_name = self.task_id_to_job_name[task.make_unique_id()]
job_status = get_job_status(
self.api_instance,
job_name,
self.namespace,
)
if job_status == JobStatus.FAILED:
raise RuntimeError(f"Task {self._gen_task_info(task)} on job {job_name} has failed.")
return True

def _is_executable(self, task: gokart.TaskOnKart) -> bool:
children = flatten(task.requires())

Expand Down
12 changes: 12 additions & 0 deletions test/integration_test/test_master_build.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import gokart
import luigi
from kubernetes import client
from luigi.task import flatten

from kannon import Kannon, TaskOnBullet

Expand Down Expand Up @@ -54,8 +55,19 @@ def _exec_gokart_task(self, task: MockTaskOnKart) -> None:
task.run()

def _exec_bullet_task(self, task: MockTaskOnBullet) -> None:
self.task_id_to_job_name[task.make_unique_id()] = "dummy_job_name"
task.run()

def _check_child_task_status(self, task: MockTaskOnBullet) -> bool:
return True

def _is_executable(self, task: MockTaskOnKart) -> bool:
children = flatten(task.requires())
for child in children:
if not child.complete():
return False
return True


class TestConsumeTaskQueue(unittest.TestCase):

Expand Down