redis: some cleanups and use of pipelines for better performance

backend/src/appenlight/celery/tasks.py

@@ -1,635 +1,650 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/
 
 import bisect
 import collections
 import math
 from datetime import datetime, timedelta
 
 import sqlalchemy as sa
 import pyelasticsearch
 
 from celery.utils.log import get_task_logger
 from zope.sqlalchemy import mark_changed
 from pyramid.threadlocal import get_current_request, get_current_registry
 from appenlight.celery import celery
 from appenlight.models.report_group import ReportGroup
 from appenlight.models import DBSession, Datastores
 from appenlight.models.report import Report
 from appenlight.models.log import Log
 from appenlight.models.metric import Metric
 from appenlight.models.event import Event
 
 from appenlight.models.services.application import ApplicationService
 from appenlight.models.services.event import EventService
 from appenlight.models.services.log import LogService
 from appenlight.models.services.report import ReportService
 from appenlight.models.services.report_group import ReportGroupService
 from appenlight.models.services.user import UserService
 from appenlight.models.tag import Tag
 from appenlight.lib import print_traceback
 from appenlight.lib.utils import parse_proto, in_batches
 from appenlight.lib.ext_json import json
 from appenlight.lib.redis_keys import REDIS_KEYS
 from appenlight.lib.enums import ReportType
 
 log = get_task_logger(__name__)
 
 sample_boundries = list(range(100, 1000, 100)) + \
                    list(range(1000, 10000, 1000)) + \
                    list(range(10000, 100000, 5000))
 
 
 def pick_sample(total_occurences, report_type=None):
     every = 1.0
     position = bisect.bisect_left(sample_boundries, total_occurences)
     if position > 0:
         if report_type == ReportType.not_found:
             divide = 10.0
         else:
             divide = 100.0
         every = sample_boundries[position - 1] / divide
     return total_occurences % every == 0
 
 
 @celery.task(queue="default", default_retry_delay=1, max_retries=2)
 def test_exception_task():
     log.error('test celery log', extra={'location': 'celery'})
     log.warning('test celery log', extra={'location': 'celery'})
     raise Exception('Celery exception test')
 
 
 @celery.task(queue="default", default_retry_delay=1, max_retries=2)
 def test_retry_exception_task():
     try:
         import time
 
         time.sleep(1.3)
         log.error('test retry celery log', extra={'location': 'celery'})
         log.warning('test retry celery log', extra={'location': 'celery'})
         raise Exception('Celery exception test')
     except Exception as exc:
         test_retry_exception_task.retry(exc=exc)
 
 
 @celery.task(queue="reports", default_retry_delay=600, max_retries=144)
-def add_reports(resource_id, params, dataset, environ=None, **kwargs):
-    proto_version = parse_proto(params.get('protocol_version', ''))
+def add_reports(resource_id, request_params, dataset, **kwargs):
+    proto_version = parse_proto(request_params.get('protocol_version', ''))
     current_time = datetime.utcnow().replace(second=0, microsecond=0)
     try:
         # we will store solr docs here for single insert
         es_report_docs = {}
         es_report_group_docs = {}
         resource = ApplicationService.by_id(resource_id)
 
         tags = []
         es_slow_calls_docs = {}
         es_reports_stats_rows = {}
         for report_data in dataset:
             # build report details for later
             added_details = 0
             report = Report()
             report.set_data(report_data, resource, proto_version)
             report._skip_ft_index = True
 
             report_group = ReportGroupService.by_hash_and_resource(
                 report.resource_id,
                 report.grouping_hash
             )
             occurences = report_data.get('occurences', 1)
             if not report_group:
                 # total reports will be +1 moment later
                 report_group = ReportGroup(grouping_hash=report.grouping_hash,
                                            occurences=0, total_reports=0,
                                            last_report=0,
                                            priority=report.priority,
                                            error=report.error,
                                            first_timestamp=report.start_time)
                 report_group._skip_ft_index = True
                 report_group.report_type = report.report_type
             report.report_group_time = report_group.first_timestamp
             add_sample = pick_sample(report_group.occurences,
                                      report_type=report_group.report_type)
             if add_sample:
                 resource.report_groups.append(report_group)
                 report_group.reports.append(report)
                 added_details += 1
                 DBSession.flush()
                 if report.partition_id not in es_report_docs:
                     es_report_docs[report.partition_id] = []
                 es_report_docs[report.partition_id].append(report.es_doc())
                 tags.extend(list(report.tags.items()))
                 slow_calls = report.add_slow_calls(report_data, report_group)
                 DBSession.flush()
                 for s_call in slow_calls:
                     if s_call.partition_id not in es_slow_calls_docs:
                         es_slow_calls_docs[s_call.partition_id] = []
                     es_slow_calls_docs[s_call.partition_id].append(
                         s_call.es_doc())
                     # try generating new stat rows if needed
             else:
                 # required for postprocessing to not fail later
                 report.report_group = report_group
 
             stat_row = ReportService.generate_stat_rows(
                 report, resource, report_group)
             if stat_row.partition_id not in es_reports_stats_rows:
                 es_reports_stats_rows[stat_row.partition_id] = []
             es_reports_stats_rows[stat_row.partition_id].append(
                 stat_row.es_doc())
 
             # see if we should mark 10th occurence of report
             last_occurences_10 = int(math.floor(report_group.occurences / 10))
             curr_occurences_10 = int(math.floor(
                 (report_group.occurences + report.occurences) / 10))
             last_occurences_100 = int(
                 math.floor(report_group.occurences / 100))
             curr_occurences_100 = int(math.floor(
                 (report_group.occurences + report.occurences) / 100))
             notify_occurences_10 = last_occurences_10 != curr_occurences_10
             notify_occurences_100 = last_occurences_100 != curr_occurences_100
             report_group.occurences = ReportGroup.occurences + occurences
             report_group.last_timestamp = report.start_time
             report_group.summed_duration = ReportGroup.summed_duration + report.duration
             summed_duration = ReportGroup.summed_duration + report.duration
             summed_occurences = ReportGroup.occurences + occurences
             report_group.average_duration = summed_duration / summed_occurences
             report_group.run_postprocessing(report)
             if added_details:
                 report_group.total_reports = ReportGroup.total_reports + 1
                 report_group.last_report = report.id
             report_group.set_notification_info(notify_10=notify_occurences_10,
                                                notify_100=notify_occurences_100)
             DBSession.flush()
             report_group.get_report().notify_channel(report_group)
             if report_group.partition_id not in es_report_group_docs:
                 es_report_group_docs[report_group.partition_id] = []
             es_report_group_docs[report_group.partition_id].append(
                 report_group.es_doc())
 
             action = 'REPORT'
             log_msg = '%s: %s %s, client: %s, proto: %s' % (
                 action,
                 report_data.get('http_status', 'unknown'),
                 str(resource),
                 report_data.get('client'),
                 proto_version)
             log.info(log_msg)
         total_reports = len(dataset)
