Big Corpus analysis update

app/daemon/__init__.py

@@ -1,4 +1,5 @@
 from app import db
+from flask import current_app
 from time import sleep
 from .corpus_utils import CheckCorporaMixin
 from .job_utils import CheckJobsMixin
@@ -8,6 +9,11 @@ import docker
 class Daemon(CheckCorporaMixin, CheckJobsMixin):
     def __init__(self):
         self.docker = docker.from_env()
+        self.docker.login(
+            username=current_app.config['NOPAQUE_DOCKER_REGISTRY_USERNAME'],
+            password=current_app.config['NOPAQUE_DOCKER_REGISTRY_PASSWORD'],
+            registry=current_app.config['NOPAQUE_DOCKER_REGISTRY']
+        )
 
     def run(self):
         while True:

app/daemon/corpus_utils.py

@@ -8,21 +8,19 @@ 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:
+        for corpus in (x for x in corpora if x.status == 'submitted'):
             self.create_build_corpus_service(corpus)
-        for corpus in queued_corpora + running_corpora:
+        for corpus in (x for x in corpora if x.status == 'queued' or x.status == 'running'):  # noqa
             self.checkout_build_corpus_service(corpus)
-        for corpus in start_analysis_corpora:
-            self.create_cqpserver_container(corpus)
-        for corpus in analysing_corpora:
+        for corpus in (x for x in corpora if x.status == 'prepared' and x.num_analysis_sessions > 0):  # noqa
+            corpus.status = 'start analysis'
+        for corpus in (x for x in corpora if x.status == 'analysing' and x.num_analysis_sessions == 0):  # noqa
+            corpus.status = 'stop analysis'
+        for corpus in (x for x in corpora if x.status == 'analysing'):
             self.checkout_analysing_corpus_container(corpus)
-        for corpus in stop_analysis_corpora:
+        for corpus in (x for x in corpora if x.status == 'start analysis'):
+            self.create_cqpserver_container(corpus)
+        for corpus in (x for x in corpora if x.status == 'stop analysis'):
             self.remove_cqpserver_container(corpus)
 
     def create_build_corpus_service(self, corpus):
@@ -32,7 +30,7 @@ class CheckCorporaMixin:
         ''' ## Constraints ## '''
         constraints = ['node.role==worker']
         ''' ## Image ## '''
-        image = current_app.config['DOCKER_IMAGE_PREFIX'] + 'cqpserver:latest'
+        image = f'{current_app.config["NOPAQUE_DOCKER_IMAGE_PREFIX"]}cqpserver:r1674'  # noqa
         ''' ## Labels ## '''
         labels = {
             'origin': current_app.config['SERVER_NAME'],
@@ -43,27 +41,24 @@ class CheckCorporaMixin:
         ''' ### 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_file_mount = f'{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'
+        corpus_data_mount = f'{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'
+        corpus_registry_mount = f'{corpus_registry_source}:{corpus_registry_target}:rw'  # noqa
         # 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)
+        name = f'build-corpus_{corpus.id}'
         ''' ## Restart policy ## '''
         restart_policy = docker.types.RestartPolicy()
         try:
@@ -78,57 +73,48 @@ class CheckCorporaMixin:
             )
         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)
+                f'Create service "{name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
             )
-        else:
-            corpus.status = 'queued'
+            return
+        corpus.status = 'queued'
 
     def checkout_build_corpus_service(self, corpus):
-        service_name = 'build-corpus_{}'.format(corpus.id)
+        service_name = f'build-corpus_{corpus.id}'
         try:
             service = self.docker.services.get(service_name)
-        except docker.errors.NotFound:
+        except docker.errors.NotFound as e:
             current_app.logger.error(
-                'Get "{}" service raised '.format(service_name)
-                + '"docker.errors.NotFound" The service does not exist. '
-                + '(corpus.status: {} -> failed)'.format(corpus.status)
+                f'Get service "{service_name}" failed '
+                + f'due to "docker.errors.NotFound": {e}'
             )
             corpus.status = 'failed'
+            return
         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.'
+                f'Get service "{service_name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
             )
