Merge pull request #623 from maertsen/initial-batch-fixes

Quality of life fixes for batch test development

checks/batch/scheduler.py

@@ -582,17 +582,16 @@ def _run_scheduler():
     logger.info("Found {} domains".format(found_domains))
 
 
-if settings.ENABLE_BATCH:
-    @app.task(name='run_batch')
-    def run():
-        """
-        Run the scheduler every interval only if it is not running already.
+@batch_shared_task
+def run():
+    """
+    Run the scheduler every interval only if it is not running already.
 
-        """
-        lock_id = redis_id.batch_scheduler_lock.id
-        lock_ttl = redis_id.batch_scheduler_lock.ttl
-        with util.memcache_lock(lock_id, lock_ttl) as acquired:
-            if acquired:
-                _run_scheduler()
-                return
-        logger.info("Already running...")
+    """
+    lock_id = redis_id.batch_scheduler_lock.id
+    lock_ttl = redis_id.batch_scheduler_lock.ttl
+    with util.memcache_lock(lock_id, lock_ttl) as acquired:
+        if acquired:
+            _run_scheduler()
+            return
+    logger.info("Already running...")

checks/batch/util.py

@@ -1,6 +1,8 @@
 # Copyright: 2019, NLnet Labs and the Internet.nl contributors
 # SPDX-License-Identifier: Apache-2.0
 import inspect
+from ipaddress import ip_address
+from json.decoder import JSONDecodeError
 import random
 import re
 from time import monotonic
@@ -731,6 +733,9 @@ def register_request(request, *args, **kwargs):
             return bad_client_request_response(
                 "'domains' is missing from the request.")
         name = json_req.get('name', 'no-name')
+    except JSONDecodeError:
+        return bad_client_request_response(
+            "Problem parsing json. Did you supply a 'type' and 'domains'?")
     except Exception:
         return general_server_error_response("Problem parsing domains.")
 
@@ -811,7 +816,9 @@ def patch_request(request, batch_request):
         BatchDomain.objects.filter(batch_request=batch_request).update(
             status=BatchDomainStatus.cancelled)
         return api_response({"request": batch_request.to_api_dict()})
-
+    except JSONDecodeError:
+        return bad_client_request_response(
+            "Problem parsing json. Did you supply a 'status'?")
     except Exception:
         return general_server_error_response(
             "Problem cancelling the batch request.")

checks/tasks/update.py

@@ -8,7 +8,7 @@
 from .. import redis_id
 from ..models import DomainTestReport, MailTestReport
 from ..batch import util
-from internetnl.celery import app
+from celery import shared_task
 
 logger = get_task_logger(__name__)
 
@@ -171,12 +171,10 @@ def _update_hof():
             cache.set(cache_id, cached_data, cache_ttl)
 
 
-# Disable HoF when on batch mode, too much DB activity.
-if not settings.ENABLE_BATCH:
-    @app.task(name='update_HoF_ranking')
-    def ranking():
-        lock_id = redis_id.hof_lock.id
-        lock_ttl = redis_id.hof_lock.ttl
-        with util.memcache_lock(lock_id, lock_ttl) as acquired:
-            if acquired:
-                _update_hof()
+@shared_task
+def update_hof():
+    lock_id = redis_id.hof_lock.id
+    lock_ttl = redis_id.hof_lock.ttl
+    with util.memcache_lock(lock_id, lock_ttl) as acquired:
+        if acquired:
+            _update_hof()

docker/docker-compose.yml

@@ -1,4 +1,4 @@
-version: '2.0'
+version: '2.1'
 
 services:
   app:
@@ -14,14 +14,16 @@ services:
       - redis
       - rabbitmq
       - postgres
+    environment:
+      - ENABLE_BATCH=${ENABLE_BATCH:-False}
 
   redis:
     network_mode: host
     image: redis:alpine
 
   rabbitmq:
     network_mode: host
-    image: rabbitmq:alpine
+    image: rabbitmq:management-alpine
 
   postgres:
     network_mode: host

docker/entrypoint.sh

@@ -4,6 +4,7 @@ set -exvu
 
 ADMIN_EMAIL=${ADMIN_EMAIL:-admin@i.dont.exist}
 CACHE_TTL=${CACHE_TTL:-200}
+ENABLE_BATCH=${ENABLE_BATCH:-False}
 POSTGRES_HOST=${POSTGRES_HOST:-localhost}
 POSTGRES_USER=${POSTGRES_USER:-internetnl}
 POSTGRES_PASSWORD=${POSTGRES_PASSWORD:-password}
