celery: decouple report notifications from alerts

backend/src/appenlight/celery/__init__.py

@@ -1,161 +1,176 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
 # App Enlight Enterprise Edition, including its added features, Support
 # services, and proprietary license terms, please see
 # https://rhodecode.com/licenses/
 
 import logging
 
 from datetime import timedelta
 from celery import Celery
 from celery.bin import Option
 from celery.schedules import crontab
 from celery.signals import worker_init, task_revoked, user_preload_options
 from celery.signals import task_prerun, task_retry, task_failure, task_success
 from kombu.serialization import register
 from pyramid.paster import bootstrap
 from pyramid.request import Request
 from pyramid.scripting import prepare
 from pyramid.settings import asbool
 from pyramid.threadlocal import get_current_request
 
 from appenlight.celery.encoders import json_dumps, json_loads
 from appenlight_client.ext.celery import register_signals
 
 log = logging.getLogger(__name__)
 
 register('date_json', json_dumps, json_loads,
          content_type='application/x-date_json',
          content_encoding='utf-8')
 
 celery = Celery()
 
 celery.user_options['preload'].add(
     Option('--ini', dest='ini', default=None,
            help='Specifies pyramid configuration file location.')
 )
 
+
 @user_preload_options.connect
 def on_preload_parsed(options, **kwargs):
     """
     This actually configures celery from pyramid config file
     """
     celery.conf['INI_PYRAMID'] = options['ini']
     import appenlight_client.client as e_client
     ini_location = options['ini']
     if not ini_location:
         raise Exception('You need to pass pyramid ini location using '
                         '--ini=filename.ini argument to the worker')
     env = bootstrap(ini_location)
     api_key = env['request'].registry.settings['appenlight.api_key']
     tr_config = env['request'].registry.settings.get(
         'appenlight.transport_config')
     CONFIG = e_client.get_config({'appenlight.api_key': api_key})
     if tr_config:
         CONFIG['appenlight.transport_config'] = tr_config
     APPENLIGHT_CLIENT = e_client.Client(CONFIG)
     # log.addHandler(APPENLIGHT_CLIENT.log_handler)
     register_signals(APPENLIGHT_CLIENT)
     celery.pyramid = env
 
 
 celery_config = {
     'CELERY_IMPORTS': ["appenlight.celery.tasks",],
     'CELERYD_TASK_TIME_LIMIT': 60,
     'CELERYD_MAX_TASKS_PER_CHILD': 1000,
     'CELERY_IGNORE_RESULT': True,
     'CELERY_ACCEPT_CONTENT': ['date_json'],
     'CELERY_TASK_SERIALIZER': 'date_json',
     'CELERY_RESULT_SERIALIZER': 'date_json',
     'BROKER_URL': None,
     'CELERYD_CONCURRENCY': None,
     'CELERY_TIMEZONE': None,
     'CELERYBEAT_SCHEDULE': {
-        'alerting': {
-            'task': 'appenlight.celery.tasks.alerting',
+        'alerting_reports': {
+            'task': 'appenlight.celery.tasks.alerting_reports',
             'schedule': timedelta(seconds=60)
         },
-        'daily_digest': {
-            'task': 'appenlight.celery.tasks.daily_digest',
-            'schedule': crontab(minute=1, hour='4,12,20')
-        },
+        'close_alerts': {
+            'task': 'appenlight.celery.tasks.close_alerts',
+            'schedule': timedelta(seconds=60)
+        }
     }
 }
 celery.config_from_object(celery_config)
 
+
 def configure_celery(pyramid_registry):
     settings = pyramid_registry.settings
     celery_config['BROKER_URL'] = settings['celery.broker_url']
     celery_config['CELERYD_CONCURRENCY'] = settings['celery.concurrency']
     celery_config['CELERY_TIMEZONE'] = settings['celery.timezone']
+
+    notifications_seconds = int(settings.get('tasks.notifications_reports.interval', 60))
+
+    celery_config['CELERYBEAT_SCHEDULE']['notifications'] = {
+        'task': 'appenlight.celery.tasks.notifications_reports',
+        'schedule': timedelta(seconds=notifications_seconds)
+    }
+
+    celery_config['CELERYBEAT_SCHEDULE']['daily_digest'] = {
+        'task': 'appenlight.celery.tasks.daily_digest',
+        'schedule': crontab(minute=1, hour='4,12,20')
+    }
+
     if asbool(settings.get('celery.always_eager')):
         celery_config['CELERY_ALWAYS_EAGER'] = True
         celery_config['CELERY_EAGER_PROPAGATES_EXCEPTIONS'] = True
 
     for plugin in pyramid_registry.appenlight_plugins.values():
         if plugin.get('celery_tasks'):
             celery_config['CELERY_IMPORTS'].extend(plugin['celery_tasks'])
         if plugin.get('celery_beats'):
             for name, config in plugin['celery_beats']:
                 celery_config['CELERYBEAT_SCHEDULE'][name] = config
     celery.config_from_object(celery_config)
 
 
 @task_prerun.connect
 def task_prerun_signal(task_id, task, args, kwargs, **kwaargs):
     if hasattr(celery, 'pyramid'):
         env = celery.pyramid
         env = prepare(registry=env['request'].registry)
         proper_base_url = env['request'].registry.settings['mailing.app_url']
-        tmp_request = Request.blank('/', base_url=proper_base_url)
+        tmp_req = Request.blank('/', base_url=proper_base_url)
         # ensure tasks generate url for right domain from config
-        env['request'].environ['HTTP_HOST'] = tmp_request.environ['HTTP_HOST']
-        env['request'].environ['SERVER_PORT'] = tmp_request.environ['SERVER_PORT']
-        env['request'].environ['SERVER_NAME'] = tmp_request.environ['SERVER_NAME']
-        env['request'].environ['wsgi.url_scheme'] = tmp_request.environ[
-            'wsgi.url_scheme']
+        env['request'].environ['HTTP_HOST'] = tmp_req.environ['HTTP_HOST']
+        env['request'].environ['SERVER_PORT'] = tmp_req.environ['SERVER_PORT']
+        env['request'].environ['SERVER_NAME'] = tmp_req.environ['SERVER_NAME']
+        env['request'].environ['wsgi.url_scheme'] = \
+            tmp_req.environ['wsgi.url_scheme']
     get_current_request().tm.begin()
 
 
 @task_success.connect
 def task_success_signal(result, **kwargs):
     get_current_request().tm.commit()
     if hasattr(celery, 'pyramid'):
         celery.pyramid["closer"]()
 
 
 @task_retry.connect
 def task_retry_signal(request, reason, einfo, **kwargs):
     get_current_request().tm.abort()
     if hasattr(celery, 'pyramid'):
         celery.pyramid["closer"]()
 
 
 @task_failure.connect
 def task_failure_signal(task_id, exception, args, kwargs, traceback, einfo,
                         **kwaargs):
     get_current_request().tm.abort()
     if hasattr(celery, 'pyramid'):
         celery.pyramid["closer"]()
 
 
 @task_revoked.connect
 def task_revoked_signal(request, terminated, signum, expired, **kwaargs):
     get_current_request().tm.abort()
     if hasattr(celery, 'pyramid'):
         celery.pyramid["closer"]()

backend/src/appenlight/celery/tasks.py

@@ -1,610 +1,662 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
 # App Enlight 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.request_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, 10000, 100))
 
 
 def pick_sample(total_occurences, report_type=1):
     every = 1.0
     position = bisect.bisect_left(sample_boundries, total_occurences)
     if position > 0:
         # 404
         if report_type == 2:
             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=999)
 def add_reports(resource_id, params, dataset, environ=None, **kwargs):
     proto_version = parse_proto(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)
         reports = []
 
         if proto_version.major < 1 and proto_version.minor < 5:
             for report_data in dataset:
                 report_details = report_data.get('report_details', [])
                 for i, detail_data in enumerate(report_details):
                     report_data.update(detail_data)
                     report_data.pop('report_details')
                     traceback = report_data.get('traceback')
                     if traceback is None:
                         report_data['traceback'] = report_data.get('frameinfo')
                     # for 0.3 api
                     error = report_data.pop('error_type', '')
                     if error:
                         report_data['error'] = error
                     if proto_version.minor < 4:
                         # convert "Unknown" slow reports to
                         # '' (from older clients)
                         if (report_data['error'] and
                                     report_data['http_status'] < 500):
                             report_data['error'] = ''
                     message = report_data.get('message')
                     if 'extra' not in report_data:
                         report_data['extra'] = []
                     if message:
                         report_data['extra'] = [('message', message), ]
                     reports.append(report_data)
         else:
             reports = dataset
 
         tags = []
         es_slow_calls_docs = {}
         es_reports_stats_rows = {}
         for report_data in reports:
             # 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)
         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)
 
         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=999)
 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=999)
 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=999)
 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=999)
 def add_logs(resource_id, request, dataset, environ=None, **kwargs):
     proto_version = request.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:
                             log.error(exc)
 
         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)
         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)
         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=999)
 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=999)
 def add_metrics(resource_id, request, 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)
         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)
         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=999)
 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=None):
