added duration argument to some tracing APIs (#550)

this fixes the breakdown tests on Windows by avoiding time.sleep(),
which has an insufficient resolution on Windows.

fixes #537
closes #550

Author: beniwohli

CHANGELOG.md

@@ -5,6 +5,7 @@
 
 ### Bugfixes
  * fixed an issue with empty responses from APM Server's config endpoint (#562, #563)
+ * fixed Windows tests by avoiding time.sleep in breakdown metrics tests (#537, #550)
 
 ## v5.1.0 
 [Check the diff](https://github.com/elastic/apm-agent-python/compare/v5.0.0...v5.1.0)

elasticapm/base.py

@@ -235,13 +235,27 @@ def queue(self, event_type, data, flush=False):
             flush = False
         self._transport.queue(event_type, data, flush)
 
-    def begin_transaction(self, transaction_type, trace_parent=None):
-        """Register the start of a transaction on the client
+    def begin_transaction(self, transaction_type, trace_parent=None, start=None):
         """
-        return self.tracer.begin_transaction(transaction_type, trace_parent=trace_parent)
+        Register the start of a transaction on the client
 
-    def end_transaction(self, name=None, result=""):
-        transaction = self.tracer.end_transaction(result, name)
+        :param transaction_type: type of the transaction, e.g. "request"
+        :param trace_parent: an optional TraceParent object for distributed tracing
+        :param start: override the start timestamp, mostly useful for testing
+        :return: the started transaction object
+        """
+        return self.tracer.begin_transaction(transaction_type, trace_parent=trace_parent, start=start)
+
+    def end_transaction(self, name=None, result="", duration=None):
+        """
+        End the current transaction.
+
+        :param name: optional name of the transaction
+        :param result: result of the transaction, e.g. "OK" or "HTTP 2xx"
+        :param duration: override duration, mostly useful for testing
+        :return: the ended transaction object
+        """
+        transaction = self.tracer.end_transaction(result, name, duration=duration)
         return transaction
 
     def close(self):

elasticapm/traces.py

@@ -97,7 +97,7 @@ def child_started(self, timestamp):
     def child_ended(self, timestamp):
         self._child_durations.stop(timestamp)
 
-    def end(self, skip_frames=0):
+    def end(self, skip_frames=0, duration=None):
         raise NotImplementedError()
 
     def label(self, **labels):
@@ -134,10 +134,13 @@ def tag(self, **tags):
 
 
 class Transaction(BaseSpan):
-    def __init__(self, tracer, transaction_type="custom", trace_parent=None, is_sampled=True):
+    def __init__(self, tracer, transaction_type="custom", trace_parent=None, is_sampled=True, start=None):
         self.id = "%016x" % random.getrandbits(64)
         self.trace_parent = trace_parent
-        self.timestamp, self.start_time = time.time(), _time_func()
+        if start:
+            self.timestamp = self.start_time = start
+        else:
+            self.timestamp, self.start_time = time.time(), _time_func()
         self.name = None
         self.duration = None
         self.result = None
@@ -165,8 +168,8 @@ def __init__(self, tracer, transaction_type="custom", trace_parent=None, is_samp
             self._transaction_metrics = None
         super(Transaction, self).__init__()
 
-    def end(self, skip_frames=0):
-        self.duration = _time_func() - self.start_time
+    def end(self, skip_frames=0, duration=None):
+        self.duration = duration if duration is not None else (_time_func() - self.start_time)
         if self._transaction_metrics:
             self._transaction_metrics.timer(
                 "transaction.duration",
@@ -202,6 +205,7 @@ def _begin_span(
         parent_span_id=None,
         span_subtype=None,
         span_action=None,
+        start=None,
     ):
         parent_span = execution_context.get_span()
         tracer = self.tracer
@@ -223,13 +227,16 @@ def _begin_span(
                 parent_span_id=parent_span_id,
                 span_subtype=span_subtype,
                 span_action=span_action,
+                start=start,
             )
             span.frames = tracer.frames_collector_func()
             self._span_counter += 1
         execution_context.set_span(span)
         return span
 
-    def begin_span(self, name, span_type, context=None, leaf=False, labels=None, span_subtype=None, span_action=None):
+    def begin_span(
+        self, name, span_type, context=None, leaf=False, labels=None, span_subtype=None, span_action=None, start=None
+    ):
         """
         Begin a new span
         :param name: name of the span
@@ -239,6 +246,7 @@ def begin_span(self, name, span_type, context=None, leaf=False, labels=None, spa
         :param labels: a flat string/string dict of labels
         :param span_subtype: sub type of the span, e.g. "postgresql"
         :param span_action: action of the span , e.g. "query"
+        :param start: timestamp, mostly useful for testing
         :return: the Span object
         """
         return self._begin_span(
@@ -250,19 +258,21 @@ def begin_span(self, name, span_type, context=None, leaf=False, labels=None, spa
             parent_span_id=None,
             span_subtype=span_subtype,
             span_action=span_action,
+            start=start,
         )
 
-    def end_span(self, skip_frames=0):
+    def end_span(self, skip_frames=0, duration=None):
         """
         End the currently active span
         :param skip_frames: numbers of frames to skip in the stack trace
+        :param duration: override duration, mostly useful for testing
         :return: the ended span
         """
         span = execution_context.get_span()
         if span is None:
             raise LookupError()
 
-        span.end(skip_frames=skip_frames)
+        span.end(skip_frames=skip_frames, duration=duration)
         return span
 
     def ensure_parent_id(self):
@@ -337,6 +347,7 @@ def __init__(
         parent_span_id=None,
         span_subtype=None,
         span_action=None,
+        start=None,
     ):
         """
         Create a new Span
@@ -350,8 +361,9 @@ def __init__(
         :param parent_span_id: override of the span ID
         :param span_subtype: sub type of the span, e.g. mysql
         :param span_action: sub type of the span, e.g. query
+        :param start: timestamp, mostly useful for testing
         """
-        self.start_time = _time_func()
+        self.start_time = start or _time_func()
         self.id = "%016x" % random.getrandbits(64)
         self.transaction = transaction
         self.name = name
@@ -405,10 +417,17 @@ def to_dict(self):
             result["stacktrace"] = self.frames
         return result
 
-    def end(self, skip_frames=0):
+    def end(self, skip_frames=0, duration=None):
+        """
+        End this span and queue it for sending.
+
+        :param skip_frames: amount of frames to skip from the beginning of the stack trace
+        :param duration: override duration, mostly useful for testing
+        :return: None
+        """
         tracer = self.transaction.tracer
         timestamp = _time_func()
-        self.duration = timestamp - self.start_time
+        self.duration = duration if duration is not None else (timestamp - self.start_time)
         if not tracer.span_frames_min_duration or self.duration >= tracer.span_frames_min_duration:
             self.frames = tracer.frames_processing_func(self.frames)[skip_frames:]
         else:
@@ -417,7 +436,7 @@ def end(self, skip_frames=0):
         tracer.queue_func(SPAN, self.to_dict())
         if self.transaction._breakdown:
             p = self.parent if self.parent else self.transaction
-            p.child_ended(timestamp)
+            p.child_ended(self.start_time + self.duration)
             self.transaction.track_span_duration(
                 self.type, self.subtype, self.duration - self._child_durations.duration
             )
@@ -434,7 +453,7 @@ def __init__(self, parent, leaf=False):
         self.leaf = leaf
         super(DroppedSpan, self).__init__()
 
-    def end(self, skip_frames=0):
+    def end(self, skip_frames=0, duration=None):
         execution_context.set_span(self.parent)
 
     def child_started(self, timestamp):
@@ -458,10 +477,14 @@ def __init__(self, frames_collector_func, frames_processing_func, queue_func, co
         else:
             self.span_frames_min_duration = config.span_frames_min_duration / 1000.0
 
-    def begin_transaction(self, transaction_type, trace_parent=None):
+    def begin_transaction(self, transaction_type, trace_parent=None, start=None):
         """
         Start a new transactions and bind it in a thread-local variable
 
+        :param transaction_type: type of the transaction, e.g. "request"
+        :param trace_parent: an optional TraceParent object
+        :param start: override the start timestamp, mostly useful for testing
+
         :returns the Transaction object
         """
         if trace_parent:
@@ -470,7 +493,7 @@ def begin_transaction(self, transaction_type, trace_parent=None):
             is_sampled = (
                 self.config.transaction_sample_rate == 1.0 or self.config.transaction_sample_rate > random.random()
             )
-        transaction = Transaction(self, transaction_type, trace_parent=trace_parent, is_sampled=is_sampled)
+        transaction = Transaction(self, transaction_type, trace_parent=trace_parent, is_sampled=is_sampled, start=start)
         if trace_parent is None:
             transaction.trace_parent = TraceParent(
                 constants.TRACE_CONTEXT_VERSION,
@@ -487,12 +510,19 @@ def _should_ignore(self, transaction_name):
                 return True
         return False
 
-    def end_transaction(self, result=None, transaction_name=None):
+    def end_transaction(self, result=None, transaction_name=None, duration=None):
+        """
+        End the current transaction and queue it for sending
+        :param result: result of the transaction, e.g. "OK" or 200
+        :param transaction_name: name of the transaction
+        :param duration: override duration, mostly useful for testing
+        :return:
+        """
         transaction = execution_context.get_transaction(clear=True)
         if transaction:
             if transaction.name is None:
                 transaction.name = transaction_name if transaction_name is not None else ""
-            transaction.end()
+            transaction.end(duration=duration)
             if self._should_ignore(transaction.name):
                 return
             if transaction.result is None:
@@ -502,7 +532,7 @@ def end_transaction(self, result=None, transaction_name=None):
 
 
 class capture_span(object):
-    __slots__ = ("name", "type", "subtype", "action", "extra", "skip_frames", "leaf", "labels")
+    __slots__ = ("name", "type", "subtype", "action", "extra", "skip_frames", "leaf", "labels", "duration", "start")
 
     def __init__(
         self,
@@ -515,6 +545,8 @@ def __init__(
         labels=None,
         span_subtype=None,
         span_action=None,
+        start=None,
+        duration=None,
     ):
         self.name = name
         self.type = span_type
@@ -532,6 +564,8 @@ def __init__(
             labels = tags
 
         self.labels = labels
+        self.start = start
+        self.duration = duration
 
     def __call__(self, func):
         self.name = self.name or get_name_from_func(func)
@@ -554,13 +588,14 @@ def __enter__(self):
                 labels=self.labels,
                 span_subtype=self.subtype,
                 span_action=self.action,
+                start=self.start,
             )
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         transaction = execution_context.get_transaction()
         if transaction and transaction.is_sampled:
             try:
-                transaction.end_span(self.skip_frames)
+                transaction.end_span(self.skip_frames, duration=self.duration)
             except LookupError:
                 logger.info("ended non-existing span %s of type %s", self.name, self.type)