def send_async(self, to=None, cc=None, bcc=None):
        from sentry.tasks.email import send_email
        fmt = options.get('system.logging-format')
        messages = self.get_built_messages(to, cc=cc, bcc=bcc)
        extra = {'message_type': self.type}
        loggable = [v for k, v in six.iteritems(self.context) if hasattr(v, 'id')]
        for context in loggable:
            extra['%s_id' % type(context).__name__.lower()] = context.id

        log_mail_queued = partial(logger.info, 'mail.queued', extra=extra)
        for message in messages:
            safe_execute(
                send_email.delay,
                message=message,
                _with_transaction=False,
            )
            extra['message_id'] = message.extra_headers['Message-Id']
            metrics.incr('email.queued', instance=self.type)
            if fmt == LoggingFormat.HUMAN:
                extra['message_to'] = self.format_to(message.to),
                log_mail_queued()
            elif fmt == LoggingFormat.MACHINE:
                for recipient in message.to:
                    extra['message_to'] = recipient
                    log_mail_queued()

def create_failed_event(cache_key, project, issues, event_id):
    """If processing failed we put the original data from the cache into a
    raw event.
    """
    # We need to get the original data here instead of passing the data in
    # from the last processing step because we do not want any
    # modifications to take place.
    delete_raw_event(project, event_id)
    data = default_cache.get(cache_key)
    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'raw'})
        error_logger.error('process.failed_raw.empty', extra={'cache_key': cache_key})
        return

    from sentry.models import RawEvent, ProcessingIssue
    raw_event = RawEvent.objects.create(
        project_id=project,
        event_id=event_id,
        datetime=datetime.utcfromtimestamp(
            data['timestamp']).replace(tzinfo=timezone.utc),
        data=data
    )

    for issue in issues:
        ProcessingIssue.objects.record_processing_issue(
            raw_event=raw_event,
            scope=issue['scope'],
            object=issue['object'],
            type=issue['type'],
            data=issue['data'],
        )

    default_cache.delete(cache_key)

def process_event(cache_key, start_time=None, **kwargs):
    from sentry.plugins import plugins

    data = default_cache.get(cache_key)

    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'process'})
        error_logger.error('process.failed.empty', extra={'cache_key': cache_key})
        return

    project = data['project']
    Raven.tags_context({
        'project': project,
    })

    # TODO(dcramer): ideally we would know if data changed by default
    has_changed = False
    for plugin in plugins.all(version=2):
        processors = safe_execute(plugin.get_event_preprocessors, data=data, _with_transaction=False)
        for processor in (processors or ()):
            result = safe_execute(processor, data)
            if result:
                data = result
                has_changed = True

    assert data['project'] == project, 'Project cannot be mutated by preprocessor'

    if has_changed:
        default_cache.set(cache_key, data, 3600)

    save_event.delay(cache_key=cache_key, data=None, start_time=start_time)

