From 3d20edc9a1ae6e76053d03f7cd1bb6b8656da5eb Mon Sep 17 00:00:00 2001 From: Aurélien Bompard Date: Dec 15 2017 09:05:45 +0000 Subject: Use Flask signals They are used to trigger reloads on widget config change and hub config change, and to create widgets when hubs are created. --- diff --git a/hubs/app.py b/hubs/app.py index 455712f..af16621 100644 --- a/hubs/app.py +++ b/hubs/app.py @@ -122,6 +122,17 @@ def check_auth(): OIDC.logout() +# Signals +import hubs.signals # noqa: E402 +import hubs.utils.taskqueue # noqa: E402 + + +@app.before_request +def create_task_queue(): + queue_name = fedmsg_config['hubs.redis.work-queue-name'] + flask.g.task_queue = hubs.utils.taskqueue.TaskQueue(queue_name) + + # Register widgets import hubs.widgets # noqa: E402 hubs.widgets.registry.register_list(app.config["WIDGETS"]) diff --git a/hubs/backend/triage.py b/hubs/backend/triage.py index 218d0dc..b9dca54 100755 --- a/hubs/backend/triage.py +++ b/hubs/backend/triage.py @@ -67,17 +67,12 @@ def triage(msg): widgets = get_widgets() - def is_widget_update(msg, widget): - """Always rebuild the cache when the widget config is updated.""" - return msg['topic'].endswith('hubs.widget.update') \ - and msg['msg']['widget']['id'] == widget.idx - log.debug("Checking should_invalidate for all widgets.") for widget in widgets: for fn_name, fn_class in widget.module.get_cached_functions().items(): # log.debug("Checking %s:%s", widget.plugin, widget.idx) fn = fn_class(widget) - if is_widget_update(msg, widget) or fn.should_invalidate(msg): + if fn.should_invalidate(msg): yield retask.task.Task(json.dumps({ 'type': 'widget-cache', 'idx': widget.idx, diff --git a/hubs/backend/worker.py b/hubs/backend/worker.py index 4cf3bcf..5252a66 100755 --- a/hubs/backend/worker.py +++ b/hubs/backend/worker.py @@ -65,6 +65,15 @@ def handle_widget_cache(db, widget_idx, fn_name): raise +def add_sse_task(sse_queue, event, data, target): + sse_task = { + "event": event, + "data": data, + "target": target, + } + sse_queue.enqueue(retask.task.Task(json.dumps(sse_task))) + + def parse_args(args): parser = argparse.ArgumentParser(description='Rebuild widget caches.') parser.add_argument("-d", "--debug", action="store_true", @@ -100,25 +109,32 @@ def main(args=None): task = queue.wait() # Wait forever... timeout is optional. log.info("Working on %r. (backlog is %r)" % (task, queue.length)) item = json.loads(task.data) - sse_task = None item_type = item.get("type", "widget-cache") if item_type == "widget-cache": handle_widget_cache(db, item['idx'], item['fn_name']) - sse_task = { - "event": "hubs:widget-updated", - "data": item["idx"], - "target": "hub/{}".format(item["hub"]), - } + add_sse_task( + sse_queue, + "hubs:widget-updated", + item["idx"], + "hub/{}".format(item["hub"]), + ) + elif item_type == "widget-update": + # Just reload the widget + add_sse_task( + sse_queue, + "hubs:widget-updated", + item["idx"], + "hub/{}".format(item["hub"]), + ) elif item_type == "notification": hubs.feed.on_new_notification(item["msg"]) username = fedmsg.meta.msg2agent(item["msg"]) - sse_task = { - "event": "hubs:new-notification", - "data": username, - "target": "user/{}".format(username), - } - if sse_task is not None: - sse_queue.enqueue(retask.task.Task(json.dumps(sse_task))) + add_sse_task( + sse_queue, + "hubs:new-notification", + username, + "user/{}".format(username), + ) log.debug(" Done.") except KeyboardInterrupt: pass diff --git a/hubs/defaults.py b/hubs/defaults.py index 4938844..15b5f74 100644 --- a/hubs/defaults.py +++ b/hubs/defaults.py @@ -5,14 +5,13 @@ import json import hubs.models -def add_user_widgets(hub, username, fullname): +def add_user_widgets(hub): """ Some defaults for an individual user's hub. """ # The feed widget works, but it needs to be cleaned up lots before we # throw it in by default. widget = hubs.models.Widget( plugin='feed', index=0, left=True, _config=json.dumps({ - 'username': username, 'message_limit': 20 })) hub.widgets.append(widget) @@ -21,7 +20,6 @@ def add_user_widgets(hub, username, fullname): widget = hubs.models.Widget( plugin='contact', index=-2, _config=json.dumps({ - 'username': username, })) hub.widgets.append(widget) @@ -36,19 +34,18 @@ def add_user_widgets(hub, username, fullname): widget = hubs.models.Widget( plugin='workflow.updates2stable', index=1, _config=json.dumps({ - 'username': username, + 'username': hub.name, })) hub.widgets.append(widget) widget = hubs.models.Widget( plugin='my_hubs', index=3, _config=json.dumps({ - 'username': username, })) hub.widgets.append(widget) widget = hubs.models.Widget( plugin='badges', index=4, _config=json.dumps({ - 'username': username, + 'username': hub.name, })) hub.widgets.append(widget) widget = hubs.models.Widget( @@ -67,7 +64,7 @@ def add_user_widgets(hub, username, fullname): widget = hubs.models.Widget( plugin='bugzilla', index=8, _config=json.dumps({ - 'username': username, + 'username': hub.name, })) hub.widgets.append(widget) hub.widgets.append( @@ -79,14 +76,11 @@ def add_user_widgets(hub, username, fullname): return hub -def add_group_widgets(hub, name, summary, +def add_group_widgets(hub, # These are all things that come from FAS apply_rules=None, - irc_channel=None, - irc_network=None, join_message=None, - mailing_list=None, - mailing_list_url=None): + ): """ Some defaults for an automatically created group hub. """ if apply_rules: @@ -118,9 +112,6 @@ def add_group_widgets(hub, name, summary, hub.widgets.append(widget) # IRC - if irc_channel and irc_network: - hub.config.chat_domain = irc_network - hub.config.chat_channel = irc_channel hub.widgets.append( hubs.models.Widget( plugin='irc', index=2, diff --git a/hubs/models.py b/hubs/models.py index 0364621..2b30812 100644 --- a/hubs/models.py +++ b/hubs/models.py @@ -31,8 +31,8 @@ import random from collections import defaultdict import bleach +import flask import sqlalchemy as sa - from sqlalchemy.orm import relation from sqlalchemy.orm import backref from sqlalchemy.orm.session import object_session @@ -42,6 +42,7 @@ import hubs.widgets from hubs.authz import ObjectAuthzMixin, AccessLevel from hubs.database import BASE, Session from hubs.utils import username2avatar +from hubs.signals import hub_created, user_created log = logging.getLogger(__name__) @@ -191,11 +192,8 @@ class Hub(ObjectAuthzMixin, BASE): hub_config = HubConfig( hub=hub, summary=fullname, avatar=username2avatar(username)) session.add(hub_config) - - hubs.defaults.add_user_widgets(hub, username, fullname) - - user = User.query.get(username) - hub.subscribe(user, role='owner') + session.flush() + hub_created.send(hub) return hub @classmethod @@ -207,10 +205,46 @@ class Hub(ObjectAuthzMixin, BASE): hub_config = HubConfig( hub=hub, summary=summary, avatar=username2avatar(name)) session.add(hub_config) - - hubs.defaults.add_group_widgets(hub, name, summary, **extra) + session.flush() + hub_created.send(hub, **extra) return hub + def on_created(self, **extra): + if self.user_hub: + hubs.defaults.add_user_widgets(self) + user = User.query.get(self.name) + self.subscribe(user, role='owner') + else: + hubs.defaults.add_group_widgets(self, **extra) + + def on_updated(self, old_config): + for widget_instance in self.widgets: + if not widget_instance.enabled: + continue + widget = widget_instance.module + new_config = self.config.__json__() + will_reload = False + cached_functions = widget.get_cached_functions() + for fn_name, fn_class in cached_functions.items(): + fn = fn_class(widget_instance) + if fn.should_invalidate_on_hub_config_change(old_config): + flask.g.task_queue.enqueue( + "widget-cache", + idx=widget_instance.idx, + hub=self.name, + fn_name=fn_name, + ) + will_reload = True + if not will_reload: + # Reload the widget if it has asked for it. + if widget.should_reload_on_hub_config_change( + old_config, new_config): + flask.g.task_queue.enqueue( + "widget-update", + idx=widget_instance.idx, + hub=self.name, + ) + def _get_auth_user_access_level(self, user): # overridden to handle user hubs. if self.user_hub and user.username == self.name: @@ -388,6 +422,28 @@ class Widget(ObjectAuthzMixin, BASE): def config(self, config): self._config = json.dumps(config) + def on_updated(self, old_config): + will_reload = False + cached_functions = self.module.get_cached_functions() + for fn_name, fn_class in cached_functions.items(): + fn = fn_class(self) + if fn.should_invalidate_on_widget_config_change(old_config): + flask.g.task_queue.enqueue( + "widget-cache", + idx=self.idx, + hub=self.hub.name, + fn_name=fn_name, + ) + will_reload = True + if not will_reload: + # Reload the widget nonetheless because the config + # change may impact rendering. + flask.g.task_queue.enqueue( + "widget-update", + idx=self.idx, + hub=self.hub.name, + ) + def _get_auth_access_level(self, user): return self.hub._get_auth_access_level(user) @@ -530,10 +586,14 @@ class User(BASE): session = Session() self = cls(username=username, fullname=fullname) session.add(self) - if Hub.query.get(username) is None: - Hub.create_user_hub(username, fullname) + session.flush() + user_created.send(self) return self + def on_created(self): + if Hub.query.get(self.username) is None: + Hub.create_user_hub(self.username, self.fullname) + class VisitCounter(BASE): __tablename__ = 'visit_counter' diff --git a/hubs/signals.py b/hubs/signals.py new file mode 100644 index 0000000..fbc7e3a --- /dev/null +++ b/hubs/signals.py @@ -0,0 +1,56 @@ +# -*- coding: utf-8 -*- +# +# Copyright © 2017 Red Hat, Inc. +# +# This copyrighted material is made available to anyone wishing to use, +# modify, copy, or redistribute it subject to the terms and conditions +# of the GNU Lesser General Public License (LGPL) version 2, or +# (at your option) any later version. This program is distributed in the +# hope that it will be useful, but WITHOUT ANY WARRANTY expressed or +# implied, including the implied warranties of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for +# more details. You should have received a copy of the GNU Lesser General +# Public License along with this program; if not, write to the Free +# Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. +# +# Any Red Hat trademarks that are incorporated in the source +# code or documentation are not subject to the GNU General Public +# License and may only be used or replicated with the express permission +# of Red Hat, Inc. +# + +from __future__ import unicode_literals + +from flask.signals import Namespace + + +hubs_signals = Namespace() + +widget_updated = hubs_signals.signal('widget-updated') +hub_created = hubs_signals.signal('hub-created') +hub_updated = hubs_signals.signal('hub-updated') +user_created = hubs_signals.signal('user-created') +# TODO: hook to FAS: +# - user_created to sync memberships +# - hub_created when its a group hub to sync memberships + + +@widget_updated.connect +def on_widget_updated(widget_instance, old_config): + widget_instance.on_updated(old_config) + + +@hub_updated.connect +def on_hub_updated(hub, old_config): + hub.on_updated(old_config) + + +@hub_created.connect +def on_hub_created(hub, **kw): + hub.on_created(**kw) + + +@user_created.connect +def on_user_created(user): + user.on_created() diff --git a/hubs/tests/__init__.py b/hubs/tests/__init__.py index 4976310..0f9750d 100644 --- a/hubs/tests/__init__.py +++ b/hubs/tests/__init__.py @@ -51,6 +51,7 @@ class APPTest(unittest.TestCase): self.populate() def tearDown(self): + self.session.rollback() hubs.database.Session.remove() self.vcr.__exit__() diff --git a/hubs/tests/test_widget_caching.py b/hubs/tests/test_widget_caching.py index 0a92421..076bb87 100644 --- a/hubs/tests/test_widget_caching.py +++ b/hubs/tests/test_widget_caching.py @@ -38,8 +38,7 @@ class CachedFunctionTest(APPTest): super(CachedFunctionTest, self).tearDown() def test_get_cache_key(self): - value = ('%d|DummyFunction|c7540fbb81683902959eb96e23be42ba' - % self.w_instance.idx) + value = '%d|DummyFunction' % self.w_instance.idx self.assertEqual( self.fn.get_cache_key(), value.encode("ascii"), diff --git a/hubs/tests/utils/test_taskqueue.py b/hubs/tests/utils/test_taskqueue.py new file mode 100644 index 0000000..dd7010d --- /dev/null +++ b/hubs/tests/utils/test_taskqueue.py @@ -0,0 +1,35 @@ +from __future__ import unicode_literals + +from flask import json +from mock import patch, Mock + +from hubs.utils.taskqueue import TaskQueue +from hubs.tests import APPTest + + +class TaskQueueTestCase(APPTest): + + def setUp(self): + super(TaskQueueTestCase, self).setUp() + self.queue_patcher = patch("hubs.utils.taskqueue.retask.queue.Queue") + Queue = self.queue_patcher.start() + self.queue = Mock() + Queue.return_value = self.queue + self.queue.connected = False + + def tearDown(self): + self.queue_patcher.stop() + super(TaskQueueTestCase, self).tearDown() + + def test_enqueue(self): + task_queue = TaskQueue("queue_name") + task_queue.enqueue("task_type", foo="bar", key="value") + self.assertTrue(self.queue.connect.called) + self.assertTrue(self.queue.enqueue.called) + task = self.queue.enqueue.call_args_list[0][0][0] + expected = { + "type": "task_type", + "foo": "bar", + "key": "value", + } + self.assertEqual(json.loads(task.data), expected) diff --git a/hubs/tests/views/test_api_hub_widget.py b/hubs/tests/views/test_api_hub_widget.py index 372ec1a..4444a0b 100644 --- a/hubs/tests/views/test_api_hub_widget.py +++ b/hubs/tests/views/test_api_hub_widget.py @@ -1,6 +1,7 @@ from __future__ import unicode_literals from flask import json +from mock import patch, Mock from hubs.app import app from hubs.models import Hub, User, Widget @@ -243,7 +244,10 @@ class TestAPIHubWidget(APPTest): response_data = json.loads(response.get_data(as_text=True)) self.assertEqual(response_data["status"], "ERROR") - def test_put_empty_data_logged_in(self): + @patch("hubs.utils.taskqueue.retask.queue.Queue") + def test_put_empty_data_logged_in(self, Queue): + queue = Mock() + Queue.return_value = queue user = FakeAuthorization('ralph') with auth_set(app, user): result = self.app.put( @@ -254,6 +258,14 @@ class TestAPIHubWidget(APPTest): self.assertEqual( json.loads(result.get_data(as_text=True)), {"status": "OK"}) + self.assertTrue(queue.enqueue.called) + task = queue.enqueue.call_args_list[0][0][0] + expected = { + "hub": "ralph", "idx": 37, + "type": "widget-cache", + "fn_name": "GetPRs", + } + self.assertEqual(json.loads(task.data), expected) def test_put_unauthorized(self): user = FakeAuthorization('decause') diff --git a/hubs/utils/taskqueue.py b/hubs/utils/taskqueue.py new file mode 100644 index 0000000..6b6c26a --- /dev/null +++ b/hubs/utils/taskqueue.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# +# Copyright © 2017 Red Hat, Inc. +# +# This copyrighted material is made available to anyone wishing to use, +# modify, copy, or redistribute it subject to the terms and conditions +# of the GNU Lesser General Public License (LGPL) version 2, or +# (at your option) any later version. This program is distributed in the +# hope that it will be useful, but WITHOUT ANY WARRANTY expressed or +# implied, including the implied warranties of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for +# more details. You should have received a copy of the GNU Lesser General +# Public License along with this program; if not, write to the Free +# Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. +# +# Any Red Hat trademarks that are incorporated in the source +# code or documentation are not subject to the GNU General Public +# License and may only be used or replicated with the express permission +# of Red Hat, Inc. +# + +from __future__ import unicode_literals + +import json + +import retask.queue +import retask.task + + +class TaskQueue(object): + + def __init__(self, name): + self.queue = retask.queue.Queue(name) + + def enqueue(self, task_type, **content): + if not self.queue.connected: + self.queue.connect() + content["type"] = task_type + self.queue.enqueue(retask.task.Task(json.dumps(content))) diff --git a/hubs/views/api/hub_config.py b/hubs/views/api/hub_config.py index defff20..4da5d95 100644 --- a/hubs/views/api/hub_config.py +++ b/hubs/views/api/hub_config.py @@ -6,6 +6,7 @@ import flask import hubs.models from hubs.app import app +from hubs.signals import hub_updated from hubs.utils.views import ( get_hub, get_user_permissions, check_hub_access, RequestValidator, require_hub_access, @@ -24,6 +25,7 @@ def api_hub_config(name): hub = get_hub(name) if flask.request.method == 'PUT': check_hub_access(hub, "config", json=True) + old_config = hub.config.__json__() request_data = flask.request.get_json() if request_data is None: return flask.jsonify({ @@ -43,6 +45,7 @@ def api_hub_config(name): flask.g.db.commit() except Exception as err: result = {"status": "ERROR", "message": str(err)} + hub_updated.send(hub, old_config=old_config) return flask.jsonify(result) data = hub.get_props() data["perms"] = get_user_permissions(hub) diff --git a/hubs/views/api/hub_widget.py b/hubs/views/api/hub_widget.py index 0871535..7421118 100644 --- a/hubs/views/api/hub_widget.py +++ b/hubs/views/api/hub_widget.py @@ -4,6 +4,7 @@ import logging import flask from hubs.app import app +from hubs.signals import widget_updated from hubs.widgets import registry from hubs.utils.views import ( create_widget_instance, configure_widget_instance, get_hub, @@ -89,6 +90,7 @@ def api_hub_widget(hub, idx): return flask.jsonify({"status": "OK", "data": result}) elif flask.request.method == "PUT": check_hub_access(hub, "config", json=True) + old_config = widget_instance.config request_data = flask.request.get_json() widget_config = request_data.get("config", {}) try: @@ -103,6 +105,7 @@ def api_hub_widget(hub, idx): except Exception as e: msg = "Could not configure this widget: {}".format(e) return flask.jsonify({"status": "ERROR", "message": msg}) + widget_updated.send(widget_instance, old_config=old_config) return flask.jsonify({"status": "OK"}) elif flask.request.method == "DELETE": # Remove the widget from the hub. diff --git a/hubs/widgets/base.py b/hubs/widgets/base.py index 6a79d39..53da560 100644 --- a/hubs/widgets/base.py +++ b/hubs/widgets/base.py @@ -46,6 +46,8 @@ class Widget(object): entirely hidden from the UI when no data is produced. As a consequence, there will also be no progress spinner when the widget is loading. + reload_on_hub_config_change (bool): Set to True to reload the widget if + the hub configuration has changed. """ name = None @@ -59,6 +61,7 @@ class Widget(object): is_large = False hidden_if_empty = False hub_types = ['user', 'group'] + reload_on_hub_config_change = False def __init__(self): if self.name is None: @@ -217,6 +220,22 @@ class Widget(object): result[fn_class.__name__] = fn_class return result + def should_reload_on_hub_config_change(self, old_config, new_config): + """Return whether the widget should be reloaded when the hub + configuration changes. + + By default it wil follow ``reload_on_hub_config_change``. Overload this + method to do a more fine-grained analysis. + + Args: + old_config (dict): the hub's old configuration. + new_config (dict): the hub's new configuration. + + Returns: + bool: Whether the widget should be reloaded. + """ + return self.reload_on_hub_config_change + @property def display_title(self): """The title of the widget box in the UI.""" diff --git a/hubs/widgets/caching.py b/hubs/widgets/caching.py index 69da12c..ce60d30 100644 --- a/hubs/widgets/caching.py +++ b/hubs/widgets/caching.py @@ -1,8 +1,6 @@ from __future__ import unicode_literals import datetime -import hashlib -import json import logging import dogpile.cache @@ -37,14 +35,19 @@ class CachedFunction(object): Attributes: instance (hubs.models.Widget): The widget instance. - invalidate_on_config_change (bool): ``True`` if the cached result - depends on the widget configuration, ``False`` otherwise. Defaults - to ``True``. E.g: the function retrieves a lot of raw data from an - external service, and config-dependant filtering is done in the - view calling the function. + invalidate_on_widget_config_change (bool): ``True`` if the cached + result depends on the widget configuration, ``False`` otherwise. + Defaults to ``True``. E.g: set to ``False`` if the function + retrieves a lot of raw data from an external service, and + config-dependant filtering is done in the view calling the + function. + invalidate_on_hub_config_change (bool): ``True`` if the cached result + depends on the hub configuration, ``False`` otherwise. Defaults + to ``False``. """ - invalidate_on_config_change = True + invalidate_on_widget_config_change = True + invalidate_on_hub_config_change = False def __init__(self, instance): self.instance = instance @@ -61,12 +64,6 @@ class CachedFunction(object): def get_cache_key(self): key_elements = [str(self.instance.idx), self.__class__.__name__] - if self.invalidate_on_config_change: - key_elements.append( - hashlib.md5( - json.dumps(self.instance.config).encode("utf-8") - ).hexdigest() - ) return "|".join(key_elements).encode('utf-8') def get_data(self): @@ -105,6 +102,36 @@ class CachedFunction(object): """ raise NotImplementedError + def should_invalidate_on_widget_config_change(self, old_config): + """Return whether the function's cache should be invalidated when the + widget configuration changes. + + By default it wil follow ``invalidate_on_widget_config_change``. + Overload this method to do a more fine-grained analysis. + + Args: + old_config (dict): the widget's old configuration. + + Returns: + bool: Whether the function's cache should be invalidated. + """ + return self.invalidate_on_widget_config_change + + def should_invalidate_on_hub_config_change(self, old_config): + """Return whether the function's cache should be invalidated when the + hub configuration changes. + + By default it wil follow ``invalidate_on_hub_config_change``. + Overload this method to do a more fine-grained analysis. + + Args: + old_config (dict): the hub's old configuration. + + Returns: + bool: Whether the function's cache should be invalidated. + """ + return self.invalidate_on_hub_config_change + def is_cached(self): """ Return a boolean indicating if the function's result is currently in diff --git a/hubs/widgets/halp/functions.py b/hubs/widgets/halp/functions.py index e288fd3..ce449b8 100644 --- a/hubs/widgets/halp/functions.py +++ b/hubs/widgets/halp/functions.py @@ -26,7 +26,7 @@ class GetRequests(CachedFunction): them. """ - invalidate_on_config_change = False + invalidate_on_widget_config_change = False TOPIC = "org.fedoraproject.prod.meetbot.meeting.item.help" def execute(self):