Asynchronous UI notification implementation

This patch makes backend and UI changes to implement the asynchronous
UI notification in WoK.

- Backend:

A push server was implemented from scratch to manage the opened websocket
connections. The push server connects to the
/run/user/<user_id>/woknotifications UNIX socket and broadcasts all messages
to all connections.

The websocket module is the same module that exists in the Kimchi
plug-in. The idea is to remove the module from Kimchi and make it
use the module from WoK. ws_proxy initialization was also added
in src/wok/server.py.

A change were made in Wok base control classes to allow every
call of log_request to also send a websocket notification in
the following format:

<METHOD>:/<plugin>/<entity>/<action>

For example, creating a new user in Ginger would trigger the
following websocket notification:

'POST:/ginger/users'

- Frontend:

In ui/js/wok.main.js two new functions were added to help the
usage of asynchronous notifications in the frontend. The idea:
a single websocket is opened per session. This opened websocket
will broadcast all incoming messages to all listeners registered.
Listeners can be added by the new wok.addNotificationListener()
method. This method will clean up any registered listener by
itself when the user changes tabs/URL.

The single websocket sends heartbeats to the backend side each
30 seconds. No reply from the backend is issued or expected. This
heartbeat is just a way to ensure that the browser does not
close the connection due to inactivity. This behavior varies from
browser to browser but this 30 second heartbeat is more than enough
to ensure that the websocket is kept alive.

- Working example in User Log:

A simple usage is provided in this patch. Changes were made in the
UI of the User Log feature to refresh the listing each time a new
log entry websocket notification is received. The idea is to allow
this code to be a working example of how other tabs can consume
the asynchronous notifications.

Signed-off-by: Daniel Henrique Barboza <[email protected]>

Author: danielhb

contrib/DEBIAN/control.in

@@ -16,6 +16,7 @@ Depends: python-cherrypy3 (>= 3.2.0),
          fonts-font-awesome,
          logrotate,
          openssl,
+         websockify,
          texlive-fonts-extra
 Build-Depends: xsltproc,
                gettext,

contrib/wok.spec.fedora.in

@@ -21,6 +21,7 @@ Requires:	fontawesome-fonts
 Requires:	open-sans-fonts
 Requires:	logrotate
 Requires:	openssl
+Requires:	python-websockify
 BuildRequires:	gettext-devel
 BuildRequires:	libxslt
 BuildRequires:	python-lxml

contrib/wok.spec.suse.in

@@ -22,6 +22,7 @@ Requires:	fontawesome-fonts
 Requires:	google-opensans-fonts
 Requires:	logrotate
 Requires:	openssl
+Requires:	python-websockify
 BuildRequires:	gettext-tools
 BuildRequires:	libxslt-tools
 BuildRequires:	python-lxml

docs/fedora-deps.md

@@ -28,7 +28,7 @@ Runtime Dependencies
     $ sudo yum install python-cherrypy python-cheetah PyPAM m2crypto \
                         python-jsonschema python-psutil python-ldap \
                         python-lxml nginx openssl open-sans-fonts \
-                        fontawesome-fonts logrotate
+                        fontawesome-fonts logrotate python-websockify
 
     # For RHEL systems, install the additional packages:
     $ sudo yum install python-ordereddict

docs/opensuse-deps.md

@@ -18,7 +18,8 @@ Runtime Dependencies
     $ sudo zypper install python-CherryPy python-Cheetah python-pam \
                           python-M2Crypto python-jsonschema python-psutil \
                           python-ldap python-lxml python-xml nginx openssl \
-                          google-opensans-fonts fontawesome-fonts logrotate
+                          google-opensans-fonts fontawesome-fonts logrotate \
+                          python-websockify
 
 Packages required for UI development
 ------------------------------------

docs/ubuntu-deps.md

@@ -18,7 +18,8 @@ Runtime Dependencies
     $ sudo apt-get install python-cherrypy3 python-cheetah python-pam \
                             python-m2crypto python-jsonschema \
                             python-psutil python-ldap python-lxml nginx \
-                            openssl fonts-font-awesome texlive-fonts-extra
+                            openssl fonts-font-awesome texlive-fonts-extra \
+                            websockify
 
 Packages required for UI development
 ------------------------------------

src/wok/config.py.in

@@ -66,10 +66,18 @@ def get_object_store():
     return os.path.join(paths.state_dir, 'objectstore')
 
 
+def get_pushserver_socket_dir():
+    return '/run/user/%s' % os.geteuid()
+
+
 def get_version():
     return "-".join([__version__, __release__])
 
 
