channelstream: separate element and basic pubsub

backend/src/appenlight/lib/utils/__init__.py

@@ -1,495 +1,495 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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
 # AppEnlight Enterprise Edition, including its added features, Support
 # services, and proprietary license terms, please see
 # https://rhodecode.com/licenses/
 
 """
 Utility functions.
 """
 import logging
 import requests
 import hashlib
 import json
 import copy
 import uuid
 import appenlight.lib.helpers as h
 from collections import namedtuple
 from datetime import timedelta, datetime, date
 from dogpile.cache.api import NO_VALUE
 from appenlight.models import Datastores
 from appenlight.validators import (LogSearchSchema,
                                    TagListSchema,
                                    accepted_search_params)
 from itsdangerous import TimestampSigner
 from ziggurat_foundations.permissions import ALL_PERMISSIONS
 from dateutil.relativedelta import relativedelta
 from dateutil.rrule import rrule, MONTHLY, DAILY
 
 log = logging.getLogger(__name__)
 
 
 Stat = namedtuple('Stat', 'start_interval value')
 
 
 def default_extractor(item):
     """
     :param item - item to extract date from
     """
     if hasattr(item, 'start_interval'):
         return item.start_interval
     return item['start_interval']
 
 
 # fast gap generator
 def gap_gen_default(start, step, itemiterator, end_time=None,
                     iv_extractor=None):
     """ generates a list of time/value items based on step and itemiterator
         if there are entries missing from iterator time/None will be returned
         instead
     :param start - datetime - what time should we start generating our values
     :param step - timedelta - stepsize
     :param itemiterator - iterable - we will check this iterable for values
     corresponding to generated steps
     :param end_time - datetime - when last step is >= end_time stop iterating
     :param iv_extractor - extracts current step from iterable items
     """
 
     if not iv_extractor:
         iv_extractor = default_extractor
 
     next_step = start
     minutes = step.total_seconds() / 60.0
     while next_step.minute % minutes != 0:
         next_step = next_step.replace(minute=next_step.minute - 1)
     for item in itemiterator:
         item_start_interval = iv_extractor(item)
         # do we have a match for current time step in our data?
         # no gen a new tuple with 0 values
         while next_step < item_start_interval:
             yield Stat(next_step, None)
             next_step = next_step + step
         if next_step == item_start_interval:
             yield Stat(item_start_interval, item)
             next_step = next_step + step
     if end_time:
         while next_step < end_time:
             yield Stat(next_step, None)
             next_step = next_step + step
 
 
 class DateTimeEncoder(json.JSONEncoder):
     """ Simple datetime to ISO encoder for json serialization"""
 
     def default(self, obj):
         if isinstance(obj, date):
             return obj.isoformat()
         if isinstance(obj, datetime):
             return obj.isoformat()
         return json.JSONEncoder.default(self, obj)
 
 
