Merge pull request #11 from appoptics/add-custom-propagator

NH-2313 Add basic TraceState handling and W3C trace context propagation

opentelemetry_distro_solarwinds/__init__.py

@@ -1 +1,10 @@
 __version__ = "0.0.1"
+
+COMMA = ","
+COMMA_W3C_SANITIZED = "...."
+EQUALS = "="
+EQUALS_W3C_SANITIZED = "####"
+SW_TRACESTATE_KEY = "sw"
+OTEL_CONTEXT_SW_OPTIONS_KEY = "sw_xtraceoptions"
+OTEL_CONTEXT_SW_SIGNATURE_KEY = "sw_signature"
+DEFAULT_SW_TRACES_EXPORTER = "solarwinds_exporter"

opentelemetry_distro_solarwinds/configurator.py

@@ -0,0 +1,156 @@
+"""Module to initialize OpenTelemetry SDK components to work with SolarWinds backend"""
+
+import logging
+from os import environ
+from pkg_resources import (
+    iter_entry_points,
+    load_entry_point
+)
+
+from opentelemetry import trace
+from opentelemetry.environment_variables import (
+    OTEL_PROPAGATORS,
+    OTEL_TRACES_EXPORTER
+)
+from opentelemetry.instrumentation.propagators import set_global_response_propagator
+from opentelemetry.propagate import set_global_textmap
+from opentelemetry.propagators.composite import CompositePropagator
+from opentelemetry.sdk._configuration import _OTelSDKConfigurator
+from opentelemetry.sdk.environment_variables import OTEL_TRACES_SAMPLER
+from opentelemetry.sdk.trace import (
+    sampling,
+    TracerProvider
+)
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+from opentelemetry_distro_solarwinds import DEFAULT_SW_TRACES_EXPORTER
+from opentelemetry_distro_solarwinds.extension.oboe import Reporter
+from opentelemetry_distro_solarwinds.response_propagator import SolarWindsTraceResponsePropagator
+
+logger = logging.getLogger(__name__)
+
+class SolarWindsConfigurator(_OTelSDKConfigurator):
+    """OpenTelemetry Configurator for initializing SolarWinds-reporting SDK components"""
+
+    # Cannot set as env default because not part of OTel Python _KNOWN_SAMPLERS
+    # https://github.com/open-telemetry/opentelemetry-python/blob/main/opentelemetry-sdk/src/opentelemetry/sdk/trace/sampling.py#L364-L380
+    _DEFAULT_SW_TRACES_SAMPLER = "solarwinds_sampler"
+
+    def _configure(self, **kwargs):
+        """Configure OTel sampler, exporter, propagator, response propagator"""
+        reporter = self._initialize_solarwinds_reporter()
+        self._configure_sampler()
+        self._configure_exporter(reporter)
+        self._configure_propagator()
+        # Set global HTTP response propagator
+        set_global_response_propagator(SolarWindsTraceResponsePropagator())
+
+    def _configure_sampler(self):
+        """Always configure SolarWinds OTel sampler"""
+        try:
+            sampler = load_entry_point(
+                "opentelemetry_distro_solarwinds",
+                "opentelemetry_traces_sampler",
+                self._DEFAULT_SW_TRACES_SAMPLER
+            )()
+        except:
+            logger.exception(
+                "Failed to load configured sampler {}".format(
+                    self._DEFAULT_SW_TRACES_SAMPLER
+                )
+            )
+            raise
+        trace.set_tracer_provider(
+            TracerProvider(sampler=sampler)
+        )
+
+    def _configure_exporter(self, reporter):
+        """Configure SolarWinds or env-specified OTel span exporter.
+        Initialization of SolarWinds exporter requires a liboboe reporter."""
+        exporter = None
+        environ_exporter_name = environ.get(OTEL_TRACES_EXPORTER)
+
+        if environ_exporter_name == DEFAULT_SW_TRACES_EXPORTER:
+            try:
+                exporter = load_entry_point(
+                    "opentelemetry_distro_solarwinds",
+                    "opentelemetry_traces_exporter",
+                    environ_exporter_name
+                )(reporter)
+            except:
+                logger.exception(
+                    "Failed to load configured exporter {} with reporter".format(
+                        environ_exporter_name
+                    )
+                )
+                raise
+        else:
+            try:
+                exporter = next(
+                    iter_entry_points(
+                        "opentelemetry_traces_exporter",
+                        environ_exporter_name
+                    )
+                ).load()()
+            except:
+                logger.exception(
+                    "Failed to load configured exporter {}".format(
+                        environ_exporter_name
+                    )
+                )
+                raise
+        span_processor = BatchSpanProcessor(exporter)
+        trace.get_tracer_provider().add_span_processor(span_processor)
+
+    def _configure_propagator(self):
+        """Configure CompositePropagator with SolarWinds and other propagators"""
+        propagators = []
+        environ_propagators_names = environ.get(OTEL_PROPAGATORS).split(",")
+        for propagator_name in environ_propagators_names:
+            try:
+                propagators.append(
+                    next(
+                        iter_entry_points("opentelemetry_propagator", propagator_name)
+                    ).load()()
+                )
+            except Exception:
+                logger.exception(
+                    "Failed to load configured propagator {}".format(
+                        propagator_name
+                    )
+                )
+                raise
+        set_global_textmap(CompositePropagator(propagators))
+
+    def _initialize_solarwinds_reporter(self) -> Reporter:
+        """Initialize SolarWinds reporter used by sampler and exporter. This establishes collector and sampling settings in a background thread."""
+        log_level = environ.get('SOLARWINDS_DEBUG_LEVEL', 3)
+        try:
+            log_level = int(log_level)
+        except ValueError:
+            log_level = 3
+        # TODO make some of these customizable
+        return Reporter(
+            hostname_alias='',
+            log_level=log_level,
+            log_file_path='',
+            max_transactions=-1,
+            max_flush_wait_time=-1,
+            events_flush_interval=-1,
+            max_request_size_bytes=-1,
+            reporter='ssl',
+            host=environ.get('SOLARWINDS_COLLECTOR', ''),
+            service_key=environ.get('SOLARWINDS_SERVICE_KEY', ''),
+            trusted_path='',
+            buffer_size=-1,
+            trace_metrics=-1,
+            histogram_precision=-1,
+            token_bucket_capacity=-1,
+            token_bucket_rate=-1,
+            file_single=0,
+            ec2_metadata_timeout=1000,
+            grpc_proxy='',
+            stdout_clear_nonblocking=0,
+            is_grpc_clean_hack_enabled=False,
+            w3c_trace_format=1,
+        )