+        redis_pipeline = Datastores.redis.pipeline(transaction=False)
         key = REDIS_KEYS['counters']['reports_per_minute'].format(current_time)
-        Datastores.redis.incr(key, total_reports)
-        Datastores.redis.expire(key, 3600 * 24)
-        key = REDIS_KEYS['counters']['reports_per_minute_per_app'].format(
-            resource_id, current_time)
-        Datastores.redis.incr(key, total_reports)
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline.incr(key, total_reports)
+        redis_pipeline.expire(key, 3600 * 24)
+        key = REDIS_KEYS['counters']['reports_per_hour_per_app'].format(
+            resource_id, current_time.replace(minute=0))
+        redis_pipeline.incr(key, total_reports)
+        redis_pipeline.expire(key, 3600 * 24 * 7)
+        redis_pipeline.sadd(
+            REDIS_KEYS['apps_that_got_new_data_per_hour'],
+            resource_id, current_time.replace(minute=0))
+        redis_pipeline.execute()
 
         add_reports_es(es_report_group_docs, es_report_docs)
         add_reports_slow_calls_es(es_slow_calls_docs)
         add_reports_stats_rows_es(es_reports_stats_rows)
         return True
     except Exception as exc:
         print_traceback(log)
         add_reports.retry(exc=exc)
 
 
 @celery.task(queue="es", default_retry_delay=600, max_retries=144)
 def add_reports_es(report_group_docs, report_docs):
     for k, v in report_group_docs.items():
         Datastores.es.bulk_index(k, 'report_group', v, id_field="_id")
     for k, v in report_docs.items():
         Datastores.es.bulk_index(k, 'report', v, id_field="_id",
                                  parent_field='_parent')
 
 
 @celery.task(queue="es", default_retry_delay=600, max_retries=144)
 def add_reports_slow_calls_es(es_docs):
     for k, v in es_docs.items():
         Datastores.es.bulk_index(k, 'log', v)
 
 
 @celery.task(queue="es", default_retry_delay=600, max_retries=144)
 def add_reports_stats_rows_es(es_docs):
     for k, v in es_docs.items():
         Datastores.es.bulk_index(k, 'log', v)
 
 
 @celery.task(queue="logs", default_retry_delay=600, max_retries=144)
-def add_logs(resource_id, request, dataset, environ=None, **kwargs):
-    proto_version = request.get('protocol_version')
+def add_logs(resource_id, request_params, dataset, **kwargs):
+    proto_version = request_params.get('protocol_version')
     current_time = datetime.utcnow().replace(second=0, microsecond=0)
 
     try:
         es_docs = collections.defaultdict(list)
         application = ApplicationService.by_id(resource_id)
         ns_pairs = []
         for entry in dataset:
             # gather pk and ns so we can remove older versions of row later
             if entry['primary_key'] is not None:
                 ns_pairs.append({"pk": entry['primary_key'],
                                  "ns": entry['namespace']})
             log_entry = Log()
             log_entry.set_data(entry, resource=application)
             log_entry._skip_ft_index = True
             application.logs.append(log_entry)
             DBSession.flush()
             # insert non pk rows first
             if entry['primary_key'] is None:
                 es_docs[log_entry.partition_id].append(log_entry.es_doc())
 
         # 2nd pass to delete all log entries from db foe same pk/ns pair
         if ns_pairs:
             ids_to_delete = []
             es_docs = collections.defaultdict(list)
             es_docs_to_delete = collections.defaultdict(list)
             found_pkey_logs = LogService.query_by_primary_key_and_namespace(
                 list_of_pairs=ns_pairs)
             log_dict = {}
             for log_entry in found_pkey_logs:
                 log_key = (log_entry.primary_key, log_entry.namespace)
                 if log_key not in log_dict:
                     log_dict[log_key] = []
                 log_dict[log_key].append(log_entry)
 
             for ns, entry_list in log_dict.items():
                 entry_list = sorted(entry_list, key=lambda x: x.timestamp)
                 # newest row needs to be indexed in es
                 log_entry = entry_list[-1]
                 # delete everything from pg and ES, leave the last row in pg
                 for e in entry_list[:-1]:
                     ids_to_delete.append(e.log_id)
                     es_docs_to_delete[e.partition_id].append(e.delete_hash)
 
                 es_docs_to_delete[log_entry.partition_id].append(
                     log_entry.delete_hash)
 
                 es_docs[log_entry.partition_id].append(log_entry.es_doc())
 
             if ids_to_delete:
                 query = DBSession.query(Log).filter(
                     Log.log_id.in_(ids_to_delete))
                 query.delete(synchronize_session=False)
             if es_docs_to_delete:
                 # batch this to avoid problems with default ES bulk limits
                 for es_index in es_docs_to_delete.keys():
                     for batch in in_batches(es_docs_to_delete[es_index], 20):
                         query = {'terms': {'delete_hash': batch}}
 
                         try:
                             Datastores.es.delete_by_query(
                                 es_index, 'log', query)
                         except pyelasticsearch.ElasticHttpNotFoundError as exc:
                             msg = 'skipping index {}'.format(es_index)
                             log.info(msg)
 
         total_logs = len(dataset)
 
         log_msg = 'LOG_NEW: %s, entries: %s, proto:%s' % (
             str(application),
             total_logs,
             proto_version)
         log.info(log_msg)
         # mark_changed(session)
