OLD | NEW |
(Empty) | |
| 1 # Copyright 2015 The Chromium Authors. All rights reserved. |
| 2 # Use of this source code is governed by a BSD-style license that can be |
| 3 # found in the LICENSE file. |
| 4 |
| 5 """Classes representing the monitoring interface for tasks or devices. |
| 6 |
| 7 In appengine, a PubSubMonitor will be automatically initialized when initialize(
) |
| 8 is called, and there is no need to initialize it directly from this class. |
| 9 """ |
| 10 |
| 11 |
| 12 import base64 |
| 13 import json |
| 14 import os |
| 15 |
| 16 from monacq.proto import metrics_pb2 |
| 17 |
| 18 from third_party import httplib2 |
| 19 from third_party.apiclient import discovery |
| 20 from third_party.oauth2client.client import GoogleCredentials |
| 21 |
| 22 |
| 23 class Monitor(object): |
| 24 """Abstract base class encapsulating the ability to collect and send metrics. |
| 25 |
| 26 This is a singleton class. There should only be one instance of a Monitor at |
| 27 a time. It will be created and initialized by process_argparse_options. It |
| 28 must exist in order for any metrics to be sent, although both Targets and |
| 29 Metrics may be initialized before the underlying Monitor. If it does not exist |
| 30 at the time that a Metric is sent, an exception will be raised. |
| 31 """ |
| 32 @staticmethod |
| 33 def _wrap_proto(data): |
| 34 """Normalize MetricsData, list(MetricsData), and MetricsCollection. |
| 35 |
| 36 Args: |
| 37 input: A MetricsData, list of MetricsData, or a MetricsCollection. |
| 38 |
| 39 Returns: |
| 40 A MetricsCollection with the appropriate data attribute set. |
| 41 """ |
| 42 if isinstance(data, metrics_pb2.MetricsCollection): |
| 43 ret = data |
| 44 elif isinstance(data, list): |
| 45 ret = metrics_pb2.MetricsCollection(data=data) |
| 46 else: |
| 47 ret = metrics_pb2.MetricsCollection(data=[data]) |
| 48 return ret |
| 49 |
| 50 def send(self, metric_pb): |
| 51 raise NotImplementedError() |
| 52 |
| 53 |
| 54 class PubSubMonitor(Monitor): |
| 55 """Class which publishes metrics to a Cloud Pub/Sub topic.""" |
| 56 |
| 57 _SCOPES = [ |
| 58 'https://www.googleapis.com/auth/pubsub', |
| 59 ] |
| 60 |
| 61 def _initialize(self, project, topic): |
| 62 creds = GoogleCredentials.get_application_default() |
| 63 creds = creds.create_scoped(self._SCOPES) |
| 64 self._http = httplib2.Http() |
| 65 creds.authorize(self._http) |
| 66 self._api = discovery.build('pubsub', 'v1', http=self._http) |
| 67 self._topic = 'projects/%s/topics/%s' % (project, topic) |
| 68 |
| 69 def __init__(self, project, topic): |
| 70 """Process monitoring related command line flags and initialize api. |
| 71 |
| 72 Args: |
| 73 project (str): the name of the Pub/Sub project to publish to. |
| 74 topic (str): the name of the Pub/Sub topic to publish to. |
| 75 """ |
| 76 self._initialize(project, topic) |
| 77 |
| 78 def send(self, metric_pb): |
| 79 """Send a metric proto to the monitoring api. |
| 80 |
| 81 Args: |
| 82 metric_pb (MetricsData or MetricsCollection): the metric protobuf to send |
| 83 """ |
| 84 proto = self._wrap_proto(metric_pb) |
| 85 body = { |
| 86 'messages': [ |
| 87 {'data': base64.b64encode(proto.SerializeToString())}, |
| 88 ], |
| 89 } |
| 90 self._api.projects().topics().publish( |
| 91 topic=self._topic, |
| 92 body=body).execute(num_retries=5) |
| 93 |
| 94 |
| 95 class NullMonitor(Monitor): |
| 96 """Class that doesn't send metrics anywhere.""" |
| 97 def send(self, metric_pb): |
| 98 pass |
OLD | NEW |