channelstream: nicer handle of connection errors for debug.

rhodecode/apps/channelstream/views.py

@@ -1,169 +1,177 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program 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 General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 import uuid
 
 from pyramid.view import view_config
 from pyramid.httpexceptions import HTTPBadRequest, HTTPForbidden, HTTPBadGateway
 
 from rhodecode.apps._base import BaseAppView
 from rhodecode.lib.channelstream import (
-    channelstream_request,
+    channelstream_request, get_channelstream_server_url,
     ChannelstreamConnectionException,
     ChannelstreamPermissionException,
     check_channel_permissions,
     get_connection_validators,
     get_user_data,
     parse_channels_info,
     update_history_from_logs,
     STATE_PUBLIC_KEYS)
 
 from rhodecode.lib.auth import NotAnonymous
 
 log = logging.getLogger(__name__)
 
 
 class ChannelstreamView(BaseAppView):
 
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         self.channelstream_config = \
             self.request.registry.rhodecode_plugins['channelstream']
         if not self.channelstream_config.get('enabled'):
             log.error('Channelstream plugin is disabled')
             raise HTTPBadRequest()
 
         return c
 
     @NotAnonymous()
     @view_config(route_name='channelstream_connect', renderer='json_ext')
     def connect(self):
+        """ handle authorization of users trying to connect """
+
         self.load_default_context()
-        """ handle authorization of users trying to connect """
         try:
             json_body = self.request.json_body
         except Exception:
             log.exception('Failed to decode json from request')
             raise HTTPBadRequest()
 
         try:
             channels = check_channel_permissions(
                 json_body.get('channels'),
                 get_connection_validators(self.request.registry))
         except ChannelstreamPermissionException:
             log.error('Incorrect permissions for requested channels')
             raise HTTPForbidden()
 
         user = self._rhodecode_user
         if user.user_id:
             user_data = get_user_data(user.user_id)
         else:
             user_data = {
                 'id': None,
                 'username': None,
                 'first_name': None,
                 'last_name': None,
                 'icon_link': None,
                 'display_name': None,
                 'display_link': None,
             }
         user_data['permissions'] = self._rhodecode_user.permissions_safe
         payload = {
             'username': user.username,
             'user_state': user_data,
             'conn_id': str(uuid.uuid4()),
             'channels': channels,
             'channel_configs': {},
             'state_public_keys': STATE_PUBLIC_KEYS,
             'info': {
                 'exclude_channels': ['broadcast']
             }
         }
         filtered_channels = [channel for channel in channels
                              if channel != 'broadcast']
         for channel in filtered_channels:
             payload['channel_configs'][channel] = {
                 'notify_presence': True,
                 'history_size': 100,
                 'store_history': True,
                 'broadcast_presence_with_user_lists': True
             }
         # connect user to server
+        channelstream_url = get_channelstream_server_url(
+            self.channelstream_config, '/connect')
         try:
-            connect_result = channelstream_request(self.channelstream_config,
-                                                   payload, '/connect')
+            connect_result = channelstream_request(
+                self.channelstream_config, payload, '/connect')
         except ChannelstreamConnectionException:
-            log.exception('Channelstream service is down')
+            log.exception(
+                'Channelstream service at {} is down'.format(channelstream_url))
             return HTTPBadGateway()
 
         connect_result['channels'] = channels
         connect_result['channels_info'] = parse_channels_info(
             connect_result['channels_info'],
             include_channel_info=filtered_channels)
         update_history_from_logs(self.channelstream_config,
                                  filtered_channels, connect_result)
         return connect_result
 
     @NotAnonymous()
     @view_config(route_name='channelstream_subscribe', renderer='json_ext')
     def subscribe(self):
         """ can be used to subscribe specific connection to other channels """
         self.load_default_context()
         try:
             json_body = self.request.json_body
         except Exception:
             log.exception('Failed to decode json from request')
             raise HTTPBadRequest()
         try:
             channels = check_channel_permissions(
                 json_body.get('channels'),
                 get_connection_validators(self.request.registry))
         except ChannelstreamPermissionException:
             log.error('Incorrect permissions for requested channels')
             raise HTTPForbidden()
         payload = {'conn_id': json_body.get('conn_id', ''),
                    'channels': channels,
                    'channel_configs': {},
                    'info': {
                        'exclude_channels': ['broadcast']}
                    }
         filtered_channels = [chan for chan in channels if chan != 'broadcast']
         for channel in filtered_channels:
             payload['channel_configs'][channel] = {
                 'notify_presence': True,
                 'history_size': 100,
                 'store_history': True,
                 'broadcast_presence_with_user_lists': True
             }