+        redis_pipeline = Datastores.redis.pipeline(transaction=False)
         key = REDIS_KEYS['counters']['logs_per_minute'].format(current_time)
-        Datastores.redis.incr(key, total_logs)
-        Datastores.redis.expire(key, 3600 * 24)
-        key = REDIS_KEYS['counters']['logs_per_minute_per_app'].format(
-            resource_id, current_time)
-        Datastores.redis.incr(key, total_logs)
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline.incr(key, total_logs)
+        redis_pipeline.expire(key, 3600 * 24)
+        key = REDIS_KEYS['counters']['logs_per_hour_per_app'].format(
+            resource_id, current_time.replace(minute=0))
+        redis_pipeline.incr(key, total_logs)
+        redis_pipeline.expire(key, 3600 * 24 * 7)
+        redis_pipeline.sadd(
+            REDIS_KEYS['apps_that_got_new_data_per_hour'],
+            resource_id, current_time.replace(minute=0))
+        redis_pipeline.execute()
         add_logs_es(es_docs)
         return True
     except Exception as exc:
         print_traceback(log)
         add_logs.retry(exc=exc)
 
 
 @celery.task(queue="es", default_retry_delay=600, max_retries=144)
 def add_logs_es(es_docs):
     for k, v in es_docs.items():
         Datastores.es.bulk_index(k, 'log', v)
 
 
 @celery.task(queue="metrics", default_retry_delay=600, max_retries=144)
-def add_metrics(resource_id, request, dataset, proto_version):
+def add_metrics(resource_id, request_params, dataset, proto_version):
     current_time = datetime.utcnow().replace(second=0, microsecond=0)
     try:
         application = ApplicationService.by_id_cached()(resource_id)
         application = DBSession.merge(application, load=False)
         es_docs = []
         rows = []
         for metric in dataset:
             tags = dict(metric['tags'])
             server_n = tags.get('server_name', metric['server_name']).lower()
             tags['server_name'] = server_n or 'unknown'
             new_metric = Metric(
                 timestamp=metric['timestamp'],
                 resource_id=application.resource_id,
                 namespace=metric['namespace'],
                 tags=tags)
             rows.append(new_metric)
             es_docs.append(new_metric.es_doc())
         session = DBSession()
         session.bulk_save_objects(rows)
         session.flush()
 
         action = 'METRICS'
         metrics_msg = '%s: %s, metrics: %s, proto:%s' % (
             action,
             str(application),
             len(dataset),
             proto_version
         )
         log.info(metrics_msg)
 
         mark_changed(session)
+        redis_pipeline = Datastores.redis.pipeline(transaction=False)
         key = REDIS_KEYS['counters']['metrics_per_minute'].format(current_time)