+        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'
+            return
+        elif corpus.status == 'running' and task_state == 'complete':
+            corpus.status = 'prepared'
+        elif corpus.status == 'running' and task_state == 'failed':
+            corpus.status = 'failed'
         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'
+            return
+        try:
+            service.remove()
+        except docker.errors.APIError as e:
+            current_app.logger.error(
+                f'Remove service "{service_name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
+            )
 
     def create_cqpserver_container(self, corpus):
         ''' # Docker container settings # '''
@@ -137,22 +123,20 @@ class CheckCorporaMixin:
         ''' ## Detach ## '''
         detach = True
         ''' ## Image ## '''
-        image = current_app.config['DOCKER_IMAGE_PREFIX'] + 'cqpserver:latest'
+        image = f'{current_app.config["NOPAQUE_DOCKER_IMAGE_PREFIX"]}cqpserver:r1674'  # noqa
         ''' ## Name ## '''
-        name = 'cqpserver_{}'.format(corpus.id)
+        name = f'cqpserver_{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_data_volume = f'{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'
+        corpus_registry_volume = f'{corpus_registry_source}:{corpus_registry_target}:rw'  # noqa
         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
@@ -162,9 +146,8 @@ class CheckCorporaMixin:
             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)
+                f'Get container "{name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
             )
             return
         else:
@@ -172,77 +155,68 @@ class CheckCorporaMixin:
                 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)
+                    f'Remove container "{name}" failed '
+                    + f'due to "docker.errors.APIError": {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.
+            self.docker.containers.run(
+                image,
+                command=command,
+                detach=detach,
+                volumes=volumes,
+                name=name,
+                network=network
+            )
+        except docker.errors.ImageNotFound as e:
             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.'
+                f'Run container "{name}" failed '
+                + f'due to "docker.errors.ImageNotFound" error: {e}'
             )
             corpus.status = 'failed'
+            return
         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)
+                f'Run container "{name}" failed '
+                + f'due to "docker.errors.APIError" error: {e}'
             )
-        else:
-            corpus.status = 'analysing'
+            return
+        corpus.status = 'analysing'
 
     def checkout_analysing_corpus_container(self, corpus):
-        container_name = 'cqpserver_{}'.format(corpus.id)
+        container_name = f'cqpserver_{corpus.id}'
         try:
             self.docker.containers.get(container_name)
-        except docker.errors.NotFound:
+        except docker.errors.NotFound as e:
             current_app.logger.error(
-                'Could not find "{}" but the corpus state is "analysing".'
+                f'Get container "{container_name}" failed '
+                + f'due to "docker.errors.NotFound": {e}'
             )
+            corpus.num_analysis_sessions = 0
             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)
+                f'Get container "{container_name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
             )
-            return
 
     def remove_cqpserver_container(self, corpus):
-        container_name = 'cqpserver_{}'.format(corpus.id)
+        container_name = f'cqpserver_{corpus.id}'
         try:
             container = self.docker.containers.get(container_name)
         except docker.errors.NotFound:
-            pass
+            corpus.status = 'prepared'
+            return
         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)
+                f'Get container "{container_name}" failed '
+                + f'due to "docker.errors.APIError": {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'
+        try:
+            container.remove(force=True)
+        except docker.errors.APIError as e:
+            current_app.logger.error(
+                f'Remove container "{container_name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
+            )

app/daemon/job_utils.py

@@ -12,15 +12,11 @@ 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:
+        for job in (x for x in jobs if x.status == 'submitted'):
             self.create_job_service(job)
-        for job in queued_jobs + running_jobs:
+        for job in (x for x in jobs if x.status in ['queued', 'running']):
             self.checkout_job_service(job)
-        for job in canceling_jobs:
+        for job in (x for x in jobs if x.status == 'canceling'):
             self.remove_job_service(job)
 
     def create_job_service(self, job):
@@ -30,26 +26,23 @@ class CheckJobsMixin:
             mem_mb = 2048
             n_cores = 2
             executable = 'file-setup'
-            image = (current_app.config['DOCKER_IMAGE_PREFIX']
-                     + 'file-setup:' + job.service_version)
+            image = f'{current_app.config["NOPAQUE_DOCKER_IMAGE_PREFIX"]}file-setup:{job.service_version}'  # noqa
         elif job.service == 'ocr':
             mem_mb = 4096
             n_cores = 4
             executable = 'ocr'
