diff --git a/README.rst b/README.rst index 03c3a32d3..ec323b5e3 100644 --- a/README.rst +++ b/README.rst @@ -104,8 +104,7 @@ Celery command and before Flower sub-command): :: API --- -Flower API enables to manage the cluster via REST API, call tasks and -receive task events in real-time via WebSockets. +Flower API enables to manage the cluster via REST API. For example you can restart worker's pool by: :: @@ -119,13 +118,6 @@ Or terminate executing task by: :: $ curl -X POST -d 'terminate=True' http://localhost:5555/api/task/revoke/8a4da87b-e12b-4547-b89a-e92e4d1f8efd -Or receive task completion events in real-time: :: - - var ws = new WebSocket("ws://localhost:5555/api/task/events/task-succeeded/"); - ws.onmessage = function (event) { - console.log(event.data); - } - For more info checkout `API Reference`_ and `examples`_. .. _API Reference: https://flower.readthedocs.io/en/latest/api.html diff --git a/examples/event-api.html b/examples/event-api.html deleted file mode 100644 index 45225bba2..000000000 --- a/examples/event-api.html +++ /dev/null @@ -1,20 +0,0 @@ - - - - -Flower event API example - - - -

Flower event API

-

This example shows how to get task events

-
- diff --git a/flower/api/__init__.py b/flower/api/__init__.py index b9951b866..de54655c9 100644 --- a/flower/api/__init__.py +++ b/flower/api/__init__.py @@ -1,25 +1,7 @@ -import tornado.websocket +import tornado.web +from ..views import BaseHandler - -class BaseWebSocketHandler(tornado.websocket.WebSocketHandler): - # listeners = [], should be created in derived class - - def open(self): - listeners = self.listeners - listeners.append(self) - - def on_message(self, message): - pass - - def on_close(self): - listeners = self.listeners - if self in listeners: - listeners.remove(self) - - @classmethod - def send_message(cls, message): - for l in cls.listeners: - l.write_message(message) - - def check_origin(self, origin): - return True +class BaseApiHandler(BaseHandler): + def prepare(self): + if self.application.options.basic_auth or self.application.options.auth: + raise tornado.web.HTTPError(405, "api is not available when auth is enabled") diff --git a/flower/api/control.py b/flower/api/control.py index 363708b4f..f86ac7d37 100644 --- a/flower/api/control.py +++ b/flower/api/control.py @@ -7,13 +7,13 @@ from tornado import gen from tornado import util -from ..views import BaseHandler +from . import BaseApiHandler logger = logging.getLogger(__name__) -class ControlHandler(BaseHandler): +class ControlHandler(BaseApiHandler): def is_worker(self, workername): return workername and workername in self.application.workers diff --git a/flower/api/events.py b/flower/api/events.py deleted file mode 100644 index 7ab9d3a31..000000000 --- a/flower/api/events.py +++ /dev/null @@ -1,35 +0,0 @@ -import sys - -from ..api import BaseWebSocketHandler - - -class EventsApiHandler(BaseWebSocketHandler): - def open(self, task_id=None): - BaseWebSocketHandler.open(self) - self.task_id = task_id - - @classmethod - def send_message(cls, event): - for l in cls.listeners: - if not l.task_id or l.task_id == event['uuid']: - l.write_message(event) - - -EVENTS = ('task-sent', 'task-received', 'task-started', 'task-succeeded', - 'task-failed', 'task-revoked', 'task-retried', 'task-custom') - - -def getClassName(eventname): - return ''.join(map(lambda x: x[0].upper() + x[1:], eventname.split('-'))) - - -# Dynamically generates handler classes -thismodule = sys.modules[__name__] -for event in EVENTS: - classname = getClassName(event) - setattr(thismodule, classname, - type(classname, (EventsApiHandler, ), {'listeners': []})) - - -__all__ = list(map(getClassName, EVENTS)) -__all__.append(getClassName) diff --git a/flower/api/tasks.py b/flower/api/tasks.py index 54b682271..85c3f358c 100644 --- a/flower/api/tasks.py +++ b/flower/api/tasks.py @@ -15,7 +15,7 @@ from celery.backends.base import DisabledBackend from ..utils import tasks -from ..views import BaseHandler +from . import BaseApiHandler from ..utils.broker import Broker from ..api.control import ControlHandler from collections import OrderedDict @@ -24,7 +24,7 @@ logger = logging.getLogger(__name__) -class BaseTaskHandler(BaseHandler): +class BaseTaskHandler(BaseApiHandler): DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f' def get_task_args(self): diff --git a/flower/events.py b/flower/events.py index ec66be6de..2268f2fd2 100644 --- a/flower/events.py +++ b/flower/events.py @@ -117,11 +117,6 @@ def event(self, event): if event_type == 'worker-offline': self.metrics.worker_online.labels(worker_name).set(0) - # Send event to api subscribers (via websockets) - classname = api.events.getClassName(event_type) - cls = getattr(api.events, classname, None) - if cls: - cls.send_message(event) class Events(threading.Thread): diff --git a/flower/urls.py b/flower/urls.py index 2d64cc0b4..7d6332f8b 100644 --- a/flower/urls.py +++ b/flower/urls.py @@ -2,7 +2,6 @@ from tornado.web import StaticFileHandler, url -from .api import events from .api import control from .api import tasks from .api import workers @@ -57,15 +56,6 @@ (r"/api/task/timeout/(.+)", control.TaskTimout), (r"/api/task/rate-limit/(.+)", control.TaskRateLimit), (r"/api/task/revoke/(.+)", control.TaskRevoke), - # Events WebSocket API - (r"/api/task/events/task-sent/(.*)", events.TaskSent), - (r"/api/task/events/task-received/(.*)", events.TaskReceived), - (r"/api/task/events/task-started/(.*)", events.TaskStarted), - (r"/api/task/events/task-succeeded/(.*)", events.TaskSucceeded), - (r"/api/task/events/task-failed/(.*)", events.TaskFailed), - (r"/api/task/events/task-revoked/(.*)", events.TaskRevoked), - (r"/api/task/events/task-retried/(.*)", events.TaskRetried), - (r"/api/task/events/task-custom/(.*)", events.TaskCustom), # Metrics (r"/metrics", monitor.Metrics), (r"/healthcheck", monitor.Healthcheck), diff --git a/tests/unit/api/test_control.py b/tests/unit/api/test_control.py index 31819f3d3..a83d9ad6a 100644 --- a/tests/unit/api/test_control.py +++ b/tests/unit/api/test_control.py @@ -1,8 +1,9 @@ from mock import MagicMock - +import mock from flower.api.control import ControlHandler from tests.unit import AsyncHTTPTestCase +from tornado.options import options class UnknownWorkerControlTests(AsyncHTTPTestCase): @@ -169,3 +170,12 @@ def test_terminate_signal(self): celery.control.revoke.assert_called_once_with('test', terminate=True, signal='SIGUSR1') + + +class ControlAuthTests(WorkerControlTests): + def test_auth(self): + with mock.patch.object(options.mockable(), 'basic_auth', ['user1:password1']): + app = self._app.capp + app.control.broadcast = MagicMock() + r = self.post('/api/worker/shutdown/test', body={}) + self.assertEqual(405, r.code) diff --git a/tests/unit/api/test_events.py b/tests/unit/api/test_events.py deleted file mode 100644 index e69de29bb..000000000