-        Datastores.redis.incr(key, len(rows))
-        Datastores.redis.expire(key, 3600 * 24)
-        key = REDIS_KEYS['counters']['metrics_per_minute_per_app'].format(
-            resource_id, current_time)
-        Datastores.redis.incr(key, len(rows))
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline.incr(key, len(rows))
+        redis_pipeline.expire(key, 3600 * 24)
+        key = REDIS_KEYS['counters']['metrics_per_hour_per_app'].format(
+            resource_id, current_time.replace(minute=0))
+        redis_pipeline.incr(key, len(rows))
+        redis_pipeline.expire(key, 3600 * 24 * 7)
+        redis_pipeline.sadd(
+            REDIS_KEYS['apps_that_got_new_data_per_hour'],
+            resource_id, current_time.replace(minute=0))
+        redis_pipeline.execute()
         add_metrics_es(es_docs)
         return True
     except Exception as exc:
         print_traceback(log)
         add_metrics.retry(exc=exc)
 
 
 @celery.task(queue="es", default_retry_delay=600, max_retries=144)
 def add_metrics_es(es_docs):
     for doc in es_docs:
         partition = 'rcae_m_%s' % doc['timestamp'].strftime('%Y_%m_%d')
         Datastores.es.index(partition, 'log', doc)
 
 
 @celery.task(queue="default", default_retry_delay=5, max_retries=2)
 def check_user_report_notifications(resource_id):
     since_when = datetime.utcnow()
     try:
         request = get_current_request()
         application = ApplicationService.by_id(resource_id)
         if not application:
             return
         error_key = REDIS_KEYS['reports_to_notify_per_type_per_app'].format(
             ReportType.error, resource_id)
         slow_key = REDIS_KEYS['reports_to_notify_per_type_per_app'].format(
             ReportType.slow, resource_id)
         error_group_ids = Datastores.redis.smembers(error_key)
         slow_group_ids = Datastores.redis.smembers(slow_key)
         Datastores.redis.delete(error_key)
         Datastores.redis.delete(slow_key)
         err_gids = [int(g_id) for g_id in error_group_ids]
         slow_gids = [int(g_id) for g_id in list(slow_group_ids)]
         group_ids = err_gids + slow_gids
         occurence_dict = {}
         for g_id in group_ids:
             key = REDIS_KEYS['counters']['report_group_occurences'].format(
                 g_id)
             val = Datastores.redis.get(key)
             Datastores.redis.delete(key)
             if val:
                 occurence_dict[g_id] = int(val)
             else:
                 occurence_dict[g_id] = 1
         report_groups = ReportGroupService.by_ids(group_ids)
         report_groups.options(sa.orm.joinedload(ReportGroup.last_report_ref))
 
         ApplicationService.check_for_groups_alert(
             application, 'alert', report_groups=report_groups,
             occurence_dict=occurence_dict)
         users = set([p.user for p in application.users_for_perm('view')])
         report_groups = report_groups.all()
         for user in users:
             UserService.report_notify(user, request, application,
                                       report_groups=report_groups,
                                       occurence_dict=occurence_dict)
         for group in report_groups:
             # marks report_groups as notified
             if not group.notified:
                 group.notified = True
     except Exception as exc:
         print_traceback(log)
         raise
 
 
 @celery.task(queue="default", default_retry_delay=5, max_retries=2)
 def check_alerts(resource_id):
     since_when = datetime.utcnow()
     try:
         request = get_current_request()
         application = ApplicationService.by_id(resource_id)
         if not application:
             return
         error_key = REDIS_KEYS[
             'reports_to_notify_per_type_per_app_alerting'].format(
             ReportType.error, resource_id)
         slow_key = REDIS_KEYS[
             'reports_to_notify_per_type_per_app_alerting'].format(
             ReportType.slow, resource_id)
         error_group_ids = Datastores.redis.smembers(error_key)
         slow_group_ids = Datastores.redis.smembers(slow_key)
         Datastores.redis.delete(error_key)
         Datastores.redis.delete(slow_key)
         err_gids = [int(g_id) for g_id in error_group_ids]
         slow_gids = [int(g_id) for g_id in list(slow_group_ids)]
         group_ids = err_gids + slow_gids
         occurence_dict = {}
         for g_id in group_ids:
             key = REDIS_KEYS['counters'][
                 'report_group_occurences_alerting'].format(
                 g_id)
             val = Datastores.redis.get(key)
             Datastores.redis.delete(key)
             if val:
                 occurence_dict[g_id] = int(val)
             else:
                 occurence_dict[g_id] = 1
         report_groups = ReportGroupService.by_ids(group_ids)
         report_groups.options(sa.orm.joinedload(ReportGroup.last_report_ref))
 
         ApplicationService.check_for_groups_alert(
             application, 'alert', report_groups=report_groups,
             occurence_dict=occurence_dict, since_when=since_when)
     except Exception as exc:
         print_traceback(log)
         raise
 
 
 @celery.task(queue="default", default_retry_delay=1, max_retries=2)
 def close_alerts():
     log.warning('Checking alerts')
     since_when = datetime.utcnow()
     try:
         event_types = [Event.types['error_report_alert'],
                        Event.types['slow_report_alert'], ]
         statuses = [Event.statuses['active']]
         # get events older than 5 min
         events = EventService.by_type_and_status(
             event_types,
             statuses,
             older_than=(since_when - timedelta(minutes=5)))
         for event in events:
             # see if we can close them
             event.validate_or_close(
                 since_when=(since_when - timedelta(minutes=1)))
     except Exception as exc:
         print_traceback(log)
         raise
 
 
 @celery.task(queue="default", default_retry_delay=600, max_retries=144)
 def update_tag_counter(tag_name, tag_value, count):
     try:
         query = DBSession.query(Tag).filter(Tag.name == tag_name).filter(
             sa.cast(Tag.value, sa.types.TEXT) == sa.cast(json.dumps(tag_value),
                                                          sa.types.TEXT))
         query.update({'times_seen': Tag.times_seen + count,
                       'last_timestamp': datetime.utcnow()},
                      synchronize_session=False)
         session = DBSession()
         mark_changed(session)
         return True
     except Exception as exc:
         print_traceback(log)
         update_tag_counter.retry(exc=exc)
 
 
 @celery.task(queue="default")
 def update_tag_counters():
     """
     Sets task to update counters for application tags
     """
     tags = Datastores.redis.lrange(REDIS_KEYS['seen_tag_list'], 0, -1)
     Datastores.redis.delete(REDIS_KEYS['seen_tag_list'])
     c = collections.Counter(tags)
     for t_json, count in c.items():
         tag_info = json.loads(t_json)
         update_tag_counter.delay(tag_info[0], tag_info[1], count)
 
 
 @celery.task(queue="default")
 def daily_digest():
     """
     Sends daily digest with top 50 error reports
     """
     request = get_current_request()
     apps = Datastores.redis.smembers(REDIS_KEYS['apps_that_had_reports'])
     Datastores.redis.delete(REDIS_KEYS['apps_that_had_reports'])
     since_when = datetime.utcnow() - timedelta(hours=8)
     log.warning('Generating daily digests')
     for resource_id in apps:
         resource_id = resource_id.decode('utf8')
         end_date = datetime.utcnow().replace(microsecond=0, second=0)
         filter_settings = {'resource': [resource_id],
                            'tags': [{'name': 'type',
                                      'value': ['error'], 'op': None}],
                            'type': 'error', 'start_date': since_when,
                            'end_date': end_date}
 
         reports = ReportGroupService.get_trending(
             request, filter_settings=filter_settings, limit=50)
 
         application = ApplicationService.by_id(resource_id)
         if application:
             users = set([p.user for p in application.users_for_perm('view')])
             for user in users:
                 user.send_digest(request, application, reports=reports,
                                  since_when=since_when)
 
 
 @celery.task(queue="default")
 def notifications_reports():
     """
     Loop that checks redis for info and then issues new tasks to celery to
     issue notifications
     """
     apps = Datastores.redis.smembers(REDIS_KEYS['apps_that_had_reports'])
     Datastores.redis.delete(REDIS_KEYS['apps_that_had_reports'])
     for app in apps:
         log.warning('Notify for app: %s' % app)
         check_user_report_notifications.delay(app.decode('utf8'))
 
 @celery.task(queue="default")
 def alerting_reports():
     """
     Loop that checks redis for info and then issues new tasks to celery to
     perform the following:
     - which applications should have new alerts opened
     """
 
     apps = Datastores.redis.smembers(REDIS_KEYS['apps_that_had_reports_alerting'])
     Datastores.redis.delete(REDIS_KEYS['apps_that_had_reports_alerting'])
     for app in apps:
         log.warning('Notify for app: %s' % app)
         check_alerts.delay(app.decode('utf8'))
 
 
 @celery.task(queue="default", soft_time_limit=3600 * 4,
              hard_time_limit=3600 * 4, max_retries=144)
 def logs_cleanup(resource_id, filter_settings):
     request = get_current_request()
     request.tm.begin()
     es_query = {
         "_source": False,
         "size": 5000,
         "query": {
             "filtered": {
                 "filter": {
                     "and": [{"term": {"resource_id": resource_id}}]
                 }
             }
         }
     }
 
     query = DBSession.query(Log).filter(Log.resource_id == resource_id)
     if filter_settings['namespace']:
         query = query.filter(Log.namespace == filter_settings['namespace'][0])
         es_query['query']['filtered']['filter']['and'].append(
             {"term": {"namespace": filter_settings['namespace'][0]}}
         )
     query.delete(synchronize_session=False)
     request.tm.commit()
     result = request.es_conn.search(es_query, index='rcae_l_*',
                                     doc_type='log', es_scroll='1m',
                                     es_search_type='scan')
     scroll_id = result['_scroll_id']
     while True:
         log.warning('log_cleanup, app:{} ns:{} batch'.format(
             resource_id,
             filter_settings['namespace']
         ))
         es_docs_to_delete = []
         result = request.es_conn.send_request(
             'POST', ['_search', 'scroll'],
             body=scroll_id, query_params={"scroll": '1m'})
         scroll_id = result['_scroll_id']
         if not result['hits']['hits']:
             break
         for doc in result['hits']['hits']:
             es_docs_to_delete.append({"id": doc['_id'],
                                       "index": doc['_index']})
 
         for batch in in_batches(es_docs_to_delete, 10):
             Datastores.es.bulk([Datastores.es.delete_op(doc_type='log',
                                                         **to_del)
                                 for to_del in batch])

