Merge pull request #27 from lirmm/bugfix/zombie_processes

Bugfix/zombie processes

docs/index.rst

@@ -1,5 +1,11 @@
 Welcome to WAVES Documentation !
 =======================================
+.. warning::
+    If you migrate from 1.6.X to 1.7, and if you have any stored password
+    A security issue has been fixed, you need to update your passwords in databases with this command:
+
+    ./manage.py updates_keys
+
 
 .. toctree::
     :maxdepth: 1

docs/installation.rst

@@ -72,12 +72,19 @@ WAVES is developed with `Django <https://www.djangoproject.com/>`_. You may need
 
     .. code-block:: bash
 
-        (.venv) user@host:~your_app$ ./manage.py wqueue start
         (.venv) user@host:~your_app$ ./manage.py runserver
+        (.venv) user@host:~your_app$ ./manage.py crontab add
 
     Go to http://127.0.0.1:8000/admin to setup your services
     WAVES-core comes with default front pages visible at http://127.0.0.1:8000
 
+    .. seealso::
+        Django crontab for other crontab setup
+
+    .. note::
+        From previous release, a known bug occured while using wqueue command. This bug block you from using this daemon queue.
+        Please use "crontab" instead, and contact us if you experience issues.
+
 
 2. Install WAVES-core inside existing Django project
 ----------------------------------------------------

manage.py

@@ -3,7 +3,7 @@
 import sys
 
 if __name__ == "__main__":
-    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "waves_core.cli")
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "waves_core.settings")
     try:
         from django.core.management import execute_from_command_line
     except ImportError:

requirements.txt

@@ -4,6 +4,7 @@ daemons==1.3.0
 django-admin-sortable2==0.6.19
 django-cors-headers==2.2.0
 django-crispy-forms==1.7.2
+django-crontab==0.7.1
 django-polymorphic==2.0.2
 djangorestframework>=3.9.1
 backports.ssl-match-hostname

waves/wcore/cron/__init__.py

@@ -0,0 +1,16 @@
+# -*- coding: utf-8 -*-
+"""
+.. See the NOTICE file distributed with this work for additional information
+   regarding copyright ownership.
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+"""
+from .process_queue import process_job_queue
+from .purge_jobs import purge_old_jobs

waves/wcore/cron/process_queue.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+"""
+.. See the NOTICE file distributed with this work for additional information
+   regarding copyright ownership.
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+"""
+from __future__ import unicode_literals
+
+import logging
+import datetime
+
+import waves.wcore.exceptions
+from waves.wcore.adaptors.const import JobStatus
+from waves.wcore.adaptors.exceptions import AdaptorException
+from waves.wcore.models import Job
+
+logger = logging.getLogger('waves.cron')
+
+
+def process_job_queue():
+    """
+    Very very simple daemon to monitor jobs queue.
+
+    - Retrieve all current non terminated job, and process according to current status.
+    - Jobs are run on a stateless process
+
+    :return: None
+    """
+    jobs = Job.objects.prefetch_related('job_inputs'). \
+        prefetch_related('outputs').filter(_status__lt=JobStatus.JOB_TERMINATED)
+    if jobs.count() > 0:
+        logger.info("Starting queue process with %i(s) unfinished jobs", jobs.count())
+    for job in jobs:
+        runner = job.adaptor
+        if runner and logger.isEnabledFor(logging.DEBUG):
+            logger.debug('[Runner]-------\n%s\n----------------', runner.dump_config())
+        try:
+            job.check_send_mail()
+            logger.debug("Launching Job %s (adapter:%s)", job, runner)
+            if job.status == JobStatus.JOB_CREATED:
+                job.run_prepare()
+                logger.debug("[PrepareJob] %s (adapter:%s)", job, runner)
+            elif job.status == JobStatus.JOB_PREPARED:
+                logger.debug("[LaunchJob] %s (adapter:%s)", job, runner)
+                job.run_launch()
+            elif job.status == JobStatus.JOB_COMPLETED:
+                job.run_results()
+                logger.debug("[JobExecutionEnded] %s (adapter:%s)", job.get_status_display(), runner)
+            else:
+                job.run_status()
+        except (waves.wcore.exceptions.WavesException, AdaptorException) as e:
+            logger.error("Error Job %s (adapter:%s-state:%s): %s", job, runner, job.get_status_display(),
+                         e.message)
+        except IOError as exc:
+            logger.error('IO error on job %s [%s]', job.slug, exc)
+            job.status = JobStatus.JOB_ERROR
+            job.save()
+        except Exception as exc:
+            logger.exception('Current job raised unrecoverable exception %s', exc)
+            job.fatal_error(exc)
+        finally:
+            logger.info("Queue job terminated at: %s", datetime.datetime.now().strftime('%A, %d %B %Y %H:%M:%I'))
+            job.check_send_mail()
+            if runner is not None:
+                runner.disconnect()

waves/wcore/cron/purge_jobs.py

