More exception handling. Remove unused database models. New common view structure!

2025-06-20 13:00:36 +00:00 · 2020-11-13 10:01:51 +01:00
parent cb9da5c7dd
commit 5a06a6b241
45 changed files with 692 additions and 1005 deletions
--- a/web/app/tasks/init.py
+++ b/web/app/tasks/init.py
@ -11,15 +11,11 @@ def check_corpora():
    corpora = Corpus.query.all()
    for corpus in filter(lambda corpus: corpus.status == 'submitted', corpora):
        corpus_utils.create_build_corpus_service(corpus)
-    for corpus in filter(lambda corpus: (corpus.status == 'queued'
-                                         or corpus.status == 'running'),
-                         corpora):
+    for corpus in filter(lambda corpus: corpus.status in ['queued', 'running'], corpora):  # noqa
        corpus_utils.checkout_build_corpus_service(corpus)
-    for corpus in filter(lambda corpus: corpus.status == 'start analysis',
-                         corpora):
+    for corpus in filter(lambda corpus: corpus.status == 'start analysis', corpora):  # noqa
        corpus_utils.create_cqpserver_container(corpus)
-    for corpus in filter(lambda corpus: corpus.status == 'stop analysis',
-                         corpora):
+    for corpus in filter(lambda corpus: corpus.status == 'stop analysis', corpora):  # noqa
        corpus_utils.remove_cqpserver_container(corpus)
    db.session.commit()

@ -28,8 +24,6 @@ def check_jobs():
    jobs = Job.query.all()
    for job in filter(lambda job: job.status == 'submitted', jobs):
        job_utils.create_job_service(job)
-    for job in filter(lambda job: job.status == 'queued', jobs):
-        job_utils.checkout_job_service(job)
-    for job in filter(lambda job: job.status == 'running', jobs):
+    for job in filter(lambda job: job.status in ['queued', 'running'], jobs):
        job_utils.checkout_job_service(job)
    db.session.commit()
--- a/web/app/tasks/corpus_utils.py
+++ b/web/app/tasks/corpus_utils.py
@ -1,4 +1,3 @@
-from flask import current_app
 from . import docker_client
 import docker
 import logging
@ -7,20 +6,14 @@ import shutil


 def create_build_corpus_service(corpus):
-    corpus_dir = os.path.join(current_app.config['DATA_DIR'],
-                              str(corpus.user_id),
-                              'corpora',
-                              str(corpus.id))
-    corpus_data_dir = os.path.join(corpus_dir, 'data')
-    corpus_file = os.path.join(corpus_dir, 'merged', 'corpus.vrt')
-    corpus_registry_dir = os.path.join(corpus_dir, 'registry')
-    if os.path.exists(corpus_data_dir):
-        shutil.rmtree(corpus_data_dir)
-    if os.path.exists(corpus_registry_dir):
-        shutil.rmtree(corpus_registry_dir)
+    corpus_data_dir = os.path.join(corpus.path, 'data')
+    shutil.rmtree(corpus_data_dir, ignore_errors=True)
    os.mkdir(corpus_data_dir)
+    corpus_registry_dir = os.path.join(corpus.path, 'registry')
+    shutil.rmtree(corpus_registry_dir, ignore_errors=True)
    os.mkdir(corpus_registry_dir)