backend/src/appenlight/lib/api.py

@@ -1,83 +1,86 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/
 
 import datetime
 import logging
 
 from pyramid.httpexceptions import HTTPForbidden, HTTPTooManyRequests
 
 from appenlight.models import Datastores
 from appenlight.models.services.config import ConfigService
 from appenlight.lib.redis_keys import REDIS_KEYS
 
 log = logging.getLogger(__name__)
 
 
 def rate_limiting(request, resource, section, to_increment=1):
     tsample = datetime.datetime.utcnow().replace(second=0, microsecond=0)
     key = REDIS_KEYS['rate_limits'][section].format(tsample,
                                                     resource.resource_id)
-    current_count = Datastores.redis.incr(key, to_increment)
-    Datastores.redis.expire(key, 3600 * 24)
+    redis_pipeline = request.registry.redis_conn.pipeline()
+    redis_pipeline.incr(key, to_increment)
+    redis_pipeline.expire(key, 3600 * 24)
+    results = redis_pipeline.execute()
+    current_count = results[0]
     config = ConfigService.by_key_and_section(section, 'global')
     limit = config.value if config else 1000
     if current_count > int(limit):
         log.info('RATE LIMITING: {}: {}, {}'.format(
             section, resource, current_count))
         abort_msg = 'Rate limits are in effect for this application'
         raise HTTPTooManyRequests(abort_msg,
                                   headers={'X-AppEnlight': abort_msg})
 
 
 def check_cors(request, application, should_return=True):
     """
     Performs a check and validation if request comes from authorized domain for
     application, otherwise return 403
     """
     origin_found = False
     origin = request.headers.get('Origin')
     if should_return:
         log.info('CORS for %s' % origin)
     if not origin:
         return False
     for domain in application.domains.split('\n'):
         if domain in origin:
             origin_found = True
     if origin_found:
         request.response.headers.add('Access-Control-Allow-Origin', origin)
         request.response.headers.add('XDomainRequestAllowed', '1')
         request.response.headers.add('Access-Control-Allow-Methods',
                                      'GET, POST, OPTIONS')
         request.response.headers.add('Access-Control-Allow-Headers',
                                      'Accept-Encoding, Accept-Language, '
                                      'Content-Type, '
                                      'Depth, User-Agent, X-File-Size, '
                                      'X-Requested-With, If-Modified-Since, '
                                      'X-File-Name, '
                                      'Cache-Control, Host, Pragma, Accept, '
                                      'Origin, Connection, '
                                      'Referer, Cookie, '
                                      'X-appenlight-public-api-key, '
                                      'x-appenlight-public-api-key')
         request.response.headers.add('Access-Control-Max-Age', '86400')
         return request.response
     else:
         return HTTPForbidden()

backend/src/appenlight/lib/redis_keys.py

