From b53c02e76689b21f2e991f0ca0c55a190c022168 Mon Sep 17 00:00:00 2001 From: Just van den Broecke Date: Tue, 20 Mar 2018 18:18:44 +0100 Subject: [PATCH 01/20] #132 first stab at healthcheck scheduler using daemon process --- GeoHealthCheck/app.py | 11 +- GeoHealthCheck/healthcheck.py | 65 +++++- GeoHealthCheck/init.py | 63 +++--- .../migrations/versions/34531bfd7cab_.py | 47 ++++ GeoHealthCheck/models.py | 125 ++++++++--- GeoHealthCheck/scheduler.py | 207 ++++++++++++++++++ GeoHealthCheck/templates/edit_resource.html | 10 + GeoHealthCheck/templates/resource.html | 6 + .../translations/en/LC_MESSAGES/messages.po | 10 + .../nl_NL/LC_MESSAGES/messages.po | 10 + pavement.py | 6 + 11 files changed, 485 insertions(+), 75 deletions(-) create mode 100644 GeoHealthCheck/migrations/versions/34531bfd7cab_.py create mode 100644 GeoHealthCheck/scheduler.py diff --git a/GeoHealthCheck/app.py b/GeoHealthCheck/app.py index a8be709a..aaebe5b6 100644 --- a/GeoHealthCheck/app.py +++ b/GeoHealthCheck/app.py @@ -161,11 +161,11 @@ def cssize_reliability(value, css_type=None): score = 'danger' panel = 'red' elif (CONFIG['GHC_RELIABILITY_MATRIX']['orange']['min'] <= number <= - CONFIG['GHC_RELIABILITY_MATRIX']['orange']['max']): + CONFIG['GHC_RELIABILITY_MATRIX']['orange']['max']): score = 'warning' panel = 'yellow' elif (CONFIG['GHC_RELIABILITY_MATRIX']['green']['min'] <= number <= - CONFIG['GHC_RELIABILITY_MATRIX']['green']['max']): + CONFIG['GHC_RELIABILITY_MATRIX']['green']['max']): score = 'success' panel = 'green' else: # should never really get here @@ -556,7 +556,7 @@ def add(): param_vals = {} for param in param_defs: if param_defs[param]['value']: - param_vals[param] =\ + param_vals[param] = \ param_defs[param]['value'] check_vars = CheckVars( probe_to_add, check_class, param_vals) @@ -744,10 +744,6 @@ def delete(resource_identifier): flash(gettext('Resource not found'), 'danger') return redirect(url_for('home', lang=g.current_lang)) - runs = Run.query.filter_by(resource_identifier=resource_identifier).all() - - for run in runs: - DB.session.delete(run) resource.clear_recipients() DB.session.delete(resource) @@ -898,6 +894,7 @@ def api_probes_avail(resource_type=None, resource_id=None): if __name__ == '__main__': # run locally, for fun import sys + logging.basicConfig() HOST = '0.0.0.0' PORT = 8000 diff --git a/GeoHealthCheck/healthcheck.py b/GeoHealthCheck/healthcheck.py index ae0c1df6..cb5624a8 100644 --- a/GeoHealthCheck/healthcheck.py +++ b/GeoHealthCheck/healthcheck.py @@ -27,13 +27,14 @@ # # ================================================================= -import datetime +from datetime import datetime import logging import json from urllib2 import urlopen from urlparse import urlparse from functools import partial from pprint import pprint +from flask_babel import gettext from owslib.wms import WebMapService from owslib.wmts import WebMapTileService @@ -44,12 +45,66 @@ from owslib.csw import CatalogueServiceWeb from owslib.sos import SensorObservationService -from flask_babel import gettext +from init import App from enums import RESOURCE_TYPES +from models import Resource, Run from probe import Probe from result import ResourceResult +from notifications import notify LOGGER = logging.getLogger(__name__) +APP = App.get_app() +DB = App.get_db() + + +# commit or rollback shorthand +def db_commit(): + err = None + try: + DB.session.commit() + except Exception as err: + DB.session.rollback() + # finally: + # DB.session.close() + return err + + +# complete handle of resource test +def run_resource(resourceid): + resource = Resource.query.filter_by(identifier=resourceid).first() + + 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) + logging.warn('error notifying: %s' % msg) + if not __name__ == '__main__': + DB.session.remove() def run_test_resource(resource): @@ -81,7 +136,7 @@ def sniff_test_resource(config, resource_type, url): raise RuntimeError(msg2) title = None - start_time = datetime.datetime.utcnow() + start_time = datetime.utcnow() message = None resource_type_map = {'OGC:WMS': [partial(WebMapService, version='1.3.0'), partial(WebMapService, version='1.1.1')], @@ -153,7 +208,7 @@ def sniff_test_resource(config, resource_type, url): title = urlparse(url).hostname elif resource_type == 'OSGeo:GeoNode': endpoints = ows - end_time = datetime.datetime.utcnow() + end_time = datetime.utcnow() delta = end_time - start_time response_time = '%s.%s' % (delta.seconds, delta.microseconds) base_tags = geonode_make_tags(url) @@ -184,7 +239,7 @@ def sniff_test_resource(config, resource_type, url): message = msg success = False - end_time = datetime.datetime.utcnow() + end_time = datetime.utcnow() delta = end_time - start_time response_time = '%s.%s' % (delta.seconds, delta.microseconds) diff --git a/GeoHealthCheck/init.py b/GeoHealthCheck/init.py index 7be7d8bb..a7017762 100644 --- a/GeoHealthCheck/init.py +++ b/GeoHealthCheck/init.py @@ -29,6 +29,7 @@ import os import sys +import logging from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_babel import Babel @@ -55,69 +56,71 @@ class App: babel_instance = None plugins_instance = None home_dir = None + count = 0 @staticmethod def init(): # Do init once - if not App.app_instance: - app = Flask(__name__) + app = Flask(__name__) - # Read and override configs - app.config.from_pyfile('config_main.py') - app.config.from_pyfile('../instance/config_site.py') + # Read and override configs + app.config.from_pyfile('config_main.py') + app.config.from_pyfile('../instance/config_site.py') - app.config['GHC_SITE_URL'] = \ - app.config['GHC_SITE_URL'].rstrip('/') + logging.basicConfig(level=logging.INFO) + if app.config['DEBUG'] is True: + logging.basicConfig(level=logging.DEBUG) - app.secret_key = app.config['SECRET_KEY'] + app.config['GHC_SITE_URL'] = \ + app.config['GHC_SITE_URL'].rstrip('/') - App.db_instance = SQLAlchemy(app) - App.babel_instance = Babel(app) + app.secret_key = app.config['SECRET_KEY'] - # Plugins (via Docker ENV) must be list, but may have been - # specified as comma-separated string, or older set notation - app.config['GHC_PLUGINS'] = to_list(app.config['GHC_PLUGINS']) - app.config['GHC_USER_PLUGINS'] = \ - to_list(app.config['GHC_USER_PLUGINS']) + App.db_instance = SQLAlchemy(app) + App.babel_instance = Babel(app) - # Concatenate core- and user-Plugins - App.plugins_instance = \ - app.config['GHC_PLUGINS'] + app.config['GHC_USER_PLUGINS'] + # Plugins (via Docker ENV) must be list, but may have been + # specified as comma-separated string, or older set notation + app.config['GHC_PLUGINS'] = to_list(app.config['GHC_PLUGINS']) + app.config['GHC_USER_PLUGINS'] = \ + to_list(app.config['GHC_USER_PLUGINS']) - # Needed to find Plugins - home_dir = os.path.dirname(os.path.abspath(__file__)) - App.home_dir = sys.path.append('%s/..' % home_dir) + # Concatenate core- and user-Plugins + App.plugins_instance = \ + app.config['GHC_PLUGINS'] + app.config['GHC_USER_PLUGINS'] - # Finally assign app-instance - App.app_instance = app - print("init.py: created GHC App instance") + # Needed to find Plugins + home_dir = os.path.dirname(os.path.abspath(__file__)) + App.home_dir = sys.path.append('%s/..' % home_dir) + + # Finally assign app-instance + App.app_instance = app + App.count += 1 + logging.info("init.py: created GHC App instance #%d" % App.count) @staticmethod def get_app(): - App.init() return App.app_instance @staticmethod def get_babel(): - App.init() return App.babel_instance @staticmethod def get_config(): - App.init() return App.app_instance.config @staticmethod def get_db(): - App.init() return App.db_instance @staticmethod def get_home_dir(): - App.init() return App.home_dir @staticmethod def get_plugins(): - App.init() return App.plugins_instance + + +App.init() diff --git a/GeoHealthCheck/migrations/versions/34531bfd7cab_.py b/GeoHealthCheck/migrations/versions/34531bfd7cab_.py new file mode 100644 index 00000000..71d76b52 --- /dev/null +++ b/GeoHealthCheck/migrations/versions/34531bfd7cab_.py @@ -0,0 +1,47 @@ +"""empty message + +Revision ID: 34531bfd7cab +Revises: bb91fb332c36 +Create Date: 2018-03-19 16:59:42.235474 + +""" +from alembic import op +import sqlalchemy as sa +from GeoHealthCheck.migrations import alembic_helpers + +# revision identifiers, used by Alembic. +revision = '34531bfd7cab' +down_revision = 'bb91fb332c36' +branch_labels = None +depends_on = None + + +def upgrade(): + if not alembic_helpers.table_has_column('resource', 'run_frequency'): + print('Column run_frequency not present in resource table, will create') + op.add_column(u'resource', sa.Column('run_frequency', sa.Integer(), + nullable=False, default=60, server_default=60)) + else: + print('Column run_frequency already present in resource table') + + if not alembic_helpers.tables_exist(['resource_lock']): + print('Table for Resource locking not present, will create') + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('resource_lock', + sa.Column('identifier', sa.Integer(), nullable=False, unique=True), + sa.Column('resource_identifier', sa.Integer(), nullable=False, unique=True), + sa.Column('owner', sa.Text, nullable=False, default='NOT SET', server_default='NOT SET'), + sa.Column('start_time', sa.DateTime, nullable=False), + sa.Column('end_time', sa.DateTime, nullable=False), + sa.ForeignKeyConstraint(['resource_identifier'], ['resource.identifier'], ), + sa.PrimaryKeyConstraint('identifier') + ) + # ### end Alembic commands ### + else: + print('Table for Resource-locking already present, will not create') + + +def downgrade(): + print('Dropping Column run_frequency from resource table') + op.drop_column(u'resource', 'run_frequency') + op.drop_table('resource_lock') diff --git a/GeoHealthCheck/models.py b/GeoHealthCheck/models.py index d26ff7ee..a33c9541 100644 --- a/GeoHealthCheck/models.py +++ b/GeoHealthCheck/models.py @@ -31,7 +31,7 @@ import json import logging from flask_babel import gettext as _ -from datetime import datetime +from datetime import datetime, timedelta from sqlalchemy import func, and_ from sqlalchemy.orm import deferred @@ -48,6 +48,21 @@ LOGGER = logging.getLogger(__name__) +# 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() + + class Run(DB.Model): """measurement of resource state""" @@ -55,7 +70,8 @@ class Run(DB.Model): resource_identifier = DB.Column(DB.Integer, DB.ForeignKey('resource.identifier')) resource = DB.relationship('Resource', - backref=DB.backref('runs', lazy='dynamic')) + backref=DB.backref('runs', lazy='dynamic', + cascade="all,delete")) checked_datetime = DB.Column(DB.DateTime, nullable=False) success = DB.Column(DB.Boolean, nullable=False) response_time = DB.Column(DB.Float, nullable=False) @@ -215,6 +231,7 @@ class dummy_value(object): @staticmethod def gettext(*args, **kwargs): return _(*args, **kwargs) + dummy_form = None v(dummy_form, dummy_value()) @@ -272,11 +289,11 @@ def is_webhook(self): @classmethod def burry_dead(cls): RN = ResourceNotification - q = DB.session.query(cls)\ - .join(RN, - RN.recipient_id == cls.id, - isouter=True)\ - .filter(RN.recipient_id.is_(None)) + q = DB.session.query(cls) \ + .join(RN, + RN.recipient_id == cls.id, + isouter=True) \ + .filter(RN.recipient_id.is_(None)) for item in q: DB.session.delete(item) @@ -288,10 +305,10 @@ def get_or_create(cls, channel, location): raise ValueError("invalid value {}: {}".format(location, err)) try: - r = DB.session.query(cls)\ - .filter(and_(cls.channel == channel, - cls.location == location))\ - .one() + r = DB.session.query(cls) \ + .filter(and_(cls.channel == channel, + cls.location == location)) \ + .one() except (MultipleResultsFound, NoResultFound,): r = cls(channel=channel, location=location) DB.session.add(r) @@ -308,12 +325,12 @@ def get_suggestions(cls, channel, for_user): Res = Resource ResNot = ResourceNotification - q = DB.session.query(Rcp.location)\ - .join(ResNot, ResNot.recipient_id == Rcp.id)\ - .join(Res, Res.identifier == ResNot.resource_id)\ - .group_by(Rcp.location)\ - .filter(and_(Res.owner_identifier == for_user, - Rcp.channel == channel)) + q = DB.session.query(Rcp.location) \ + .join(ResNot, ResNot.recipient_id == Rcp.id) \ + .join(Res, Res.identifier == ResNot.resource_id) \ + .group_by(Rcp.location) \ + .filter(and_(Res.owner_identifier == for_user, + Rcp.channel == channel)) return [item[0] for item in q] @@ -347,6 +364,7 @@ class Resource(DB.Model): owner = DB.relationship('User', backref=DB.backref('username2', lazy='dynamic')) tags = DB.relationship('Tag', secondary=resource_tags, backref='resource') + run_frequency = DB.Column(DB.Integer, default=60) def __init__(self, owner, resource_type, title, url, tags): self.resource_type = resource_type @@ -471,26 +489,26 @@ def clear_recipients(self, channel=None, burry_dead=True): # clear specific channel to_delete = self.get_recipients(channel) if to_delete: - to_del_rcp = DB.session.query(RN)\ - .join(Rcp, - Rcp.id == RN.recipient_id)\ - .filter( - and_(RN.resource_id == - self.identifier, - Rcp.channel == channel, - Rcp.location.in_(to_delete)) - ) + to_del_rcp = DB.session.query(RN) \ + .join(Rcp, + Rcp.id == RN.recipient_id) \ + .filter( + and_(RN.resource_id == + self.identifier, + Rcp.channel == channel, + Rcp.location.in_(to_delete)) + ) else: to_del_rcp = [] else: # remove all m2m connections for Resource<->Recipient - to_del_rcp = DB.session.query(RN)\ - .join(Rcp, - Rcp.id == RN.recipient_id)\ - .filter( - RN.resource_id == - self.identifier, - ) + to_del_rcp = DB.session.query(RN) \ + .join(Rcp, + Rcp.id == RN.recipient_id) \ + .filter( + RN.resource_id == + self.identifier, + ) for rcp_ntf in to_del_rcp: DB.session.delete(rcp_ntf) if burry_dead: @@ -527,6 +545,47 @@ def dump_recipients(self): return out +class ResourceLock(DB.Model): + """lock resource for multiprocessing runs""" + + identifier = DB.Column(DB.Integer, + primary_key=True, autoincrement=False, unique=True) + resource_identifier = DB.Column( + DB.Integer, DB.ForeignKey('resource.identifier'), unique=True) + resource = DB.relationship('Resource', + backref=DB.backref('locks', lazy='dynamic', + cascade="all,delete")) + owner = DB.Column(DB.Text, nullable=False, default='NOT SET') + + start_time = DB.Column(DB.DateTime, nullable=False) + end_time = DB.Column(DB.DateTime, nullable=False) + + def __init__(self, resource, owner, interval_mins): + self.identifier = resource.identifier + self.resource = resource + self.owner = owner + self.init_datetimes(interval_mins) + + def init_datetimes(self, interval_mins): + self.start_time = datetime.utcnow() + self.end_time = self.start_time + timedelta(minutes=interval_mins) + + def has_expired(self): + now = datetime.utcnow() + return now > self.end_time + + def obtain(self, owner, frequency): + if not self.has_expired(): + return False + + self.owner = owner + self.init_datetimes(frequency) + return True + + def __repr__(self): + return '' % self.identifier + + class User(DB.Model): """user accounts""" diff --git a/GeoHealthCheck/scheduler.py b/GeoHealthCheck/scheduler.py new file mode 100644 index 00000000..4b8b2f41 --- /dev/null +++ b/GeoHealthCheck/scheduler.py @@ -0,0 +1,207 @@ +# coding=utf-8 +# ================================================================= +# +# Authors: Tom Kralidis +# Just van den Broecke +# +# Copyright (c) 2014 Tom Kralidis +# +# Permission is hereby granted, free of charge, to any person +# obtaining a copy of this software and associated documentation +# files (the "Software"), to deal in the Software without +# restriction, including without limitation the rights to use, +# copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the +# Software is furnished to do so, subject to the following +# conditions: +# +# The above copyright notice and this permission notice shall be +# included in all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES +# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND +# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR +# OTHER DEALINGS IN THE SOFTWARE. +# +# ================================================================= + +import logging +import os +import random +import string +from models import Resource, ResourceLock, flush_runs +from healthcheck import run_resource +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.jobstores.base import JobLookupError +from init import App + +LOGGER = logging.getLogger(__name__) +DB = App.get_db() + +# Create scheduler +scheduler = BackgroundScheduler() + + +# commit or rollback shorthand +def db_commit(): + err = None + try: + DB.session.commit() + except Exception as err: + DB.session.rollback() + # finally: + # DB.session.close() + return err + + +def run_job(resource_id, frequency): + # Generate unique id + # https://stackoverflow.com/questions/2257441/random-string-generation-with-upper-case-letters-and-digits-in-python + uuid = '%d-%s' % (os.getpid(), ''.join(random.choice( + string.ascii_uppercase + string.digits) for _ in range(8))) + + resource = Resource.query.filter_by(identifier=resource_id).first() + + if not resource: + stop_job(resource_id) + return + + lock = ResourceLock.query.filter_by(identifier=resource_id).first() + + if not lock: + # No lock at all on Resource: (hope) we're first + # obtain fresh lock, back-off if failed + LOGGER.info('%d No Lock at all: obtain new' % resource_id) + + lock = ResourceLock(resource, uuid, frequency) + DB.session.add(lock) + lock_err = db_commit() + if lock_err: + # Another process may have been there first! + LOGGER.info('%d Cannot obtain Lock %s' % + (resource_id, str(lock_err))) + return + else: + # Lock is there, look if available + LOGGER.info('%d Lock avail: try obtaining..' % resource_id) + if not lock.obtain(uuid, frequency): + LOGGER.info('%d Cannot obtain' % resource_id) + return + else: + LOGGER.info('%d Lock obtained, delete and renew' % resource_id) + DB.session.delete(lock) + lock_err = db_commit() + if lock_err: + # Another process may have been there first! + LOGGER.info('%d Lock Delete failed' % resource_id) + return + + LOGGER.info('%d Lock deleted, add new' % resource_id) + # (hope) we're first + # obtain fresh lock, back-off if failed + lock = ResourceLock(resource, uuid, frequency) + DB.session.add(lock) + lock_err = db_commit() + if lock_err: + # Another process may have been there first! + LOGGER.info('%d Lock Add failed' % resource_id) + return + + # Check if we really own the lock + LOGGER.info('%d Lock Add OK' % resource_id) + lock = ResourceLock.query.filter_by( + identifier=resource_id).first() + + if lock.owner != uuid: + LOGGER.info('%d Lock Add OK, not owner: back-off' + % resource_id) + return + + if lock: + try: + run_resource(resource_id) + LOGGER.info('%d run_resource OK' % resource_id) + finally: + pass + + +def start_schedule(): + # Cold start every cron of every Resource + for resource in Resource.query.all(): + add_job(resource) + + # change configuration + scheduler.configure(job_defaults={ + 'coalesce': False, + 'max_instances': 100000 + }) + + scheduler.add_job(flush_runs, 'interval', minutes=60) + scheduler.add_job(check_schedule, 'interval', minutes=5) + + scheduler.start() + import atexit + atexit.register(lambda: stop_schedule()) + + +def check_schedule(): + LOGGER.info('Checking Job schedules') + # Check the schedule for changed jobs + for resource in Resource.query.all(): + job = scheduler.get_job(str(resource.identifier)) + if job is None: + add_job(resource) + + current_freq = job.args[1] + + # Run frequency changed? + if current_freq != resource.run_frequency: + # Reschedule Job + update_job(resource) + + +def update_job(resource): + stop_job(resource.identifier) + + # Add job to Scheduler + add_job(resource) + + +def add_job(resource): + LOGGER.info('Starting job for resource=%d' % resource.identifier) + freq = resource.run_frequency + + scheduler.add_job( + run_job, 'interval', args=[resource.identifier, freq], + minutes=freq, + id=str(resource.identifier)) + + +def stop_job(resource_id): + LOGGER.info('Stopping job for resource=%d' % resource_id) + + # Try to remove job from scheduler + try: + scheduler.remove_job(str(resource_id)) + except JobLookupError: + pass + + +def stop_schedule(): + LOGGER.info('Stopping Scheduler') + scheduler.shutdown() + + +if __name__ == '__main__': + import time + + # Start scheduler + start_schedule() + + while True: + print("This prints once a minute.") + time.sleep(60) # Delay for 1 minute (60 seconds). diff --git a/GeoHealthCheck/templates/edit_resource.html b/GeoHealthCheck/templates/edit_resource.html index bb9722fb..bc45d251 100644 --- a/GeoHealthCheck/templates/edit_resource.html +++ b/GeoHealthCheck/templates/edit_resource.html @@ -88,6 +88,12 @@

[{{ _('Edit') }}] {{ resour + + {{ _('Run Every') }} + + + + Probes @@ -258,6 +264,9 @@

[{{ _('Edit') }}] {{ resour // Collect tags var new_tags = $('#resource_tags').val(); + // Collect test_frequency + var new_frequency = $('input[name="resource_frequency_value"]').val(); + // Collect active var new_active = $('#input_resource_active').prop('checked'); @@ -335,6 +344,7 @@

[{{ _('Edit') }}] {{ resour title: new_title, active: new_active, tags: new_tags, + run_frequency: new_frequency, probes: new_probes, notify_emails: new_notify_emails, notify_webhooks: new_notify_webhooks diff --git a/GeoHealthCheck/templates/resource.html b/GeoHealthCheck/templates/resource.html index 723f75fb..dc42185c 100644 --- a/GeoHealthCheck/templates/resource.html +++ b/GeoHealthCheck/templates/resource.html @@ -54,6 +54,12 @@