@@ -0,0 +1,37 @@
+# -*- coding: utf-8 -*-
+"""
+.. See the NOTICE file distributed with this work for additional information
+   regarding copyright ownership.
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+"""
+from __future__ import unicode_literals
+
+import datetime
+import logging
+from itertools import chain
+
+from waves.wcore.models import Job
+
+logger = logging.getLogger('waves.cron')
+
+
+def purge_old_jobs():
+    from waves.wcore.settings import waves_settings
+
+    logger.info("Purge job launched at: %s", datetime.datetime.now().strftime('%A, %d %B %Y %H:%M:%I'))
+    date_anonymous = datetime.date.today() - datetime.timedelta(waves_settings.KEEP_ANONYMOUS_JOBS)
+    date_registered = datetime.date.today() - datetime.timedelta(waves_settings.KEEP_REGISTERED_JOBS)
+    anonymous = Job.objects.filter(client__isnull=True, updated__lt=date_anonymous)
+    registered = Job.objects.filter(client__isnull=False, updated__lt=date_registered)
+    for job in list(chain(*[anonymous, registered])):
+        logger.info('Deleting job %s created on %s', job.slug, job.created)
+        job.delete()
+    logger.info("Purge job terminated at: %s", datetime.datetime.now().strftime('%A, %d %B %Y %H:%M:%I'))

waves/wcore/management/daemoncommand.py

@@ -57,6 +57,10 @@ def handle(self, **options):
         :param options: list of possible django command options
         :return: Nothing
         """
+        # FIXME process queue unterminated processes
+        import warnings
+        warnings.warn("This method has a known bug: please use crontab setup see Docs")
+        exit(0)
         try:
             self.action = options.pop('action')
             if self._class is None:

waves_core/crontab.py

@@ -13,10 +13,6 @@
 
 CLI_LOG_LEVEL = 'WARNING'
 
-INSTALLED_APPS += [
-    'django_crontab'
-]
-
 LOGGING = {
     'version': 1,
     'disable_existing_loggers': False,
@@ -33,7 +29,7 @@
         },
         'log_file': {
             'class': 'logging.handlers.RotatingFileHandler',
-            'filename': os.path.join(LOG_DIR, 'waves-cli.log'),
+            'filename': os.path.join(LOG_DIR, 'waves-cron.log'),
             'formatter': 'verbose',
             'backupCount': 10,
             'maxBytes': 1024 * 1024 * 5
@@ -48,34 +44,19 @@
         },
         'waves': {
             'handlers': ['log_file'],
-            'level': CLI_LOG_LEVEL,
+            'level': 'WARNING',
             'propagate': True,
         },
         'django_crontab': {
             'handlers': ['log_file'],
             'propagate': True,
-            'level': CLI_LOG_LEVEL,
+            'level': 'WARNING',
         },
-        'waves.daemon': {
+        'waves.cron': {
             'handlers': ['log_file'],
             'propagate': False,
-            'level': 'INFO',
-        },
-        'daemons': {
-            'handlers': ['console'],
-            'propagate': False,
-            'level': 'INFO',
-        },
+            'level': 'WARNING',
+        }
 
     }
-}
-
-# CRONTAB JOBS
-CRONTAB_COMMAND_SUFFIX = '2>&1'
-CRONTAB_COMMAND_PREFIX = ''
-CRONTAB_DJANGO_SETTINGS_MODULE = 'waves_core.crontab'
-CRONTAB_LOCK_JOBS = True
-CRONJOBS = [
-    ('*/5 * * * *', 'django.core.management.call_command', ['wqueue', 'start']),
-    ('0 * * * *', 'django.core.management.call_command', ['wpurge', 'start']),
-]
+}

waves_core/settings.py

@@ -10,8 +10,10 @@
 https://docs.djangoproject.com/en/1.11/ref/settings/
 """
 from __future__ import unicode_literals
-import os
+
 import logging.config
+import os
+
 from django.contrib import messages
 
 # python 3 compatibility
@@ -50,7 +52,8 @@
     'crispy_forms',
     'rest_framework',
     'corsheaders',
-    'adminsortable2'
+    'adminsortable2',
+    'django_crontab'
 ]
 
 MIDDLEWARE = [
@@ -250,6 +253,7 @@
 }
 
 configFile = join(BASE_DIR, 'tests', 'settings.ini')
+
 try:
     assert (isfile(configFile))
     Config = ConfigParser.SafeConfigParser()
@@ -276,3 +280,13 @@
 except AssertionError:
     # Don't load variables from ini files they are initialized elsewhere (i.e lab ci)
     pass
+
+# CRONTAB JOBS
+CRONTAB_COMMAND_SUFFIX = '2>&1'
+CRONTAB_COMMAND_PREFIX = ''
+CRONTAB_DJANGO_SETTINGS_MODULE = 'waves_core.crontab'
+CRONTAB_LOCK_JOBS = True
+CRONJOBS = [
+    ('* * * * *', 'waves.wcore.cron.process_job_queue'),
+    ('*/10 * * * *', 'waves.wcore.cron.purge_old_jobs')
+]