feat(eventstream): Add Snuba backend

src/sentry/eventstream/snuba.py

@@ -0,0 +1,19 @@
+from __future__ import absolute_import
+
+from sentry.eventstream.base import EventStream
+from sentry.utils import snuba
+
+
+class SnubaEventStream(EventStream):
+    def publish(self, group, event, is_new, is_sample, is_regression, is_new_group_environment, primary_hash, skip_consume=False):
+        snuba.insert_raw([{
+            'group_id': event.group_id,
+            'event_id': event.event_id,
+            'project_id': event.project_id,
+            'message': event.message,
+            'platform': event.platform,
+            'datetime': event.datetime,
+            'data': event.data.data,
+            'primary_hash': primary_hash,
+        }])
+        super(SnubaEventStream, self).publish(group, event, is_new, is_sample, is_regression, is_new_group_environment, primary_hash, skip_consume)

src/sentry/utils/snuba.py

@@ -5,7 +5,6 @@ from contextlib import contextmanager
 from dateutil.parser import parse as parse_datetime
 from itertools import chain
 from operator import or_
-import json
 import six
 import time
 import urllib3
@@ -15,7 +14,7 @@ from django.db.models import Q
 
 from sentry.event_manager import HASH_RE
 from sentry.models import Environment, Group, GroupHash, GroupHashTombstone, GroupRelease, Release, ReleaseProject
-from sentry.utils import metrics
+from sentry.utils import metrics, json
 from sentry.utils.dates import to_timestamp
 from functools import reduce
 
@@ -377,3 +376,15 @@ def get_related_project_ids(column, ids):
                 project_field + '__isnull': False,
             }).values_list(project_field, flat=True)
     return []
+
+
+def insert_raw(data):
+    data = json.dumps(data)
+    try:
+        with timer('snuba_insert_raw'):
+            return _snuba_pool.urlopen(
+                'POST', '/tests/insert',
+                body=data,
+            )
+    except urllib3.exceptions.HTTPError as err:
+        raise SnubaError(err)