Azure log exporter (#668)

Implemented the actual log conversion and exporting logic.
Moved the network/storage transmission logic from trace exporter to a shared mixin, applied to both traces and logs.
Avoid common names (Object and prototype), use BaseObject and _default instead.

Author: reyang

contrib/opencensus-ext-azure/CHANGELOG.md

@@ -2,7 +2,8 @@
 
 ## Unreleased
 - Added log exporter
-  ([#657](https://github.com/census-instrumentation/opencensus-python/pull/657))
+  ([#657](https://github.com/census-instrumentation/opencensus-python/pull/657),
+  [#668](https://github.com/census-instrumentation/opencensus-python/pull/668))
 - Added persistent storage support
   ([#640](https://github.com/census-instrumentation/opencensus-python/pull/640))
 - Changed AzureExporter constructor signature to use kwargs

contrib/opencensus-ext-azure/examples/logs/correlated.py

@@ -27,7 +27,6 @@
 # TODO: you need to specify the instrumentation key in the
 # APPINSIGHTS_INSTRUMENTATIONKEY environment variable.
 handler = AzureLogHandler()
-handler.setFormatter(logging.Formatter('%(traceId)s %(spanId)s %(message)s'))
 logger.addHandler(handler)
 
 tracer = Tracer(exporter=AzureExporter(), sampler=ProbabilitySampler(1.0))

contrib/opencensus-ext-azure/examples/logs/error.py

@@ -0,0 +1,33 @@
+# Copyright 2019, OpenCensus Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from opencensus.ext.azure.log_exporter import AzureLogHandler
+
+logger = logging.getLogger(__name__)
+# TODO: you need to specify the instrumentation key in the
+# APPINSIGHTS_INSTRUMENTATIONKEY environment variable.
+logger.addHandler(AzureLogHandler())
+
+
+def main():
+    try:
+        return 1 / 0  # generate a ZeroDivisionError
+    except Exception:
+        logger.exception('Captured an exception.')
+
+
+if __name__ == '__main__':
+    main()

contrib/opencensus-ext-azure/opencensus/ext/azure/common/__init__.py

@@ -15,11 +15,11 @@
 import os
 import sys
 
-from opencensus.ext.azure.common.protocol import Object
+from opencensus.ext.azure.common.protocol import BaseObject
 
 
-class Options(Object):
-    prototype = Object(
+class Options(BaseObject):
+    _default = BaseObject(
         endpoint='https://dc.services.visualstudio.com/v2/track',
         export_interval=15.0,
         grace_period=5.0,

contrib/opencensus-ext-azure/opencensus/ext/azure/common/protocol.py

@@ -13,9 +13,9 @@
 # limitations under the License.
 
 
-class Object(dict):
+class BaseObject(dict):
     def __init__(self, *args, **kwargs):
-        super(Object, self).__init__(*args, **kwargs)
+        super(BaseObject, self).__init__(*args, **kwargs)
         for key in kwargs:
             self[key] = kwargs[key]
 
@@ -25,9 +25,9 @@ def __repr__(self):
             for item in self.items():
                 if item[0] not in tmp:
                     tmp[item[0]] = item[1]
-            if self.prototype == self:
+            if self._default == self:
                 break
-            self = self.prototype
+            self = self._default
         return repr(tmp)
 
     def __setattr__(self, name, value):
@@ -36,25 +36,25 @@ def __setattr__(self, name, value):
     def __getattr__(self, name):
         try:
             return self[name]
-        except KeyError as ex:
+        except KeyError:
             raise AttributeError("'{}' object has no attribute {}".format(
                 type(self).__name__,
-                ex,
+                name,
             ))
 
     def __getitem__(self, key):
-        if self.prototype is self:
-            return super(Object, self).__getitem__(key)
+        if self._default is self:
+            return super(BaseObject, self).__getitem__(key)
         if key in self:
-            return super(Object, self).__getitem__(key)
-        return self.prototype[key]
+            return super(BaseObject, self).__getitem__(key)
+        return self._default[key]
 
 
-Object.prototype = Object()
+BaseObject._default = BaseObject()
 
 
-class Data(Object):
-    prototype = Object(
+class Data(BaseObject):
+    _default = BaseObject(
         baseData=None,
         baseType=None,
     )
@@ -65,8 +65,8 @@ def __init__(self, *args, **kwargs):
         self.baseType = self.baseType
 
 
-class Envelope(Object):
-    prototype = Object(
+class Envelope(BaseObject):
+    _default = BaseObject(
         ver=1,
         name='',
         time='',
@@ -84,8 +84,8 @@ def __init__(self, *args, **kwargs):
         self.time = self.time
 
 
-class Event(Object):
-    prototype = Object(
+class Event(BaseObject):
+    _default = BaseObject(
         ver=2,
         name='',
         properties=None,
@@ -98,8 +98,24 @@ def __init__(self, *args, **kwargs):
         self.name = self.name
 
 
-class Message(Object):
-    prototype = Object(
+class ExceptionData(BaseObject):
+    _default = BaseObject(
+        ver=2,
+        exceptions=[],
+        severityLevel=None,
+        problemId=None,
+        properties=None,
+        measurements=None,
+    )
+
+    def __init__(self, *args, **kwargs):
+        super(ExceptionData, self).__init__(*args, **kwargs)
+        self.ver = self.ver
+        self.exceptions = self.exceptions
+
+
+class Message(BaseObject):
+    _default = BaseObject(
         ver=2,
         message='',
         severityLevel=None,
@@ -113,8 +129,8 @@ def __init__(self, *args, **kwargs):
         self.message = self.message
 
 
-class RemoteDependency(Object):
-    prototype = Object(
+class RemoteDependency(BaseObject):
+    _default = BaseObject(
         ver=2,
         name='',
         id='',
@@ -136,8 +152,8 @@ def __init__(self, *args, **kwargs):
         self.duration = self.duration
 
 
-class Request(Object):
-    prototype = Object(
+class Request(BaseObject):
+    _default = BaseObject(
         ver=2,
         id='',
         duration='',

contrib/opencensus-ext-azure/opencensus/ext/azure/common/transport.py

@@ -0,0 +1,135 @@
+# Copyright 2019, OpenCensus Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import logging
+import requests
+
+from opencensus.trace import execution_context
+
+logger = logging.getLogger(__name__)
+
+
+class TransportMixin(object):
+    def _transmit_from_storage(self):
+        for blob in self.storage.gets():
+            # give a few more seconds for blob lease operation
+            # to reduce the chance of race (for perf consideration)
+            if blob.lease(self.options.timeout + 5):
+                envelopes = blob.get()  # TODO: handle error
+                result = self._transmit(envelopes)
+                if result > 0:
+                    blob.lease(result)
+                else:
+                    blob.delete(silent=True)
+
+    def _transmit(self, envelopes):
+        """
+        Transmit the data envelopes to the ingestion service.
+        Return a negative value for partial success or non-retryable failure.
+        Return 0 if all envelopes have been successfully ingested.
+        Return the next retry time in seconds for retryable failure.
+        This function should never throw exception.
+        """
+        # TODO: prevent requests being tracked
+        blacklist_hostnames = execution_context.get_opencensus_attr(
+            'blacklist_hostnames',
+        )
+        execution_context.set_opencensus_attr(
+            'blacklist_hostnames',
+            ['dc.services.visualstudio.com'],
+        )
+        try:
+            response = requests.post(
+                url=self.options.endpoint,
+                data=json.dumps(envelopes),
+                headers={
+                    'Accept': 'application/json',
+                    'Content-Type': 'application/json; charset=utf-8',
+                },
+                timeout=self.options.timeout,
+            )
+        except Exception as ex:  # TODO: consider RequestException
+            logger.warning('Transient client side error %s.', ex)
+            # client side error (retryable)
+            return self.options.minimum_retry_interval
+        finally:
+            execution_context.set_opencensus_attr(
+                'blacklist_hostnames',
+                blacklist_hostnames,
+            )
+        text = 'N/A'
+        data = None
+        try:
+            text = response.text
+        except Exception as ex:
+            logger.warning('Error while reading response body %s.', ex)
+        else:
+            try:
+                data = json.loads(text)
+            except Exception:
+                pass
+        if response.status_code == 200:
+            logger.info('Transmission succeeded: %s.', text)
+            return 0
+        if response.status_code == 206:  # Partial Content
+            # TODO: store the unsent data
+            if data:
+                try:
+                    resend_envelopes = []
+                    for error in data['errors']:
+                        if error['statusCode'] in (
+                                429,  # Too Many Requests
+                                500,  # Internal Server Error
+                                503,  # Service Unavailable
+                        ):
+                            resend_envelopes.append(envelopes[error['index']])
+                        else:
+                            logger.error(
+                                'Data drop %s: %s %s.',
+                                error['statusCode'],
+                                error['message'],
+                                envelopes[error['index']],
+                            )
+                    if resend_envelopes:
+                        self.storage.put(resend_envelopes)
+                except Exception as ex:
+                    logger.error(
+                        'Error while processing %s: %s %s.',
+                        response.status_code,
+                        text,
+                        ex,
+                    )
+                return -response.status_code
+            # cannot parse response body, fallback to retry
+        if response.status_code in (
+                206,  # Partial Content
+                429,  # Too Many Requests
+                500,  # Internal Server Error
+                503,  # Service Unavailable
+        ):
+            logger.warning(
+                'Transient server side error %s: %s.',
+                response.status_code,
+                text,
+            )
+            # server side error (retryable)
+            return self.options.minimum_retry_interval
+        logger.error(
+            'Non-retryable server side error %s: %s.',
+            response.status_code,
+            text,
+        )
+        # server side error (non-retryable)
+        return -response.status_code

contrib/opencensus-ext-azure/opencensus/ext/azure/common/utils.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import datetime
 import locale
 import os
 import platform
@@ -24,6 +25,7 @@
 
 from opencensus.common.version import __version__ as opencensus_version
 from opencensus.common.utils import timestamp_to_microseconds
+from opencensus.common.utils import to_iso_str
 from opencensus.ext.azure.common.version import __version__ as ext_version
 
 azure_monitor_context = {
@@ -57,5 +59,9 @@ def timestamp_to_duration(start_time, end_time):
     return microseconds_to_duration(duration_us)
 
 
+def timestamp_to_iso_str(timestamp):
+    return to_iso_str(datetime.datetime.utcfromtimestamp(timestamp))
+
+
 def url_to_dependency_name(url):
     return urlparse(url).netloc