@@ -29,6 +30,7 @@ ldns-dane -n -T verify ${LDNS_DANE_VALIDATION_DOMAIN} 443 || echo >&2 "ERROR: Pl
 # overridden at container creation time.
 sed \
     -e "s|DEBUG = False|DEBUG = True|g" \
+    -e "s|ENABLE_BATCH = False|ENABLE_BATCH = ${ENABLE_BATCH}|g" \
     -e "s|localhost:15672|${RABBITMQ_HOST}:15672|g" \
     -e "s|localhost:6379|${REDIS_HOST}:6379|g" \
     -e "s|BROKER_URL = 'amqp://guest@localhost//'|BROKER_URL = 'amqp://guest@${RABBITMQ_HOST}//'|g" \
@@ -41,6 +43,29 @@ sed \
     -e "s|CACHE_TTL = .*|CACHE_TTL = ${CACHE_TTL}|g" \
     ${APP_PATH}/internetnl/settings.py-dist > ${APP_PATH}/internetnl/settings.py
 
+# configure Django logging
+cat << EOF >> ${APP_PATH}/internetnl/settings.py
+if DEBUG:
+    LOGGING = {
+        'version': 1,
+        'disable_existing_loggers': False,
+        'handlers': {
+            'file': {
+                'level': 'INFO',
+                'class': 'logging.FileHandler',
+                'filename': 'django.log',
+            },
+        },
+        'loggers': {
+            'django': {
+                'handlers': ['file'],
+                'level': 'INFO',
+                'propagate': True,
+            },
+        },
+    }
+EOF
+
 # Prepare translations for use
 cd ${APP_PATH}/checks
 ../manage.py compilemessages
@@ -52,18 +77,44 @@ docker/postgres-ping.sh postgresql://${POSTGRES_USER}@${POSTGRES_HOST}/${POSTGRE
 # Prepare the database for use
 ./manage.py migrate
 
+# Optional steps for the batch dev environment
+if [ ${ENABLE_BATCH} == "True" ]; then
+    # create indexes
+    ./manage.py api_create_db_indexes
+    # guarantee the existence of a test_user in the db
+    ./manage.py api_users register -u test_user -n test_user -o test_user -e test_user || :
+    # generate API documentation
+    cp ${APP_PATH}/internetnl/batch_api_doc_conf.py{-dist,}
+    ln -sf ${APP_PATH}/checks/static ${APP_PATH}/static # static/ is not served, checks/static is
+    ./manage.py api_generate_doc # creates openapi.yaml in static/
+fi
+
 # Start Celery
-celery -A internetnl multi start \
-    worker db_worker slow_db_worker \
-    -c:1 5 -c:2 1 -Q:2 db_worker -c:3 3 -Q:3 slow_db_worker \
-    -l info --without-gossip --time-limit=300 --pidfile='/app/%n.pid' \
-    --logfile='/app/%n%I.log' -P eventlet &
+if [ ${ENABLE_BATCH} == "True" ]; then
+    celery -A internetnl multi start \
+	worker db_worker slow_db_worker \
+    batch_scheduler batch_main batch_callback batch_slow \
+	-c:1 5 -Q:1 celery -c:2 1 -Q:2 db_worker -c:3 3 -Q:3 slow_db_worker \
+    -c:4 1 -Q batch_scheduler -c:5 5 -Q:5 batch_main -c:6 1 -Q:6 batch_callback -c:7 1 -Q:7 batch_slow \
+        -l info --without-gossip --time-limit=300 --pidfile='/app/%n.pid' \
+        --logfile='/app/%n%I.log' -P eventlet &
+else
+    celery -A internetnl multi start \
+        worker db_worker slow_db_worker \
+        -c:1 5 -c:2 1 -Q:2 db_worker -c:3 3 -Q:3 slow_db_worker \
+        -l info --without-gossip --time-limit=300 --pidfile='/app/%n.pid' \
+        --logfile='/app/%n%I.log' -P eventlet &
+fi
 
 # Start Celery Beat
 celery -A internetnl beat &
 
 # Wait a little while for all 3 Celery worker groups to become ready
-docker/celery-ping.sh 3
+if [ ${ENABLE_BATCH} == "True" ]; then
+    docker/celery-ping.sh 7 20
+else
+    docker/celery-ping.sh 3
+fi
 
 # Tail the Celery log files so that they appear in Docker logs output
 tail -F -n 1000 *.log &

internetnl/celery.py

@@ -9,7 +9,7 @@
 
 app = Celery('internetnl')
 