@@ -1,67 +1,68 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/
 
 BASE = 'appenlight:data:{}'
 
 REDIS_KEYS = {
     'tasks': {
         'add_reports_lock': BASE.format('add_reports_lock:{}'),
         'add_logs_lock': BASE.format('add_logs_lock:{}'),
     },
     'counters': {
         'reports_per_minute': BASE.format('reports_per_minute:{}'),
-        'reports_per_minute_per_app': BASE.format(
-            'reports_per_minute_per_app:{}:{}'),
+        'reports_per_hour_per_app': BASE.format(
+            'reports_per_hour_per_app:{}:{}'),
         'reports_per_type': BASE.format('reports_per_type:{}'),
         'logs_per_minute': BASE.format('logs_per_minute:{}'),
-        'logs_per_minute_per_app': BASE.format(
-            'logs_per_minute_per_app:{}:{}'),
+        'logs_per_hour_per_app': BASE.format(
+            'logs_per_hour_per_app:{}:{}'),
         'metrics_per_minute': BASE.format('metrics_per_minute:{}'),
-        'metrics_per_minute_per_app': BASE.format(
-            'metrics_per_minute_per_app:{}:{}'),
+        'metrics_per_hour_per_app': BASE.format(
+            'metrics_per_hour_per_app:{}:{}'),
         'report_group_occurences': BASE.format('report_group_occurences:{}'),
         'report_group_occurences_alerting': BASE.format(
             'report_group_occurences_alerting:{}'),
         'report_group_occurences_10th': BASE.format(
             'report_group_occurences_10th:{}'),
         'report_group_occurences_100th': BASE.format(
             'report_group_occurences_100th:{}'),
     },
     'rate_limits': {
         'per_application_reports_rate_limit': BASE.format(
             'per_application_reports_limit:{}:{}'),
         'per_application_logs_rate_limit': BASE.format(
             'per_application_logs_rate_limit:{}:{}'),
         'per_application_metrics_rate_limit': BASE.format(
             'per_application_metrics_rate_limit:{}:{}'),
     },
+    'apps_that_got_new_data_per_hour': BASE.format('apps_that_got_new_data_per_hour'),
     'apps_that_had_reports': BASE.format('apps_that_had_reports'),
     'apps_that_had_error_reports': BASE.format('apps_that_had_error_reports'),
     'apps_that_had_reports_alerting': BASE.format(
         'apps_that_had_reports_alerting'),
     'apps_that_had_error_reports_alerting': BASE.format(
         'apps_that_had_error_reports_alerting'),
     'reports_to_notify_per_type_per_app': BASE.format(
         'reports_to_notify_per_type_per_app:{}:{}'),
     'reports_to_notify_per_type_per_app_alerting': BASE.format(
         'reports_to_notify_per_type_per_app_alerting:{}:{}'),
     'seen_tag_list': BASE.format('seen_tag_list')
 }

backend/src/appenlight/models/report_group.py

@@ -1,267 +1,268 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/
 
 import logging
 import sqlalchemy as sa
 
 from datetime import datetime
 
 from pyramid.threadlocal import get_current_request
 from sqlalchemy.dialects.postgresql import JSON
 from ziggurat_foundations.models.base import BaseModel
 
 from appenlight.models import Base, get_db_session, Datastores
 from appenlight.lib.enums import ReportType
 from appenlight.lib.rule import Rule
 from appenlight.lib.redis_keys import REDIS_KEYS
 from appenlight.models.report import REPORT_TYPE_MATRIX
 
 log = logging.getLogger(__name__)
 
 
 class ReportGroup(Base, BaseModel):
     __tablename__ = 'reports_groups'
     __table_args__ = {'implicit_returning': False}
 
     id = sa.Column(sa.BigInteger(), nullable=False, primary_key=True)
     resource_id = sa.Column(sa.Integer(),
                             sa.ForeignKey('applications.resource_id',
                                           onupdate='CASCADE',
                                           ondelete='CASCADE'),
                             nullable=False,
                             index=True)
     priority = sa.Column(sa.Integer, nullable=False, index=True, default=5,
                          server_default='5')
     first_timestamp = sa.Column(sa.DateTime(), default=datetime.utcnow,
                                 server_default=sa.func.now())
     last_timestamp = sa.Column(sa.DateTime(), default=datetime.utcnow,
                                server_default=sa.func.now())
     error = sa.Column(sa.UnicodeText(), index=True)
     grouping_hash = sa.Column(sa.String(40), default='')
     triggered_postprocesses_ids = sa.Column(JSON(), nullable=False,
                                             default=list)
     report_type = sa.Column(sa.Integer, default=1)
     total_reports = sa.Column(sa.Integer, default=1)
     last_report = sa.Column(sa.Integer)
     occurences = sa.Column(sa.Integer, default=1)
     average_duration = sa.Column(sa.Float, default=0)
     summed_duration = sa.Column(sa.Float, default=0)
     read = sa.Column(sa.Boolean(), index=True, default=False)
     fixed = sa.Column(sa.Boolean(), index=True, default=False)
     notified = sa.Column(sa.Boolean(), index=True, default=False)
     public = sa.Column(sa.Boolean(), index=True, default=False)
 
     reports = sa.orm.relationship('Report',
                                   lazy='dynamic',
                                   backref='report_group',
                                   cascade="all, delete-orphan",
                                   passive_deletes=True,
                                   passive_updates=True, )
 
     comments = sa.orm.relationship('ReportComment',
                                    lazy='dynamic',
                                    backref='report',
                                    cascade="all, delete-orphan",
                                    passive_deletes=True,
                                    passive_updates=True,
                                    order_by="ReportComment.comment_id")
 
     assigned_users = sa.orm.relationship('User',
                                          backref=sa.orm.backref(
                                              'assigned_reports_relation',
                                              lazy='dynamic',
                                              order_by=sa.desc(
                                                  "reports_groups.id")
                                          ),
                                          passive_deletes=True,
                                          passive_updates=True,
                                          secondary='reports_assignments',
                                          order_by="User.user_name")
 
     stats = sa.orm.relationship('ReportStat',
                                 lazy='dynamic',
                                 backref='report',
                                 passive_deletes=True,
                                 passive_updates=True, )
 
     last_report_ref = sa.orm.relationship('Report',
                                           uselist=False,
                                           primaryjoin="ReportGroup.last_report "
                                                       "== Report.id",
                                           foreign_keys="Report.id",
                                           cascade="all, delete-orphan",
                                           passive_deletes=True,
                                           passive_updates=True, )
 
     def __repr__(self):
         return '<ReportGroup id:{}>'.format(self.id)
 
     def get_report(self, report_id=None, public=False):
         """
         Gets report with specific id or latest report if id was not specified
         """
         from .report import Report
 
         if not report_id:
             return self.last_report_ref
         else:
             return self.reports.filter(Report.id == report_id).first()
 
     def get_public_url(self, request, _app_url=None):
         url = request.route_url('/', _app_url=_app_url)
         return (url + 'ui/report/%s') % self.id
 
     def run_postprocessing(self, report):
         """
         Alters report group priority based on postprocessing configuration
         """
         request = get_current_request()
         get_db_session(None, self).flush()
         for action in self.application.postprocess_conf:
             get_db_session(None, self).flush()
             rule_obj = Rule(action.rule, REPORT_TYPE_MATRIX)
             report_dict = report.get_dict(request)
             # if was not processed yet
             if (rule_obj.match(report_dict) and
                         action.pkey not in self.triggered_postprocesses_ids):
                 action.postprocess(self)
                 # this way sqla can track mutation of list
                 self.triggered_postprocesses_ids = \
                     self.triggered_postprocesses_ids + [action.pkey]
 
         get_db_session(None, self).flush()
         # do not go out of bounds
         if self.priority < 1:
             self.priority = 1
         if self.priority > 10:
             self.priority = 10
 
     def get_dict(self, request):
         instance_dict = super(ReportGroup, self).get_dict()
         instance_dict['server_name'] = self.get_report().tags.get(
             'server_name')
         instance_dict['view_name'] = self.get_report().tags.get('view_name')
         instance_dict['resource_name'] = self.application.resource_name
         instance_dict['report_type'] = self.get_report().report_type
         instance_dict['url_path'] = self.get_report().url_path
         instance_dict['front_url'] = self.get_report().get_public_url(request)
         del instance_dict['triggered_postprocesses_ids']
         return instance_dict
 
     def es_doc(self):
         return {
             '_id': str(self.id),
             'pg_id': str(self.id),
             'resource_id': self.resource_id,
             'error': self.error,
             'fixed': self.fixed,
             'public': self.public,
             'read': self.read,
             'priority': self.priority,
             'occurences': self.occurences,
             'average_duration': self.average_duration,
             'summed_duration': self.summed_duration,
             'first_timestamp': self.first_timestamp,
             'last_timestamp': self.last_timestamp
         }
 
     def set_notification_info(self, notify_10=False, notify_100=False):
         """
         Update redis notification maps for notification job
         """
         current_time = datetime.utcnow().replace(second=0, microsecond=0)
         # global app counter
         key = REDIS_KEYS['counters']['reports_per_type'].format(
             self.report_type, current_time)
