fix(profiling/pprof): provide timestamp to export() (#1312)

Currently, the Exporters have to guess what the time range is for the provided
events.
This is not optimal in cases where we don't have any events — even if it
_should_ not happen in practice.

By forcing the caller of the exporter to provide the collection time range, we
can generate data that are actually more precise.

Co-authored-by: Kyle Verhoog <[email protected]>

Author: jd

ddtrace/profiling/exporter/__init__.py

@@ -10,10 +10,12 @@ class Exporter(object):
     """Exporter base class."""
 
     @staticmethod
-    def export(events):
+    def export(events, start_time_ns, end_time_ns):
         """Export events.
 
         :param events: List of events to export.
+        :param start_time_ns: The start time of recording.
+        :param end_time_ns: The end time of recording.
         """
         raise NotImplementedError
 
@@ -23,6 +25,6 @@ class NullExporter(Exporter):
     """Exporter that does nothing."""
 
     @staticmethod
-    def export(events):
+    def export(events, start_time_ns, end_time_ns):
         """Discard events."""
         pass

ddtrace/profiling/exporter/file.py

@@ -12,15 +12,17 @@ class PprofFileExporter(pprof.PprofExporter):
     prefix = attr.ib()
     _increment = attr.ib(default=1, init=False, repr=False)
 
-    def export(self, events):
+    def export(self, events, start_time_ns, end_time_ns):
         """Export events to pprof file.
 
         The file name is based on the prefix passed to init. The process ID number and type of export is then added as a
         suffix.
 
         :param events: The event dictionary from a `ddtrace.profiling.recorder.Recorder`.
+        :param start_time_ns: The start time of recording.
+        :param end_time_ns: The end time of recording.
         """
-        profile = super(PprofFileExporter, self).export(events)
+        profile = super(PprofFileExporter, self).export(events, start_time_ns, end_time_ns)
         with gzip.open(self.prefix + (".%d.%d" % (os.getpid(), self._increment)), "wb") as f:
             f.write(profile.SerializeToString())
         self._increment += 1

ddtrace/profiling/exporter/http.py

@@ -121,10 +121,12 @@ def _get_tags(service):
                     tags[key] = value
         return tags
 
-    def export(self, events):
+    def export(self, events, start_time_ns, end_time_ns):
         """Export events to an HTTP endpoint.
 
         :param events: The event dictionary from a `ddtrace.profiling.recorder.Recorder`.
+        :param start_time_ns: The start time of recording.
+        :param end_time_ns: The end time of recording.
         """
         if not self.endpoint:
             raise InvalidEndpoint("Endpoint is empty")
@@ -144,16 +146,18 @@ def export(self, events):
         }
 
         exceptions = []
