nopaque/web/app/tasks/job_utils.py

180 lines
8.7 KiB
Python
Raw Normal View History

2020-11-09 15:14:19 +00:00
from datetime import datetime
2020-11-13 12:33:32 +00:00
from werkzeug.utils import secure_filename
from .. import db, mail
from ..email import create_message
from ..models import Job, JobResult
2020-11-09 15:14:19 +00:00
import docker
import logging
import json
import os
class CheckJobsMixin:
def check_jobs(self):
jobs = Job.query.all()
canceling_jobs = list(filter(lambda job: job.status == 'canceling', jobs))
queued_jobs = list(filter(lambda job: job.status == 'queued', jobs))
running_jobs = list(filter(lambda job: job.status == 'running', jobs))
submitted_jobs = list(filter(lambda job: job.status == 'submitted', jobs))
for job in submitted_jobs:
self.create_job_service(job)
for job in queued_jobs + running_jobs:
self.checkout_job_service(job)
for job in canceling_jobs:
self.remove_job_service(job)
2020-11-09 15:14:19 +00:00
def create_job_service(self, job):
cmd = '{} -i /files -o /files/output'.format(job.service)
if job.service == 'file-setup':
cmd += ' -f {}'.format(secure_filename(job.title))
2021-03-26 11:15:33 +00:00
ressources = docker.types.Resources(cpu_reservation=4 * (10 ** 9), mem_reservation=4096 * (10 ** 6)) # noqa
elif job.service == 'nlp':
ressources = docker.types.Resources(cpu_reservation=2 * (10 ** 9), mem_reservation=4096 * (10 ** 6)) # noqa
elif job.service == 'ocr':
ressources = docker.types.Resources(cpu_reservation=4 * (10 ** 9), mem_reservation=8192 * (10 ** 6)) # noqa
cmd += ' --log-dir /files'
cmd += ' --zip [{}]_{}'.format(job.service, secure_filename(job.title))
cmd += ' ' + ' '.join(json.loads(job.service_args))
service_kwargs = {'command': cmd,
'constraints': ['node.role==worker'],
'labels': {'origin': 'nopaque',
'type': 'service.{}'.format(job.service),
'job_id': str(job.id)},
'mounts': [job.path + ':/files:rw'],
'name': 'job_{}'.format(job.id),
2021-03-26 11:15:33 +00:00
'resources': ressources,
'restart_policy': docker.types.RestartPolicy()}
2021-03-26 11:15:33 +00:00
service_image = 'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/{}:{}'.format(job.service, job.service_version) # noqa
try:
self.docker.services.create(service_image, **service_kwargs)
except docker.errors.APIError as e:
logging.error(
'Create "{}" service raised '.format(service_kwargs['name'])
+ '"docker.errors.APIError" The server returned an error. '
+ 'Details: {}'.format(e)
)
2020-11-09 15:14:19 +00:00
return
else:
job.status = 'queued'
2021-03-26 11:15:33 +00:00
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
self.buffer_user_patch_operation(job, patch_operation)
finally:
self.send_job_notification(job)
2020-11-09 15:14:19 +00:00
def checkout_job_service(self, job):
service_name = 'job_{}'.format(job.id)
2020-11-13 12:33:32 +00:00
try:
service = self.docker.services.get(service_name)
except docker.errors.NotFound:
logging.error('Get "{}" service raised '.format(service_name)
+ '"docker.errors.NotFound" The service does not exist. '
+ '(job.status: {} -> failed)'.format(job.status))
job.status = 'failed'
2021-03-26 11:15:33 +00:00
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
self.buffer_user_patch_operation(job, patch_operation)
2020-11-13 12:33:32 +00:00
except docker.errors.APIError as e:
2020-11-19 11:31:29 +00:00
logging.error(
'Get "{}" service raised '.format(service_name)
2020-11-19 11:31:29 +00:00
+ '"docker.errors.APIError" The server returned an error. '
+ 'Details: {}'.format(e)
)
2020-11-13 12:33:32 +00:00
return
except docker.errors.InvalidVersion:
logging.error(
'Get "{}" service raised '.format(service_name)
+ '"docker.errors.InvalidVersion" One of the arguments is '
+ 'not supported with the current API version.'
)
return
else:
service_tasks = service.tasks()
if not service_tasks:
return
task_state = service_tasks[0].get('Status').get('State')
if job.status == 'queued' and task_state != 'pending':
job.status = 'running'
2021-03-26 11:15:33 +00:00
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
self.buffer_user_patch_operation(job, patch_operation)
elif job.status == 'running' and task_state in ['complete', 'failed']:
try:
service.remove()
except docker.errors.APIError as e:
logging.error(
'Remove "{}" service raised '.format(service_name)
2021-03-26 11:15:33 +00:00
+ '"docker.errors.APIError" The server returned an error. ' # noqa
+ 'Details: {}'.format(e)
)
return
else:
if task_state == 'complete':
results_dir = os.path.join(job.path, 'output')
result_files = filter(lambda x: x.endswith('.zip'),
os.listdir(results_dir))
for result_file in result_files:
2021-03-26 11:15:33 +00:00
job_result = JobResult(filename=result_file, job=job) # noqa
db.session.add(job_result)
db.session.flush()
db.session.refresh(job_result)
2021-03-26 11:15:33 +00:00
patch_operation = {'op': 'add', 'path': '/jobs/{}/results/{}'.format(job.id, job_result.id), 'value': job_result.to_dict()} # noqa
self.buffer_user_patch_operation(job, patch_operation) # noqa
job.end_date = datetime.utcnow()
2021-03-26 11:15:33 +00:00
patch_operation = {'op': 'replace', 'path': '/jobs/{}/end_date'.format(job.id), 'value': job.end_date.timestamp()} # noqa
self.buffer_user_patch_operation(job, patch_operation)
job.status = task_state
2021-03-26 11:15:33 +00:00
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
self.buffer_user_patch_operation(job, patch_operation)
finally:
self.send_job_notification(job)
def remove_job_service(self, job):
service_name = 'job_{}'.format(job.id)
2020-11-13 12:33:32 +00:00
try:
service = self.docker.services.get(service_name)
except docker.errors.NotFound:
job.status = 'canceled'
2021-03-26 11:15:33 +00:00
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
self.buffer_user_patch_operation(job, patch_operation)
2020-11-13 12:33:32 +00:00
except docker.errors.APIError as e:
2020-11-19 11:31:29 +00:00
logging.error(
'Get "{}" service raised '.format(service_name)
2020-11-19 11:31:29 +00:00
+ '"docker.errors.APIError" The server returned an error. '
+ 'Details: {}'.format(e)
)
return
except docker.errors.InvalidVersion:
logging.error(
'Get "{}" service raised '.format(service_name)
+ '"docker.errors.InvalidVersion" One of the arguments is '
+ 'not supported with the current API version.'
)
return
else:
try:
service.update(mounts=None)
except docker.errors.APIError as e:
logging.error(
'Update "{}" service raised '.format(service_name)
+ '"docker.errors.APIError" The server returned an error. '
+ 'Details: {}'.format(e)
)
return
try:
service.remove()
except docker.errors.APIError as e:
logging.error(
'Remove "{}" service raised '.format(service_name)
+ '"docker.errors.APIError" The server returned an error. '
+ 'Details: {}'.format(e)
)
2020-11-19 08:41:22 +00:00
def send_job_notification(self, job):
if job.creator.setting_job_status_mail_notifications == 'none':
return
if (job.creator.setting_job_status_mail_notifications == 'end'
and job.status not in ['complete', 'failed']):
return
msg = create_message(job.creator.email,
2021-03-26 11:15:33 +00:00
'Status update for your Job "{}"'.format(job.title), # noqa
'tasks/email/notification', job=job)
mail.send(msg)