-        Datastores.redis.incr(key)
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline = Datastores.redis.pipeline()
+        redis_pipeline.incr(key)
+        redis_pipeline.expire(key, 3600 * 24)
         # detailed app notification for alerts and notifications
-        Datastores.redis.sadd(REDIS_KEYS['apps_that_had_reports'],
-                              self.resource_id)
-        Datastores.redis.sadd(REDIS_KEYS['apps_that_had_reports_alerting'],
-                              self.resource_id)
+        redis_pipeline.sadd(
+            REDIS_KEYS['apps_that_had_reports'], self.resource_id)
+        redis_pipeline.sadd(
+            REDIS_KEYS['apps_that_had_reports_alerting'], self.resource_id)
         # only notify for exceptions here
         if self.report_type == ReportType.error:
-            Datastores.redis.sadd(
-                REDIS_KEYS['apps_that_had_reports'],
-                self.resource_id)
-            Datastores.redis.sadd(
+            redis_pipeline.sadd(
+                REDIS_KEYS['apps_that_had_reports'], self.resource_id)
+            redis_pipeline.sadd(
                 REDIS_KEYS['apps_that_had_error_reports_alerting'],
                 self.resource_id)
         key = REDIS_KEYS['counters']['report_group_occurences'].format(self.id)
-        Datastores.redis.incr(key)
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline.incr(key)
+        redis_pipeline.expire(key, 3600 * 24)
         key = REDIS_KEYS['counters']['report_group_occurences_alerting'].format(self.id)
-        Datastores.redis.incr(key)
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline.incr(key)
+        redis_pipeline.expire(key, 3600 * 24)
 
         if notify_10:
             key = REDIS_KEYS['counters'][
                 'report_group_occurences_10th'].format(self.id)
-            Datastores.redis.setex(key, 3600 * 24, 1)
+            redis_pipeline.setex(key, 3600 * 24, 1)
         if notify_100:
             key = REDIS_KEYS['counters'][
                 'report_group_occurences_100th'].format(self.id)
-            Datastores.redis.setex(key, 3600 * 24, 1)
+            redis_pipeline.setex(key, 3600 * 24, 1)
 
         key = REDIS_KEYS['reports_to_notify_per_type_per_app'].format(
             self.report_type, self.resource_id)
-        Datastores.redis.sadd(key, self.id)
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline.sadd(key, self.id)
+        redis_pipeline.expire(key, 3600 * 24)
         key = REDIS_KEYS['reports_to_notify_per_type_per_app_alerting'].format(
             self.report_type, self.resource_id)
-        Datastores.redis.sadd(key, self.id)
-        Datastores.redis.expire(key, 3600 * 24)
+        redis_pipeline.sadd(key, self.id)
+        redis_pipeline.expire(key, 3600 * 24)
+        redis_pipeline.execute()
 
     @property
     def partition_id(self):
         return 'rcae_r_%s' % self.first_timestamp.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_group',
                             data, 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_group',
                             data, id=target.id)
 
 
 def after_delete(mapper, connection, target):
     query = {'term': {'group_id': target.id}}
     # TODO: routing seems unnecessary, need to test a bit more
     #Datastores.es.delete_by_query(target.partition_id, 'report', query,
     #                              query_params={'routing':str(target.id)})
     Datastores.es.delete_by_query(target.partition_id, 'report', query)
     query = {'term': {'pg_id': target.id}}
     Datastores.es.delete_by_query(target.partition_id, 'report_group', query)
 
 
 sa.event.listen(ReportGroup, 'after_insert', after_insert)
 sa.event.listen(ReportGroup, 'after_update', after_update)
 sa.event.listen(ReportGroup, 'after_delete', after_delete)

backend/src/appenlight/views/admin/admin.py