opentelemetry_distro_solarwinds/distro.py

@@ -1,24 +1,52 @@
-"""Module to configure OpenTelemetry agent to work with SolarWinds backend"""
+"""Module to configure OpenTelemetry to work with SolarWinds backend"""
 
-from opentelemetry import trace
+import logging
+from os import environ
+
+from opentelemetry.environment_variables import (
+    OTEL_PROPAGATORS,
+    OTEL_TRACES_EXPORTER
+)
 from opentelemetry.instrumentation.distro import BaseDistro
-from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
-from opentelemetry_distro_solarwinds.exporter import SolarWindsSpanExporter
-from opentelemetry_distro_solarwinds.sampler import ParentBasedAoSampler
+from opentelemetry_distro_solarwinds import DEFAULT_SW_TRACES_EXPORTER
 
+logger = logging.getLogger(__name__)
 
 class SolarWindsDistro(BaseDistro):
-    """SolarWinds custom distro for OpenTelemetry agents.
+    """OpenTelemetry Distro for SolarWinds reporting environment"""
+
+    _TRACECONTEXT_PROPAGATOR = "tracecontext"
+    _SW_PROPAGATOR = "solarwinds_propagator"
+    _DEFAULT_SW_PROPAGATORS = [
+        _TRACECONTEXT_PROPAGATOR,
+        "baggage",
+        _SW_PROPAGATOR,
+    ]
 
-    With this custom distro, the following functionality is introduced:
-        - no functionality added at this time
-    """
     def _configure(self, **kwargs):
-        # automatically make use of custom SolarWinds sampler
-        trace.set_tracer_provider(
-            TracerProvider(sampler=ParentBasedAoSampler()))
-        # Automatically configure the SolarWinds Span exporter
-        span_exporter = BatchSpanProcessor(SolarWindsSpanExporter())
-        trace.get_tracer_provider().add_span_processor(span_exporter)
+        """Configure OTel exporter and propagators"""
+        environ.setdefault(OTEL_TRACES_EXPORTER, DEFAULT_SW_TRACES_EXPORTER)
+
+        environ_propagators = environ.get(
+            OTEL_PROPAGATORS,
+            ",".join(self._DEFAULT_SW_PROPAGATORS)
+        ).split(",")
+        # If not using the default propagators,
+        # can any arbitrary list BUT
+        # (1) must include tracecontext and solarwinds_propagator
+        # (2) tracecontext must be before solarwinds_propagator
+        if environ_propagators != self._DEFAULT_SW_PROPAGATORS:
+            if not self._TRACECONTEXT_PROPAGATOR in environ_propagators or \
+                not self._SW_PROPAGATOR in environ_propagators:
+                raise ValueError("Must include tracecontext and solarwinds_propagator in OTEL_PROPAGATORS to use SolarWinds Observability.")
+
+            if environ_propagators.index(self._SW_PROPAGATOR) \
+                < environ_propagators.index(self._TRACECONTEXT_PROPAGATOR):
+                raise ValueError("tracecontext must be before solarwinds_propagator in OTEL_PROPAGATORS to use SolarWinds Observability.")
+        environ[OTEL_PROPAGATORS] = ",".join(environ_propagators)
+
+        logger.debug("Configured SolarWindsDistro: {}, {}".format(
+            environ.get(OTEL_TRACES_EXPORTER),
+            environ.get(OTEL_PROPAGATORS)
+        ))

opentelemetry_distro_solarwinds/exporter.py

