mirror of
https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git
synced 2025-06-12 00:50:40 +00:00
Rename TaskRunner to Daemon (because it essentially is one) and restructure cli
This commit is contained in:
Patrick Jentsch
20
app/daemon/__init__.py
Normal file
20
app/daemon/__init__.py
Normal file
@ -0,0 +1,20 @@
|
||||
from app import db
|
||||
from time import sleep
|
||||
from .corpus_utils import CheckCorporaMixin
|
||||
from .job_utils import CheckJobsMixin
|
||||
import docker
|
||||
|
||||
|
||||
class Daemon(CheckCorporaMixin, CheckJobsMixin):
|
||||
def __init__(self):
|
||||
self.docker = docker.from_env()
|
||||
|
||||
def run(self):
|
||||
while True:
|
||||
try:
|
||||
self.check_corpora()
|
||||
self.check_jobs()
|
||||
db.session.commit()
|
||||
except:
|
||||
pass
|
||||
sleep(1.5)
|
||||
248
app/daemon/corpus_utils.py
Normal file
248
app/daemon/corpus_utils.py
Normal file
@ -0,0 +1,248 @@
|
||||
from flask import current_app
|
||||
from ..models import Corpus
|
||||
import docker
|
||||
import os
|
||||
import shutil
|
||||
|
||||
|
||||
class CheckCorporaMixin:
|
||||
def check_corpora(self):
|
||||
corpora = Corpus.query.all()
|
||||
queued_corpora = list(filter(lambda corpus: corpus.status == 'queued', corpora)) # noqa
|
||||
running_corpora = list(filter(lambda corpus: corpus.status == 'running', corpora)) # noqa
|
||||
start_analysis_corpora = list(filter(lambda corpus: corpus.status == 'start analysis', corpora)) # noqa
|
||||
analysing_corpora = list(filter(lambda corpus: corpus.status == 'analysing', corpora)) # noqa
|
||||
stop_analysis_corpora = list(filter(lambda corpus: corpus.status == 'stop analysis', corpora)) # noqa
|
||||
submitted_corpora = list(filter(lambda corpus: corpus.status == 'submitted', corpora)) # noqa
|
||||
for corpus in submitted_corpora:
|
||||
self.create_build_corpus_service(corpus)
|
||||
for corpus in queued_corpora + running_corpora:
|
||||
self.checkout_build_corpus_service(corpus)
|
||||
for corpus in start_analysis_corpora:
|
||||
self.create_cqpserver_container(corpus)
|
||||
for corpus in analysing_corpora:
|
||||
self.checkout_analysing_corpus_container(corpus)
|
||||
for corpus in stop_analysis_corpora:
|
||||
self.remove_cqpserver_container(corpus)
|
||||
|
||||
def create_build_corpus_service(self, corpus):
|
||||
''' # Docker service settings # '''
|
||||
''' ## Command ## '''
|
||||
command = 'docker-entrypoint.sh build-corpus'
|
||||
''' ## Constraints ## '''
|
||||
constraints = ['node.role==worker']
|
||||
''' ## Image ## '''
|
||||
image = current_app.config['DOCKER_IMAGE_PREFIX'] + 'cqpserver:latest'
|
||||
''' ## Labels ## '''
|
||||
labels = {
|
||||
'origin': current_app.config['SERVER_NAME'],
|
||||
'type': 'build-corpus',
|
||||
'corpus_id': str(corpus.id)
|
||||
}
|
||||
''' ## Mounts ## '''
|
||||
''' ### Corpus file mount ### '''
|
||||
corpus_file_source = os.path.join(corpus.path, 'merged', 'corpus.vrt')
|
||||
corpus_file_target = '/root/files/corpus.vrt'
|
||||
corpus_file_mount = \
|
||||
corpus_file_source + ':' + corpus_file_target + ':ro'
|
||||
''' ### Corpus data mount ### '''
|
||||
corpus_data_source = os.path.join(corpus.path, 'data')
|
||||
corpus_data_target = '/corpora/data'
|
||||
corpus_data_mount = \
|
||||
corpus_data_source + ':' + corpus_data_target + ':rw'
|
||||
# Make sure that their is no data in the corpus data directory
|
||||
shutil.rmtree(corpus_data_source, ignore_errors=True)
|
||||
os.mkdir(corpus_data_source)
|
||||
''' ### Corpus registry mount ### '''
|
||||
corpus_registry_source = os.path.join(corpus.path, 'registry')
|
||||
corpus_registry_target = '/usr/local/share/cwb/registry'
|
||||
corpus_registry_mount = \
|
||||
corpus_registry_source + ':' + corpus_registry_target + ':rw'
|
||||
# Make sure that their is no data in the corpus registry directory
|
||||
shutil.rmtree(corpus_registry_source, ignore_errors=True)
|
||||
os.mkdir(corpus_registry_source)
|
||||
mounts = [corpus_file_mount, corpus_data_mount, corpus_registry_mount]
|
||||
''' ## Name ## '''
|
||||
name = 'build-corpus_{}'.format(corpus.id)
|
||||
''' ## Restart policy ## '''
|
||||
restart_policy = docker.types.RestartPolicy()
|
||||
try:
|
||||
self.docker.services.create(
|
||||
image,
|
||||
command=command,
|
||||
constraints=constraints,
|
||||
labels=labels,
|
||||
mounts=mounts,
|
||||
name=name,
|
||||
restart_policy=restart_policy
|
||||
)
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Create "{}" service raised '.format(name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
else:
|
||||
corpus.status = 'queued'
|
||||
|
||||
def checkout_build_corpus_service(self, corpus):
|
||||
service_name = 'build-corpus_{}'.format(corpus.id)
|
||||
try:
|
||||
service = self.docker.services.get(service_name)
|
||||
except docker.errors.NotFound:
|
||||
current_app.logger.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.NotFound" The service does not exist. '
|
||||
+ '(corpus.status: {} -> failed)'.format(corpus.status)
|
||||
)
|
||||
corpus.status = 'failed'
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
except docker.errors.InvalidVersion:
|
||||
current_app.logger.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.InvalidVersion" One of the arguments is '
|
||||
+ 'not supported with the current API version.'
|
||||
)
|
||||
else:
|
||||
service_tasks = service.tasks()
|
||||
if not service_tasks:
|
||||
return
|
||||
task_state = service_tasks[0].get('Status').get('State')
|
||||
if corpus.status == 'queued' and task_state != 'pending':
|
||||
corpus.status = 'running'
|
||||
elif (corpus.status == 'running'
|
||||
and task_state in ['complete', 'failed']):
|
||||
try:
|
||||
service.remove()
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Remove "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. ' # noqa
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
corpus.status = \
|
||||
'prepared' if task_state == 'complete' else 'failed'
|
||||
|
||||
def create_cqpserver_container(self, corpus):
|
||||
''' # Docker container settings # '''
|
||||
''' ## Command ## '''
|
||||
command = 'cqpserver'
|
||||
''' ## Detach ## '''
|
||||
detach = True
|
||||
''' ## Image ## '''
|
||||
image = current_app.config['DOCKER_IMAGE_PREFIX'] + 'cqpserver:latest'
|
||||
''' ## Name ## '''
|
||||
name = 'cqpserver_{}'.format(corpus.id)
|
||||
''' ## Network ## '''
|
||||
network = 'nopaque_default'
|
||||
''' ## Volumes ## '''
|
||||
''' ### Corpus data volume ### '''
|
||||
corpus_data_source = os.path.join(corpus.path, 'data')
|
||||
corpus_data_target = '/corpora/data'
|
||||
corpus_data_volume = \
|
||||
corpus_data_source + ':' + corpus_data_target + ':rw'
|
||||
''' ### Corpus registry volume ### '''
|
||||
corpus_registry_source = os.path.join(corpus.path, 'registry')
|
||||
corpus_registry_target = '/usr/local/share/cwb/registry'
|
||||
corpus_registry_volume = \
|
||||
corpus_registry_source + ':' + corpus_registry_target + ':rw'
|
||||
volumes = [corpus_data_volume, corpus_registry_volume]
|
||||
# Check if a cqpserver container already exists. If this is the case,
|
||||
# remove it and create a new one
|
||||
try:
|
||||
container = self.docker.containers.get(name)
|
||||
except docker.errors.NotFound:
|
||||
pass
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Get "{}" container raised '.format(name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
try:
|
||||
container.remove(force=True)
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Remove "{}" container raised '.format(name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
try:
|
||||
self.docker.containers.run(image, command=command, detach=detach,
|
||||
volumes=volumes, name=name,
|
||||
network=network)
|
||||
except docker.errors.ContainerError:
|
||||
# This case should not occur, because detach is True.
|
||||
current_app.logger.error(
|
||||
'Run "{}" container raised '.format(name)
|
||||
+ '"docker.errors.ContainerError" The container exits with a '
|
||||
+ 'non-zero exit code and detach is False.'
|
||||
)
|
||||
corpus.status = 'failed'
|
||||
except docker.errors.ImageNotFound:
|
||||
current_app.logger.error(
|
||||
'Run "{}" container raised '.format(name)
|
||||
+ '"docker.errors.ImageNotFound" The specified image does not '
|
||||
+ 'exist.'
|
||||
)
|
||||
corpus.status = 'failed'
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Run "{}" container raised '.format(name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
else:
|
||||
corpus.status = 'analysing'
|
||||
|
||||
def checkout_analysing_corpus_container(self, corpus):
|
||||
container_name = 'cqpserver_{}'.format(corpus.id)
|
||||
try:
|
||||
self.docker.containers.get(container_name)
|
||||
except docker.errors.NotFound:
|
||||
current_app.logger.error(
|
||||
'Could not find "{}" but the corpus state is "analysing".'
|
||||
)
|
||||
corpus.status = 'prepared'
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Get "{}" container raised '.format(container_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
|
||||
def remove_cqpserver_container(self, corpus):
|
||||
container_name = 'cqpserver_{}'.format(corpus.id)
|
||||
try:
|
||||
container = self.docker.containers.get(container_name)
|
||||
except docker.errors.NotFound:
|
||||
pass
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Get "{}" container raised '.format(container_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
try:
|
||||
container.remove(force=True)
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Remove "{}" container raised '.format(container_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
corpus.status = 'prepared'
|
||||
199
app/daemon/job_utils.py
Normal file
199
app/daemon/job_utils.py
Normal file
@ -0,0 +1,199 @@
|
||||
from datetime import datetime
|
||||
from flask import current_app
|
||||
from werkzeug.utils import secure_filename
|
||||
from .. import db
|
||||
from ..models import Job, JobResult
|
||||
import docker
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
|
||||
|
||||
class CheckJobsMixin:
|
||||
def check_jobs(self):
|
||||
jobs = Job.query.all()
|
||||
canceling_jobs = list(filter(lambda job: job.status == 'canceling', jobs)) # noqa
|
||||
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)) # noqa
|
||||
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)
|
||||
|
||||
def create_job_service(self, job):
|
||||
''' # Docker service settings # '''
|
||||
''' ## Service specific settings ## '''
|
||||
if job.service == 'file-setup':
|
||||
mem_mb = 2048
|
||||
n_cores = 2
|
||||
executable = 'file-setup'
|
||||
image = (current_app.config['DOCKER_IMAGE_PREFIX']
|
||||
+ 'file-setup:' + job.service_version)
|
||||
elif job.service == 'ocr':
|
||||
mem_mb = 4096
|
||||
n_cores = 4
|
||||
executable = 'ocr'
|
||||
image = (current_app.config['DOCKER_IMAGE_PREFIX']
|
||||
+ 'ocr:' + job.service_version)
|
||||
elif job.service == 'nlp':
|
||||
mem_mb = 2048
|
||||
n_cores = 2
|
||||
executable = 'nlp'
|
||||
image = (current_app.config['DOCKER_IMAGE_PREFIX']
|
||||
+ 'nlp:' + job.service_version)
|
||||
''' ## Command ## '''
|
||||
command = '{} -i /input -o /output'.format(executable)
|
||||
command += ' --log-dir /input'
|
||||
command += ' --mem-mb {}'.format(mem_mb)
|
||||
command += ' --n-cores {}'.format(n_cores)
|
||||
command += ' --zip [' + job.service + ']_' + secure_filename(job.title)
|
||||
command += ' ' + ' '.join(json.loads(job.service_args))
|
||||
''' ## Constraints ## '''
|
||||
constraints = ['node.role==worker']
|
||||
''' ## Labels ## '''
|
||||
labels = {
|
||||
'origin': current_app.config['SERVER_NAME'],
|
||||
'type': 'job',
|
||||
'job_id': str(job.id)
|
||||
}
|
||||
''' ## Mounts ## '''
|
||||
''' ### Input mount ### '''
|
||||
input_mount_source = job.path
|
||||
input_mount_target = '/input'
|
||||
if job.service == 'file-setup':
|
||||
input_mount_target += '/' + secure_filename(job.title)
|
||||
input_mount = input_mount_source + ':' + input_mount_target + ':rw'
|
||||
''' ### Output mount ### '''
|
||||
output_mount_source = os.path.join(job.path, 'output')
|
||||
output_mount_target = '/output'
|
||||
output_mount = output_mount_source + ':' + output_mount_target + ':rw'
|
||||
# Make sure that their is no data in the output directory
|
||||
shutil.rmtree(output_mount_source, ignore_errors=True)
|
||||
os.makedirs(output_mount_source)
|
||||
mounts = [input_mount, output_mount]
|
||||
''' ## Name ## '''
|
||||
name = 'job_{}'.format(job.id)
|
||||
''' ## Resources ## '''
|
||||
resources = docker.types.Resources(
|
||||
cpu_reservation=n_cores * (10 ** 9),
|
||||
mem_reservation=mem_mb * (10 ** 6)
|
||||
)
|
||||
''' ## Restart policy ## '''
|
||||
restart_policy = docker.types.RestartPolicy()
|
||||
try:
|
||||
self.docker.services.create(
|
||||
image,
|
||||
command=command,
|
||||
constraints=constraints,
|
||||
labels=labels,
|
||||
mounts=mounts,
|
||||
name=name,
|
||||
resources=resources,
|
||||
restart_policy=restart_policy
|
||||
)
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Create "{}" service raised '.format(name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
job.status = 'queued'
|
||||
|
||||
def checkout_job_service(self, job):
|
||||
service_name = 'job_{}'.format(job.id)
|
||||
try:
|
||||
service = self.docker.services.get(service_name)
|
||||
except docker.errors.NotFound:
|
||||
current_app.logger.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.NotFound" The service does not exist. '
|
||||
+ '(job.status: {} -> failed)'.format(job.status)
|
||||
)
|
||||
job.status = 'failed'
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
except docker.errors.InvalidVersion:
|
||||
current_app.logger.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'
|
||||
elif job.status == 'running' and task_state in ['complete', 'failed']: # noqa
|
||||
try:
|
||||
service.remove()
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Remove "{}" service raised '.format(service_name)
|
||||
+ '"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:
|
||||
job_result = JobResult(filename=result_file, job=job) # noqa
|
||||
db.session.add(job_result)
|
||||
db.session.flush()
|
||||
db.session.refresh(job_result)
|
||||
job.end_date = datetime.utcnow()
|
||||
job.status = task_state
|
||||
|
||||
def remove_job_service(self, job):
|
||||
service_name = 'job_{}'.format(job.id)
|
||||
try:
|
||||
service = self.docker.services.get(service_name)
|
||||
except docker.errors.NotFound:
|
||||
job.status = 'canceled'
|
||||
except docker.errors.APIError as e:
|
||||
current_app.logger.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
except docker.errors.InvalidVersion:
|
||||
current_app.logger.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:
|
||||
current_app.logger.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:
|
||||
current_app.logger.error(
|
||||
'Remove "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
Reference in New Issue
Block a user