+
+        channelstream_url = get_channelstream_server_url(
+            self.channelstream_config, '/subscribe')
         try:
             connect_result = channelstream_request(
                 self.channelstream_config, payload, '/subscribe')
         except ChannelstreamConnectionException:
-            log.exception('Channelstream service is down')
+            log.exception(
+                'Channelstream service at {} is down'.format(channelstream_url))
             return HTTPBadGateway()
         # include_channel_info will limit history only to new channel
         # to not overwrite histories on other channels in client
         connect_result['channels_info'] = parse_channels_info(
             connect_result['channels_info'],
             include_channel_info=filtered_channels)
-        update_history_from_logs(self.channelstream_config,
-                                 filtered_channels, connect_result)
+        update_history_from_logs(
+            self.channelstream_config, filtered_channels, connect_result)
         return connect_result

rhodecode/lib/channelstream.py

@@ -1,255 +1,259 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2018 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program 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 General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import hashlib
 import itsdangerous
 import logging
 import requests
 import datetime
 
 from dogpile.core import ReadWriteMutex
 from pyramid.threadlocal import get_current_registry
 
 import rhodecode.lib.helpers as h
 from rhodecode.lib.auth import HasRepoPermissionAny
 from rhodecode.lib.ext_json import json
 from rhodecode.model.db import User
 
 log = logging.getLogger(__name__)
 
 LOCK = ReadWriteMutex()
 
 STATE_PUBLIC_KEYS = ['id', 'username', 'first_name', 'last_name',
                      'icon_link', 'display_name', 'display_link']
 
 
 class ChannelstreamException(Exception):
     pass
 
 
 class ChannelstreamConnectionException(ChannelstreamException):
     pass
 
 
 class ChannelstreamPermissionException(ChannelstreamException):
     pass
 
 
+def get_channelstream_server_url(config, endpoint):
+    return 'http://{}{}'.format(config['server'], endpoint)
+
+
 def channelstream_request(config, payload, endpoint, raise_exc=True):
     signer = itsdangerous.TimestampSigner(config['secret'])
     sig_for_server = signer.sign(endpoint)
     secret_headers = {'x-channelstream-secret': sig_for_server,
                       'x-channelstream-endpoint': endpoint,
                       'Content-Type': 'application/json'}
-    req_url = 'http://{}{}'.format(config['server'], endpoint)
+    req_url = get_channelstream_server_url(config, endpoint)
     response = None
     try:
         response = requests.post(req_url, data=json.dumps(payload),
                                  headers=secret_headers).json()
     except requests.ConnectionError:
-        log.exception('ConnectionError happened')
+        log.exception('ConnectionError occurred for endpoint %s', req_url)
         if raise_exc:
-            raise ChannelstreamConnectionException()
+            raise ChannelstreamConnectionException(req_url)
     except Exception:
