Events: Send vote data to event-collector

r2/r2/controllers/api.py

@@ -563,7 +563,8 @@ def POST_submit(self, form, jquery, url, selftext, kind, title,
             l._commit()
             l.set_url_cache()
 
-        queries.queue_vote(c.user, l, True, request.ip, cheater=c.cheater)
+        queries.queue_vote(c.user, l, True, request.ip, cheater=c.cheater,
+                           event_data=g.events.event_base(request, c))
 
         if sr.should_ratelimit(c.user, 'link'):
             c.user.clog_quota('link', l)
@@ -1884,7 +1885,8 @@ def POST_comment(self, commentform, jquery, parent, comment):
             item, inbox_rel = Comment._new(c.user, link, parent_comment,
                                            comment, request.ip)
             queries.queue_vote(c.user, item, True, request.ip,
-                               cheater=c.cheater)
+                               cheater=c.cheater,
+                               event_data=g.events.event_base(request, c))
 
         if is_message:
             queries.new_message(item, inbox_rel)
@@ -2060,8 +2062,8 @@ def POST_vote(self, dir, thing, vote_info):
                else None)
 
         queries.queue_vote(user, thing, dir, request.ip, vote_info=vote_info,
-                           store=store,
-                           cheater=c.cheater)
+                           store=store, cheater=c.cheater,
+                           event_data=g.events.event_base(request, c))
 
     @require_oauth2_scope("modconfig")
     @validatedForm(VSrModerator(perms='config'),

r2/r2/lib/db/queries.py

@@ -1683,8 +1683,8 @@ def _run_commentstree(msgs, chan):
 vote_names_by_dir = {True: "1", None: "0", False: "-1"}
 vote_dirs_by_name = {v: k for k, v in vote_names_by_dir.iteritems()}
 
-def queue_vote(user, thing, dir, ip, vote_info=None,
-               cheater = False, store = True):
+def queue_vote(user, thing, dir, ip, vote_info=None, cheater=False, store=True,
+               event_data=None):
     # set the vote in memcached so the UI gets updated immediately
     key = prequeued_vote_key(user, thing)
     grace_period = int(g.vote_queue_grace_period.total_seconds())
@@ -1722,7 +1722,7 @@ def queue_vote(user, thing, dir, ip, vote_info=None,
             "ip": ip,
             "info": vote_info,
             "cheater": cheater,
-            "event": None,
+            "event": event_data,
         }
         amqp.add_item(qname, json.dumps(vote))
 

r2/r2/lib/eventcollector.py

@@ -24,6 +24,7 @@
 import hmac
 import json
 import time
+import uuid
 
 import pytz
 import requests
@@ -70,7 +71,7 @@ def vote_event(self, vote, old_vote=None, event_base=None, request=None,
             event_base = Event.base_from_request(request, context)
         event_base["event_topic"] = "vote"
         event_base["event_name"] = "vote_server"
-        event_base["event_ts"] = epoch_timestamp(vote._date)
+        event_base["event_ts"] = str(epoch_timestamp(vote._date))
         event_base["vote_target"] = vote._thing2._fullname
         event_base["vote_direction"] = self.VOTES[vote._name]
         if old_vote:
@@ -80,7 +81,7 @@ def vote_event(self, vote, old_vote=None, event_base=None, request=None,
         if event_base["vote_type"] == "link" and vote._thing2.is_self:
             event_base["vote_type"] = "self"
         event_base["sr"] = vote._thing2.subreddit_slow.name
-        event_base["sr_id"] = vote._thing2.subreddit_slow._id
+        event_base["sr_id"] = str(vote._thing2.subreddit_slow._id)
 
         self.save_event(event_base)
 
@@ -97,11 +98,12 @@ class Event(dict):
         "user_agent",
         "ip",
         "domain",
+        "uuid",
     )
     @classmethod
     def base_from_request(cls, request, context, **kw):
         if context.user_is_loggedin:
-            user_id = context.user._id
+            user_id = str(context.user._id)
             loid = None
         else:
             user_id = None
@@ -125,21 +127,27 @@ def base_from_request(cls, request, context, **kw):
     @classmethod
     def base(cls, event_name=None, timestamp=None, user_agent=None, ip=None,
               domain=None, user_id=None, loid=None, oauth2_client_id=None,
-              **kw):
+              event_uuid=None, **kw):
         ret = cls(kw)
 
-        ret["event_name"] = event_name
-        ret["timestamp"] = timestamp
-        ret["user_agent"] = user_agent
-        ret["ip"] = ip
-        ret["domain"] = domain
-
-        if user_id:
+        if event_uuid is None:
+            ret["uuid"] = str(uuid.uuid4())
+
+        if event_name is not None:
+            ret["event_name"] = event_name
+        if timestamp is not None:
+            ret["event_ts"] = timestamp
+        if user_agent is not None:
+            ret["user_agent"] = user_agent
+        if ip is not None:
+            ret["ip"] = ip
+        if domain is not None:
+            ret["domain"] = domain
+        if user_id is not None:
             ret["user_id"] = user_id
-        if loid:
+        if loid is not None:
             ret["loid"] = loid
-
-        if oauth2_client_id:
+        if oauth2_client_id is not None:
             ret["oauth_client_id"] = oauth2_client_id
 
         return ret

r2/r2/models/vote.py

@@ -314,12 +314,15 @@ def cast_vote(sub, obj, vote_info, timer, date):
             timer.intermediate("incr_sr_counts")
 
     # write the vote to cassandra
-    VotesByAccount.copy_from(vote, vote_info)
+    VotesByAccount.copy_from(vote, vote_info["info"])
     timer.intermediate("cassavotes")
 
     vote._thing2.update_search_index(boost_only=True)
     timer.intermediate("update_search_index")
 
+    if "event" in vote_info and vote_info["event"]:
+        g.events.vote_event(vote, old_vote, event_base=vote_info["event"])
+
     return vote
 
 

upstart/reddit-consumer-event_collector_q.conf

@@ -10,5 +10,5 @@ respawn limit 10 5
 nice 10
 script
     . /etc/default/reddit
-    wrap-job paster run --proctitle event_collector_q$x $REDDIT_INI $REDDIT_ROOT/r2/lib/eventcollector.py -c 'process_events(g, limit=200)'
+    wrap-job paster run --proctitle event_collector_q$x $REDDIT_INI $REDDIT_ROOT/r2/lib/eventcollector.py -c 'from pylons import g; process_events(g, limit=100)'
 end script