@@ -5,26 +5,25 @@
 """
 
 import logging
-import os
 
 from opentelemetry.sdk.trace.export import SpanExporter
 
-from opentelemetry_distro_solarwinds.extension.oboe import (Context, Metadata,
-                                                            Reporter)
-from opentelemetry_distro_solarwinds.ot_ao_transformer import transform_id
+from opentelemetry_distro_solarwinds.extension.oboe import (
+    Context,
+    Metadata
+)
+from opentelemetry_distro_solarwinds.w3c_transformer import W3CTransformer
 
 logger = logging.getLogger(__file__)
 
 
 class SolarWindsSpanExporter(SpanExporter):
-    """SolarWinds span exporter.
-
-    Reports instrumentation data to the SolarWinds backend.
+    """SolarWinds custom span exporter for the SolarWinds backend.
+    Initialization requires a liboboe reporter.
     """
-    def __init__(self, *args, **kw_args):
+    def __init__(self, reporter, *args, **kw_args):
         super().__init__(*args, **kw_args)
-        self.reporter = None
-        self._initialize_solarwinds_reporter()
+        self.reporter = reporter
 
     def export(self, spans):
         """Export to AO events and report via liboboe.
@@ -36,31 +35,31 @@ def export(self, spans):
             md = self._build_metadata(span.get_span_context())
             if span.parent and span.parent.is_valid:
                 # If there is a parent, we need to add an edge to this parent to this entry event
-                logger.debug("Continue trace from %s", md.toString())
+                logger.debug("Continue trace from {}".format(md.toString()))
                 parent_md = self._build_metadata(span.parent)
-                evt = Context.startTrace(md, int(span.start_time / 1000),
+                evt = Context.createEntry(md, int(span.start_time / 1000),
                                          parent_md)
             else:
                 # In OpenTelemrtry, there are no events with individual IDs, but only a span ID
                 # and trace ID. Thus, the entry event needs to be generated such that it has the
                 # same op ID as the span ID of the OTel span.
-                logger.debug("Start a new trace %s", md.toString())
-                evt = Context.startTrace(md, int(span.start_time / 1000))
+                logger.debug("Start a new trace {}".format(md.toString()))
+                evt = Context.createEntry(md, int(span.start_time / 1000))
             evt.addInfo('Layer', span.name)
             evt.addInfo('Language', 'Python')
             for k, v in span.attributes.items():
                 evt.addInfo(k, v)
-            self.reporter.sendReport(evt)
+            self.reporter.sendReport(evt, False)
 
             for event in span.events:
                 if event.name == 'exception':
                     self._report_exception_event(event)
                 else:
                     self._report_info_event(event)
 
-            evt = Context.stopTrace(int(span.end_time / 1000))
+            evt = Context.createExit(int(span.end_time / 1000))
             evt.addInfo('Layer', span.name)
-            self.reporter.sendReport(evt)
+            self.reporter.sendReport(evt, False)
 
     def _report_exception_event(self, event):
         evt = Context.createEvent(int(event.timestamp / 1000))
@@ -76,7 +75,7 @@ def _report_exception_event(self, event):
             if k not in ('exception.type', 'exception.message',
                          'exception.stacktrace'):
                 evt.addInfo(k, v)
-        self.reporter.sendReport(evt)
+        self.reporter.sendReport(evt, False)
 
     def _report_info_event(self, event):
         print("Found info event")
@@ -86,40 +85,10 @@ def _report_info_event(self, event):
         evt.addInfo('Label', 'info')
         for k, v in event.attributes.items():
             evt.addInfo(k, v)
-        self.reporter.sendReport(evt)
-
-    def _initialize_solarwinds_reporter(self):
-        """Initialize liboboe."""
-        log_level = os.environ.get('SOLARWINDS_DEBUG_LEVEL', 3)
-        try:
-            log_level = int(log_level)
-        except ValueError:
-            log_level = 3
-        self.reporter = Reporter(
-            hostname_alias='',
-            log_level=log_level,
-            log_file_path='',
-            max_transactions=-1,
-            max_flush_wait_time=-1,
-            events_flush_interval=-1,
-            max_request_size_bytes=-1,
-            reporter='ssl',
-            host=os.environ.get('SOLARWINDS_COLLECTOR', ''),
-            service_key=os.environ.get('SOLARWINDS_SERVICE_KEY', ''),
-            trusted_path='',
-            buffer_size=-1,
-            trace_metrics=-1,
-            histogram_precision=-1,
-            token_bucket_capacity=-1,
-            token_bucket_rate=-1,
-            file_single=0,
-            ec2_metadata_timeout=1000,
-            grpc_proxy='',
-            stdout_clear_nonblocking=0,
-            is_grpc_clean_hack_enabled=False,
-            w3c_trace_format=1,
-        )
+        self.reporter.sendReport(evt, False)
 
     @staticmethod
     def _build_metadata(span_context):
-        return Metadata.fromString(transform_id(span_context))
+        return Metadata.fromString(
+            W3CTransformer.traceparent_from_context(span_context)
+        )