@@ -1,200 +1,199 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/
 
 import logging
 import os
 import pkg_resources
 
 from datetime import datetime, timedelta
 
 import psutil
 import redis
 
 from pyramid.view import view_config
 from appenlight.models import DBSession
 from appenlight.models import Datastores
 from appenlight.lib.redis_keys import REDIS_KEYS
 
 
 def bytes2human(total):
     giga = 1024.0 ** 3
     mega = 1024.0 ** 2
     kilo = 1024.0
     if giga <= total:
         return '{:0.1f}G'.format(total / giga)
     elif mega <= total:
         return '{:0.1f}M'.format(total / mega)
     else:
         return '{:0.1f}K'.format(total / kilo)
 
 
 log = logging.getLogger(__name__)
 
 
 @view_config(route_name='section_view',
              match_param=['section=admin_section', 'view=system'],
              renderer='json', permission='root_administration')
 def system(request):
     current_time = datetime.utcnow(). \
                        replace(second=0, microsecond=0) - timedelta(minutes=1)
     # global app counter
-
-    processed_reports = Datastores.redis.get(
+    processed_reports = request.registry.redis_conn.get(
         REDIS_KEYS['counters']['reports_per_minute'].format(current_time))
     processed_reports = int(processed_reports) if processed_reports else 0
-    processed_logs = Datastores.redis.get(
+    processed_logs = request.registry.redis_conn.get(
         REDIS_KEYS['counters']['logs_per_minute'].format(current_time))
     processed_logs = int(processed_logs) if processed_logs else 0
-    processed_metrics = Datastores.redis.get(
+    processed_metrics = request.registry.redis_conn.get(
         REDIS_KEYS['counters']['metrics_per_minute'].format(current_time))
     processed_metrics = int(processed_metrics) if processed_metrics else 0
 
     waiting_reports = 0
     waiting_logs = 0
     waiting_metrics = 0
     waiting_other = 0
 
     if 'redis' in request.registry.settings['celery.broker_type']:
         redis_client = redis.StrictRedis.from_url(
             request.registry.settings['celery.broker_url'])
         waiting_reports = redis_client.llen('reports')
         waiting_logs = redis_client.llen('logs')
         waiting_metrics = redis_client.llen('metrics')
         waiting_other = redis_client.llen('default')
 
     # process
     def replace_inf(val):
         return val if val != psutil.RLIM_INFINITY else 'unlimited'
 
     p = psutil.Process()
     fd = p.rlimit(psutil.RLIMIT_NOFILE)
     memlock = p.rlimit(psutil.RLIMIT_MEMLOCK)
     self_info = {
         'fds': {'soft': replace_inf(fd[0]),
                 'hard': replace_inf(fd[1])},
         'memlock': {'soft': replace_inf(memlock[0]),
                     'hard': replace_inf(memlock[1])},
     }
 
     # disks
     disks = []
     for part in psutil.disk_partitions(all=False):
         if os.name == 'nt':
             if 'cdrom' in part.opts or part.fstype == '':
                 continue
         usage = psutil.disk_usage(part.mountpoint)
         disks.append({
             'device': part.device,
             'total': bytes2human(usage.total),
             'used': bytes2human(usage.used),
             'free': bytes2human(usage.free),
             'percentage': int(usage.percent),
             'mountpoint': part.mountpoint,
             'fstype': part.fstype
         })
 
     # memory
     memory_v = psutil.virtual_memory()
     memory_s = psutil.swap_memory()
 
     memory = {
         'total': bytes2human(memory_v.total),
         'available': bytes2human(memory_v.available),
         'percentage': memory_v.percent,
         'used': bytes2human(memory_v.used),
         'free': bytes2human(memory_v.free),
         'active': bytes2human(memory_v.active),
         'inactive': bytes2human(memory_v.inactive),
         'buffers': bytes2human(memory_v.buffers),
         'cached': bytes2human(memory_v.cached),
         'swap_total': bytes2human(memory_s.total),
         'swap_used': bytes2human(memory_s.used)
     }
 
     # load
     system_load = os.getloadavg()
 
     # processes
     min_mem = 1024 * 1024 * 40  # 40MB
     process_info = []
     for p in psutil.process_iter():
         mem_used = p.get_memory_info().rss
         if mem_used < min_mem:
             continue
         process_info.append({'owner': p.username(),
                              'pid': p.pid,
                              'cpu': round(p.get_cpu_percent(interval=0), 1),
                              'mem_percentage': round(p.get_memory_percent(),
                                                      1),
                              'mem_usage': bytes2human(mem_used),
                              'name': p.name(),
                              'command': ' '.join(p.cmdline())
                              })
     process_info = sorted(process_info, key=lambda x: x['mem_percentage'],
                           reverse=True)
 
     # pg tables
 
     db_size_query = '''
     SELECT tablename, pg_total_relation_size(tablename::text) size
     FROM pg_tables WHERE tablename NOT LIKE 'pg_%' AND 
     tablename NOT LIKE 'sql_%' ORDER BY size DESC;'''
 
     db_tables = []
     for row in DBSession.execute(db_size_query):
         db_tables.append({"size_human": bytes2human(row.size),
                           "table_name": row.tablename})
 
     # es indices
     es_indices = []
     result = Datastores.es.send_request(
         'GET', ['_stats', 'store, docs'], query_params={})
     for ix, stats in result['indices'].items():
         size = stats['primaries']['store']['size_in_bytes']
         es_indices.append({'name': ix,
                            'size': size,
                            'size_human': bytes2human(size)})
 
     # packages
 
     packages = ({'name': p.project_name, 'version': p.version}
                 for p in pkg_resources.working_set)
 
     return {'db_tables': db_tables,
             'es_indices': sorted(es_indices,
                                  key=lambda x: x['size'], reverse=True),
             'process_info': process_info,
             'system_load': system_load,
             'disks': disks,
             'memory': memory,
             'packages': sorted(packages, key=lambda x: x['name'].lower()),
             'current_time': current_time,
             'queue_stats': {
                 'processed_reports': processed_reports,
                 'processed_logs': processed_logs,
                 'processed_metrics': processed_metrics,
                 'waiting_reports': waiting_reports,
                 'waiting_logs': waiting_logs,
                 'waiting_metrics': waiting_metrics,
                 'waiting_other': waiting_other
             },
             'self_info': self_info
             }