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

Generalized references to LSF #1413

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

Generalized references to LSF #1413

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 4 additions & 4 deletions notifier/events/run_finished_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ def __init__(
failed,
total,
operator_run_id,
lsf_log_location,
run_log_location,
input_json_location,
job_group_id,
):
Expand All @@ -38,7 +38,7 @@ def __init__(
self.failed = failed
self.total = total
self.operator_run_id = operator_run_id
self.lsf_log_location = lsf_log_location
self.run_log_location = run_log_location
self.input_json_location = input_json_location
self.job_group_id = job_group_id

Expand All @@ -60,7 +60,7 @@ def __str__(self):
{tags}
Status: {status}
Link: {link}
LSF Log Location: {lsf_log_location}
Run Log Location: {run_log_location}
inputs.json Location: {inputs_json_location}
Datadog link: {datadog_link}

Expand Down Expand Up @@ -122,7 +122,7 @@ def __str__(self):
tags=tags,
run_status=status,
output_directory=self.output_directory,
lsf_log_location=self.lsf_log_location,
run_log_location=self.run_log_location,
inputs_json_location=self.input_json_location,
rerun_info=rerun_str,
)
43 changes: 23 additions & 20 deletions runner/tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -618,7 +618,7 @@ def check_statuses_on_ridgeback(execution_ids):


@shared_task(bind=True)
def fail_job(self, run_id, error_message, lsf_log_location=None, input_json_location=None):
def fail_job(self, run_id, error_message, run_log_location=None, input_json_location=None):
lock_id = "run_lock_%s" % run_id
with memcache_task_lock(lock_id, self.app.oid) as acquired:
if acquired:
Expand All @@ -643,7 +643,7 @@ def fail_job(self, run_id, error_message, lsf_log_location=None, input_json_loca
send_notification.delay(ci_review)

_upload_qc_report(run.run_obj)
_job_finished_notify(run, lsf_log_location, input_json_location)
_job_finished_notify(run, run_log_location, input_json_location)
else:
run_id, output_directory, execution_id = restart_run
submit_job.delay(run_id, output_directory, execution_id)
Expand All @@ -652,7 +652,7 @@ def fail_job(self, run_id, error_message, lsf_log_location=None, input_json_loca


@shared_task(bind=True)
def complete_job(self, run_id, outputs, lsf_log_location=None, inputs_json_location=None):
def complete_job(self, run_id, outputs, run_log_location=None, inputs_json_location=None):
lock_id = "run_lock_%s" % run_id
with memcache_task_lock(lock_id, self.app.oid) as acquired:
if acquired:
Expand All @@ -673,7 +673,7 @@ def complete_job(self, run_id, outputs, lsf_log_location=None, inputs_json_locat
job_group = run.job_group
job_group_id = str(job_group.id) if job_group else None

_job_finished_notify(run, lsf_log_location, inputs_json_location)
_job_finished_notify(run, run_log_location, inputs_json_location)

for trigger in run.run_obj.operator_run.operator.from_triggers.filter(run_type=TriggerRunType.INDIVIDUAL):
create_jobs_from_chaining.delay(
Expand All @@ -697,7 +697,7 @@ def _upload_qc_report(run):
logger.error("Operator on_job_fail failed", e)


def _job_finished_notify(run, lsf_log_location=None, input_json_location=None):
def _job_finished_notify(run, run_log_location=None, input_json_location=None):
job_group_notifier = run.job_group_notifier
job_group_notifier_id = str(job_group_notifier.id) if job_group_notifier else None

Expand Down Expand Up @@ -749,7 +749,7 @@ def _job_finished_notify(run, lsf_log_location=None, input_json_location=None):
failed_runs,
total_runs,
operator_run_id,
lsf_log_location,
run_log_location,
input_json_location,
str(run.run_obj.job_group.id),
)
Expand All @@ -772,7 +772,7 @@ def running_job(self, run_id):


@shared_task(bind=True)
def terminate_job(self, run_id, lsf_log_location, inputs_json_location):
def terminate_job(self, run_id, run_log_location, inputs_json_location):
lock_id = "run_lock_%s" % run_id
with memcache_task_lock(lock_id, self.app.oid) as acquired:
if acquired:
Expand All @@ -782,7 +782,7 @@ def terminate_job(self, run_id, lsf_log_location, inputs_json_location):
run.status = RunStatus.TERMINATED
run.save()
run_obj = RunObjectFactory.from_db(run_id)
_job_finished_notify(run_obj, lsf_log_location, inputs_json_location)
_job_finished_notify(run_obj, run_log_location, inputs_json_location)
else:
logger.warning("Run %s is processing by another worker" % run_id)

Expand Down Expand Up @@ -965,19 +965,21 @@ def check_jobs_status():
if status["status"] == "FAILED":
logger.error(format_log("Job failed ", obj=run))
message = dict(details=status.get("message"))
lsf_log_location = status.get("message", {}).get("log")
run_log_location = status.get("message", {}).get("log")
inputs_location = None
if lsf_log_location:
inputs_location = lsf_log_location.replace("lsf.log", "input.json")
fail_job.delay(str(run.id), message, lsf_log_location, inputs_location)
if run_log_location:
run_log_filename = os.path.basename(run_log_location)
inputs_location = run_log_location.replace(run_log_filename, "input.json")
fail_job.delay(str(run.id), message, run_log_location, inputs_location)
continue
if status["status"] == "COMPLETED":
logger.info(format_log("Job completed", obj=run))
lsf_log_location = status.get("message", {}).get("log")
run_log_location = status.get("message", {}).get("log")
inputs_location = None
if lsf_log_location:
inputs_location = lsf_log_location.replace("lsf.log", "input.json")
complete_job.delay(str(run.id), status["outputs"], lsf_log_location, inputs_location)
if run_log_location:
run_log_filename = os.path.basename(run_log_location)
inputs_location = run_log_location.replace(run_log_filename, "input.json")
complete_job.delay(str(run.id), status["outputs"], run_log_location, inputs_location)
continue
if status["status"] == "CREATED":
logger.info(format_log("Job created", obj=run))
Expand All @@ -991,11 +993,12 @@ def check_jobs_status():
continue
if status["status"] == "TERMINATED":
logger.info(format_log("Job terminated", obj=run))
lsf_log_location = status.get("message", {}).get("log")
run_log_location = status.get("message", {}).get("log")
inputs_location = None
if lsf_log_location:
inputs_location = lsf_log_location.replace("lsf.log", "input.json")
terminate_job.delay(str(run.id), lsf_log_location, inputs_location)
if run_log_location:
run_log_filename = os.path.basename(run_log_location)
inputs_location = run_log_location.replace(run_log_filename, "input.json")
terminate_job.delay(str(run.id), run_log_location, inputs_location)
else:
logger.info("Run lock not acquired for run: %s" % str(run.id))

Expand Down