Add HttpsMonitor

infra_libs/ts_mon/common/monitors.py

 # Copyright 2015 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Classes representing the monitoring interface for tasks or devices."""
 
 
 import base64
+import httplib2
 import json
 import logging
 import socket
 import traceback
 
 from googleapiclient import discovery
 from googleapiclient import errors
+from infra_libs import httplib2_utils
+from infra_libs.ts_mon.common import http_metrics
+from infra_libs.ts_mon.common import pb_to_popo
+from infra_libs.ts_mon.protos import metrics_pb2
 from oauth2client import gce
 from oauth2client.client import GoogleCredentials
 from oauth2client.file import Storage
-import httplib2
-
-from infra_libs import httplib2_utils
-from infra_libs.ts_mon.common import http_metrics
-from infra_libs.ts_mon.protos import metrics_pb2
-
 
 # Special string that can be passed through as the credentials path to use the
 # default Appengine or GCE service account.
 APPENGINE_CREDENTIALS = ':appengine'
 GCE_CREDENTIALS = ':gce'
 
 
 class Monitor(object):
   """Abstract base class encapsulating the ability to collect and send metrics.
 
   This is a singleton class. There should only be one instance of a Monitor at
   a time. It will be created and initialized by process_argparse_options. It
   must exist in order for any metrics to be sent, although both Targets and
   Metrics may be initialized before the underlying Monitor. If it does not exist
   at the time that a Metric is sent, an exception will be raised.
   """
+
+  _SCOPES = []
+
   @staticmethod
   def _wrap_proto(data):
     """Normalize MetricsData, list(MetricsData), and MetricsCollection.
 
     Args:
       input: A MetricsData, list of MetricsData, or a MetricsCollection.
 
     Returns:
       A MetricsCollection with the appropriate data attribute set.
     """
     if isinstance(data, metrics_pb2.MetricsCollection):
       ret = data
     elif isinstance(data, list):
       ret = metrics_pb2.MetricsCollection(data=data)
     else:
       ret = metrics_pb2.MetricsCollection(data=[data])
     return ret
 
-  def send(self, metric_pb):
-    raise NotImplementedError()
-
-
-class PubSubMonitor(Monitor):
-  """Class which publishes metrics to a Cloud Pub/Sub topic."""
-
-  _SCOPES = [
-      'https://www.googleapis.com/auth/pubsub',
-  ]
-
-  TIMEOUT = 10  # seconds
-
   def _load_credentials(self, credentials_file_path):
     if credentials_file_path == GCE_CREDENTIALS:
       return gce.AppAssertionCredentials(self._SCOPES)
     if credentials_file_path == APPENGINE_CREDENTIALS:  # pragma: no cover
       # This import doesn't work outside appengine, so delay it until it's used.
       from oauth2client import appengine
       from google.appengine.api import app_identity
       logging.info('Initializing with service account %s',
                    app_identity.get_service_account_name())
       return appengine.AppAssertionCredentials(self._SCOPES)
 
     with open(credentials_file_path, 'r') as credentials_file:
       credentials_json = json.load(credentials_file)
     if credentials_json.get('type', None):
       credentials = GoogleCredentials.from_stream(credentials_file_path)
       credentials = credentials.create_scoped(self._SCOPES)
       return credentials
     return Storage(credentials_file_path).get()
 
+  def send(self, metric_pb):
+    raise NotImplementedError()
+
+class HttpsMonitor(Monitor):
+
+  _SCOPES = [
+    'https://www.googleapis.com/auth/prodxmon'
+  ]
+
+  def __init__(self, endpoint, credentials_file_path, http=None):
+    self._endpoint = endpoint
+    credentials = self._load_credentials(credentials_file_path)
+    if http is None:
+      http = httplib2_utils.RetriableHttp(
+                 httplib2_utils.InstrumentedHttp('acq-mon-api'))

[Sergey Berezin, 2016/07/06 22:35:49]

nit: 4 spaces from the start of the previous line. I.e.:

  http = f(
      arg1, ...)

[tnn, 2016/07/06 22:44:36]

Done.

+    self._http = credentials.authorize(http)
+
+  def encodeToJson(self, metric_pb):
+    return json.dumps({ 'resource': pb_to_popo.convert(metric_pb) })
+
+  def send(self, metric_pb):
+    body = self.encodeToJson(self._wrap_proto(metric_pb))
+
+    try:
+      resp, content = self._http.request(self._endpoint, method='POST',
+                                         body=body)
+      if resp.status != 200:
+        logging.warning('HttpsMonitor.send received status %d: %s', resp,
+                        content)
+    except (ValueError, errors.Error,
+            socket.timeout, socket.error, socket.herror, socket.gaierror,
+            httplib2.HttpLib2Error):
+      logging.warning('HttpsMonitor.send failed: %s\n',
+                      traceback.format_exc())
+
+
+class PubSubMonitor(Monitor):
+  """Class which publishes metrics to a Cloud Pub/Sub topic."""
+
+  _SCOPES = [
+      'https://www.googleapis.com/auth/pubsub',
+  ]
+
+  TIMEOUT = 10  # seconds
+
   def _initialize(self):
     creds = self._load_credentials(self._credsfile)
     creds.authorize(self._http)
     self._api = discovery.build('pubsub', 'v1', http=self._http)
 
   def _update_init_metrics(self, status):
     if not self._use_instrumented_http:
       return
     fields = {'name': 'acq-mon-api-pubsub',
               'client': 'discovery',
@@ skipping 85 matching lines @@
     logging.info('Flushing monitoring metrics:\n%s', text)
     if self._fh is not None:
       self._fh.write(text + '\n\n')
       self._fh.flush()
 
 
 class NullMonitor(Monitor):
   """Class that doesn't send metrics anywhere."""
   def send(self, metric_pb):
     pass