Refactor exporters to accept SpanData objects instead of Spans (#144)

Author: wkiser

opencensus/trace/exporters/base.py

@@ -20,13 +20,24 @@ class Exporter(object):
 
     Subclasses of :class:`Exporter` must override :meth:`export`.
     """
-    def emit(self, trace):
-        """Emit the trace."""
+
+    def emit(self, span_datas):
+        """
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to emit
+        """
         raise NotImplementedError
 
-    def export(self, trace):
+    def export(self, span_datas):
         """Export the trace. Send trace to transport, and transport will call
         exporter.emit() to actually send the trace to the specified tracing
         backend.
+
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to export
         """
         raise NotImplementedError

opencensus/trace/exporters/file_exporter.py

@@ -16,6 +16,7 @@
 
 import json
 
+from opencensus.trace import span_data
 from opencensus.trace.exporters import base
 from opencensus.trace.exporters.transports import sync
 
@@ -47,14 +48,25 @@ def __init__(self, file_name=DEFAULT_FILENAME,
         self.transport = transport(self)
         self.file_mode = file_mode
 
-    def emit(self, trace):
+    def emit(self, span_datas):
         """
-        :type trace: dict
-        :param trace: Trace collected.
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to emit
         """
         with open(self.file_name, self.file_mode) as file:
-            trace_str = json.dumps(trace)
+            # convert to the legacy trace json for easier refactoring
+            # TODO: refactor this to use the span data directly
+            legacy_trace_json = span_data.format_legacy_trace_json(span_datas)
+            trace_str = json.dumps(legacy_trace_json)
             file.write(trace_str)
 
-    def export(self, trace):
-        self.transport.export(trace)
+    def export(self, span_datas):
+        """
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to export
+        """
+        self.transport.export(span_datas)

opencensus/trace/exporters/logging_exporter.py

@@ -16,6 +16,7 @@
 
 import logging
 
+from opencensus.trace import span_data
 from opencensus.trace.exporters import base
 from opencensus.trace.exporters.transports import sync
 
@@ -63,12 +64,23 @@ def __init__(self, handler=None, transport=sync.SyncTransport):
         self.logger.setLevel(logging.INFO)
         self.transport = transport(self)
 
-    def emit(self, trace):
+    def emit(self, span_datas):
         """
-        :type traces: dict
-        :param traces: Trace collected.
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to emit
         """
-        self.logger.info(trace)
+        # convert to the legacy trace json for easier refactoring
+        # TODO: refactor this to use the span data directly
+        legacy_trace_json = span_data.format_legacy_trace_json(span_datas)
+        self.logger.info(legacy_trace_json)
 
-    def export(self, trace):
-        self.transport.export(trace)
+    def export(self, span_datas):
+        """
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to export
+        """
+        self.transport.export(span_datas)

opencensus/trace/exporters/print_exporter.py

@@ -28,20 +28,24 @@ class PrintExporter(base.Exporter):
                       :class:`.SyncTransport`. The other option is
                       :class:`.BackgroundThreadTransport`.
     """
+
     def __init__(self, transport=sync.SyncTransport):
         self.transport = transport(self)
 
-    def emit(self, trace):
-        """export the traces by printing it out.
-
-        :type trace: dict
-        :param trace: Trace collected.
-
-        :rtype: dict
-        :returns: Trace printed.
+    def emit(self, span_datas):
         """
-        print(trace)
-        return trace
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to emit
+        """
+        print(span_datas)
 
-    def export(self, trace):
-        self.transport.export(trace)
+    def export(self, span_datas):
+        """
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to export
+        """
+        self.transport.export(span_datas)

opencensus/trace/exporters/stackdriver_exporter.py

@@ -14,14 +14,14 @@
 
 import os
 
-from opencensus.trace.attributes import Attributes
+from google.cloud.trace.client import Client
+
 from opencensus.trace import attributes_helper
+from opencensus.trace import span_data
+from opencensus.trace.attributes import Attributes
 from opencensus.trace.exporters import base
 from opencensus.trace.exporters.transports import sync
 
-from google.cloud.trace.client import Client
-
-
 # OpenCensus Version
 VERSION = '0.1.3'
 
@@ -132,33 +132,46 @@ def __init__(self, client=None, project_id=None,
         self.project_id = client.project
         self.transport = transport(self)
 
-    def emit(self, spans):
+    def emit(self, span_datas):
         """
-        :type spans: dict
-        :param spans: Spans collected.
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to emit
         """
         name = 'projects/{}'.format(self.project_id)
-        stackdriver_spans = self.translate_to_stackdriver(spans)
+
+        # convert to the legacy trace json for easier refactoring
+        # TODO: refactor this to use the span data directly
+        trace = span_data.format_legacy_trace_json(span_datas)
+
+        stackdriver_spans = self.translate_to_stackdriver(trace)
         self.client.batch_write_spans(name, stackdriver_spans)
 
-    def export(self, trace):
-        self.transport.export(trace)
+    def export(self, span_datas):
+        """
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to export
+        """
+        self.transport.export(span_datas)
 
-    def translate_to_stackdriver(self, spans):
+    def translate_to_stackdriver(self, trace):
         """Translate the spans json to Stackdriver format.
 
         See: https://cloud.google.com/trace/docs/reference/v2/rest/v2/
              projects.traces/batchWrite
 
-        :type spans: dict
-        :param spans: Spans collected.
+        :type trace: dict
+        :param trace: Trace dictionary
 
         :rtype: dict
         :returns: Spans in Google Cloud StackDriver Trace format.
         """
-        set_attributes(spans)
-        spans_json = spans.get('spans')
-        trace_id = spans.get('traceId')
+        set_attributes(trace)
+        spans_json = trace.get('spans')
+        trace_id = trace.get('traceId')
         spans_list = []
 
         for span in spans_json:

opencensus/trace/exporters/transports/background_thread.py

@@ -81,34 +81,27 @@ def _get_items(self):
     def _thread_main(self):
         """The entry point for the worker thread.
 
-        Pulls pending spans off the queue and writes them in batches to
-        the specified tracing backend using the exporter.
+        Pulls pending SpanData tuples off the queue and writes them in
+        batches to the specified tracing backend using the exporter.
         """
         print('Background thread started.')
 
         quit_ = False
 
         while True:
             items = self._get_items()
-            trace_id = None
-            spans = []
+            span_datas = []
 
             for item in items:
                 if item is _WORKER_TERMINATOR:
                     quit_ = True
                     # Continue processing items, don't break, try to process
                     # all items we got back before quitting.
                 else:
-                    trace_id = item.get('traceId')
-                    spans.extend(item.get('spans'))
+                    span_datas.extend(item)
 
-            if spans and trace_id:
-                spans_json = {
-                    'traceId': trace_id,
-                    'spans': spans,
-                }
-
-                self.exporter.emit(spans_json)
+            if span_datas:
+                self.exporter.emit(span_datas)
 
             for _ in range(len(items)):
                 self._queue.task_done()
@@ -181,9 +174,9 @@ def _export_pending_spans(self):
         else:
             print('Failed to send pending spans.')
 
-    def enqueue(self, trace):
-        """Queues a trace to be written by the background thread."""
-        self._queue.put_nowait(trace)
+    def enqueue(self, span_datas):
+        """Queues span_datas to be written by the background thread."""
+        self._queue.put_nowait(span_datas)
 
     def flush(self):
         """Submit any pending spans."""
@@ -214,9 +207,9 @@ def __init__(self, exporter, grace_period=_DEFAULT_GRACE_PERIOD,
         self.worker = _Worker(exporter, grace_period, max_batch_size)
         self.worker.start()
 
-    def export(self, trace):
+    def export(self, span_datas):
         """Put the trace to be exported into queue."""
-        self.worker.enqueue(trace)
+        self.worker.enqueue(span_datas)
 
     def flush(self):
         """Submit any pending traces."""

opencensus/trace/exporters/transports/base.py

@@ -20,8 +20,8 @@ class Transport(object):
 
     Subclasses of :class:`Transport` must override :meth:`export`.
     """
-    def export(self, trace):
-        """Export the trace by calling exporter.emit()."""
+    def export(self, span_datas):
+        """Export the SpanData tuples by calling exporter.emit()."""
         raise NotImplementedError
 
     def flush(self):

opencensus/trace/exporters/transports/sync.py

@@ -19,5 +19,5 @@ class SyncTransport(base.Transport):
     def __init__(self, exporter):
         self.exporter = exporter
 
-    def export(self, trace):
-        self.exporter.emit(trace)
+    def export(self, span_datas):
+        self.exporter.emit(span_datas)

opencensus/trace/exporters/zipkin_exporter.py

@@ -14,12 +14,14 @@
 
 """Export the spans data to Zipkin Collector."""
 
+import calendar
 import datetime
 import json
 import logging
+
 import requests
-import calendar
 
+from opencensus.trace import span_data
 from opencensus.trace.exporters import base
 from opencensus.trace.exporters.transports import sync
 
@@ -90,12 +92,18 @@ def get_url(self):
             self.port,
             self.endpoint)
 