+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, since_when=since_when)
+            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,
-                                      since_when=since_when)
+                                      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
 
 
[e-mail](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(since_when=None):
+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=999)
 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 alerting():
+def notifications_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
-    - which currently opened alerts should be closed
+    issue notifications
     """
-    start_time = datetime.utcnow()
-    # transactions are needed for mailer
     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'))
-    # clear app ids from set
-    close_alerts.delay(since_when=start_time)
+
[e-mail](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'))
 
 
[e-mail](queue="default", soft_time_limit=3600 * 4, hard_time_limit=3600 * 4,
-      max_retries=999)
[e-mail](queue="default", soft_time_limit=3600 * 4,
+             hard_time_limit=3600 * 4, max_retries=999)
 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/__init__.py

@@ -1,53 +1,55 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
 # App Enlight Enterprise Edition, including its added features, Support
 # services, and proprietary license terms, please see
 # https://rhodecode.com/licenses/
 
 """Miscellaneous support packages for {{project}}.
 """
 import random
 import string
 import importlib
 
 from appenlight_client.exceptions import get_current_traceback
 
 
 def generate_random_string(chars=10):
     return ''.join(random.sample(string.ascii_letters * 2 + string.digits,
                                   chars))
 
 
 def to_integer_safe(input):
     try:
         return int(input)
     except (TypeError, ValueError,):
         return None
 
+
 def print_traceback(log):
     traceback = get_current_traceback(skip=1, show_hidden_frames=True,
                                       ignore_system_exceptions=True)
     exception_text = traceback.exception
     log.error(exception_text)
     log.error(traceback.plaintext)
     del traceback
 
+
 def get_callable(import_string):
     import_module, indexer_callable = import_string.split(':')
     return getattr(importlib.import_module(import_module),
                    indexer_callable)

backend/src/appenlight/lib/redis_keys.py

@@ -1,59 +1,67 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
 # App Enlight 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_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:{}:{}'),
         'metrics_per_minute': BASE.format('metrics_per_minute:{}'),
         'metrics_per_minute_per_app': BASE.format(
             'metrics_per_minute_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_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,251 +1,264 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
 # App Enlight 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)
-        # detailed app notification
+        # 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)
         # only notify for exceptions here
         if self.report_type == ReportType.error:
-            Datastores.redis.sadd(REDIS_KEYS['apps_that_had_reports'],
+            Datastores.redis.sadd(
+                REDIS_KEYS['apps_that_had_reports'],
+                self.resource_id)
+            Datastores.redis.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)
+        key = REDIS_KEYS['counters']['report_group_occurences_alerting'].format(self.id)
+        Datastores.redis.incr(key)
+        Datastores.redis.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)
         if notify_100:
             key = REDIS_KEYS['counters'][
                 'report_group_occurences_100th'].format(self.id)
             Datastores.redis.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)
+        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)
 
     @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}}
     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/models/services/user.py

@@ -1,158 +1,156 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
 # App Enlight Enterprise Edition, including its added features, Support
 # services, and proprietary license terms, please see
 # https://rhodecode.com/licenses/
 
 import logging
 import pyramid_mailer
 import pyramid.renderers
 import sqlalchemy as sa
 
 from collections import namedtuple
 from datetime import datetime
 
 from appenlight.lib.rule import Rule
 from appenlight.models import get_db_session
 from appenlight.models.integrations import IntegrationException
 from appenlight.models.report import REPORT_TYPE_MATRIX
 from appenlight.models.user import User
 from appenlight.models.services.base import BaseService
 from paginate_sqlalchemy import SqlalchemyOrmPage
 from pyramid.threadlocal import get_current_registry
 
 log = logging.getLogger(__name__)
 
 GroupOccurence = namedtuple('GroupOccurence', ['occurences', 'group'])
 
 
 class UserService(BaseService):
     @classmethod
     def all(cls, db_session=None):
         return get_db_session(db_session).query(User).order_by(User.user_name)
 
     @classmethod
     def send_email(cls, request, recipients, variables, template,
                    immediately=False, silent=False):
         html = pyramid.renderers.render(template, variables, request)
         title = variables.get('email_title',
                               variables.get('title', "No Title"))
         title = title.replace('\r', '').replace('\n', '')
         sender = "{} <{}>".format(
             request.registry.settings['mailing.from_name'],
             request.registry.settings['mailing.from_email'])
         message = pyramid_mailer.message.Message(
             subject=title, sender=sender, recipients=recipients, html=html)
         if immediately:
             try:
                 request.registry.mailer.send_immediately(message)
             except Exception as e:
                 log.warning('Exception %s' % e)
                 if not silent:
                     raise
         else:
             request.registry.mailer.send(message)
 
     @classmethod
     def get_paginator(cls, page=1, item_count=None, items_per_page=50,
                       order_by=None, filter_settings=None,
                       exclude_columns=None, db_session=None):
         registry = get_current_registry()
         if not exclude_columns:
             exclude_columns = []
         if not filter_settings:
             filter_settings = {}
         db_session = get_db_session(db_session)
         q = db_session.query(User)
         if filter_settings.get('order_col'):
             order_col = filter_settings.get('order_col')
             if filter_settings.get('order_dir') == 'dsc':
                 sort_on = 'desc'
             else:
                 sort_on = 'asc'
             q = q.order_by(getattr(sa, sort_on)(getattr(User, order_col)))
         else:
             q = q.order_by(sa.desc(User.registered_date))
             # remove urlgen or it never caches count
         cache_params = dict(filter_settings)
         cache_params.pop('url', None)
         cache_params.pop('url_maker', None)
 
         @registry.cache_regions.redis_min_5.cache_on_arguments()
         def estimate_users(cache_key):
             o_q = q.order_by(False)
             return o_q.count()
 
         item_count = estimate_users(cache_params)
         # if the number of pages is low we may want to invalidate the count to
         # provide 'real time' update - use case -
         # errors just started to flow in
         if item_count < 1000:
             item_count = estimate_users.refresh(cache_params)
         paginator = SqlalchemyOrmPage(q, page=page,
                                       item_count=item_count,
                                       items_per_page=items_per_page,
                                       **filter_settings)
         return paginator
 
     @classmethod
     def get_valid_channels(cls, user):
         return [channel for channel in user.alert_channels
                 if channel.channel_validated]
 
     @classmethod
     def report_notify(cls, user, request, application, report_groups,
-                      occurence_dict, since_when=None, db_session=None):
+                      occurence_dict, db_session=None):
         db_session = get_db_session(db_session)
         if not report_groups:
             return True
-
-        if not since_when:
             since_when = datetime.utcnow()
         for channel in cls.get_valid_channels(user):
             confirmed_groups = []
 
             for group in report_groups:
                 occurences = occurence_dict.get(group.id, 1)
                 for action in channel.channel_actions:
                     not_matched = (
                         action.resource_id and action.resource_id !=
                         application.resource_id)
                     if action.type != 'report' or not_matched:
                         continue
                     should_notify = (action.action == 'always' or
                                      not group.notified)
                     rule_obj = Rule(action.rule, REPORT_TYPE_MATRIX)
                     report_dict = group.get_report().get_dict(request)
                     if rule_obj.match(report_dict) and should_notify:
                         item = GroupOccurence(occurences, group)
                         if item not in confirmed_groups:
                             confirmed_groups.append(item)
 
             # send individual reports
             total_confirmed = len(confirmed_groups)
             if not total_confirmed:
                 continue
             try:
                 channel.notify_reports(resource=application,
                                        user=user,
                                        request=request,
                                        since_when=since_when,
                                        reports=confirmed_groups)
             except IntegrationException as e:
                 log.warning('%s' % e)

backend/src/appenlight/templates/ini/production.ini.jinja2

@@ -1,176 +1,184 b''
 [app:appenlight]
 use = egg:appenlight
 reload_templates = false
 debug_authorization = false
 debug_notfound = false
 debug_routematch = false
 debug_templates = false
 default_locale_name = en
 sqlalchemy.url = {{ appenlight_dbstring }}
 sqlalchemy.pool_size = 10
 sqlalchemy.max_overflow = 50
 sqlalchemy.echo = false
 jinja2.directories = appenlight:templates
 jinja2.filters = nl2br = appenlight.lib.jinja2_filters.nl2br
 
 #includes
 appenlight.includes =
 
 #redis
 redis.url = redis://localhost:6379/0
 redis.redlock.url = redis://localhost:6379/3
 
 #elasticsearch
 elasticsearch.nodes = http://127.0.0.1:9200
 
 #dirs
 webassets.dir = %(here)s/webassets/
 
 # encryption
 encryption_secret = {{appenlight_encryption_secret}}
 
 #authtkt
 # uncomment if you use SSL
 # authtkt.secure = true
 authtkt.secret = {{appenlight_authtkt_secret}}
 # session settings
 redis.sessions.secret = {{appenlight_redis_session_secret}}
 redis.sessions.timeout = 86400
 
 # session cookie settings
 redis.sessions.cookie_name = appenlight
 redis.sessions.cookie_max_age = 2592000
 redis.sessions.cookie_path = /
 redis.sessions.cookie_domain =
 # uncomment if you use SSL
 redis.sessions.cookie_secure = True
 redis.sessions.cookie_httponly = True
 redis.sessions.cookie_on_exception = True
 redis.sessions.prefix = appenlight:session:
 
 #cache
 cache.regions = default_term, second, short_term, long_term
 cache.type = ext:memcached
 cache.url = 127.0.0.1:11211
 cache.lock_dir = %(here)s/data/cache/lock
 cache.second.expire = 1
 cache.short_term.expire = 60
 cache.default_term.expire = 300
 
 #mailing
 mailing.app_url = https://{{appenlight_domain}}
 mailing.from_name = App Enlight
 mailing.from_email = no-reply@{{appenlight_domain}}
 
 ###
 # Authomatic configuration
 ###
 
 authomatic.secret =
 authomatic.pr.facebook.app_id =
 authomatic.pr.facebook.secret =
 authomatic.pr.twitter.key =
 authomatic.pr.twitter.secret =
 authomatic.pr.google.key =
 authomatic.pr.google.secret =
 authomatic.pr.github.key =
 authomatic.pr.github.secret =
 authomatic.pr.github.scope  =
 authomatic.pr.bitbucket.key =
 authomatic.pr.bitbucket.secret =
 
 #ziggurat
 ziggurat_foundations.model_locations.User = appenlight.models.user:User
 ziggurat_foundations.sign_in.username_key = sign_in_user_name
 ziggurat_foundations.sign_in.password_key = sign_in_user_password
 ziggurat_foundations.sign_in.came_from_key = came_from
 
 #cometd
 cometd.server = http://127.0.0.1:8088
 cometd.secret = secret
 cometd.ws_url = wss://{{appenlight_domain}}/channelstream
 
 # for celery
 appenlight.api_key =
 appenlight.transport_config =
 appenlight.public_api_key =
 
 # celery
 celery.broker_type = redis
 celery.broker_url = redis://localhost:6379/3
 celery.concurrency = 2
 celery.timezone = UTC
 
+# tasks
+
+# how often run alerting tasks (60s default)
+tasks.notifications_reports.interval = 60
+
 [filter:paste_prefix]
 use = egg:PasteDeploy#prefix
 
 
 [filter:appenlight_client]
 use = egg:appenlight_client
 appenlight.api_key =
+
+# appenlight.transport_config = http://127.0.0.1:6543?threaded=1&timeout=5&verify=0
+# by default uses api.appenlight.com server
 appenlight.transport_config =
 appenlight.report_local_vars = true
 appenlight.report_404 = true
 appenlight.timing.dbapi2_psycopg2 = 0.3
 
 
 
 [pipeline:main]
 pipeline = paste_prefix
            appenlight_client
            appenlight
 
 
 
 [server:main]
 use = egg:gunicorn#main
 host = 0.0.0.0:6543, unix:/tmp/appenlight.sock
 workers = 2
 timeout = 90
 max_requests = 10000
 
 [server:api]
 use = egg:gunicorn#main
 host = 0.0.0.0:6553, unix:/tmp/api.appenlight.sock
 workers = 2
 max_requests = 10000
 
 
 # Begin logging configuration
 
 [loggers]
 keys = root, appenlight, sqlalchemy
 
 [handlers]
 keys = console
 
 [formatters]
 keys = generic
 
 [logger_root]
 level = WARN
 handlers = console
 
 [logger_appenlight]
 level = WARN
 handlers =
 qualname = appenlight
 
 [logger_sqlalchemy]
 level = WARN
 handlers =
 qualname = sqlalchemy.engine
 # "level = INFO" logs SQL queries.
 # "level = DEBUG" logs SQL queries and results.
 # "level = WARN" logs neither.  (Recommended for production systems.)
 
 [handler_console]
 class = StreamHandler
 args = (sys.stderr,)
 level = NOTSET
 formatter = generic
 
 [formatter_generic]
 format = %(asctime)s %(levelname)-5.5s [%(name)s][%(threadName)s] %(message)s
 
 # End logging configuration