Cron by ressource

GeoHealthCheck/app.py

@@ -29,6 +29,7 @@
 #
 # =================================================================
 
+from __future__ import print_function
 import csv
 import logging
 from datetime import datetime, timedelta
@@ -49,6 +50,15 @@
 from factory import Factory
 from util import render_template2, send_email
 import views
+import atexit
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from models import run_resource, flush_runs
+from apscheduler.jobstores.base import JobLookupError
+
+# Create scheduler
+scheduler = BackgroundScheduler()
+
 
 # Module globals for convenience
 LOGGER = logging.getLogger(__name__)
@@ -84,6 +94,33 @@ def db_commit():
     return err
 
 
+def start_crons():
+    # Cold start evry cron of evry ressource
+    for resource in Resource.query.all():
+        freq = resource.test_frequency
+        if freq is None:
+            freq = 60
+        scheduler.add_job(
+                          run_resource, 'interval', [resource.identifier],
+                          minutes=freq,
+                          id=str(resource.identifier))
+
+    # change configuration
+    scheduler.configure(job_defaults={
+        'coalesce': False,
+        'max_instances': 100000
+        })
+
+    scheduler.add_job(flush_runs, 'interval', minutes=1)
+
+    scheduler.start()
+    atexit.register(lambda: scheduler.shutdown())
+
+
+# Start scheduler
+start_crons()
+
+
 @APP.before_request
 def before_request():
     g.user = current_user
@@ -640,6 +677,17 @@ def update(resource_identifier):
         if err:
             status = str(err)
 
+    # Try to remove job from cron
+    try:
+        scheduler.remove_job(str(resource_identifier))
+    except JobLookupError:
+        pass
+    # Add jop to cron
+    scheduler.add_job(
+                      run_resource, 'interval', [resource.identifier],
+                      minutes=resource.test_frequency,
+                      id=str(resource_identifier))
+
     return jsonify({'status': status})
 
 
@@ -708,6 +756,12 @@ def delete(resource_identifier):
 
     DB.session.delete(resource)
 
+    # Delete cron job associated with this resource
+    try:
+        scheduler.remove_job(str(resource_identifier))
+    except JobLookupError:
+        pass
+
     try:
         DB.session.commit()
         flash(gettext('Resource deleted'), 'success')

GeoHealthCheck/config_main.py

@@ -37,7 +37,8 @@
 # Replace None with 'your secret key string' in quotes
 SECRET_KEY = None
 
-GHC_RETENTION_DAYS = 30
+# days, seconds, microseconds, milliseconds, minutes, hours, weeks
+GHC_RETENTION_DAYS = (30, 0, 0, 0, 0, 0, 0)
 GHC_RUN_FREQUENCY = 'hourly'
 GHC_SELF_REGISTER = False
 GHC_NOTIFICATIONS = False

GeoHealthCheck/models.py

@@ -30,7 +30,7 @@
 
 import json
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta
 from sqlalchemy import func
 
 from sqlalchemy.orm import deferred
