Merge branch 'grunt-js-folder' of https://github.com/coreos-inc/quay into ackbar

util/analytics.py

@@ -1,8 +1,9 @@
 import json
 import logging
 
-from multiprocessing import Process, Queue
-from mixpanel import Consumer, Mixpanel
+from Queue import Queue
+from threading import Thread
+from mixpanel import BufferedConsumer, Mixpanel
 
 
 logger = logging.getLogger(__name__)
@@ -17,24 +18,23 @@ class MixpanelQueingConsumer(object):
     self._mp_queue.put(json.dumps([endpoint, json_message]))
 
 
-class SendToMixpanel(Process):
+class SendToMixpanel(Thread):
   def __init__(self, request_queue):
-    Process.__init__(self)
+    Thread.__init__(self)
+    self.daemon = True
 
     self._mp_queue = request_queue
-    self._consumer = Consumer()
-    self.daemon = True
+    self._consumer = BufferedConsumer()
 
   def run(self):
     logger.debug('Starting mixpanel sender process.')
     while True:
       mp_request = self._mp_queue.get()
-      logger.debug('Got queued mixpanel reqeust.')
+      logger.debug('Got queued mixpanel request.')
       try:
         self._consumer.send(*json.loads(mp_request))
       except:
-        # Make sure we don't crash if Mixpanel request fails.
-        pass
+        logger.exception('Failed to send Mixpanel request.')
 
 
 class FakeMixpanel(object):

util/queuemetrics.py

@@ -1,7 +1,9 @@
 import logging
 import boto
 
-from multiprocessing import Process, Queue
+from Queue import Queue
+from threading import Thread
+
 
 logger = logging.getLogger(__name__)
 
@@ -12,6 +14,7 @@ class NullReporter(object):
 
 
 class QueueingCloudWatchReporter(object):
+  """ QueueingCloudWatchReporter reports metrics to the "SendToCloudWatch" process """
   def __init__(self, request_queue, namespace, need_capacity_name, build_percent_name):
     self._namespace = namespace
     self._need_capacity_name = need_capacity_name
@@ -34,26 +37,37 @@ class QueueingCloudWatchReporter(object):
                         unit='Percent')
 
 
-class SendToCloudWatch(Process):
+class SendToCloudWatch(Thread):
+  """ SendToCloudWatch loops indefinitely and pulls metrics off of a queue then sends it to
+  CloudWatch. """
   def __init__(self, request_queue, aws_access_key, aws_secret_key):
-    Process.__init__(self)
+    Thread.__init__(self)
+    self.daemon = True
+
     self._aws_access_key = aws_access_key
     self._aws_secret_key = aws_secret_key
     self._put_metrics_queue = request_queue
-    self.daemon = True
 
   def run(self):
-    logger.debug('Starting cloudwatch sender process.')
-    connection = boto.connect_cloudwatch(self._aws_access_key, self._aws_secret_key)
+    try:
+      logger.debug('Starting CloudWatch sender process.')
+      connection = boto.connect_cloudwatch(self._aws_access_key, self._aws_secret_key)
+    except:
+      logger.exception('Failed to connect to CloudWatch.')
+
     while True:
       put_metric_args, kwargs = self._put_metrics_queue.get()
-      logger.debug('Got queued put metrics reqeust.')
-      connection.put_metric_data(*put_metric_args, **kwargs)
+      logger.debug('Got queued put metrics request.')
+      try:
+        connection.put_metric_data(*put_metric_args, **kwargs)
+      except:
+        logger.exception('Failed to write to CloudWatch')
 
 
 class QueueMetrics(object):
   def __init__(self, app=None):
     self.app = app
+    self.sender = None
     if app is not None:
       self.state = self.init_app(app)
     else:
@@ -72,8 +86,7 @@ class QueueMetrics(object):
       request_queue = Queue()
       reporter = QueueingCloudWatchReporter(request_queue, namespace, req_capacity_name,
                                             build_percent_name)
-      sender = SendToCloudWatch(request_queue, access_key, secret_key)
-      sender.start()
+      self.sender = SendToCloudWatch(request_queue, access_key, secret_key)
     else:
       reporter = NullReporter()
 