-app.config_from_object('django.conf:settings')
+app.config_from_object('django.conf:settings', namespace='CELERY')
 
 app.autodiscover_tasks()
 
@@ -20,15 +20,15 @@ def debug_task(self):
 if app.conf.ENABLE_BATCH:
     app.conf.beat_schedule = {
         'run_batch': {
-                'task': 'tasks.run_batch',
-                'schedule': app.conf.BATCH_SCHEDULER_INTERVAL,
+                'task': 'checks.batch.scheduler.run',
+                'schedule': app.conf.BATCH_SCHEDULER_INTERVAL
         }
     }
 else:
     # Disable HoF when on batch mode, too much DB activity.
     app.conf.beat_schedule = {
         'generate_HoF': {
-                'task': 'update_HoF_ranking',
-                'schedule': crontab(hour='*', minute='*/10', day_of_week='*'),
+                'task': 'checks.tasks.update.update_hof',
+                'schedule': app.conf.HOF_UPDATE_INTERVAL
         }
     }

internetnl/settings.py-dist

@@ -55,7 +55,7 @@ if ENABLE_BATCH:
 
     # Note that all the following queues need to be defined in the celery
     # service configuration.
-    CELERY_ROUTES = {
+    CELERY_BATCH_TASK_ROUTES = {
         'checks.tasks.dnssec.batch_mail_callback': {'queue': 'batch_callback'},
         'checks.tasks.dnssec.batch_mail_is_secure': {'queue': 'batch_main'},
         'checks.tasks.dnssec.batch_web_callback': {'queue': 'batch_callback'},
@@ -85,8 +85,10 @@ if ENABLE_BATCH:
         'checks.tasks.appsecpriv.batch_web_appsecpriv': {'queue': 'batch_main'},
         'checks.tasks.appsecpriv.batch_web_callback': {'queue': 'batch_callback'},
 
-        'checks.batch.util.batch_async_generate_results': {'queue': 'worker_slow'},
-        'checks.batch.util.batch_async_register': {'queue': 'worker_slow'},
+        'checks.batch.util.batch_async_generate_results': {'queue': 'batch_slow'},
+        'checks.batch.util.batch_async_register': {'queue': 'batch_slow'},
+
+        'checks.batch.scheduler.run': {'queue': 'batch_scheduler'},
     }
 
     # Custom results for the /results endpoint.
@@ -226,10 +228,10 @@ STATICFILES_DIRS = [
 
 # --- Celery configuration
 #
-BROKER_URL = 'amqp://guest@localhost//'
+CELERY_BROKER_URL = 'amqp://guest@localhost//'
 CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
-CELERY_TASK_RESULT_EXPIRES = 7200
-BROKER_HEARTBEAT = 0  # Workaround for https://github.com/celery/celery/issues/4817
+CELERY_RESULT_EXPIRES = 7200
+CELERY_BROKER_HEARTBEAT = 0  # Workaround for https://github.com/celery/celery/issues/4817
 CELERY_TASK_ACKS_LATE = True
 CELERY_WORKER_PREFETCH_MULTIPLIER = 1
 
@@ -246,7 +248,7 @@ CELERY_ACCEPT_CONTENT = ['pickle']
 
 # Note that all the following queues need to be defined in the celery
 # service configuration.
-CELERY_ROUTES = {
+CELERY_TASK_ROUTES = {
         'checks.tasks.dnssec.mail_callback': {'queue': 'db_worker'},
         'checks.tasks.dnssec.web_callback': {'queue': 'db_worker'},
 
@@ -262,9 +264,12 @@ CELERY_ROUTES = {
 
         'checks.views.shared.run_stats_queries': {'queue': 'slow_db_worker'},
         'checks.views.shared.update_running_status': {'queue': 'slow_db_worker'},
-        'checks.tasks.update.ranking': {'queue': 'slow_db_worker'},
+        'checks.tasks.update.update_hof': {'queue': 'slow_db_worker'},
 }
 
+if ENABLE_BATCH:
+    CELERY_TASK_ROUTES.update(CELERY_BATCH_TASK_ROUTES)
+
 # Shared task timings
 SHARED_TASK_SOFT_TIME_LIMIT_HIGH = 90
 SHARED_TASK_TIME_LIMIT_HIGH = 100
@@ -335,6 +340,8 @@ MATOMO_SITEID = "site_id"
 # Used for subdomain tracking eg. *.internet.nl
 MATOMO_SUBDOMAIN_TRACKING = ""
 
+# --- HoF update interval
+HOF_UPDATE_INTERVAL = 600 # seconds
 
 # --- Extra manual HoF page(s)
 #