-            image = (current_app.config['DOCKER_IMAGE_PREFIX']
-                     + 'ocr:' + job.service_version)
+            image = f'{current_app.config["NOPAQUE_DOCKER_IMAGE_PREFIX"]}ocr:{job.service_version}'  # noqa
         elif job.service == 'nlp':
             mem_mb = 2048
             n_cores = 2
             executable = 'nlp'
-            image = (current_app.config['DOCKER_IMAGE_PREFIX']
-                     + 'nlp:' + job.service_version)
+            image = f'{current_app.config["NOPAQUE_DOCKER_IMAGE_PREFIX"]}nlp:{job.service_version}'  # noqa
         ''' ## Command ## '''
-        command = '{} -i /input -o /output'.format(executable)
+        command = f'{executable} -i /input -o /output'
         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 += f' --mem-mb {mem_mb}'
+        command += f' --n-cores {n_cores}'
+        command += f' --zip [{job.service}]_{secure_filename(job.title)}'
         command += ' ' + ' '.join(json.loads(job.service_args))
         ''' ## Constraints ## '''
         constraints = ['node.role==worker']
@@ -64,18 +57,18 @@ class CheckJobsMixin:
         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'
+            input_mount_target += f'/{secure_filename(job.title)}'
+        input_mount = f'{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'
+        output_mount = f'{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)
+        name = f'job_{job.id}'
         ''' ## Resources ## '''
         resources = docker.types.Resources(
             cpu_reservation=n_cores * (10 ** 9),
@@ -96,104 +89,83 @@ class CheckJobsMixin:
             )
         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)
+                f'Create service "{name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
             )
             return
-        else:
-            job.status = 'queued'
+        job.status = 'queued'
 
     def checkout_job_service(self, job):
-        service_name = 'job_{}'.format(job.id)
+        service_name = f'job_{job.id}'
         try:
             service = self.docker.services.get(service_name)
-        except docker.errors.NotFound:
+        except docker.errors.NotFound as e:
             current_app.logger.error(
-                'Get "{}" service raised '.format(service_name)
-                + '"docker.errors.NotFound" The service does not exist. '
-                + '(job.status: {} -> failed)'.format(job.status)
+                f'Get service "{service_name}" failed '
+                + f'due to "docker.errors.NotFound": {e}'
             )
             job.status = 'failed'
+            return
         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)
+                f'Get service "{service_name}" failed '
+                + f'due to "docker.errors.APIError": {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.'
-            )
+        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'
+            return
+        elif job.status == 'running' and task_state == 'complete':
+            job.status = 'complete'
+            results_dir = os.path.join(job.path, 'output')
+            result_files = [x for x in os.listdir(results_dir) if x.endswith('.zip')]  # noqa
+            for result_file in result_files:
+                job_result = JobResult(filename=result_file, job=job)
+                db.session.add(job_result)
+                db.session.flush()
+                db.session.refresh(job_result)
+        elif job.status == 'running' and task_state == 'failed':
+            job.status = 'failed'
         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
+            return
+        job.end_date = datetime.utcnow()
+        try:
+            service.remove()
+        except docker.errors.APIError as e:
+            current_app.logger.error(
+                f'Remove service "{service_name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
+            )
 
     def remove_job_service(self, job):
-        service_name = 'job_{}'.format(job.id)
+        service_name = f'job_{job.id}'
         try:
             service = self.docker.services.get(service_name)
         except docker.errors.NotFound:
             job.status = 'canceled'
+            return
         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)
+                f'Get service "{service_name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
             )
             return
-        except docker.errors.InvalidVersion:
+        try:
+            service.update(mounts=None)
+        except docker.errors.APIError as e:
             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.'
+                f'Update service "{service_name}" failed '
+                + f'due to "docker.errors.APIError": {e}'
             )
             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)
-                )
+        try:
+            service.remove()
+        except docker.errors.APIError as e:
+            current_app.logger.error(
+                f'Remove "{service_name}" service failed '
+                + f'due to "docker.errors.APIError": {e}'
+            )