+def get_wstokens_dir():
+    return os.path.join(paths.state_dir, 'ws-tokens')
+
+
 class Paths(object):
 
     def __init__(self):

src/wok/control/base.py

@@ -163,7 +163,9 @@ def wrapper(*args, **kwargs):
                 # log request
                 code = self.getRequestMessage(method, action_name)
                 reqParams = utf8_dict(self.log_args, request)
-                log_id = log_request(code, reqParams, details, method, status)
+                log_id = log_request(code, reqParams, details, method, status,
+                                     class_name=get_class_name(self),
+                                     action_name=action_name)
                 if status == 202:
                     save_request_log_id(log_id, action_result['id'])
 
@@ -218,7 +220,8 @@ def index(self, *args, **kargs):
             # log request
             if method not in LOG_DISABLED_METHODS and status != 202:
                 code = self.getRequestMessage(method)
-                log_request(code, self.log_args, details, method, status)
+                log_request(code, self.log_args, details, method, status,
+                            class_name=get_class_name(self))
 
         return result
 
@@ -306,7 +309,8 @@ def delete(self):
         code = self.getRequestMessage(method)
         reqParams = utf8_dict(self.log_args)
         log_id = log_request(code, reqParams, None, method,
-                             cherrypy.response.status)
+                             cherrypy.response.status,
+                             class_name=get_class_name(self))
         save_request_log_id(log_id, task['id'])
 
         return wok.template.render("Task", task)
@@ -458,7 +462,8 @@ def index(self, *args, **kwargs):
                 # log request
                 code = self.getRequestMessage(method)
                 reqParams = utf8_dict(self.log_args, params)
-                log_request(code, reqParams, details, method, status)
+                log_request(code, reqParams, details, method, status,
+                            class_name=get_class_name(self))
 
 
 class AsyncCollection(Collection):
@@ -486,7 +491,8 @@ def create(self, params, *args):
         code = self.getRequestMessage(method)
         reqParams = utf8_dict(self.log_args, params)
         log_id = log_request(code, reqParams, None, method,
-                             cherrypy.response.status)
+                             cherrypy.response.status,
+                             class_name=get_class_name(self))
         save_request_log_id(log_id, task['id'])
 
         return wok.template.render("Task", task)

src/wok/model/notifications.py

@@ -1,7 +1,7 @@
 #
 # Project Wok
 #
-# Copyright IBM Corp, 2016
+# Copyright IBM Corp, 2016-2017
 #
 # This library is free software; you can redistribute it and/or
 # modify it under the terms of the GNU Lesser General Public

src/wok/pushserver.py