def _do_preprocess_event(cache_key, data, start_time, event_id, process_event):
    if cache_key:
        data = default_cache.get(cache_key)

    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'pre'}, skip_internal=False)
        error_logger.error('preprocess.failed.empty', extra={'cache_key': cache_key})
        return

    data = CanonicalKeyDict(data)
    project = data['project']

    with configure_scope() as scope:
        scope.set_tag("project", project)

    if should_process(data):
        process_event.delay(cache_key=cache_key, start_time=start_time, event_id=event_id)
        return

    # If we get here, that means the event had no preprocessing needed to be done
    # so we can jump directly to save_event
    if cache_key:
        data = None
    save_event.delay(
        cache_key=cache_key, data=data, start_time=start_time, event_id=event_id,
        project_id=project
    )

    def after(self, event, state):
        service = self.get_option('service')

        extra = {'event_id': event.id}
        if not service:
            self.logger.info('rules.fail.is_configured', extra=extra)
            return

        app = None
        try:
            app = SentryApp.objects.get(slug=service)
        except SentryApp.DoesNotExist:
            pass

        if app:
            kwargs = {'sentry_app': app}
            metrics.incr('notifications.sent', instance=app.slug, skip_internal=False)
            yield self.future(notify_sentry_app, **kwargs)
        else:
            plugin = plugins.get(service)
            if not plugin.is_enabled(self.project):
                extra['project_id'] = self.project.id
                self.logger.info('rules.fail.is_enabled', extra=extra)
                return

            group = event.group

            if not plugin.should_notify(group=group, event=event):
                extra['group_id'] = group.id
                self.logger.info('rule.fail.should_notify', extra=extra)
                return

            metrics.incr('notifications.sent', instance=plugin.slug, skip_internal=False)
            yield self.future(plugin.rule_notify)

    def _request(self, method, path, headers=None, data=None, params=None,
                 auth=None, json=True, allow_text=None, allow_redirects=None,
                 timeout=None):

        if allow_text is None:
            allow_text = self.allow_text

        if allow_redirects is None:
            allow_redirects = self.allow_redirects

        if allow_redirects is None:  # is still None
            allow_redirects = method.upper() == 'GET'

        if timeout is None:
            timeout = 30

        full_url = self.build_url(path)
        host = urlparse(full_url).netloc
        session = build_session()
        try:
            resp = getattr(session, method.lower())(
                url=full_url,
                headers=headers,
                json=data if json else None,
                data=data if not json else None,
                params=params,
                auth=auth,
                verify=self.verify_ssl,
                allow_redirects=allow_redirects,
                timeout=timeout,
            )
            resp.raise_for_status()
        except ConnectionError as e:
            metrics.incr('integrations.http_response', tags={
                'host': host,
                'status': 'connection_error'
            })
            raise ApiHostError.from_exception(e)
        except Timeout as e:
            metrics.incr('integrations.http_response', tags={
                'host': host,
                'status': 'timeout'
            })
            raise ApiTimeoutError.from_exception(e)
        except HTTPError as e:
            resp = e.response
            if resp is None:
                track_response_code(host, 'unknown')
                self.logger.exception('request.error', extra={
                    'url': full_url,
                })
                raise ApiError('Internal Error')
            track_response_code(host, resp.status_code)
            raise ApiError.from_response(resp)

        track_response_code(host, resp.status_code)
        if resp.status_code == 204:
            return {}

        return BaseApiResponse.from_response(resp, allow_text=allow_text)

def send_messages(messages, fail_silently=False):
    connection = get_connection(fail_silently=fail_silently)
    sent = connection.send_messages(messages)
    metrics.incr("email.sent", len(messages))
    for message in messages:
        logger.info(name="sentry.mail", event="mail.sent", message_id=message.extra_headers["Message-Id"])
    return sent

    def __init__(self, *args, **kwargs):
        StacktraceProcessor.__init__(self, *args, **kwargs)

        # If true, the project has been opted into using the symbolicator
        # service for native symbolication, which also means symbolic is not
        # used at all anymore.
        # The (iOS) symbolserver is still used regardless of this value.
        self.use_symbolicator = _is_symbolicator_enabled(self.project, self.data)

        metrics.incr('native.use_symbolicator', tags={'value': self.use_symbolicator})

        self.arch = cpu_name_from_data(self.data)
        self.signal = signal_from_data(self.data)

        self.sym = None
        self.difs_referenced = set()

        images = get_path(self.data, 'debug_meta', 'images', default=(),
                          filter=self._is_valid_image)

        if images:
            self.available = True
            self.sdk_info = get_sdk_from_event(self.data)
            self.object_lookup = ObjectLookup(images)
            self.images = images
        else:
            self.available = False

    def after(self, event, state):
        service = self.get_option('service')

        extra = {
            'event_id': event.id
        }
        if not service:
            self.logger.info('rules.fail.is_configured', extra=extra)
            return

        plugin = plugins.get(service)
        if not plugin.is_enabled(self.project):
            extra['project_id'] = self.project.id
            self.logger.info('rules.fail.is_enabled', extra=extra)
            return

        group = event.group

        if not plugin.should_notify(group=group, event=event):
            extra['group_id'] = group.id
            self.logger.info('rule.fail.should_notify', extra=extra)
            return

        metrics.incr('notifications.sent', instance=plugin.slug)
        yield self.future(plugin.rule_notify)

    def process(self, key):
        client = self.cluster.get_routing_client()
        lock_key = self._make_lock_key(key)
        # prevent a stampede due to the way we use celery etas + duplicate
        # tasks
        if not client.set(lock_key, '1', nx=True, ex=10):
            metrics.incr('buffer.revoked', tags={'reason': 'locked'})
            self.logger.info('Skipped process on %s; unable to get lock', key)
            return

        conn = self.cluster.get_local_client_for_key(key)
        pipe = conn.pipeline()
        pipe.hgetall(key)
        pipe.zrem(self.pending_key, key)
        pipe.delete(key)
        values = pipe.execute()[0]

        if not values:
            metrics.incr('buffer.revoked', tags={'reason': 'empty'})
            self.logger.info('Skipped process on %s; no values found', key)
            return

        model = import_string(values['m'])
        filters = pickle.loads(values['f'])
        incr_values = {}
        extra_values = {}
        for k, v in values.iteritems():
            if k.startswith('i+'):
                incr_values[k[2:]] = int(v)
            elif k.startswith('e+'):
                extra_values[k[2:]] = pickle.loads(v)

        super(RedisBuffer, self).process(model, incr_values, filters, extra_values)