-        profile = super(PprofHTTPExporter, self).export(events)
+        profile = super(PprofHTTPExporter, self).export(events, start_time_ns, end_time_ns)
         s = six.BytesIO()
         with gzip.GzipFile(fileobj=s, mode="wb") as gz:
             gz.write(profile.SerializeToString())
         fields = {
             "runtime-id": RUNTIME_ID,
             "recording-start": (
-                datetime.datetime.utcfromtimestamp(profile.time_nanos // 10e8).isoformat() + "Z"
+                datetime.datetime.utcfromtimestamp(start_time_ns).replace(microsecond=0).isoformat() + "Z"
+            ).encode(),
+            "recording-end": (
+                datetime.datetime.utcfromtimestamp(end_time_ns).replace(microsecond=0).isoformat() + "Z"
             ).encode(),
-            "recording-end": (datetime.datetime.utcnow().replace(microsecond=0).isoformat() + "Z").encode(),
             "runtime": PYTHON_IMPLEMENTATION,
             "format": b"pprof",
             "type": b"cpu+alloc+exceptions",

ddtrace/profiling/exporter/pprof.py

@@ -263,16 +263,16 @@ def max_none(a, b):
             return a
         return max(a, b)
 
-    def export(self, events):
+    def export(self, events, start_time_ns, end_time_ns):
         """Convert events to pprof format.
 
         :param events: The event dictionary from a `ddtrace.profiling.recorder.Recorder`.
-        :return: A dict where key is the type of profiling and value is the profile objects in protobuf format.
+        :param start_time_ns: The start time of recording.
+        :param end_time_ns: The end time of recording.
+        :return: A protobuf Profile object.
         """
         program_name = self._get_program_name()
 
-        start_time_ns = None
-        stop_time_ns = None
         sum_period = 0
         nb_event = 0
 
@@ -282,8 +282,6 @@ def export(self, events):
         stack_events = []
         for event in events.get(stack.StackSampleEvent, []):
             stack_events.append(event)
-            start_time_ns = self.min_none(event.timestamp, start_time_ns)
-            stop_time_ns = self.max_none(event.timestamp, stop_time_ns)
             sum_period += event.sampling_period
             nb_event += 1
 
@@ -354,8 +352,6 @@ def export(self, events):
                 traces.extend(event.snapshot.traces._traces)
                 # Assume they are all the same
                 traceback_limit = event.snapshot.traceback_limit
-                start_time_ns = self.min_none(event.timestamp, start_time_ns)
-                stop_time_ns = self.max_none(event.timestamp, stop_time_ns)
                 # Ignore period for memory events are it's not a time-based sampling
 
             if nb_events:
@@ -369,10 +365,7 @@ def export(self, events):
         else:
             period = None
 
-        if stop_time_ns is not None and start_time_ns is not None:
-            duration_ns = stop_time_ns - start_time_ns
-        else:
-            duration_ns = None
+        duration_ns = end_time_ns - start_time_ns
 
         return converter._build_profile(
             start_time_ns=start_time_ns,

ddtrace/profiling/scheduler.py

@@ -1,6 +1,7 @@
 # -*- encoding: utf-8 -*-
 import logging
 
+from ddtrace import compat
 from ddtrace.profiling import _attr
 from ddtrace.profiling import _periodic
 from ddtrace.profiling import _traceback
@@ -18,6 +19,7 @@ class Scheduler(object):
     exporters = attr.ib()
     interval = attr.ib(factory=_attr.from_env("DD_PROFILING_UPLOAD_INTERVAL", 60, float))
     _periodic = attr.ib(init=False, default=None)
+    _last_export = attr.ib(init=False, default=None)
 
     def __enter__(self):
         self.start()
@@ -29,6 +31,7 @@ def start(self):
             self.interval, self.flush, name="%s:%s" % (__name__, self.__class__.__name__)
         )
         LOG.debug("Starting scheduler")
+        self._last_export = compat.time_ns()
         self._periodic.start()
         LOG.debug("Scheduler started")
 
@@ -53,10 +56,12 @@ def flush(self):
         """Flush events from recorder to exporters."""
         LOG.debug("Flushing events")
         events = self.recorder.reset()
+        start = self._last_export
+        self._last_export = compat.time_ns()
         total_events = sum(len(v) for v in events.values())
         for exp in self.exporters:
             try:
-                exp.export(events)
+                exp.export(events, start, self._last_export)
             except exporter.ExportError as e:
                 LOG.error("Unable to export %d events: %s", total_events, _traceback.format_exception(e))
             except Exception:

tests/profiling/exporter/test_file.py

@@ -9,5 +9,5 @@
 def test_export(tmp_path):
     filename = str(tmp_path / "pprof")
     exp = file.PprofFileExporter(filename)
-    exp.export(test_pprof.TEST_EVENTS)
+    exp.export(test_pprof.TEST_EVENTS, 0, 1)
     test_main.check_pprof_file(filename + "." + str(os.getpid()) + ".1")

tests/profiling/exporter/test_http.py

@@ -63,7 +63,7 @@ def do_POST(self):
             items[part.get_param("name", header="content-disposition")].append(part.get_payload(decode=True))
         for key, check in {
             "recording-start": lambda x: x[0] == b"1970-01-01T00:00:00Z",
-            "recording-end": lambda x: x[0].startswith(b"20") and x[0].endswith(b"Z"),
+            "recording-end": lambda x: b"1970-01-01T00:00:01Z",
             "runtime": lambda x: x[0] == platform.python_implementation().encode(),
             "format": lambda x: x[0] == b"pprof",
             "type": lambda x: x[0] == b"cpu+alloc+exceptions",
@@ -139,7 +139,7 @@ def test_wrong_api_key(endpoint_test_server):
     # This is mostly testing our test server, not the exporter
     exp = http.PprofHTTPExporter(_ENDPOINT, "this is not the right API key")
     with pytest.raises(http.UploadFailed) as t:
-        exp.export(test_pprof.TEST_EVENTS)
+        exp.export(test_pprof.TEST_EVENTS, 0, 1)
         e = t.exceptions[0]
         assert isinstance(e, http.RequestFailed)
         assert e.response.status == 400
@@ -148,19 +148,19 @@ def test_wrong_api_key(endpoint_test_server):
 
 def test_export(endpoint_test_server):
     exp = http.PprofHTTPExporter(_ENDPOINT, _API_KEY)
-    exp.export(test_pprof.TEST_EVENTS)
+    exp.export(test_pprof.TEST_EVENTS, 0, 1)
 
 
 def test_export_no_endpoint(endpoint_test_server):
     exp = http.PprofHTTPExporter(endpoint="")
     with pytest.raises(http.InvalidEndpoint):
-        exp.export(test_pprof.TEST_EVENTS)
+        exp.export(test_pprof.TEST_EVENTS, 0, 1)
 
 
 def test_export_server_down():
     exp = http.PprofHTTPExporter("http://localhost:2", _API_KEY)
     with pytest.raises(http.UploadFailed) as t:
-        exp.export(test_pprof.TEST_EVENTS)
+        exp.export(test_pprof.TEST_EVENTS, 0, 1)
         e = t.exceptions[0]
         assert isinstance(e, (IOError, OSError))
         assert e.errno in (61, 99)
@@ -169,15 +169,15 @@ def test_export_server_down():
 def test_export_timeout(endpoint_test_timeout_server):
     exp = http.PprofHTTPExporter(_TIMEOUT_ENDPOINT, _API_KEY, timeout=1)
     with pytest.raises(http.UploadFailed) as t:
-        exp.export(test_pprof.TEST_EVENTS)
+        exp.export(test_pprof.TEST_EVENTS, 0, 1)
     e = t.value.exceptions[0]
     assert isinstance(e, socket.timeout)
 
 
 def test_export_reset(endpoint_test_reset_server):
     exp = http.PprofHTTPExporter(_RESET_ENDPOINT, _API_KEY, timeout=1)
     with pytest.raises(http.UploadFailed) as t:
-        exp.export(test_pprof.TEST_EVENTS)
+        exp.export(test_pprof.TEST_EVENTS, 0, 1)
     e = t.value.exceptions[0]
     if six.PY3:
         assert isinstance(e, ConnectionResetError)

tests/profiling/exporter/test_pprof.py

@@ -395,7 +395,7 @@ def test_ppprof_exporter():
     exp = pprof.PprofExporter()
     exp._get_program_name = mock.Mock()
     exp._get_program_name.return_value = "bonjour"
-    exports = exp.export(TEST_EVENTS)
+    exports = exp.export(TEST_EVENTS, 1, 7)
     if tracemalloc:
         if stack.FEATURES["stack-exceptions"]:
             filename = "test-pprof-exporter_tracemalloc+stack-exceptions.txt"
@@ -410,7 +410,7 @@ def test_ppprof_exporter():
 
 def test_pprof_exporter_empty():
     exp = pprof.PprofExporter()
-    export = exp.export({})
+    export = exp.export({}, 0, 1)
     assert len(export.sample) == 0
 
 
@@ -575,5 +575,5 @@ def test_ppprof_memory_exporter():
   unit: 8
 }
 """ == str(
-            exp.export(events)
+            exp.export(events, 1, 2)
         )