Merge pull request #311 from n1analytics/develop

release 1.9.2

Jenkinsfile.groovy

@@ -17,6 +17,16 @@ DockerUtils dockerUtils
 GitCommit gitCommit
 String composeProject
 
+String composeCmd(String composeProject, String cmd) {
+    return sh (
+            script: """ docker-compose --no-ansi \
+                                -f tools/docker-compose.yml \
+                                -f tools/ci.yml \
+                                -p ${composeProject} ${cmd}""",
+            returnStdout: true
+    ).trim()
+}
+
 node('docker&&multicore&&ram') {
 
   stage ('Initialisation') {
@@ -36,19 +46,19 @@ node('docker&&multicore&&ram') {
       try {
         dir("backend") {
           String imageNameLabel = QuayIORepo.ENTITY_SERVICE_APP.getRepo() + ":latest"
-          dockerUtils.dockerCommand("build -t " + imageNameLabel + " .")
+          dockerUtils.dockerCommand("build -t ${imageNameLabel} .")
         }
         dir("frontend") {
           String imageNameLabel = QuayIORepo.ENTITY_SERVICE_NGINX.getRepo() + ":latest"
-          dockerUtils.dockerCommand("build -t " + imageNameLabel + " .")
+          dockerUtils.dockerCommand("build -t ${imageNameLabel} .")
         }
         dir("docs") {
           String imageNameLabel = "quay.io/n1analytics/entity-docs:latest"
-          dockerUtils.dockerCommand("build -t " + imageNameLabel + " .")
+          dockerUtils.dockerCommand("build -t ${imageNameLabel} .")
         }
         dir("benchmarking") {
           String imageNameLabel = "quay.io/n1analytics/entity-benchmark:latest"
-          dockerUtils.dockerCommand("build -t " + imageNameLabel + " .")
+          dockerUtils.dockerCommand("build -t ${imageNameLabel} .")
         }
         gitCommit.setSuccessStatus(gitContextDockerBuild)
       } catch (err) {
@@ -62,15 +72,8 @@ node('docker&&multicore&&ram') {
       gitCommit.setInProgressStatus(gitContextComposeDeploy);
       try {
         echo("Start all the containers (including tests)")
-        sh """
-        docker-compose -v
-        docker-compose \
-            -f tools/docker-compose.yml \
-            -f tools/ci.yml \
-            -p ${composeProject} up -d \
-            db minio redis backend db_init worker nginx tests
-        """
-
+        sh "docker-compose -v"
+          composeCmd(composeProject, "up -d db minio redis backend db_init worker nginx tests")
         gitCommit.setSuccessStatus(gitContextComposeDeploy)
       } catch (err) {
         print("Error in compose deploy stage:\n" + err)
@@ -82,15 +85,16 @@ node('docker&&multicore&&ram') {
     stage('Integration Tests') {
       gitCommit.setInProgressStatus(gitContextIntegrationTests);
       try {
-        DockerContainer containerTests = new DockerContainer(dockerUtils, composeProject + "_tests_1")
-        sleep 2
+        testContainerID = composeCmd(composeProject,"ps -q tests")
+        DockerContainer containerTests = new DockerContainer(dockerUtils, testContainerID)
+        sleep 30
         timeout(time: 60, unit: 'MINUTES') {
-          containerTests.watchLogs()
 
-          sh("docker logs " + composeProject + "_nginx_1" + " &> nginx.log")
-          sh("docker logs " + composeProject + "_backend_1" + " &> backend.log")
-          sh("docker logs " + composeProject + "_worker_1" + " &> worker.log")
-          sh("docker logs " + composeProject + "_db_1" + " &> db.log")
+          containerTests.watchLogs()
+          composeCmd(composeProject, "logs nginx &> nginx.log")
+          composeCmd(composeProject, "logs backend &> backend.log")
+          composeCmd(composeProject, "logs worker &> worker.log")
+          composeCmd(composeProject, "logs db &> db.log")
 
           archiveArtifacts artifacts: "*.log", fingerprint: false
           if (containerTests.getExitCode() != "0") {
@@ -212,8 +216,7 @@ node('docker&&multicore&&ram') {
     stage("Cleaning") {
       try {
         dockerUtils.dockerLogoutQuayIOWithoutFail()
-        String cmdTearDown = "docker-compose -f tools/docker-compose.yml -f tools/ci.yml -p " + composeProject + " down -v"
-        sh cmdTearDown
+        composeCmd(composeProject, " down -v")
       } catch(Exception err) {
         print("Error in cleaning stage, but do nothing about it:\n" + err)
         // Do nothing on purpose.
@@ -279,8 +282,7 @@ node('helm && kubectl') {
                     -f values.yaml -f minimal-values.yaml -f test-versions.yaml \
                     --set api.app.debug=true \
                     --set api.ingress.enabled=false \
-                    --set api.certManager.enabled=false \
-                    --set provision.redis=true
+                    --set api.certManager.enabled=false                    
                 """
               // give the cluster a chance to provision volumes etc, assign an IP to the service, then create a new job to test it
               sleep(time: 120, unit: "SECONDS")
@@ -292,7 +294,7 @@ node('helm && kubectl') {
 
             pvc = createK8sTestJob(DEPLOYMENT, QuayIORepo.ENTITY_SERVICE_APP.getRepo() + ":" + TAG, serviceIP)
 
-            sleep(time: 60, unit: "SECONDS")
+            sleep(time: 120, unit: "SECONDS")
 
             def jobPodName = sh(script: """
                 kubectl get pods -l deployment=${DEPLOYMENT} -o jsonpath="{.items[0].metadata.name}"
@@ -482,6 +484,7 @@ String createK8sTestJob(String deploymentName, String imageNameWithTag, String s
                                                           "-m",
                                                           "pytest",
                                                           "entityservice/tests",
+                                                          "-x",
                                                           "--junit-xml=/mnt/results.xml"
                                                   ],
                                                   "volumeMounts": [[

backend/entityservice/VERSION

@@ -1 +1 @@
-v1.9.1
+v1.9.2

backend/entityservice/__init__.py

@@ -31,10 +31,6 @@
 
 # Logging setup
 logger = structlog.get_logger()
-if config.LOGFILE is not None:
-    fileHandler = logging.FileHandler(config.LOGFILE)
-    fileHandler.setLevel(logging.INFO)
-    fileHandler.setFormatter(config.fileFormat)
 consoleHandler = logging.StreamHandler()
 consoleHandler.setLevel(logging.DEBUG)
 structlog.configure(
@@ -62,7 +58,10 @@
     app.logger.setLevel(gunicorn_logger.level)
 
 if config.LOGFILE is not None:
-   app.logger.addHandler(fileHandler)
+    fileHandler = logging.FileHandler(config.LOGFILE)
+    fileHandler.setLevel(logging.INFO)
+    fileHandler.setFormatter(config.fileFormat)
+    app.logger.addHandler(fileHandler)
 
 # Tracer setup (currently just trace all requests)
 flask_tracer = FlaskTracer(initialize_tracer, True, app)

backend/entityservice/api_def/swagger.yaml

@@ -193,10 +193,6 @@ paths:
         If the `result_type` is `"mapping"` or `"similarity_scores"` then the results can be accessed with the
         `result_token``token`, which is provided when initially creating the mapping.
 
-        - `"permutation"`\
-        If the `result_type` is `"permutation"` then the data provider can access their respective permutation and the
-        encrypted mask with their respective `receipt_token`, which they obtain when adding data to the mapping.
-
         - `"permutations"`\
         If the `result_type` is `permutations`, then the data providers can access their respective permutation with
         their individual `receipt_token`, which they obtain when adding data to the mapping.

backend/entityservice/async_worker.py

@@ -1,25 +1,25 @@
 import logging
 
 from celery import Celery
-from celery.signals import task_prerun, task_postrun
+from celery.signals import task_prerun
 import structlog
 
-from entityservice.settings import Config as config
+from entityservice.settings import Config
 
 
 celery = Celery('tasks',
-                broker=config.BROKER_URL,
-                backend=config.CELERY_RESULT_BACKEND
+                broker=Config.BROKER_URL,
+                backend=Config.CELERY_RESULT_BACKEND
                 )
 
 celery.conf.CELERY_TASK_SERIALIZER = 'json'
 celery.conf.CELERY_ACCEPT_CONTENT = ['json']
 celery.conf.CELERY_RESULT_SERIALIZER = 'json'
-celery.conf.CELERY_ANNOTATIONS = config.CELERY_ANNOTATIONS
-celery.conf.CELERYD_PREFETCH_MULTIPLIER = config.CELERYD_PREFETCH_MULTIPLIER
-celery.conf.CELERYD_MAX_TASKS_PER_CHILD = config.CELERYD_MAX_TASKS_PER_CHILD
-celery.conf.CELERY_ACKS_LATE = config.CELERY_ACKS_LATE
-celery.conf.CELERY_ROUTES = config.CELERY_ROUTES
+celery.conf.CELERY_ANNOTATIONS = Config.CELERY_ANNOTATIONS
+celery.conf.CELERYD_PREFETCH_MULTIPLIER = Config.CELERYD_PREFETCH_MULTIPLIER
+celery.conf.CELERYD_MAX_TASKS_PER_CHILD = Config.CELERYD_MAX_TASKS_PER_CHILD
+celery.conf.CELERY_ACKS_LATE = Config.CELERY_ACKS_LATE
+celery.conf.CELERY_ROUTES = Config.CELERY_ROUTES
 
 
 structlog.configure(
@@ -35,11 +35,16 @@
     logger_factory=structlog.stdlib.LoggerFactory(),
 )
 
+# Set logging level of other python libraries that we use
+logging.getLogger('celery').setLevel(logging.WARNING)
+logging.getLogger('jaeger_tracing').setLevel(logging.WARNING)
+
+# Set up our logging
 logger = structlog.wrap_logger(logging.getLogger('celery.es'))
-if config.DEBUG:
-    logging.getLogger('celery').setLevel(logging.INFO)
+if Config.DEBUG:
     logging.getLogger('celery.es').setLevel(logging.DEBUG)
-    logging.getLogger('jaeger_tracing').setLevel(logging.WARNING)
+    logging.getLogger('celery').setLevel(logging.INFO)
+
 
 logger.info("Setting up celery worker")
 logger.debug("Debug logging enabled")
@@ -52,4 +57,3 @@ def configure_structlog(sender, body=None, **kwargs):
         task_id=kwargs['task_id'],
         task_name=sender.__name__
     )
-

backend/entityservice/database/authorization.py

@@ -15,4 +15,4 @@ def check_project_auth(db, project_id, results_token):
 
 
 def check_update_auth(db, update_token):
-    return get_dataprovider_id(db, update_token) is not None
+    return get_dataprovider_id(db, update_token) is not None

backend/entityservice/database/insertions.py

@@ -1,7 +1,6 @@
-# Insertion Queries
-
 import psycopg2
 import psycopg2.extras
+
 from entityservice.database.util import execute_returning_id, logger
 from entityservice.errors import RunDeleted
 

backend/entityservice/database/metrics.py

@@ -1,6 +1,6 @@
 from datetime import timedelta
 
-from entityservice.database.util import query_db, execute_returning_id, logger
+from entityservice.database.util import query_db, execute_returning_id
 
 
 def get_latest_rate(db):
@@ -37,3 +37,13 @@ def insert_comparison_rate(cur, rate):
         RETURNING id;
         """
     return execute_returning_id(cur, insertion_stmt, [rate])
+
+
+def get_elapsed_run_times(db):
+    elapsed_run_time_query = """
+        select run_id, project as project_id, (time_completed - time_started) as elapsed
+        from runs
+        WHERE
+          runs.state='completed'
+        """
+    return query_db(db, elapsed_run_time_query)

backend/entityservice/database/selections.py

@@ -130,6 +130,20 @@ def get_runs(db, project_id):
     return query_db(db, select_query, [project_id], one=False)
 
 
+def get_run_state_for_update(db, run_id):
+    """
+    Get the current run state and acquire a row lock
+    (or fail fast if row already locked)
+    """
+    sql_query = """
+        SELECT state from runs
+        WHERE 
+            run_id = %s
+        FOR UPDATE NOWAIT
+        """
+    return query_db(db, sql_query, [run_id], one=True)['state']
+
+
 def get_run(db, run_id):
     sql_query = """
         SELECT * from runs

backend/entityservice/database/util.py

@@ -1,12 +1,11 @@
-import logging
 import time
 
 import psycopg2
 import psycopg2.extras
 from flask import current_app, g
 from structlog import get_logger
-from entityservice import database as db
-from entityservice.errors import DatabaseInconsistent, DBResourceMissing
+
+from entityservice import database
 from entityservice.settings import Config as config
 
 logger = get_logger()
@@ -102,5 +101,5 @@ def get_db():
     conn = getattr(g, 'db', None)
     if conn is None:
         logger.debug("Caching a new database connection in application context")
-        conn = g.db = db.connect_db()
+        conn = g.db = database.connect_db()
     return conn

backend/entityservice/models/project.py

@@ -1,7 +1,8 @@
+from structlog import get_logger
+
 from entityservice.messages import INVALID_RESULT_TYPE_MESSAGE
 from entityservice.utils import generate_code
 import entityservice.database as db
-from structlog import get_logger
 
 logger = get_logger()
 

backend/entityservice/models/run.py

@@ -1,10 +1,10 @@
+from structlog import get_logger
+
 import entityservice.database as db
-from entityservice import app
 from entityservice import cache
 from entityservice.settings import Config as config
 from entityservice.utils import generate_code
 
-from structlog import get_logger
 
 logger = get_logger()
 

backend/entityservice/settings.py

@@ -36,7 +36,7 @@ class Config(object):
     DATABASE_PASSWORD = os.getenv('DATABASE_PASSWORD', '')
 
     BROKER_URL = os.getenv('CELERY_BROKER_URL',
-                                'redis://:{}@{}:6379/0'.format(REDIS_PASSWORD, REDIS_SERVER))
+                           'redis://:{}@{}:6379/0'.format(REDIS_PASSWORD, REDIS_SERVER))
 
     CELERY_RESULT_BACKEND = BROKER_URL
     CELERY_ANNOTATIONS = {

backend/entityservice/tasks/base_task.py

@@ -1,15 +1,16 @@
-from entityservice.async_worker import celery, logger
-from entityservice.database import logger, DBConn, update_run_mark_failure
+import time
+from abc import ABC
 
-from entityservice.tracing import create_tracer
 from opentracing.propagation import Format
+import psycopg2
 
-import time
+from entityservice.async_worker import celery, logger
+from entityservice.database import DBConn, update_run_mark_failure
+from entityservice.tracing import create_tracer
 from entityservice.errors import DBResourceMissing
-import psycopg2
 
 
-class BaseTask(celery.Task):
+class BaseTask(celery.Task, ABC):
     """Abstract base class for all tasks in Entity Service"""
 
     abstract = True
@@ -86,7 +87,7 @@ def __call__(self, *args, **kwargs):
             return super(TracedTask, self).__call__(*args, **kwargs)
 
     def after_return(self, status, retval, task_id, args, kwargs, einfo):
-        time.sleep(2) # jaeger bug
+        time.sleep(2)  # jaeger bug
         self.tracer.close()
         self._tracer = None
         return super(TracedTask, self).after_return(status, retval, task_id, args, kwargs, einfo)