Implement TelemetryProcessors for Azure exporters (#851)

Author: lzchen

contrib/opencensus-ext-azure/README.rst

@@ -37,6 +37,8 @@ This example shows how to send a warning level log to Azure Monitor.
     logger.addHandler(AzureLogHandler(connection_string='InstrumentationKey=<your-instrumentation_key-here>'))
     logger.warning('Hello, World!')
 
+Correlation
+###########
 
 You can enrich the logs with trace IDs and span IDs by using the `logging integration <../opencensus-ext-logging>`_.
 
@@ -73,6 +75,9 @@ You can enrich the logs with trace IDs and span IDs by using the `logging integr
         logger.warning('In the span')
     logger.warning('After the span')
 
+Custom Properties
+#################
+
 You can also add custom properties to your log messages in the *extra* keyword argument using the custom_dimensions field.
 
 WARNING: For this feature to work, you need to pass a dictionary to the custom_dimensions field. If you pass arguments of any other type, the logger will ignore them.
@@ -89,6 +94,34 @@ WARNING: For this feature to work, you need to pass a dictionary to the custom_d
     properties = {'custom_dimensions': {'key_1': 'value_1', 'key_2': 'value_2'}}
     logger.warning('action', extra=properties)
 
+Modifying Logs
+##############
+
+* You can pass a callback function to the exporter to process telemetry before it is exported.
+* Your callback function can return `False` if you do not want this envelope exported.
+* Your callback function must accept an [envelope](https://github.com/census-instrumentation/opencensus-python/blob/master/contrib/opencensus-ext-azure/opencensus/ext/azure/common/protocol.py#L86) data type as its parameter.
+* You can see the schema for Azure Monitor data types in the envelopes [here](https://github.com/census-instrumentation/opencensus-python/blob/master/contrib/opencensus-ext-azure/opencensus/ext/azure/common/protocol.py).
+* The `AzureLogHandler` handles `ExceptionData` and `MessageData` data types.
+
+.. code:: python
+
+    import logging
+
+    from opencensus.ext.azure.log_exporter import AzureLogHandler
+
+    logger = logging.getLogger(__name__)
+
+    # Callback function to append '_hello' to each log message telemetry
+    def callback_function(envelope):
+        envelope.data.baseData.message += '_hello'
+        return True
+
+    handler = AzureLogHandler(connection_string='InstrumentationKey=<your-instrumentation_key-here>')
+    handler.add_telemetry_processor(callback_function)
+    logger.addHandler(handler)
+    logger.warning('Hello, World!')
+
+
 Metrics
 ~~~~~~~
 
@@ -145,6 +178,9 @@ The **Azure Monitor Metrics Exporter** allows you to export metrics to `Azure Mo
     if __name__ == "__main__":
         main()
 
+Standard Metrics
+################
+
 The exporter also includes a set of standard metrics that are exported to Azure Monitor by default.
 
 .. code:: python
@@ -179,6 +215,67 @@ Below is a list of standard metrics that are currently available:
 - Process CPU Usage (percentage)
 - Process Private Bytes (bytes)
 
+Modifying Metrics
+#################
+
+* You can pass a callback function to the exporter to process telemetry before it is exported.
+* Your callback function can return `False` if you do not want this envelope exported.
+* Your callback function must accept an [envelope](https://github.com/census-instrumentation/opencensus-python/blob/master/contrib/opencensus-ext-azure/opencensus/ext/azure/common/protocol.py#L86) data type as its parameter.
+* You can see the schema for Azure Monitor data types in the envelopes [here](https://github.com/census-instrumentation/opencensus-python/blob/master/contrib/opencensus-ext-azure/opencensus/ext/azure/common/protocol.py).
+* The `MetricsExporter` handles `MetricData` data types.
+
+.. code:: python
+
+    import time
+
+    from opencensus.ext.azure import metrics_exporter
+    from opencensus.stats import aggregation as aggregation_module
+    from opencensus.stats import measure as measure_module
+    from opencensus.stats import stats as stats_module
+    from opencensus.stats import view as view_module
+    from opencensus.tags import tag_map as tag_map_module
+
+    stats = stats_module.stats
+    view_manager = stats.view_manager
+    stats_recorder = stats.stats_recorder
+
+    CARROTS_MEASURE = measure_module.MeasureInt("carrots",
+                                                "number of carrots",
+                                                "carrots")
+    CARROTS_VIEW = view_module.View("carrots_view",
+                                    "number of carrots",
+                                    [],
+                                    CARROTS_MEASURE,
+                                    aggregation_module.CountAggregation())
+
+    # Callback function to only export the metric if value is greater than 0
+    def callback_function(envelope):
+        return envelope.data.baseData.metrics[0].value > 0
+
+    def main():
+        # Enable metrics
+        # Set the interval in seconds in which you want to send metrics
+        exporter = metrics_exporter.new_metrics_exporter(connection_string='InstrumentationKey=<your-instrumentation-key-here>')
+        exporter.add_telemetry_processor(callback_function)
+        view_manager.register_exporter(exporter)
+
+        view_manager.register_view(CARROTS_VIEW)
+        mmap = stats_recorder.new_measurement_map()
+        tmap = tag_map_module.TagMap()
+
+        mmap.measure_int_put(CARROTS_MEASURE, 1000)
+        mmap.record(tmap)
+        # Default export interval is every 15.0s
+        # Your application should run for at least this amount
+        # of time so the exporter will meet this interval
+        # Sleep can fulfill this
+        time.sleep(60)
+
+        print("Done recording metrics")
+
+    if __name__ == "__main__":
+        main()
+
 Trace
 ~~~~~
 
@@ -197,13 +294,18 @@ This example shows how to send a span "hello" to Azure Monitor.
     from opencensus.trace.tracer import Tracer
 
     tracer = Tracer(
-        exporter=AzureExporter(connection_string='InstrumentationKey=<your-instrumentation-key-here>'),
+        exporter=AzureExporter(
+            connection_string='InstrumentationKey=<your-instrumentation-key-here>'
+        ),
         sampler=ProbabilitySampler(1.0)
     )
 
     with tracer.span(name='hello'):
         print('Hello, World!')
 
+Integrations
+############
+
 OpenCensus also supports several `integrations <https://github.com/census-instrumentation/opencensus-python#integration>`_ which allows OpenCensus to integrate with third party libraries.
 
 This example shows how to integrate with the `requests <https://2.python-requests.org/en/master/>`_ library.
@@ -225,14 +327,45 @@ This example shows how to integrate with the `requests <https://2.python-request
     config_integration.trace_integrations(['requests'])
     tracer = Tracer(
         exporter=AzureExporter(
-            # TODO: replace this with your own instrumentation key.
-            instrumentation_key='00000000-0000-0000-0000-000000000000',
+            connection_string='InstrumentationKey=<your-instrumentation-key-here>',
         ),
         sampler=ProbabilitySampler(1.0),
     )
     with tracer.span(name='parent'):
         response = requests.get(url='https://www.wikipedia.org/wiki/Rabbit')
 
+Modifying Traces
+################
+
+* You can pass a callback function to the exporter to process telemetry before it is exported.
+* Your callback function can return `False` if you do not want this envelope exported.
+* Your callback function must accept an [envelope](https://github.com/census-instrumentation/opencensus-python/blob/master/contrib/opencensus-ext-azure/opencensus/ext/azure/common/protocol.py#L86) data type as its parameter.
+* You can see the schema for Azure Monitor data types in the envelopes [here](https://github.com/census-instrumentation/opencensus-python/blob/master/contrib/opencensus-ext-azure/opencensus/ext/azure/common/protocol.py).
+* The `AzureExporter` handles `Data` data types.
+
+.. code:: python
+
+    import requests
+
+    from opencensus.ext.azure.trace_exporter import AzureExporter
+    from opencensus.trace import config_integration
+    from opencensus.trace.samplers import ProbabilitySampler
+    from opencensus.trace.tracer import Tracer
+
+    config_integration.trace_integrations(['requests'])
+
+    # Callback function to add os_type: linux to span properties
+    def callback_function(envelope):
+        envelope.data.baseData.properties['os_type'] = 'linux'
+        return True
+
+    exporter = AzureExporter(
+        connection_string='InstrumentationKey=<your-instrumentation-key-here>'
+    )
+    exporter.add_telemetry_processor(callback_function)
+    tracer = Tracer(exporter=exporter, sampler=ProbabilitySampler(1.0))
+    with tracer.span(name='parent'):
+        response = requests.get(url='https://www.wikipedia.org/wiki/Rabbit')
 
 References
 ----------

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

@@ -0,0 +1,63 @@
+# 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
+
+logger = logging.getLogger(__name__)
+
+
+class ProcessorMixin(object):
+    """ProcessorMixin adds the ability to process telemetry processors
+
+    Telemetry processors are functions that are called before exporting of
+    telemetry to possibly modify the envelope contents.
+    """
+
+    def add_telemetry_processor(self, processor):
+        """Adds telemetry processor to the collection. Telemetry processors
+        will be called one by one before telemetry item is pushed for sending
+        and in the order they were added.
+
+        :param processor: The processor to add.
+        """
+        self._telemetry_processors.append(processor)
+
+    def clear_telemetry_processors(self):
+        """Removes all telemetry processors"""
+        self._telemetry_processors = []
+
+    def apply_telemetry_processors(self, envelopes):
+        """Applies all telemetry processors in the order they were added.
+
+        This function will return the list of envelopes to be exported after
+        each processor has been run sequentially. Individual processors can
+        throw exceptions and fail, but the applying of all telemetry processors
+        will proceed (not fast fail). Processors also return True if envelope
+        should be included for exporting, False otherwise.
+
+        :param envelopes: The envelopes to apply each processor to.
+        """
+        filtered_envelopes = []
+        for envelope in envelopes:
+            accepted = True
+            for processor in self._telemetry_processors:
+                try:
+                    if processor(envelope) is False:
+                        accepted = False
+                        break
+                except Exception as ex:
+                    logger.warning('Telemetry processor failed with: %s.', ex)
+            if accepted:
+                filtered_envelopes.append(envelope)
+        return filtered_envelopes

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

@@ -20,6 +20,7 @@
 
 from opencensus.common.schedule import Queue, QueueEvent, QueueExitEvent
 from opencensus.ext.azure.common import Options, utils
+from opencensus.ext.azure.common.processor import ProcessorMixin
 from opencensus.ext.azure.common.protocol import (
     Data,
     Envelope,
@@ -118,7 +119,7 @@ def filter(self, record):
         return random.random() < self.probability
 
 
-class AzureLogHandler(TransportMixin, BaseLogHandler):
+class AzureLogHandler(TransportMixin, ProcessorMixin, BaseLogHandler):
     """Handler for logging to Microsoft Azure Monitor.
 
     :param options: Options for the log handler.
@@ -137,6 +138,7 @@ def __init__(self, **options):
             maintenance_period=self.options.storage_maintenance_period,
             retention_period=self.options.storage_retention_period,
         )
+        self._telemetry_processors = []
         super(AzureLogHandler, self).__init__()
         self.addFilter(SamplingFilter(self.options.logging_sampling_rate))
 
@@ -148,6 +150,7 @@ def _export(self, batch, event=None):  # pragma: NO COVER
         try:
             if batch:
                 envelopes = [self.log_record_to_envelope(x) for x in batch]
+                envelopes = self.apply_telemetry_processors(envelopes)
                 result = self._transmit(envelopes)
                 if result > 0:
                     self.storage.put(envelopes, result)

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

@@ -19,6 +19,7 @@
 
 from opencensus.common import utils as common_utils
 from opencensus.ext.azure.common import Options, utils
+from opencensus.ext.azure.common.processor import ProcessorMixin
 from opencensus.ext.azure.common.protocol import (
     Data,
     DataPoint,
@@ -35,7 +36,7 @@
 logger = logging.getLogger(__name__)
 
 
-class MetricsExporter(object):
+class MetricsExporter(ProcessorMixin):
     """Metrics exporter for Microsoft Azure Monitor."""
 
     def __init__(self, options=None):
@@ -46,6 +47,8 @@ def __init__(self, options=None):
         if self.options.max_batch_size <= 0:
             raise ValueError('Max batch size must be at least 1.')
         self.max_batch_size = self.options.max_batch_size
+        self._telemetry_processors = []
+        super(MetricsExporter, self).__init__()
 
     def export_metrics(self, metrics):
         if metrics:
@@ -75,6 +78,7 @@ def export_metrics(self, metrics):
                 batched_envelopes = list(common_utils.window(
                     envelopes, self.max_batch_size))
                 for batch in batched_envelopes:
+                    batch = self.apply_telemetry_processors(batch)
                     self._transmit_without_retry(batch)
 
     def create_data_points(self, time_series, metric_descriptor):

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

@@ -17,6 +17,7 @@
 from opencensus.common.schedule import QueueExitEvent
 from opencensus.ext.azure.common import Options, utils
 from opencensus.ext.azure.common.exporter import BaseExporter
+from opencensus.ext.azure.common.processor import ProcessorMixin
 from opencensus.ext.azure.common.protocol import (
     Data,
     Envelope,
@@ -37,7 +38,7 @@
 __all__ = ['AzureExporter']
 
 
-class AzureExporter(TransportMixin, BaseExporter):
+class AzureExporter(BaseExporter, ProcessorMixin, TransportMixin):
     """An exporter that sends traces to Microsoft Azure Monitor.
 
     :param options: Options for the exporter.
@@ -52,6 +53,7 @@ def __init__(self, **options):
             maintenance_period=self.options.storage_maintenance_period,
             retention_period=self.options.storage_retention_period,
         )
+        self._telemetry_processors = []
         super(AzureExporter, self).__init__(**options)
 
     def span_data_to_envelope(self, sd):
@@ -152,6 +154,7 @@ def emit(self, batch, event=None):
         try:
             if batch:
                 envelopes = [self.span_data_to_envelope(sd) for sd in batch]
+                envelopes = self.apply_telemetry_processors(envelopes)
                 result = self._transmit(envelopes)
                 if result > 0:
                     self.storage.put(envelopes, result)