-        log.exception('Exception related to channelstream happened')
+        log.exception('Exception related to Channelstream happened')
         if raise_exc:
             raise ChannelstreamConnectionException()
     return response
 
 
 def get_user_data(user_id):
     user = User.get(user_id)
     return {
         'id': user.user_id,
         'username': user.username,
         'first_name': user.first_name,
         'last_name': user.last_name,
         'icon_link': h.gravatar_url(user.email, 60),
         'display_name': h.person(user, 'username_or_name_or_email'),
         'display_link': h.link_to_user(user),
         'notifications': user.user_data.get('notification_status', True)
     }
 
 
 def broadcast_validator(channel_name):
     """ checks if user can access the broadcast channel """
     if channel_name == 'broadcast':
         return True
 
 
 def repo_validator(channel_name):
     """ checks if user can access the broadcast channel """
     channel_prefix = '/repo$'
     if channel_name.startswith(channel_prefix):
         elements = channel_name[len(channel_prefix):].split('$')
         repo_name = elements[0]
         can_access = HasRepoPermissionAny(
             'repository.read',
             'repository.write',
             'repository.admin')(repo_name)
         log.debug(
             'permission check for %s channel resulted in %s',
             repo_name, can_access)
         if can_access:
             return True
     return False
 
 
 def check_channel_permissions(channels, plugin_validators, should_raise=True):
     valid_channels = []
 
     validators = [broadcast_validator, repo_validator]
     if plugin_validators:
         validators.extend(plugin_validators)
     for channel_name in channels:
         is_valid = False
         for validator in validators:
             if validator(channel_name):
                 is_valid = True
                 break
         if is_valid:
             valid_channels.append(channel_name)
         else:
             if should_raise:
                 raise ChannelstreamPermissionException()
     return valid_channels
 
 
 def get_channels_info(self, channels):
     payload = {'channels': channels}
     # gather persistence info
     return channelstream_request(self._config(), payload, '/info')
 
 
 def parse_channels_info(info_result, include_channel_info=None):
     """
     Returns data that contains only secure information that can be
     presented to clients
     """
     include_channel_info = include_channel_info or []
 
     user_state_dict = {}
     for userinfo in info_result['users']:
         user_state_dict[userinfo['user']] = {
             k: v for k, v in userinfo['state'].items()
             if k in STATE_PUBLIC_KEYS
             }
 
     channels_info = {}
 
     for c_name, c_info in info_result['channels'].items():
         if c_name not in include_channel_info:
             continue
         connected_list = []
         for userinfo in c_info['users']:
             connected_list.append({
                 'user': userinfo['user'],
                 'state': user_state_dict[userinfo['user']]
             })
         channels_info[c_name] = {'users': connected_list,
                                  'history': c_info['history']}
 
     return channels_info
 
 
 def log_filepath(history_location, channel_name):
     hasher = hashlib.sha256()
     hasher.update(channel_name.encode('utf8'))
     filename = '{}.log'.format(hasher.hexdigest())
     filepath = os.path.join(history_location, filename)
     return filepath
 
 
 def read_history(history_location, channel_name):
     filepath = log_filepath(history_location, channel_name)
     if not os.path.exists(filepath):
         return []
     history_lines_limit = -100
     history = []
     with open(filepath, 'rb') as f:
         for line in f.readlines()[history_lines_limit:]:
             try:
                 history.append(json.loads(line))
             except Exception:
                 log.exception('Failed to load history')
     return history
 
 
 def update_history_from_logs(config, channels, payload):
     history_location = config.get('history.location')
     for channel in channels:
         history = read_history(history_location, channel)
         payload['channels_info'][channel]['history'] = history
 
 
 def write_history(config, message):
     """ writes a messge to a base64encoded filename """
     history_location = config.get('history.location')
     if not os.path.exists(history_location):
         return
     try:
         LOCK.acquire_write_lock()
         filepath = log_filepath(history_location, message['channel'])
         with open(filepath, 'ab') as f:
             json.dump(message, f)
             f.write('\n')
     finally:
         LOCK.release_write_lock()
 
 
 def get_connection_validators(registry):
     validators = []
     for k, config in registry.rhodecode_plugins.iteritems():
         validator = config.get('channelstream', {}).get('connect_validator')
         if validator:
             validators.append(validator)
     return validators
 
 
 def post_message(channel, message, username, registry=None):
 
     if not registry:
         registry = get_current_registry()
 
     log.debug('Channelstream: sending notification to channel %s', channel)
     rhodecode_plugins = getattr(registry, 'rhodecode_plugins', {})
     channelstream_config = rhodecode_plugins.get('channelstream', {})
     if channelstream_config.get('enabled'):
         payload = {
             'type': 'message',
             'timestamp': datetime.datetime.utcnow(),
             'user': 'system',
             'exclude_users': [username],
             'channel': channel,
             'message': {
                 'message': message,
                 'level': 'success',
                 'topic': '/notifications'
             }
         }
 
         try:
             return channelstream_request(
                 channelstream_config, [payload], '/message',
                 raise_exc=False)
         except ChannelstreamException:
             log.exception('Failed to send channelstream data')
             raise