def preprocess_event(cache_key=None, data=None, start_time=None, **kwargs):
    from sentry.plugins import plugins

    if cache_key:
        data = default_cache.get(cache_key)

    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'pre'})
        logger.error('Data not available in preprocess_event (cache_key=%s)', cache_key)
        return

    project = data['project']

    # TODO(dcramer): ideally we would know if data changed by default
    has_changed = False
    for plugin in plugins.all(version=2):
        for processor in (safe_execute(plugin.get_event_preprocessors) or ()):
            result = safe_execute(processor, data)
            if result:
                data = result
                has_changed = True

    assert data['project'] == project, 'Project cannot be mutated by preprocessor'

    if has_changed and cache_key:
        default_cache.set(cache_key, data, 3600)

    if cache_key:
        data = None
    save_event.delay(cache_key=cache_key, data=data, start_time=start_time)

def _do_preprocess_event(cache_key, data, start_time, event_id,
                         process_event):
    if cache_key:
        data = default_cache.get(cache_key)

    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'pre'})
        error_logger.error('preprocess.failed.empty',
            extra={'cache_key': cache_key})
        return

    project = data['project']
    Raven.tags_context({
        'project': project,
    })

    if should_process(data):
        process_event.delay(cache_key=cache_key, start_time=start_time,
            event_id=event_id)
        return

    # If we get here, that means the event had no preprocessing needed to be done
    # so we can jump directly to save_event
    if cache_key:
        data = None
    save_event.delay(cache_key=cache_key, data=data, start_time=start_time,
        event_id=event_id)

def preprocess_event(cache_key=None, data=None, start_time=None, **kwargs):
    from sentry.plugins import plugins

    if cache_key:
        data = default_cache.get(cache_key)

    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'pre'})
        error_logger.error('preprocess.failed.empty', extra={'cache_key': cache_key})
        return

    project = data['project']
    Raven.tags_context({
        'project': project,
    })

    # Iterate over all plugins looking for processors based on the input data
    # plugins should yield a processor function only if it actually can operate
    # on the input data, otherwise it should yield nothing
    for plugin in plugins.all(version=2):
        processors = safe_execute(plugin.get_event_preprocessors, data=data, _with_transaction=False)
        for processor in (processors or ()):
            # On the first processor found, we just defer to the process_event
            # queue to handle the actual work.
            process_event.delay(cache_key=cache_key, start_time=start_time)
            return

    # If we get here, that means the event had no preprocessing needed to be done
    # so we can jump directly to save_event
    if cache_key:
        data = None
    save_event.delay(cache_key=cache_key, data=data, start_time=start_time)