@@ -82,5 +95,11 @@ class QueueMetrics(object):
     app.extensions['queuemetrics'] = reporter
     return reporter
 
+  def run(self):
+    logger.debug('Asked to start CloudWatch reporter')
+    if self.sender is not None:
+      logger.debug('Starting CloudWatch reporter')
+      self.sender.start()
+
   def __getattr__(self, name):
     return getattr(self.state, name, None)

util/useremails.py

@@ -1,5 +1,6 @@
 import logging
 import traceback
+import json
 
 from flask.ext.mail import Message
 
@@ -13,7 +14,42 @@ template_env = get_template_env("emails")
 class CannotSendEmailException(Exception):
   pass
 
-def send_email(recipient, subject, template_file, parameters):
+class GmailAction(object):
+  """ Represents an action that can be taken in Gmail in response to the email. """
+  def __init__(self, metadata):
+    self.metadata = metadata
+
+  @staticmethod
+  def confirm(name, url, description):
+    return GmailAction({
+      "@context": "http://schema.org",
+      "@type": "EmailMessage",
+      "action": {
+        "@type": 'ConfirmAction',
+        "name": name,
+        "handler": {
+          "@type": "HttpActionHandler",
+          "url": get_app_url() + '/' + url
+        }
+      },
+      "description": description
+    })
+
+  @staticmethod
+  def view(name, url, description):
+    return GmailAction({
+      "@context": "http://schema.org",
+      "@type": "EmailMessage",
+      "action": {
+        "@type": 'ViewAction',
+        "name": name,
+        "url": get_app_url() + '/' + url
+      },
+      "description": description
+    })
+
+
+def send_email(recipient, subject, template_file, parameters, action=None):
   app_title = app.config['REGISTRY_TITLE_SHORT']
   app_url = get_app_url()
 
@@ -29,7 +65,8 @@ def send_email(recipient, subject, template_file, parameters):
     'app_logo': 'https://quay.io/static/img/quay-logo.png', # TODO: make this pull from config
     'app_url': app_url,
     'app_title': app_title,
-    'app_link': app_link_handler
+    'app_link': app_link_handler,
+    'action_metadata': json.dumps(action.metadata) if action else None
   })
 
   rendered_html = template_env.get_template(template_file + '.html').render(parameters)
@@ -61,25 +98,34 @@ def send_change_email(username, email, token):
   })
 
 def send_confirmation_email(username, email, token):
+  action = GmailAction.confirm('Confirm E-mail', 'confirm?code=' + token,
+                               'Verification of e-mail address')
+
   send_email(email, 'Please confirm your e-mail address', 'confirmemail', {
     'username': username,
     'token': token
-  })
+  }, action=action)
 
 def send_repo_authorization_email(namespace, repository, email, token):
+  action = GmailAction.confirm('Verify E-mail', 'authrepoemail?code=' + token,
+                               'Verification of e-mail address')
+
   subject = 'Please verify your e-mail address for repository %s/%s' % (namespace, repository)
   send_email(email, subject, 'repoauthorizeemail', {
     'namespace': namespace,
     'repository': repository,
     'token': token
-  })
+  }, action=action)
 
 def send_recovery_email(email, token):
+  action = GmailAction.view('Recover Account', 'recovery?code=' + token,
+                            'Recovery of an account')
+
   subject = 'Account recovery'
   send_email(email, subject, 'recovery', {
     'email': email,
     'token': token
-  })
+  }, action=action)
 
 def send_payment_failed(email, username):
   send_email(email, 'Subscription Payment Failure', 'paymentfailure', {
@@ -87,12 +133,15 @@ def send_payment_failed(email, username):
   })
 
 def send_org_invite_email(member_name, member_email, orgname, team, adder, code):
+  action = GmailAction.view('Join %s' % team, 'confirminvite?code=' + code,
+                            'Invitation to join a team')
+
   send_email(member_email, 'Invitation to join team', 'teaminvite', {
     'inviter': adder,
     'token': code,
     'organization': orgname,
     'teamname': team
-  })
+  }, action=action)
 
 
 def send_invoice_email(email, contents):