-    def emit(self, trace):
-        """Send trace to Zipkin server, default using the v2 API.
+    def emit(self, span_datas):
+        """Send SpanData tuples to Zipkin server, default using the v2 API.
 
-        :type trace: dict
-        :param trace: Trace data in dictionary format.
+        :type span_datas: list of :class:
+            `~opencensus.trace.span_data.SpanData`
+        :param list of opencensus.trace.span_data.SpanData span_datas:
+            SpanData tuples to emit
         """
+        # convert to the legacy trace json for easier refactoring
+        # TODO: refactor this to use the span data directly
+        trace = span_data.format_legacy_trace_json(span_datas)
+
         trace_id = trace.get('traceId')
         spans = trace.get('spans')
 
@@ -113,8 +121,8 @@ def emit(self, trace):
         except Exception as e:  # pragma: NO COVER
             logging.error(getattr(e, 'message', e))
 
-    def export(self, trace):
-        self.transport.export(trace)
+    def export(self, span_datas):
+        self.transport.export(span_datas)
 
     def translate_to_zipkin(self, trace_id, spans):
         """Translate the opencensus spans to zipkin spans.

opencensus/trace/span.py

@@ -25,6 +25,12 @@
 from opencensus.trace.utils import _get_truncatable_str
 
 
+class SpanKind(object):
+    UNSPECIFIED = 0
+    SERVER = 1
+    CLIENT = 2
+
+
 class Span(object):
     """A span is an individual timed event which forms a node of the trace
     tree. Each span has its name, span id and parent id. The parent id
@@ -85,6 +91,11 @@ class Span(object):
                            method in the tracer class to finish a span. If no
                            tracer is passed in, then just finish the span using
                            the finish method in the Span class.
+
+    :type span_kind: int
+    :param span_kind: (Optional) Highly recommended flag that denotes the type
+                        of span (valid values defined by :class:
+                        `opencensus.trace.span.SpanKind`)
     """
 
     def __init__(
@@ -100,7 +111,8 @@ def __init__(
             links=None,
             status=None,
             same_process_as_parent_span=None,
-            context_tracer=None):
+            context_tracer=None,
+            span_kind=SpanKind.UNSPECIFIED):
         self.name = name
         self.parent_span = parent_span
         self.start_time = start_time
@@ -132,6 +144,7 @@ def __init__(
         self.same_process_as_parent_span = same_process_as_parent_span
         self._child_spans = []
         self.context_tracer = context_tracer
+        self.span_kind = span_kind
 
     @property
     def children(self):