@@ -69,7 +69,6 @@ def __init__(self, resource, result,
         self.checked_datetime = checked_datetime
         self.message = result.message
         self.report = result.get_report()
-
     # JSON string object specifying report for the Run
     # See http://docs.sqlalchemy.org/en/latest/orm/mapped_attributes.html
     _report = DB.Column("report", DB.Text, default={})
@@ -200,6 +199,7 @@ class Resource(DB.Model):
     owner = DB.relationship('User',
                             backref=DB.backref('username2', lazy='dynamic'))
     tags = DB.relationship('Tag', secondary=resource_tags, backref='resource')
+    test_frequency = DB.Column(DB.Integer)
 
     def __init__(self, owner, resource_type, title, url, tags):
         self.resource_type = resource_type
@@ -497,6 +497,62 @@ def db_commit():
         print(msg)
 
 
+# complete handle of resource test
+def run_resource(resourceid):
+    resource = Resource.query.filter_by(identifier=resourceid).first()
+
+    APP = App.get_app()
+    from healthcheck import run_test_resource
+
+    if not resource.active:
+        # Exit test of resource if it's not active
+        return
+
+    # Get the status of the last run,
+    # assume success if there is none
+    last_run_success = True
+    last_run = resource.last_run
+    if last_run:
+        last_run_success = last_run.success
+
+    # Run test
+    result = run_test_resource(resource)
+
+    run1 = Run(resource, result, datetime.utcnow())
+
+    DB.session.add(run1)
+
+    # commit or rollback each run to avoid long-lived transactions
+    # see https://github.com/geopython/GeoHealthCheck/issues/14
+    db_commit()
+
+    if APP.config['GHC_NOTIFICATIONS']:
+        # Attempt notification
+        try:
+            notify(APP.config, resource, run1, last_run_success)
+        except Exception as err:
+            # Don't bail out on failure in order to commit the Run
+            msg = str(err)
+            print('error notifying: %s' % msg)
+    if not __name__ == '__main__':
+        DB.session.remove()
+
+
+# Complete handle of old runs deletion
+def flush_runs():
+    APP = App.get_app()
+    retention_time = timedelta(*APP.config['GHC_RETENTION_DAYS'])
+
+    all_runs = Run.query.all()
+    for run in all_runs:
+        how_old = (datetime.utcnow() - run.checked_datetime)
+        if how_old > retention_time:
+            DB.session.delete(run)
+    db_commit()
+
+    DB.session.remove()
+
+
 if __name__ == '__main__':
     import sys
 
@@ -556,58 +612,14 @@ def db_commit():
         elif sys.argv[1] == 'run':
             print('START - Running health check tests on %s'
                   % datetime.utcnow().isoformat())
-            from healthcheck import run_test_resource
 
+            print(Resource.query.all())
             for resource in Resource.query.all():  # run all tests
-                print('Testing %s %s' %
-                      (resource.resource_type, resource.url))
-
-                if not resource.active:
-                    print('Resource is not active. Skipping')
-                    continue
-
-                # Get the status of the last run,
-                # assume success if there is none
-                last_run_success = True
-                last_run = resource.last_run
-                if last_run:
-                    last_run_success = last_run.success
-
-                # Run test
-                result = run_test_resource(resource)
-
-                run1 = Run(resource, result)
-
-                print('Adding Run: success=%s, response_time=%ss\n'
-                      % (str(run1.success), run1.response_time))
-
-                DB.session.add(run1)
-
-                # commit or rollback each run to avoid long-lived transactions
-                # see https://github.com/geopython/GeoHealthCheck/issues/14
-                db_commit()
-
-                if APP.config['GHC_NOTIFICATIONS']:
-                    # Attempt notification
-                    try:
-                        notify(APP.config, resource, run1, last_run_success)
-                    except Exception as err:
-                        # Don't bail out on failure in order to commit the Run
-                        msg = str(err)
-                        print('error notifying: %s' % msg)
+                run_resource(resource.identifier)
 
             print('END - Running health check tests on %s'
                   % datetime.utcnow().isoformat())
         elif sys.argv[1] == 'flush':
-            retention_days = int(APP.config['GHC_RETENTION_DAYS'])
-            print('Flushing runs older than %d days' %
-                  retention_days)
-            all_runs = Run.query.all()
-            for run in all_runs:
-                days_old = (datetime.utcnow() - run.checked_datetime).days
-                if days_old > retention_days:
-                    print('Run older than %d days. Deleting' % days_old)
-                    DB.session.delete(run)
-            db_commit()
+            flush_runs()
 
         DB.session.remove()

GeoHealthCheck/templates/edit_resource.html

@@ -44,6 +44,12 @@ <h1 class="page-header">[{{ _('Edit') }}] <span id="resource_title_h1">{{ resour
                     </select>
                 </td>
             </tr>
+            <tr>
+                <th>Test frenqucy</th>
+                <td>
+                    <input type="number" id="input_resource_frequency" name="resource_frenquency_value" value="{{ resource.test_frequency }}" style="width: 100%;"/>
+                </td>
+            </tr>
             <tr>
                 <th>Probes</th>
 
@@ -204,7 +210,10 @@ <h1 class="page-header">[{{ _('Edit') }}] <span id="resource_title_h1">{{ resour
 
         // Collect tags
         var new_tags = $('#resource_tags').val();
-
+
+        // Collect test_frequency
+        var new_frequency = $('input[name="resource_frenquency_value"]').val();
+
         // Collect active
         var new_active = $('#input_resource_active').prop('checked');
 
@@ -270,6 +279,7 @@ <h1 class="page-header">[{{ _('Edit') }}] <span id="resource_title_h1">{{ resour
                 title: new_title,
                 active: new_active,
                 tags: new_tags,
+                test_frequency:new_frequency,
                 probes: new_probes
             }),
             contentType: "application/json; charset=utf-8",

GeoHealthCheck/templates/resource.html

@@ -80,6 +80,12 @@ <h3 class="page-header">{{ _('Monitoring Period') }}: {{ resource.first_run.chec
                     <button class="btn btn-{{ resource.reliability| cssize_reliability }} btn-sm nohover">{{ resource.reliability|round2 }}%</button>
                 </td>
             </tr>
+            <tr>
+              <th>{{ _('Test frenquency (minutes)') }}</th>
+              <td>
+              {{ resource.test_frequency }}
+              </td>
+            </tr>
         </table>
     </div>
     <div class="row">

requirements.txt

@@ -6,3 +6,4 @@ Flask-Script==2.0.5
 Flask-SQLAlchemy==2.1
 OWSLib
 Sphinx
+APScheduler