-    service_args = {
+    corpus_file = os.path.join(corpus.path, 'merged', 'corpus.vrt')
+    service_kwargs = {
        'command': 'docker-entrypoint.sh build-corpus',
        'constraints': ['node.role==worker'],
        'labels': {'origin': 'nopaque',
@ -32,30 +25,34 @@ def create_build_corpus_service(corpus):
        'name': 'build-corpus_{}'.format(corpus.id),
        'restart_policy': docker.types.RestartPolicy()
    }
-    service_image = \
-        'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'
+    service_image = 'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'  # noqa
    try:
-        docker_client.services.create(service_image, **service_args)
+        docker_client.services.create(service_image, **service_kwargs)
    except docker.errors.APIError as e:
-        logging.error('create_build_corpus_service({}): '.format(corpus.id)
-                      + '{} (status: {} -> failed)'.format(e, corpus.status))
-        corpus.status = 'failed'
+        logging.error('Create "{}" service raised '.format(service_kwargs['name'])  # noqa
+                      + '[docker-APIError] The server returned an error. '
+                      + 'Details: {}'.format(e))
    else:
        corpus.status = 'queued'
-    finally:
-        # TODO: send email
-        pass


 def checkout_build_corpus_service(corpus):
    service_name = 'build-corpus_{}'.format(corpus.id)
    try:
        service = docker_client.services.get(service_name)
-    except docker.errors.NotFound as e:
-        logging.error('checkout_build_corpus_service({}):'.format(corpus.id)
-                      + ' {} (stauts: {} -> failed)'.format(e, corpus.status))
+    except docker.errors.NotFound:
+        logging.error('Get "{}" service raised '.format(service_name)
+                      + '[docker-NotFound] The service does not exist. '
+                      + '(corpus.status: {} -> failed)'.format(corpus.status))
        corpus.status = 'failed'
-    # TODO: handle docker.errors.APIError and docker.errors.InvalidVersion
+    except docker.errors.APIError as e:
+        logging.error('Get "{}" service raised '.format(service_name)
+                      + '[docker-APIError] The server returned an error. '
+                      + 'Details: {}'.format(e))
+    except docker.errors.InvalidVersion:
+        logging.error('Get "{}" service raised '.format(service_name)
+                      + '[docker-InvalidVersion] One of the arguments is '
+                      + 'not supported with the current API version.')
    else:
        service_tasks = service.tasks()
        if not service_tasks:
@ -63,25 +60,23 @@ def checkout_build_corpus_service(corpus):
        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 == 'complete':
-            service.remove()
-            corpus.status = 'prepared'
-        elif corpus.status == 'running' and task_state == 'failed':
-            service.remove()
-            corpus.status = task_state
-    finally:
-        # TODO: send email
-        pass
+        elif corpus.status == 'running' and task_state in ['complete', 'failed']:  # noqa
+            try:
+                service.remove()
+            except docker.errors.APIError as e:
+                logging.error('Remove "{}" service raised '.format(service_name)  # noqa
+                              + '[docker-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(corpus):
-    corpus_dir = os.path.join(current_app.config['DATA_DIR'],
-                              str(corpus.user_id),
-                              'corpora',
-                              str(corpus.id))
-    corpus_data_dir = os.path.join(corpus_dir, 'data')
-    corpus_registry_dir = os.path.join(corpus_dir, 'registry')
-    container_args = {
+    corpus_data_dir = os.path.join(corpus.path, 'data')
+    corpus_registry_dir = os.path.join(corpus.path, 'registry')
+    container_kwargs = {
        'command': 'cqpserver',
        'detach': True,
        'volumes': [corpus_data_dir + ':/corpora/data:rw',
@ -89,20 +84,43 @@ def create_cqpserver_container(corpus):
        'name': 'cqpserver_{}'.format(corpus.id),
        'network': 'nopaque_default'
    }
-    container_image = \
-        'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'
+    container_image = 'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'  # noqa
+    # Check if a cqpserver container already exists. If this is the case,
+    # remove it and create a new one
    try:
-        container = docker_client.containers.get(container_args['name'])
+        container = docker_client.containers.get(container_kwargs['name'])
    except docker.errors.NotFound:
        pass
-    except docker.errors.DockerException:
+    except docker.errors.APIError as e:
+        logging.error('Get "{}" container raised '.format(container_kwargs['name'])
+                      + '[docker-APIError] The server returned an error. '
+                      + 'Details: {}'.format(e))
        return
    else:
-        container.remove(force=True)
+        try:
+            container.remove(force=True)
+        except docker.errors.APIError as e:
+            logging.error('Remove "{}" container raised '.format(container_kwargs['name'])
+                          + '[docker-APIError] The server returned an error. '
+                          + 'Details: {}'.format(e))
+            return
    try:
-        docker_client.containers.run(container_image, **container_args)
-    except docker.errors.DockerException:
-        return
+        docker_client.containers.run(container_image, **container_kwargs)
+    except docker.errors.ContainerError:
+        # This case should not occur, because detach is True.
+        logging.error('Run "{}" container raised '.format(container_kwargs['name'])
+                      + '[docker-ContainerError] The container exits with a '
+                      + 'non-zero exit code and detach is False.')
+        corpus.status = 'failed'
+    except docker.errors.ImageNotFound:
+        logging.error('Run "{}" container raised '.format(container_kwargs['name'])
+                      + '[docker-ImageNotFound] The specified image does not '
+                      + 'exist.')
+        corpus.status = 'failed'
+    except docker.errors.APIError as e:
+        logging.error('Run "{}" container raised '.format(container_kwargs['name'])
+                      + '[docker-APIError] The server returned an error. '
+                      + 'Details: {}'.format(e))
    else:
        corpus.status = 'analysing'

@ -113,8 +131,17 @@ def remove_cqpserver_container(corpus):
        container = docker_client.containers.get(container_name)
    except docker.errors.NotFound:
        pass
-    except docker.errors.DockerException:
+    except docker.errors.APIError as e:
+        logging.error('Get "{}" container raised '.format(container_name)
+                      + '[docker-APIError] The server returned an error. '
+                      + 'Details: {}'.format(e))
        return
    else:
-        container.remove(force=True)
+        try:
+            container.remove(force=True)
+        except docker.errors.APIError as e:
+            logging.error('Remove "{}" container raised '.format(container_name)
+                          + '[docker-APIError] The server returned an error. '
+                          + 'Details: {}'.format(e))
+            return
    corpus.status = 'prepared'
--- a/web/app/tasks/job_utils.py
+++ b/web/app/tasks/job_utils.py
@ -1,7 +1,7 @@
 from datetime import datetime
-from flask import current_app
 from . import docker_client
 from .. import db
+from ..email import create_message, send
 from ..models import JobResult
 import docker
 import logging
@ -10,51 +10,60 @@ import os


 def create_job_service(job):
-    job_dir = os.path.join(current_app.config['DATA_DIR'],
-                           str(job.user_id),
-                           'jobs',
-                           str(job.id))
    cmd = '{} -i /files -o /files/output'.format(job.service)
    if job.service == 'file-setup':
        cmd += ' -f {}'.format(job.secure_filename)
    cmd += ' --log-dir /files'
    cmd += ' --zip [{}]_{}'.format(job.service, job.secure_filename)
    cmd += ' ' + ' '.join(json.loads(job.service_args))
-    service_args = {'command': cmd,
-                    'constraints': ['node.role==worker'],
-                    'labels': {'origin': 'nopaque',
-                               'type': 'service.{}'.format(job.service),
-                               'job_id': str(job.id)},
-                    'mounts': [job_dir + ':/files:rw'],
-                    'name': 'job_{}'.format(job.id),
-                    'resources': docker.types.Resources(
-                        cpu_reservation=job.n_cores * (10 ** 9),
-                        mem_reservation=job.mem_mb * (10 ** 6)),
-                    'restart_policy': docker.types.RestartPolicy()}
+    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),
+                      'resources': docker.types.Resources(
+                          cpu_reservation=job.n_cores * (10 ** 9),
+                          mem_reservation=job.mem_mb * (10 ** 6)
+                      ),
+                      'restart_policy': docker.types.RestartPolicy()}
    service_image = ('gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/'
                     + job.service + ':' + job.service_version)
    try:
-        docker_client.services.create(service_image, **service_args)
+        docker_client.services.create(service_image, **service_kwargs)
    except docker.errors.APIError as e:
-        logging.error('create_job_service({}): {} '.format(job.id, e)
-                      + '(status: {} -> failed)'.format(job.status))
-        job.status = 'failed'
+        logging.error('Create "{}" service raised '.format(service_kwargs['name'])  # noqa
+                      + '[docker-APIError] The server returned an error. '
+                      + 'Details: {}'.format(e))
    else:
        job.status = 'queued'
-    finally:
-        # TODO: send email
-        pass
+        msg = create_message(
+            job.creator.email,
+            'Status update for your Job "{}"'.format(job.title),
+            'tasks/email/notification',
+            job=job
+        )
+        send(msg)


 def checkout_job_service(job):
    service_name = 'job_{}'.format(job.id)
    try:
        service = docker_client.services.get(service_name)
-    except docker.errors.NotFound as e:
-        logging.error('checkout_job_service({}): {} '.format(job.id, e)
-                      + '(status: {} -> submitted)'.format(job.status))
-        job.status = 'submitted'
-    # TODO: handle docker.errors.APIError and docker.errors.InvalidVersion
+    except docker.errors.NotFound:
+        logging.error('Get "{}" service raised '.format(service_name)
+                      + '[docker-NotFound] The service does not exist. '
+                      + '(job.status: {} -> failed)'.format(job.status))
+        job.status = 'failed'
+    except docker.errors.APIError as e:
+        logging.error('Get "{}" service raised '.format(service_name)
+                      + '[docker-APIError] The server returned an error. '
+                      + 'Details: {}'.format(e))
+    except docker.errors.InvalidVersion:
+        logging.error('Get "{}" service raised '.format(service_name)
+                      + '[docker-InvalidVersion] One of the arguments is '
+                      + 'not supported with the current API version.')
    else:
        service_tasks = service.tasks()
        if not service_tasks:
@ -62,22 +71,16 @@ def checkout_job_service(job):
        task_state = service_tasks[0].get('Status').get('State')
        if job.status == 'queued' and task_state != 'pending':
            job.status = 'running'
-        elif job.status == 'queued' and task_state == 'complete':
+        elif job.status == 'running' and task_state == 'complete':
            service.remove()
            job.end_date = datetime.utcnow()
            job.status = task_state
            if task_state == 'complete':
-                results_dir = os.path.join(current_app.config['DATA_DIR'],
-                                           str(job.user_id),
-                                           'jobs',
-                                           str(job.id),
-                                           'output')
-                results = filter(lambda x: x.endswith('.zip'),
-                                 os.listdir(results_dir))
-                for result in results:
-                    job_result = JobResult(dir=results_dir,
-                                           filename=result,
-                                           job_id=job.id)
+                job_results_dir = os.path.join(job.path, 'output')
+                job_results = filter(lambda x: x.endswith('.zip'),
+                                     os.listdir(job_results_dir))
+                for job_result in job_results:
+                    job_result = JobResult(filename=job_result, job=job)
                    db.session.add(job_result)
        elif job.status == 'running' and task_state == 'failed':
            service.remove()
@ -85,6 +88,13 @@ def checkout_job_service(job):
            job.status = task_state
    finally:
        # TODO: send email
+        msg = create_message(
+            job.creator.email,
+            '[nopaque] Status update for your Job "{}"'.format(job.title),
+            'tasks/email/notification',
+            job=job
+        )
+        send(msg)
        pass