fix(profiling): call before_flush() before calling ddup.upload() (#11258)

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Author: taegyunkim

ddtrace/profiling/collector/memalloc.py

@@ -4,6 +4,7 @@
 import os
 import threading
 import typing  # noqa:F401
+from typing import Optional
 
 
 try:
@@ -69,20 +70,22 @@ def __init__(
         self,
         recorder: Recorder,
         _interval: float = _DEFAULT_INTERVAL,
-        _max_events: int = config.memory.events_buffer,
-        max_nframe: int = config.max_frames,
-        heap_sample_size: int = config.heap.sample_size,
-        ignore_profiler: bool = config.ignore_profiler,
-        _export_libdd_enabled: bool = config.export.libdd_enabled,
+        _max_events: Optional[int] = None,
+        max_nframe: Optional[int] = None,
+        heap_sample_size: Optional[int] = None,
+        ignore_profiler: Optional[bool] = None,
+        _export_libdd_enabled: Optional[bool] = None,
     ):
         super().__init__(recorder=recorder)
         self._interval: float = _interval
         # TODO make this dynamic based on the 1. interval and 2. the max number of events allowed in the Recorder
-        self._max_events: int = _max_events
-        self.max_nframe: int = max_nframe
-        self.heap_sample_size: int = heap_sample_size
-        self.ignore_profiler: bool = ignore_profiler
-        self._export_libdd_enabled: bool = _export_libdd_enabled
+        self._max_events: int = _max_events if _max_events is not None else config.memory.events_buffer
+        self.max_nframe: int = max_nframe if max_nframe is not None else config.max_frames
+        self.heap_sample_size: int = heap_sample_size if heap_sample_size is not None else config.heap.sample_size
+        self.ignore_profiler: bool = ignore_profiler if ignore_profiler is not None else config.ignore_profiler
+        self._export_libdd_enabled: bool = (
+            _export_libdd_enabled if _export_libdd_enabled is not None else config.export.libdd_enabled
+        )
 
     def _start_service(self):
         # type: (...) -> None

ddtrace/profiling/scheduler.py

@@ -52,6 +52,12 @@ def flush(self):
         # type: (...) -> None
         """Flush events from recorder to exporters."""
         LOG.debug("Flushing events")
+        if self.before_flush is not None:
+            try:
+                self.before_flush()
+            except Exception:
+                LOG.error("Scheduler before_flush hook failed", exc_info=True)
+
         if self._export_libdd_enabled:
             ddup.upload()
 
@@ -61,11 +67,6 @@ def flush(self):
             self._last_export = compat.time_ns()
             return
 
-        if self.before_flush is not None:
-            try:
-                self.before_flush()
-            except Exception:
-                LOG.error("Scheduler before_flush hook failed", exc_info=True)
         events: EventsType = {}
         if self.recorder:
             events = self.recorder.reset()

releasenotes/notes/profiling-live-heap-libdd-49ff3a5d6405fd7e.yaml

@@ -0,0 +1,7 @@
+---
+fixes:
+  - |
+    profiling: fixes an issue where enabling native exporter via
+    ``DD_PROFILING_EXPORT_LIBDD_ENABLED``, ``DD_PROFILING_TIMELINE_ENABLED``
+    or ``DD_PROFILING_STACK_V2_ENABLED`` turned off live heap profiling.
+

tests/profiling_v2/collector/test_memalloc.py

@@ -0,0 +1,29 @@
+import os
+
+from ddtrace.profiling import Profiler
+from ddtrace.settings.profiling import config
+from tests.profiling.collector import pprof_utils
+
+
+def _allocate_1k():
+    return [object() for _ in range(1000)]
+
+
+def test_heap_samples_collected(tmp_path, monkeypatch):
+    # Test for https://github.com/DataDog/dd-trace-py/issues/11069
+    test_name = "test_heap"
+    pprof_prefix = str(tmp_path / test_name)
+    monkeypatch.setattr(config, "output_pprof", pprof_prefix)
+    monkeypatch.setattr(config, "max_frames", 32)
+    monkeypatch.setattr(config.memory, "events_buffer", 10)
+    monkeypatch.setattr(config.heap, "sample_size", 1024)
+    output_filename = pprof_prefix + "." + str(os.getpid())
+
+    p = Profiler()
+    p.start()
+    x = _allocate_1k()  # noqa: F841
+    p.stop()
+
+    profile = pprof_utils.parse_profile(output_filename)
+    samples = pprof_utils.get_samples_with_value_type(profile, "heap-space")
+    assert len(samples) > 0