Merge remote-tracking branch 'upstream/main' into span_compression

Author: basepi

CHANGELOG.asciidoc

@@ -47,6 +47,7 @@ endif::[]
 * Lambda fixes to align with the cross-agent spec {pull}1489[#1489]
 * Lambda fix for custom `service_name` {pull}1493[#1493]
 * Change default for `stack_trace_limit` from 500 to 50 {pull}1492[#1492]
+* Switch all duration handling to use `datetime.timedelta` objects {pull}1488[#1488]
 
 
 [[release-notes-6.x]]

elasticapm/base.py

@@ -43,6 +43,7 @@
 import urllib.parse
 import warnings
 from copy import deepcopy
+from datetime import timedelta
 from typing import Optional, Tuple
 
 import elasticapm
@@ -307,6 +308,8 @@ def end_transaction(self, name=None, result="", duration=None):
  :param duration: override duration, mostly useful for testing
  :return: the ended transaction object
  """
+ if duration is not None and not isinstance(duration, timedelta):
+ duration = timedelta(seconds=duration)
  transaction = self.tracer.end_transaction(result, name, duration=duration)
  return transaction
 

elasticapm/conf/__init__.py

@@ -36,6 +36,7 @@
 import re
 import socket
 import threading
+from datetime import timedelta
 
 from elasticapm.conf.constants import BASE_SANITIZE_FIELD_NAMES
 from elasticapm.utils import compat, starmatch_to_regex
@@ -232,6 +233,39 @@ def __set__(self, instance, value):
  instance._values[self.dict_key] = bool(value)
 
 
+class _DurationConfigValue(_ConfigValue):
+ units = (
+ ("us", 0.000001),
+ ("ms", 0.001),
+ ("s", 1),
+ ("m", 60),
+ )
+
+ def __init__(self, dict_key, allow_microseconds=False, unitless_factor=None, **kwargs):
+ self.type = None # no type coercion
+ used_units = self.units if allow_microseconds else self.units[1:]
+ pattern = "|".join(unit[0] for unit in used_units)
+ unit_multipliers = dict(used_units)
+ unit_required = ""
+ if unitless_factor:
+ unit_multipliers[None] = unitless_factor
+ unit_required = "?"
+ regex = f"((?:-)?\\d+)({pattern}){unit_required}$"
+ verbose_pattern = f"\\d+({pattern}){unit_required}$"
+ duration_validator = UnitValidator(
+ regex=regex, verbose_pattern=verbose_pattern, unit_multipliers=unit_multipliers
+ )
+ validators = kwargs.pop("validators", [])
+ validators.insert(0, duration_validator)
+ super().__init__(dict_key, validators=validators, **kwargs)
+
+ def __set__(self, config_instance, value):
+ value = self._validate(config_instance, value)
+ value = timedelta(seconds=float(value))
+ self._callback_if_changed(config_instance, value)
+ config_instance._values[self.dict_key] = value
+
+
 class RegexValidator(object):
  def __init__(self, regex, verbose_pattern=None):
  self.regex = regex
@@ -283,16 +317,13 @@ def __call__(self, value, field_name):
  value = float(value)
  except ValueError:
  raise ConfigurationError("{} is not a float".format(value), field_name)
- multiplier = 10 ** self.precision
+ multiplier = 10**self.precision
  rounded = math.floor(value * multiplier + 0.5) / multiplier
  if rounded == 0 and self.minimum and value != 0:
  rounded = self.minimum
  return rounded
 
 
-duration_validator = UnitValidator(
- r"^((?:-)?\d+)(us|ms|s|m)$", r"\d+(us|ms|s|m)", {"us": 0.001, "ms": 1, "s": 1000, "m": 60000}
-)
 size_validator = UnitValidator(
  r"^(\d+)(b|kb|mb|gb)$", r"\d+(b|KB|MB|GB)", {"b": 1, "kb": 1024, "mb": 1024 * 1024, "gb": 1024 * 1024 * 1024}
 )
@@ -557,50 +588,38 @@ class Config(_ConfigBase):
  "elasticapm.metrics.sets.cpu.CPUMetricSet",
  ],
  )
- metrics_interval = _ConfigValue(
+ metrics_interval = _DurationConfigValue(
  "METRICS_INTERVAL",
- type=int,
- validators=[duration_validator, ExcludeRangeValidator(1, 999, "{range_start} - {range_end} ms")],
- default=30000,
+ validators=[ExcludeRangeValidator(0.001, 0.999, "{range_start} - {range_end} s")],
+ default=timedelta(seconds=30),
  )
  breakdown_metrics = _BoolConfigValue("BREAKDOWN_METRICS", default=True)
  prometheus_metrics = _BoolConfigValue("PROMETHEUS_METRICS", default=False)
  prometheus_metrics_prefix = _ConfigValue("PROMETHEUS_METRICS_PREFIX", default="prometheus.metrics.")
  disable_metrics = _ListConfigValue("DISABLE_METRICS", type=starmatch_to_regex, default=[])
  central_config = _BoolConfigValue("CENTRAL_CONFIG", default=True)
  api_request_size = _ConfigValue("API_REQUEST_SIZE", type=int, validators=[size_validator], default=768 * 1024)
- api_request_time = _ConfigValue("API_REQUEST_TIME", type=int, validators=[duration_validator], default=10 * 1000)
+ api_request_time = _DurationConfigValue("API_REQUEST_TIME", default=timedelta(seconds=10))
  transaction_sample_rate = _ConfigValue(
  "TRANSACTION_SAMPLE_RATE", type=float, validators=[PrecisionValidator(4, 0.0001)], default=1.0
  )
  transaction_max_spans = _ConfigValue("TRANSACTION_MAX_SPANS", type=int, default=500)
  stack_trace_limit = _ConfigValue("STACK_TRACE_LIMIT", type=int, default=50)
- span_frames_min_duration = _ConfigValue(
- "SPAN_FRAMES_MIN_DURATION",
- default=5,
- validators=[
- UnitValidator(r"^((?:-)?\d+)(ms|s|m)?$", r"\d+(ms|s|m)", {"ms": 1, "s": 1000, "m": 60000, None: 1})
- ],
- type=int,
+ span_frames_min_duration = _DurationConfigValue(
+ "SPAN_FRAMES_MIN_DURATION", default=timedelta(seconds=0.005), unitless_factor=0.001
  )
  span_compression_enabled = _BoolConfigValue("SPAN_COMPRESSION_ENABLED", default=True)
- span_compression_exact_match_max_duration = _ConfigValue(
+ span_compression_exact_match_max_duration = _DurationConfigValue(
  "SPAN_COMPRESSION_EXACT_MATCH_MAX_DURATION",
- default=50,
- validators=[duration_validator],
- type=int,
+ default=timedelta(seconds=0.05),
  )
- span_compression_same_kind_max_duration = _ConfigValue(
+ span_compression_same_kind_max_duration = _DurationConfigValue(
  "SPAN_COMPRESSION_SAME_KIND_MAX_DURATION",
- default=0,
- validators=[duration_validator],
- type=int,
+ default=timedelta(seconds=0),
  )
- exit_span_min_duration = _ConfigValue(
+ exit_span_min_duration = _DurationConfigValue(
  "EXIT_SPAN_MIN_DURATION",
- default=0,
- validators=[duration_validator],
- type=float,
+ default=timedelta(seconds=0),
  )
  collect_local_variables = _ConfigValue("COLLECT_LOCAL_VARIABLES", default="errors")
  source_lines_error_app_frames = _ConfigValue("SOURCE_LINES_ERROR_APP_FRAMES", type=int, default=5)

elasticapm/metrics/base_metrics.py

@@ -107,7 +107,7 @@ def stop_thread(self):
 
  @property
  def collect_interval(self):
- return self.client.config.metrics_interval / 1000.0
+ return self.client.config.metrics_interval.total_seconds()
 
  @property
  def ignore_patterns(self):
@@ -391,8 +391,8 @@ class Timer(BaseMetric):
  __slots__ = BaseMetric.__slots__ + ("_val", "_count", "_lock", "_unit")
 
  def __init__(self, name=None, reset_on_collect=False, unit=None):
- self._val = 0
- self._count = 0
+ self._val: float = 0
+ self._count: int = 0
  self._unit = unit
  self._lock = threading.Lock()
  super(Timer, self).__init__(name, reset_on_collect=reset_on_collect)

elasticapm/traces.py

@@ -35,6 +35,7 @@
 import time
 import timeit
 from collections import defaultdict
+from datetime import timedelta
 from types import TracebackType
 from typing import Any, Callable, Dict, Optional, Tuple, Type, Union
 
@@ -68,7 +69,7 @@ def __init__(self, obj: "BaseSpan"):
  self.obj = obj
  self._nesting_level: int = 0
  self._start: float = 0
- self._duration: float = 0
+ self._duration: timedelta = timedelta(seconds=0)
  self._lock = threading.Lock()
 
  def start(self, timestamp: float):
@@ -81,10 +82,10 @@ def stop(self, timestamp: float):
  with self._lock:
  self._nesting_level -= 1
  if self._nesting_level == 0:
- self._duration += timestamp - self._start
+ self._duration += timedelta(seconds=timestamp - self._start)
 
  @property
- def duration(self) -> float:
+ def duration(self) -> timedelta:
  return self._duration
 
 
@@ -97,7 +98,7 @@ def __init__(self, labels=None, start=None):
  self.compression_buffer_lock = threading.Lock()
  self.start_time: float = time_to_perf_counter(start) if start is not None else _time_func()
  self.ended_time: Optional[float] = None
- self.duration: Optional[float] = None
+ self.duration: Optional[timedelta] = None
  if labels:
  self.label(**labels)
 
@@ -117,9 +118,9 @@ def child_ended(self, child: SpanType):
  self.compression_buffer.report()
  self.compression_buffer = child
 
- def end(self, skip_frames: int = 0, duration: Optional[float] = None):
+ def end(self, skip_frames: int = 0, duration: Optional[timedelta] = None):
  self.ended_time = _time_func()
- self.duration = duration if duration is not None else (self.ended_time - self.start_time)
+ self.duration = duration if duration is not None else timedelta(seconds=self.ended_time - self.start_time)
  if self.compression_buffer:
  self.compression_buffer.report()
  self.compression_buffer = None
@@ -226,7 +227,7 @@ def __init__(
  self._breakdown = None
  super(Transaction, self).__init__(start=start)
 
- def end(self, skip_frames: int = 0, duration: Optional[float] = None):
+ def end(self, skip_frames: int = 0, duration: Optional[timedelta] = None):
  super().end(skip_frames, duration)
  if self._breakdown:
  for (span_type, span_subtype), timer in self._span_timers.items():
@@ -239,15 +240,15 @@ def end(self, skip_frames: int = 0, duration: Optional[float] = None):
  labels["span.subtype"] = span_subtype
  val = timer.val
  self._breakdown.timer("span.self_time", reset_on_collect=True, unit="us", **labels).update(
- int(val[0] * 1000000), val[1]
+ val[0], val[1]
  )
  if self.is_sampled:
  self._breakdown.timer(
  "span.self_time",
  reset_on_collect=True,
  unit="us",
  **{"span.type": "app", "transaction.name": self.name, "transaction.type": self.transaction_type},
- ).update(int((self.duration - self._child_durations.duration) * 1000000))
+ ).update((self.duration - self._child_durations.duration).total_seconds() * 1_000_000)
 
  def _begin_span(
  self,
@@ -369,9 +370,9 @@ def to_dict(self) -> dict:
  "trace_id": self.trace_parent.trace_id,
  "name": encoding.keyword_field(self.name or ""),
  "type": encoding.keyword_field(self.transaction_type),
- "duration": self.duration * 1000, # milliseconds
+ "duration": self.duration.total_seconds() * 1000,
  "result": encoding.keyword_field(str(self.result)),
- "timestamp": int(self.timestamp * 1000000), # microseconds
+ "timestamp": int(self.timestamp * 1_000_000), # microseconds
  "outcome": self.outcome,
  "sampled": self.is_sampled,
  "span_count": {"started": self._span_counter, "dropped": self.dropped_spans},
@@ -381,7 +382,7 @@ def to_dict(self) -> dict:
  {
  "destination_service_resource": resource,
  "outcome": outcome,
- "duration": {"count": v["count"], "sum": {"us": int(v["duration.sum.us"] * 1000000)}},
+ "duration": {"count": v["count"], "sum": {"us": int(v["duration.sum.us"])}},
  }
  for (resource, outcome), v in self._dropped_span_statistics.items()
  ]
@@ -419,7 +420,7 @@ def track_span_duration(self, span_type, span_subtype, self_duration):
  # TODO: once asynchronous spans are supported, we should check if the transaction is already finished
  # TODO: and, if it has, exit without tracking.
  with self._span_timers_lock:
- self._span_timers[(span_type, span_subtype)].update(self_duration)
+ self._span_timers[(span_type, span_subtype)].update(self_duration.total_seconds() * 1_000_000)
 
  @property
  def is_sampled(self) -> bool:
@@ -448,7 +449,7 @@ def track_dropped_span(self, span: SpanType):
  resource = span.context["destination"]["service"]["resource"]
  stats = self._dropped_span_statistics[(resource, span.outcome)]
  stats["count"] += 1
- stats["duration.sum.us"] += span.duration
+ stats["duration.sum.us"] += int(span.duration.total_seconds() * 1_000_000)
  except KeyError:
  pass
 
@@ -551,7 +552,7 @@ def to_dict(self) -> dict:
  "subtype": encoding.keyword_field(self.subtype),
  "action": encoding.keyword_field(self.action),
  "timestamp": int(self.timestamp * 1000000), # microseconds
- "duration": self.duration * 1000, # milliseconds
+ "duration": self.duration.total_seconds() * 1000,
  "outcome": self.outcome,
  }
  if self.transaction.sample_rate is not None:
@@ -586,7 +587,7 @@ def to_dict(self) -> dict:
  if self.composite:
  result["composite"] = {
  "compression_strategy": self.composite["compression_strategy"],
- "sum": self.composite["sum"] * 1000,
+ "sum": self.composite["sum"].total_seconds() * 1000,
  "count": self.composite["count"],
  }
  return result
@@ -650,14 +651,14 @@ def end(self, skip_frames: int = 0, duration: Optional[float] = None):
 
  p = self.parent if self.parent else self.transaction
  if self.transaction._breakdown:
- p._child_durations.stop(self.start_time + self.duration)
+ p._child_durations.stop(self.start_time + self.duration.total_seconds())
  self.transaction.track_span_duration(
  self.type, self.subtype, self.duration - self._child_durations.duration
  )
  p.child_ended(self)
 
  def report(self) -> None:
- if self.discardable and (self.duration * 1000) < self.transaction.config_exit_span_min_duration:
+ if self.discardable and self.duration < self.transaction.config_exit_span_min_duration:
  self.transaction.track_dropped_span(self)
  self.transaction.dropped_spans += 1
  else:
@@ -674,7 +675,7 @@ def try_to_compress(self, sibling: SpanType) -> bool:
  self.composite = {"compression_strategy": compression_strategy, "count": 1, "sum": self.duration}
  self.composite["count"] += 1
  self.composite["sum"] += sibling.duration
- self.duration = sibling.ended_time - self.start_time
+ self.duration = timedelta(seconds=sibling.ended_time - self.start_time)
  self.transaction._span_counter -= 1
  return True
 
@@ -820,11 +821,11 @@ def __init__(self, frames_collector_func, frames_processing_func, queue_func, co
  self._ignore_patterns = [re.compile(p) for p in config.transactions_ignore_patterns or []]
 
  @property
- def span_frames_min_duration(self):
- if self.config.span_frames_min_duration in (-1, None):
+ def span_frames_min_duration(self) -> Optional[timedelta]:
+ if self.config.span_frames_min_duration in (timedelta(seconds=-1), None):
  return None
  else:
- return self.config.span_frames_min_duration / 1000.0
+ return self.config.span_frames_min_duration
 
  def begin_transaction(self, transaction_type, trace_parent=None, start=None, auto_activate=True):
  """
@@ -918,7 +919,7 @@ def __init__(
  span_subtype: Optional[str] = None,
  span_action: Optional[str] = None,
  start: Optional[int] = None,
- duration: Optional[float] = None,
+ duration: Optional[Union[float, timedelta]] = None,
  sync: Optional[bool] = None,
  ):
  self.name = name
@@ -937,6 +938,8 @@ def __init__(
  self.leaf = leaf
  self.labels = labels
  self.start = start
+ if duration and not isinstance(duration, timedelta):
+ duration = timedelta(seconds=duration)
  self.duration = duration
  self.sync = sync