@@ -0,0 +1,164 @@
+#
+# Project Wok
+#
+# Copyright IBM Corp, 2017
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty 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 library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+#
+
+import cherrypy
+import os
+import select
+import socket
+import threading
+
+import wok.websocket as websocket
+from wok.config import get_pushserver_socket_dir
+from wok.utils import wok_log
+
+
+BASE_DIRECTORY = get_pushserver_socket_dir()
+TOKEN_NAME = 'woknotifications'
+END_OF_MESSAGE_MARKER = '//EOM//'
+push_server = None
+
+
+def start_push_server():
+    global push_server
+
+    if not push_server:
+        push_server = PushServer()
+
+
+def send_websocket_notification(message):
+    global push_server
+
+    if push_server:
+        push_server.send_notification(message)
+
+
+def send_wok_notification(uri, entity, method, action_name=None):
+    app_name = 'wok'
+    app = cherrypy.tree.apps.get(uri)
+    if app:
+        app_name = app.root.domain
+
+    source = '/%s/%s' % (app_name, entity)
+    if action_name:
+        source = '%s/%s' % (source, action_name)
+    message = '%s:%s' % (method, source)
+    send_websocket_notification(message)
+
+
+class PushServer(object):
+
+    def set_socket_file(self):
+        if not os.path.isdir(BASE_DIRECTORY):
+            try:
+                os.mkdir(BASE_DIRECTORY)
+            except OSError:
+                raise RuntimeError('PushServer base UNIX socket dir %s '
+                                   'not found.' % BASE_DIRECTORY)
+
+        self.server_addr = os.path.join(BASE_DIRECTORY, TOKEN_NAME)
+
+        if os.path.exists(self.server_addr):
+            try:
+                os.remove(self.server_addr)
+            except:
+                raise RuntimeError('There is an existing connection in %s' %
+                                   self.server_addr)
+
+    def __init__(self):
+        self.set_socket_file()
+
+        websocket.add_proxy_token(TOKEN_NAME, self.server_addr, True)
+
+        self.connections = []
+
+        self.server_running = True
+        self.server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.server_socket.setsockopt(socket.SOL_SOCKET,
+                                      socket.SO_REUSEADDR, 1)
+        self.server_socket.bind(self.server_addr)
+        self.server_socket.listen(10)
+        wok_log.info('Push server created on address %s' % self.server_addr)
+
+        self.connections.append(self.server_socket)
+        cherrypy.engine.subscribe('stop', self.close_server, 1)
+
+        server_loop = threading.Thread(target=self.listen)
+        server_loop.setDaemon(True)
+        server_loop.start()
+
+    def listen(self):
+        try:
+            while self.server_running:
+                read_ready, _, _ = select.select(self.connections,
+                                                 [], [], 1)
+                for sock in read_ready:
+                    if not self.server_running:
+                        break
+
+                    if sock == self.server_socket:
+
+                        new_socket, addr = self.server_socket.accept()
+                        self.connections.append(new_socket)
+                    else:
+                        try:
+                            data = sock.recv(4096)
+                        except:
+                            try:
+                                self.connections.remove(sock)
+                            except ValueError:
+                                pass
+
+                            continue
+                        if data and data == 'CLOSE':
+                            sock.send('ACK')
+                            try:
+                                self.connections.remove(sock)
+                            except ValueError:
+                                pass
+                            sock.close()
+
+        except Exception as e:
+            raise RuntimeError('Exception ocurred in listen() of pushserver '
+                               'module: %s' % e.message)
+
+    def send_notification(self, message):
+        message += END_OF_MESSAGE_MARKER
+        for sock in self.connections:
+            if sock != self.server_socket:
+                try:
+                    sock.send(message)
+                except IOError as e:
+                    if 'Broken pipe' in str(e):
+                        sock.close()
+                        try:
+                            self.connections.remove(sock)
+                        except ValueError:
+                            pass
+
+    def close_server(self):
+        try:
+            self.server_running = False
+            self.server_socket.shutdown(socket.SHUT_RDWR)
+            self.server_socket.close()
+            os.remove(self.server_addr)
+        except:
+            pass
+        finally:
+            cherrypy.engine.unsubscribe('stop', self.close_server)

src/wok/reqlogger.py

@@ -1,7 +1,7 @@
 #
 # Project Wok
 #
-# Copyright IBM Corp, 2016
+# Copyright IBM Corp, 2016-2017
 #
 # This library is free software; you can redistribute it and/or
 # modify it under the terms of the GNU Lesser General Public
@@ -34,6 +34,7 @@
 from wok.config import get_log_download_path, paths
 from wok.exception import InvalidParameter, OperationFailed
 from wok.message import WokMessage
+from wok.pushserver import send_wok_notification
 from wok.stringutils import ascii_dict
 from wok.utils import remove_old_files
 
@@ -68,9 +69,11 @@
 # AsyncTask handling
 ASYNCTASK_REQUEST_METHOD = 'TASK'
 
+NEW_LOG_ENTRY_MESSAGE = 'new_log_entry'
+
 
 def log_request(code, params, exception, method, status, app=None, user=None,
-                ip=None):
+                ip=None, class_name=None, action_name=None):
     '''
     Add an entry to user request log
 
@@ -114,6 +117,10 @@ def log_request(code, params, exception, method, status, app=None, user=None,
         ip=ip
     ).log()
 
+    if class_name:
+        send_wok_notification(app, class_name, method, action_name)
+    send_wok_notification('', 'logs', 'POST')
+
     return log_id
 
 

src/wok/root.py

@@ -34,6 +34,7 @@
 from wok.control.base import Resource
 from wok.control.utils import parse_request, validate_params
 from wok.exception import OperationFailed, UnauthorizedError, WokException
+from wok.pushserver import send_wok_notification
 from wok.reqlogger import log_request
 
 
@@ -235,6 +236,7 @@ def _raise_timeout(user_id):
             status = e.getHttpStatusCode()
             raise cherrypy.HTTPError(401, e.message)
         finally:
+            send_wok_notification('', 'login', 'POST')
             log_request(code, params, details, method, status)
 
         return json.dumps(user_info)
@@ -247,6 +249,7 @@ def logout(self):
 
         auth.logout()
 
+        send_wok_notification('', 'logout', 'POST')
         log_request(code, params, None, method, 200, user=params['username'])
 
         return '{}'