def task_revoked_handler(task, expired=False, **kwargs):
    if not isinstance(task, basestring):
        task = _get_task_name(task)
    if expired:
        metrics.incr('jobs.expired', instance=task)
    else:
        metrics.incr('jobs.revoked', instance=task)

def save_event(cache_key=None, data=None, start_time=None, event_id=None, **kwargs):
    """
    Saves an event to the database.
    """
    from sentry.event_manager import EventManager

    if cache_key:
        data = default_cache.get(cache_key)

    if event_id is None and data is not None:
        event_id = data['event_id']

    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'post'})
        return

    project = data.pop('project')

    delete_raw_event(project, event_id)

    Raven.tags_context({
        'project': project,
    })

    try:
        manager = EventManager(data)
        manager.save(project)
    finally:
        if cache_key:
            default_cache.delete(cache_key)
        if start_time:
            metrics.timing('events.time-to-process', time() - start_time,
                           instance=data['platform'])

    def get_interfaces(self):
        was_renormalized = _should_skip_to_python(self.event_id)

        metrics.incr('event.get_interfaces',
                     tags={'rust_renormalized': was_renormalized})

        return CanonicalKeyView(get_interfaces(self.data, rust_renormalized=was_renormalized))

def create_failed_event(cache_key, project_id, issues, event_id, start_time=None):
    """If processing failed we put the original data from the cache into a
    raw event.  Returns `True` if a failed event was inserted
    """
    reprocessing_active = ProjectOption.objects.get_value(
        project_id, 'sentry:reprocessing_active', REPROCESSING_DEFAULT
    )

    # The first time we encounter a failed event and the hint was cleared
    # we send a notification.
    sent_notification = ProjectOption.objects.get_value(
        project_id, 'sentry:sent_failed_event_hint', False
    )
    if not sent_notification:
        project = Project.objects.get_from_cache(id=project_id)
        Activity.objects.create(
            type=Activity.NEW_PROCESSING_ISSUES,
            project=project,
            datetime=to_datetime(start_time),
            data={'reprocessing_active': reprocessing_active,
                  'issues': issues},
        ).send_notification()
        ProjectOption.objects.set_value(project, 'sentry:sent_failed_event_hint', True)

    # If reprocessing is not active we bail now without creating the
    # processing issues
    if not reprocessing_active:
        return False

    # We need to get the original data here instead of passing the data in
    # from the last processing step because we do not want any
    # modifications to take place.
    delete_raw_event(project_id, event_id)
    data = default_cache.get(cache_key)
    if data is None:
        metrics.incr('events.failed', tags={'reason': 'cache', 'stage': 'raw'})
        error_logger.error('process.failed_raw.empty', extra={'cache_key': cache_key})
        return True

    from sentry.models import RawEvent, ProcessingIssue
    raw_event = RawEvent.objects.create(
        project_id=project_id,
        event_id=event_id,
        datetime=datetime.utcfromtimestamp(data['timestamp']).replace(tzinfo=timezone.utc),
        data=data
    )

    for issue in issues:
        ProcessingIssue.objects.record_processing_issue(
            raw_event=raw_event,
            scope=issue['scope'],
            object=issue['object'],
            type=issue['type'],
            data=issue['data'],
        )

    default_cache.delete(cache_key)

    return True

def send_messages(messages, fail_silently=False):
    connection = get_connection(fail_silently=fail_silently)
    sent = connection.send_messages(messages)
    metrics.incr("email.sent", len(messages))
    for message in messages:
        extra = {"message_id": message.extra_headers["Message-Id"]}
        logger.info("mail.sent", extra=extra)
    return sent

    def after(self, event, state):
        group = event.group

        for plugin in self.get_plugins():
            if not safe_execute(plugin.should_notify, group=group, event=event):
                continue

            metrics.incr('notifications.sent', instance=plugin.slug)
            yield self.future(plugin.rule_notify)

 def close(self):
     StacktraceProcessor.close(self)
     if self.dsyms_referenced:
         metrics.incr('dsyms.processed',
                      amount=len(self.dsyms_referenced),
                      instance=self.project.id)
     if self.sym is not None:
         self.sym.close()
         self.sym = None