-def cometd_request(secret, endpoint, payload, throw_exceptions=False,
+def channelstream_request(secret, endpoint, payload, throw_exceptions=False,
                    servers=None):
     responses = []
     if not servers:
         servers = []
 
     signer = TimestampSigner(secret)
     sig_for_server = signer.sign(endpoint)
     for secret, server in [(s['secret'], s['server']) for s in servers]:
         response = {}
         secret_headers = {'x-channelstream-secret': sig_for_server,
                           'x-channelstream-endpoint': endpoint,
                           'Content-Type': 'application/json'}
         url = '%s%s' % (server, endpoint)
         try:
             response = requests.post(url,
                                      data=json.dumps(payload,
                                                      cls=DateTimeEncoder),
                                      headers=secret_headers,
                                      verify=False,
                                      timeout=2).json()
         except requests.exceptions.RequestException as e:
             if throw_exceptions:
                 raise
         responses.append(response)
     return responses
 
 
 def add_cors_headers(response):
     # allow CORS
     response.headers.add('Access-Control-Allow-Origin', '*')
     response.headers.add('XDomainRequestAllowed', '1')
     response.headers.add('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
     # response.headers.add('Access-Control-Allow-Credentials', 'true')
     response.headers.add('Access-Control-Allow-Headers',
                          'Content-Type, Depth, User-Agent, X-File-Size, X-Requested-With, If-Modified-Since, X-File-Name, Cache-Control, Pragma, Origin, Connection, Referer, Cookie')
     response.headers.add('Access-Control-Max-Age', '86400')
 
 
 from sqlalchemy.sql import compiler
 from psycopg2.extensions import adapt as sqlescape
 
 
 # or use the appropiate escape function from your db driver
 
 def compile_query(query):
     dialect = query.session.bind.dialect
     statement = query.statement
     comp = compiler.SQLCompiler(dialect, statement)
     comp.compile()
     enc = dialect.encoding
     params = {}
     for k, v in comp.params.items():
         if isinstance(v, str):
             v = v.encode(enc)
         params[k] = sqlescape(v)
     return (comp.string.encode(enc) % params).decode(enc)
 
 
 def convert_es_type(input_data):
     """
     This might need to convert some text or other types to corresponding ES types
     """
     return str(input_data)
 
 
 ProtoVersion = namedtuple('ProtoVersion', ['major', 'minor', 'patch'])
 
 
 def parse_proto(input_data):
     try:
         parts = [int(x) for x in input_data.split('.')]
         while len(parts) < 3:
             parts.append(0)
         return ProtoVersion(*parts)
     except Exception as e:
         log.info('Unknown protocol version: %s' % e)
     return ProtoVersion(99, 99, 99)
 
 
 def es_index_name_limiter(start_date=None, end_date=None, months_in_past=6,
                           ixtypes=None):
     """
     This function limits the search to 6 months by default so we don't have to
     query 300 elasticsearch indices for 20 years of historical data for example
     """
 
     # should be cached later
     def get_possible_names():
         return list(Datastores.es.aliases().keys())
 
     possible_names = get_possible_names()
     es_index_types = []
     if not ixtypes:
         ixtypes = ['reports', 'metrics', 'logs']
     for t in ixtypes:
         if t == 'reports':
             es_index_types.append('rcae_r_%s')
         elif t == 'logs':
             es_index_types.append('rcae_l_%s')
         elif t == 'metrics':
             es_index_types.append('rcae_m_%s')
         elif t == 'uptime':
             es_index_types.append('rcae_u_%s')
         elif t == 'slow_calls':
             es_index_types.append('rcae_sc_%s')
 
     if start_date:
         start_date = copy.copy(start_date)
     else:
         if not end_date:
             end_date = datetime.utcnow()
         start_date = end_date + relativedelta(months=months_in_past * -1)
 
     if not end_date:
         end_date = start_date + relativedelta(months=months_in_past)
 
     index_dates = list(rrule(MONTHLY,
                              dtstart=start_date.date().replace(day=1),
                              until=end_date.date(),
                              count=36))
     index_names = []
     for ix_type in es_index_types:
         to_extend = [ix_type % d.strftime('%Y_%m') for d in index_dates
                      if ix_type % d.strftime('%Y_%m') in possible_names]
         index_names.extend(to_extend)
         for day in list(rrule(DAILY, dtstart=start_date.date(),
                               until=end_date.date(), count=366)):
             ix_name = ix_type % day.strftime('%Y_%m_%d')
             if ix_name in possible_names:
                 index_names.append(ix_name)
     return index_names
 
 
 def build_filter_settings_from_query_dict(
         request, params=None, override_app_ids=None,
         resource_permissions=None):
     """
     Builds list of normalized search terms for ES from query params
     ensuring application list is restricted to only applications user
     has access to
 
     :param params (dictionary)
     :param override_app_ids - list of application id's to use instead of
     applications user normally has access to
     """
     params = copy.deepcopy(params)
     applications = []
     if not resource_permissions:
         resource_permissions = ['view']
 
     if request.user:
         applications = request.user.resources_with_perms(
             resource_permissions, resource_types=['application'])
 
     # CRITICAL - this ensures our resultset is limited to only the ones
     # user has view permissions
     all_possible_app_ids = set([app.resource_id for app in applications])
 
     # if override is preset we force permission for app to be present
     # this allows users to see dashboards and applications they would
     # normally not be able to
 
     if override_app_ids:
         all_possible_app_ids = set(override_app_ids)
 
     schema = LogSearchSchema().bind(resources=all_possible_app_ids)
     tag_schema = TagListSchema()
     filter_settings = schema.deserialize(params)
     tag_list = []
     for k, v in list(filter_settings.items()):
         if k in accepted_search_params:
             continue
         tag_list.append({"name": k, "value": v, "op": 'eq'})
         # remove the key from filter_settings
         filter_settings.pop(k, None)
     tags = tag_schema.deserialize(tag_list)
     filter_settings['tags'] = tags
     return filter_settings
 
 
 def gen_uuid():
     return str(uuid.uuid4())
 
 
 def gen_uuid4_sha_hex():
     return hashlib.sha1(uuid.uuid4().bytes).hexdigest()
 
 
 def permission_tuple_to_dict(data):
     out = {
         "user_name": None,
         "perm_name": data.perm_name,
         "owner": data.owner,
         "type": data.type,
         "resource_name": None,
         "resource_type": None,
         "resource_id": None,
         "group_name": None,
         "group_id": None
     }
     if data.user:
         out["user_name"] = data.user.user_name
     if data.perm_name == ALL_PERMISSIONS:
         out['perm_name'] = '__all_permissions__'
     if data.resource:
         out['resource_name'] = data.resource.resource_name
         out['resource_type'] = data.resource.resource_type
         out['resource_id'] = data.resource.resource_id
     if data.group:
         out['group_name'] = data.group.group_name
         out['group_id'] = data.group.id
     return out
 
 
 def get_cached_buckets(request, stats_since, end_time, fn, cache_key,
                        gap_gen=None, db_session=None, step_interval=None,
                        iv_extractor=None,
                        rerange=False, *args, **kwargs):
     """ Takes "fn" that should return some data and tries to load the data
     dividing it into daily buckets - if the stats_since and end time give a
     delta bigger than 24hours, then only "todays" data is computed on the fly
 
     :param request: (request) request object
     :param stats_since: (datetime) start date of buckets range
     :param end_time: (datetime) end date of buckets range - utcnow() if None
     :param fn: (callable) callable to use to populate buckets should have
     following signature:
         def get_data(request, since_when, until, *args, **kwargs):
 
     :param cache_key: (string) cache key that will be used to build bucket
     caches
     :param gap_gen: (callable) gap generator - should return step intervals
     to use with out `fn` callable
     :param db_session: (Session) sqlalchemy session
     :param step_interval: (timedelta) optional step interval if we want to
     override the default determined from total start/end time delta
     :param iv_extractor: (callable) used to get step intervals from data
     returned by `fn` callable
     :param rerange: (bool) handy if we want to change ranges from hours to
     days when cached data is missing - will shorten execution time if `fn`
     callable supports that and we are working with multiple rows - like metrics
     :param args:
     :param kwargs:
 
     :return: iterable
     """
     if not end_time:
         end_time = datetime.utcnow().replace(second=0, microsecond=0)
     delta = end_time - stats_since
     # if smaller than 3 days we want to group by 5min else by 1h,
     # for 60 min group by min
     if not gap_gen:
         gap_gen = gap_gen_default
     if not iv_extractor:
         iv_extractor = default_extractor
 
     # do not use custom interval if total time range with new iv would exceed
     # end time
     if not step_interval or stats_since + step_interval >= end_time:
         if delta < h.time_deltas.get('12h')['delta']:
             step_interval = timedelta(seconds=60)
         elif delta < h.time_deltas.get('3d')['delta']:
             step_interval = timedelta(seconds=60 * 5)
         elif delta > h.time_deltas.get('2w')['delta']:
             step_interval = timedelta(days=1)
         else:
             step_interval = timedelta(minutes=60)
 
     if step_interval >= timedelta(minutes=60):
         log.info('cached_buckets:{}: adjusting start time '
                  'for hourly or daily intervals'.format(cache_key))
         stats_since = stats_since.replace(hour=0, minute=0)
 
     ranges = [i.start_interval for i in list(gap_gen(stats_since,
                                                      step_interval, [],
                                                      end_time=end_time))]
     buckets = {}
     storage_key = 'buckets:' + cache_key + '{}|{}'
     # this means we basicly cache per hour in 3-14 day intervals but i think
     # its fine at this point - will be faster than db access anyways
 
     if len(ranges) >= 1:
         last_ranges = [ranges[-1]]
     else:
         last_ranges = []
     if step_interval >= timedelta(minutes=60):
         for r in ranges:
             k = storage_key.format(step_interval.total_seconds(), r)
             value = request.registry.cache_regions.redis_day_30.get(k)
             # last buckets are never loaded from cache
             is_last_result = (
                 r >= end_time - timedelta(hours=6) or r in last_ranges)
             if value is not NO_VALUE and not is_last_result:
                 log.info("cached_buckets:{}: "
                          "loading range {} from cache".format(cache_key, r))
                 buckets[r] = value
             else:
                 log.info("cached_buckets:{}: "
                          "loading range {} from storage".format(cache_key, r))
                 range_size = step_interval
                 if (step_interval == timedelta(minutes=60) and
                         not is_last_result and rerange):
                     range_size = timedelta(days=1)
                     r = r.replace(hour=0, minute=0)
                     log.info("cached_buckets:{}: "
                              "loading collapsed "
                              "range {} {}".format(cache_key, r,
                                                   r + range_size))
                 bucket_data = fn(
                     request, r, r + range_size, step_interval,
                     gap_gen, bucket_count=len(ranges), *args, **kwargs)
                 for b in bucket_data:
                     b_iv = iv_extractor(b)
                     buckets[b_iv] = b
                     k2 = storage_key.format(
                         step_interval.total_seconds(), b_iv)
                     request.registry.cache_regions.redis_day_30.set(k2, b)
         log.info("cached_buckets:{}: saving cache".format(cache_key))
     else:
         # bucket count is 1 for short time ranges <= 24h from now
         bucket_data = fn(request, stats_since, end_time, step_interval,
                          gap_gen, bucket_count=1, *args, **kwargs)
         for b in bucket_data:
             buckets[iv_extractor(b)] = b
     return buckets
 
 
 def get_cached_split_data(request, stats_since, end_time, fn, cache_key,
                           db_session=None, *args, **kwargs):
     """ Takes "fn" that should return some data and tries to load the data
     dividing it into 2 buckets - cached "since_from" bucket and "today"
     bucket - then the data can be reduced into single value
 
     Data is cached if the stats_since and end time give a delta bigger
     than 24hours - then only 24h is computed on the fly
     """
     if not end_time:
         end_time = datetime.utcnow().replace(second=0, microsecond=0)
     delta = end_time - stats_since
 
     if delta >= timedelta(minutes=60):
         log.info('cached_split_data:{}: adjusting start time '
                  'for hourly or daily intervals'.format(cache_key))
         stats_since = stats_since.replace(hour=0, minute=0)
 
     storage_key = 'buckets_split_data:' + cache_key + ':{}|{}'
     old_end_time = end_time.replace(hour=0, minute=0)
 
     final_storage_key = storage_key.format(delta.total_seconds(),
                                            old_end_time)
     older_data = None
 
     cdata = request.registry.cache_regions.redis_day_7.get(
         final_storage_key)
 
     if cdata:
         log.info("cached_split_data:{}: found old "
                  "bucket data".format(cache_key))
         older_data = cdata
 
     if (stats_since < end_time - h.time_deltas.get('24h')['delta'] and
             not cdata):
         log.info("cached_split_data:{}: didn't find the "
                  "start bucket in cache so load older data".format(cache_key))
         recent_stats_since = old_end_time
         older_data = fn(request, stats_since, recent_stats_since,
                         db_session=db_session, *args, **kwargs)
         request.registry.cache_regions.redis_day_7.set(final_storage_key,
                                                        older_data)
     elif stats_since < end_time - h.time_deltas.get('24h')['delta']:
         recent_stats_since = old_end_time
     else:
         recent_stats_since = stats_since
 
     log.info("cached_split_data:{}: loading fresh "
              "data bucksts from last 24h ".format(cache_key))
     todays_data = fn(request, recent_stats_since, end_time,
                      db_session=db_session, *args, **kwargs)
     return older_data, todays_data
 
 
 def in_batches(seq, size):
     """
     Splits am iterable into batches of specified size
     :param seq (iterable)
     :param size integer
     """
     return (seq[pos:pos + size] for pos in range(0, len(seq), size))

backend/src/appenlight/models/report.py

@@ -1,512 +1,511 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  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
 # AppEnlight Enterprise Edition, including its added features, Support
 # services, and proprietary license terms, please see
 # https://rhodecode.com/licenses/
 
 from datetime import datetime
 import math
 import uuid
 import hashlib
 import copy
 import urllib.parse
 import logging
 import sqlalchemy as sa
 
 from appenlight.models import Base, Datastores
 from appenlight.lib.utils.date_utils import convert_date
 from appenlight.lib.utils import convert_es_type
 from appenlight.models.slow_call import SlowCall
-from appenlight.lib.utils import cometd_request
+from appenlight.lib.utils import channelstream_request
 from appenlight.lib.enums import ReportType, Language
 from pyramid.threadlocal import get_current_registry, get_current_request
 from sqlalchemy.dialects.postgresql import JSON
 from ziggurat_foundations.models.base import BaseModel
 
 log = logging.getLogger(__name__)
 
 REPORT_TYPE_MATRIX = {
     'http_status': {"type": 'int',
                     "ops": ('eq', 'ne', 'ge', 'le',)},
     'group:priority': {"type": 'int',
                        "ops": ('eq', 'ne', 'ge', 'le',)},
     'duration': {"type": 'float',
                  "ops": ('ge', 'le',)},
     'url_domain': {"type": 'unicode',
                    "ops": ('eq', 'ne', 'startswith', 'endswith', 'contains',)},
     'url_path': {"type": 'unicode',
                  "ops": ('eq', 'ne', 'startswith', 'endswith', 'contains',)},
     'error': {"type": 'unicode',
               "ops": ('eq', 'ne', 'startswith', 'endswith', 'contains',)},
     'tags:server_name': {"type": 'unicode',
                          "ops": ('eq', 'ne', 'startswith', 'endswith',
                                  'contains',)},
     'traceback': {"type": 'unicode',
                   "ops": ('contains',)},
     'group:occurences': {"type": 'int',
                          "ops": ('eq', 'ne', 'ge', 'le',)}
 }
 
 
 class Report(Base, BaseModel):
     __tablename__ = 'reports'
     __table_args__ = {'implicit_returning': False}
 
     id = sa.Column(sa.Integer, nullable=False, primary_key=True)
     group_id = sa.Column(sa.BigInteger,
                          sa.ForeignKey('reports_groups.id', ondelete='cascade',
                                        onupdate='cascade'))
     resource_id = sa.Column(sa.Integer(), nullable=False, index=True)
     report_type = sa.Column(sa.Integer(), nullable=False, index=True)
     error = sa.Column(sa.UnicodeText(), index=True)
     extra = sa.Column(JSON(), default={})
     request = sa.Column(JSON(), nullable=False, default={})
     ip = sa.Column(sa.String(39), index=True, default='')
     username = sa.Column(sa.Unicode(255), default='')
     user_agent = sa.Column(sa.Unicode(255), default='')
     url = sa.Column(sa.UnicodeText(), index=True)
     request_id = sa.Column(sa.Text())
     request_stats = sa.Column(JSON(), nullable=False, default={})
     traceback = sa.Column(JSON(), nullable=False, default=None)
     traceback_hash = sa.Column(sa.Text())
     start_time = sa.Column(sa.DateTime(), default=datetime.utcnow,
                            server_default=sa.func.now())
     end_time = sa.Column(sa.DateTime())
     duration = sa.Column(sa.Float, default=0)
     http_status = sa.Column(sa.Integer, index=True)
     url_domain = sa.Column(sa.Unicode(100), index=True)
     url_path = sa.Column(sa.Unicode(255), index=True)
     tags = sa.Column(JSON(), nullable=False, default={})
     language = sa.Column(sa.Integer(), default=0)
     # this is used to determine partition for the report
     report_group_time = sa.Column(sa.DateTime(), default=datetime.utcnow,
                                   server_default=sa.func.now())
 
     logs = sa.orm.relationship(
         'Log',
         lazy='dynamic',
         passive_deletes=True,
         passive_updates=True,
         primaryjoin="and_(Report.request_id==Log.request_id, "
                     "Log.request_id != None, Log.request_id != '')",
         foreign_keys='[Log.request_id]')
 
     slow_calls = sa.orm.relationship('SlowCall',
                                      backref='detail',
                                      cascade="all, delete-orphan",
                                      passive_deletes=True,
                                      passive_updates=True,
                                      order_by='SlowCall.timestamp')
 
     def set_data(self, data, resource, protocol_version=None):
         self.http_status = data['http_status']
         self.priority = data['priority']
         self.error = data['error']
         report_language = data.get('language', '').lower()
         self.language = getattr(Language, report_language, Language.unknown)
         # we need temp holder here to decide later
         # if we want to to commit the tags if report is marked for creation
         self.tags = {
             'server_name': data['server'],
             'view_name': data['view_name']
         }
         if data.get('tags'):
             for tag_tuple in data['tags']:
                 self.tags[tag_tuple[0]] = tag_tuple[1]
         self.traceback = data['traceback']
         stripped_traceback = self.stripped_traceback()
         tb_repr = repr(stripped_traceback).encode('utf8')
         self.traceback_hash = hashlib.sha1(tb_repr).hexdigest()
         url_info = urllib.parse.urlsplit(
             data.get('url', ''), allow_fragments=False)
         self.url_domain = url_info.netloc[:128]
         self.url_path = url_info.path[:2048]
         self.occurences = data['occurences']
         if self.error:
             self.report_type = ReportType.error
         else:
             self.report_type = ReportType.slow
 
         # but if its status 404 its 404 type
         if self.http_status in [404, '404'] or self.error == '404 Not Found':
             self.report_type = ReportType.not_found
             self.error = ''
 
         self.generate_grouping_hash(data.get('appenlight.group_string',
                                              data.get('group_string')),
                                     resource.default_grouping,
                                     protocol_version)
 
         # details
         if data['http_status'] in [404, '404']:
             data = {"username": data["username"],
                     "ip": data["ip"],
                     "url": data["url"],
                     "user_agent": data["user_agent"]}
             if data.get('HTTP_REFERER') or data.get('http_referer'):
                 data['HTTP_REFERER'] = data.get(
                     'HTTP_REFERER', '') or data.get('http_referer', '')
 
         self.resource_id = resource.resource_id
         self.username = data['username']
         self.user_agent = data['user_agent']
         self.ip = data['ip']
         self.extra = {}
         if data.get('extra'):
             for extra_tuple in data['extra']:
                 self.extra[extra_tuple[0]] = extra_tuple[1]
 
         self.url = data['url']
         self.request_id = data.get('request_id', '').replace('-', '') or str(
             uuid.uuid4())
         request_data = data.get('request', {})
 
         self.request = request_data
         self.request_stats = data.get('request_stats', {})
         traceback = data.get('traceback')
         if not traceback:
             traceback = data.get('frameinfo')
         self.traceback = traceback
         start_date = convert_date(data.get('start_time'))
         if not self.start_time or self.start_time < start_date:
             self.start_time = start_date
 
         self.end_time = convert_date(data.get('end_time'), False)
         self.duration = 0
 
         if self.start_time and self.end_time:
             d = self.end_time - self.start_time
             self.duration = d.total_seconds()
 
         # update tags with other vars
         if self.username:
             self.tags['user_name'] = self.username
         self.tags['report_language'] = Language.key_from_value(self.language)
 
     def add_slow_calls(self, data, report_group):
         slow_calls = []
         for call in data.get('slow_calls', []):
             sc_inst = SlowCall()
             sc_inst.set_data(call, resource_id=self.resource_id,
                              report_group=report_group)
             slow_calls.append(sc_inst)
             self.slow_calls.extend(slow_calls)
         return slow_calls
 
     def get_dict(self, request, details=False, exclude_keys=None,
                  include_keys=None):
         from appenlight.models.services.report_group import ReportGroupService
         instance_dict = super(Report, self).get_dict()
         instance_dict['req_stats'] = self.req_stats()
         instance_dict['group'] = {}
         instance_dict['group']['id'] = self.report_group.id
         instance_dict['group'][
             'total_reports'] = self.report_group.total_reports
         instance_dict['group']['last_report'] = self.report_group.last_report
         instance_dict['group']['priority'] = self.report_group.priority
         instance_dict['group']['occurences'] = self.report_group.occurences
         instance_dict['group'][
             'last_timestamp'] = self.report_group.last_timestamp
         instance_dict['group'][
             'first_timestamp'] = self.report_group.first_timestamp
         instance_dict['group']['public'] = self.report_group.public
         instance_dict['group']['fixed'] = self.report_group.fixed
         instance_dict['group']['read'] = self.report_group.read
         instance_dict['group'][
             'average_duration'] = self.report_group.average_duration
 
         instance_dict[
             'resource_name'] = self.report_group.application.resource_name
         instance_dict['report_type'] = self.report_type
 
         if instance_dict['http_status'] == 404 and not instance_dict['error']:
             instance_dict['error'] = '404 Not Found'
 
         if details:
             instance_dict['affected_users_count'] = \
                 ReportGroupService.affected_users_count(self.report_group)
             instance_dict['top_affected_users'] = [
                 {'username': u.username, 'count': u.count} for u in
                 ReportGroupService.top_affected_users(self.report_group)]
             instance_dict['application'] = {'integrations': []}
             for integration in self.report_group.application.integrations:
                 if integration.front_visible:
                     instance_dict['application']['integrations'].append(
                         {'name': integration.integration_name,
                          'action': integration.integration_action})
             instance_dict['comments'] = [c.get_dict() for c in
                                          self.report_group.comments]
 
             instance_dict['group']['next_report'] = None
             instance_dict['group']['previous_report'] = None
             next_in_group = self.get_next_in_group(request)
             previous_in_group = self.get_previous_in_group(request)
             if next_in_group:
                 instance_dict['group']['next_report'] = next_in_group
             if previous_in_group:
                 instance_dict['group']['previous_report'] = previous_in_group
 
             # slow call ordering
             def find_parent(row, data):
                 for r in reversed(data):
                     try:
                         if (row['timestamp'] > r['timestamp'] and
                                     row['end_time'] < r['end_time']):
                             return r
                     except TypeError as e:
                         log.warning('reports_view.find_parent: %s' % e)
                 return None
 
             new_calls = []
             calls = [c.get_dict() for c in self.slow_calls]
             while calls:
                 # start from end
                 for x in range(len(calls) - 1, -1, -1):
                     parent = find_parent(calls[x], calls)
                     if parent:
                         parent['children'].append(calls[x])
                     else:
                         # no parent at all? append to new calls anyways
                         new_calls.append(calls[x])
                         # print 'append', calls[x]
                     del calls[x]
                     break
             instance_dict['slow_calls'] = new_calls
 
         instance_dict['front_url'] = self.get_public_url(request)
 
         exclude_keys_list = exclude_keys or []
         include_keys_list = include_keys or []
         for k in list(instance_dict.keys()):
             if k == 'group':
                 continue
             if (k in exclude_keys_list or
                     (k not in include_keys_list and include_keys)):
                 del instance_dict[k]
         return instance_dict
 
     def get_previous_in_group(self, request):
         query = {
             "size": 1,
             "query": {
                 "filtered": {
                     "filter": {
                         "and": [{"term": {"group_id": self.group_id}},
                                 {"range": {"pg_id": {"lt": self.id}}}]
                     }
                 }
             },
             "sort": [
                 {"_doc": {"order": "desc"}},
             ],
         }
         result = request.es_conn.search(query, index=self.partition_id,
                                         doc_type='report')
         if result['hits']['total']:
             return result['hits']['hits'][0]['_source']['pg_id']
 
     def get_next_in_group(self, request):
         query = {
             "size": 1,
             "query": {
                 "filtered": {
                     "filter": {
                         "and": [{"term": {"group_id": self.group_id}},
                                 {"range": {"pg_id": {"gt": self.id}}}]
                     }
                 }
             },
             "sort": [
                 {"_doc": {"order": "asc"}},
             ],
         }
         result = request.es_conn.search(query, index=self.partition_id,
                                         doc_type='report')
         if result['hits']['total']:
             return result['hits']['hits'][0]['_source']['pg_id']
 
     def get_public_url(self, request=None, report_group=None, _app_url=None):
         """
         Returns url that user can use to visit specific report
         """
         if not request:
             request = get_current_request()
         url = request.route_url('/', _app_url=_app_url)
         if report_group:
             return (url + 'ui/report/%s/%s') % (report_group.id, self.id)
         return (url + 'ui/report/%s/%s') % (self.group_id, self.id)
 
     def req_stats(self):
         stats = self.request_stats.copy()
         stats['percentages'] = {}
         stats['percentages']['main'] = 100.0
         main = stats.get('main', 0.0)
         if not main:
             return None
         for name, call_time in stats.items():
             if ('calls' not in name and 'main' not in name and
                         'percentages' not in name):
                 stats['main'] -= call_time
                 stats['percentages'][name] = math.floor(
                     (call_time / main * 100.0))
                 stats['percentages']['main'] -= stats['percentages'][name]
         if stats['percentages']['main'] < 0.0:
             stats['percentages']['main'] = 0.0
             stats['main'] = 0.0
         return stats
 
     def generate_grouping_hash(self, hash_string=None, default_grouping=None,
                                protocol_version=None):
         """
         Generates SHA1 hash that will be used to group reports together
         """
         if not hash_string:
             location = self.tags.get('view_name') or self.url_path;
             server_name = self.tags.get('server_name') or ''
             if default_grouping == 'url_traceback':
                 hash_string = '%s_%s_%s' % (self.traceback_hash, location,
                                              self.error)
                 if self.language == Language.javascript:
                     hash_string = '%s_%s' % (self.traceback_hash, self.error)
 
             elif default_grouping == 'traceback_server':
                 hash_string = '%s_%s' % (self.traceback_hash, server_name)
                 if self.language == Language.javascript:
                     hash_string = '%s_%s' % (self.traceback_hash, server_name)
             else:
                 hash_string = '%s_%s' % (self.error, location)
         binary_string = hash_string.encode('utf8')
         self.grouping_hash = hashlib.sha1(binary_string).hexdigest()
         return self.grouping_hash
 
     def stripped_traceback(self):
         """
         Traceback without local vars
         """
         stripped_traceback = copy.deepcopy(self.traceback)
 
         if isinstance(stripped_traceback, list):
             for row in stripped_traceback:
                 row.pop('vars', None)
         return stripped_traceback
 
     def notify_channel(self, report_group):
         """
         Sends notification to websocket channel
         """
         settings = get_current_registry().settings
-        log.info('notify cometd')
+        log.info('notify channelstream')
         if self.report_type != ReportType.error:
             return
         payload = {
             'type': 'message',
             "user": '__system__',
             "channel": 'app_%s' % self.resource_id,
             'message': {
-                'type': 'report',
+                'topic': 'front_dashboard.new_topic',
                 'report': {
                     'group': {
                         'priority': report_group.priority,
                         'first_timestamp': report_group.first_timestamp,
                         'last_timestamp': report_group.last_timestamp,
                         'average_duration': report_group.average_duration,
                         'occurences': report_group.occurences
                     },
                     'report_id': self.id,
                     'group_id': self.group_id,
                     'resource_id': self.resource_id,
                     'http_status': self.http_status,
                     'url_domain': self.url_domain,
                     'url_path': self.url_path,
                     'error': self.error or '',
                     'server': self.tags.get('server_name'),
                     'view_name': self.tags.get('view_name'),
                     'front_url': self.get_public_url(),
                 }
             }
 
         }
-
-        cometd_request(settings['cometd.secret'], '/message', [payload],
+        channelstream_request(settings['cometd.secret'], '/message', [payload],
                        servers=[settings['cometd_servers']])
 
     def es_doc(self):
         tags = {}
         tag_list = []
         for name, value in self.tags.items():
             name = name.replace('.', '_')
             tag_list.append(name)
             tags[name] = {
                 "values": convert_es_type(value),
                 "numeric_values": value if (
                     isinstance(value, (int, float)) and
                     not isinstance(value, bool)) else None}
 
         if 'user_name' not in self.tags and self.username:
             tags["user_name"] = {"value": [self.username],
                                  "numeric_value": None}
         return {
             '_id': str(self.id),
             'pg_id': str(self.id),
             'resource_id': self.resource_id,
             'http_status': self.http_status or '',
             'start_time': self.start_time,
             'end_time': self.end_time,
             'url_domain': self.url_domain if self.url_domain else '',
             'url_path': self.url_path if self.url_path else '',
             'duration': self.duration,
             'error': self.error if self.error else '',
             'report_type': self.report_type,
             'request_id': self.request_id,
             'ip': self.ip,
             'group_id': str(self.group_id),
             '_parent': str(self.group_id),
             'tags': tags,
             'tag_list': tag_list
         }
 
     @property
     def partition_id(self):
         return 'rcae_r_%s' % self.report_group_time.strftime('%Y_%m')
 
 
 def after_insert(mapper, connection, target):
     if not hasattr(target, '_skip_ft_index'):
         data = target.es_doc()
         data.pop('_id', None)
         Datastores.es.index(target.partition_id, 'report', data,
                             parent=target.group_id, id=target.id)
 
 
 def after_update(mapper, connection, target):
     if not hasattr(target, '_skip_ft_index'):
         data = target.es_doc()
         data.pop('_id', None)
         Datastores.es.index(target.partition_id, 'report', data,
                             parent=target.group_id, id=target.id)
 
 
 def after_delete(mapper, connection, target):
     if not hasattr(target, '_skip_ft_index'):
         query = {'term': {'pg_id': target.id}}
         Datastores.es.delete_by_query(target.partition_id, 'report', query)
 
 
 sa.event.listen(Report, 'after_insert', after_insert)
 sa.event.listen(Report, 'after_update', after_update)
 sa.event.listen(Report, 'after_delete', after_delete)