test(profiling): unflake test_copy_memory_stats

tests/profiling/collector/pprof_utils.py

@@ -285,6 +285,17 @@ def parse_newest_profile(
     return profile
 
 
+def get_internal_metadata_files(filename_prefix: str) -> list[str]:
+    """Internal metadata files with the given prefix (which includes the pid), oldest upload first.
+
+    Files are named <filename_prefix>.<counter>.internal_metadata.json without
+    padding, so a lexicographic sort would place upload 10 before upload 2.
+    """
+    files = glob.glob(filename_prefix + ".*.internal_metadata.json")
+    files.sort(key=lambda f: int(f.rsplit(".", 3)[-3]))
+    return files
+
+
 def get_sample_type_index(profile: pprof_pb2.Profile, value_type: str) -> int:
     return next(
         i for i, sample_type in enumerate(profile.sample_type) if profile.string_table[sample_type.type] == value_type

tests/profiling/collector/test_asyncio_task_count.py

@@ -11,13 +11,13 @@
 def test_asyncio_task_count_present():
     """asyncio_task_count is present and positive when asyncio tasks are active."""
     import asyncio
-    import glob
     import json
     import os
     import time
 
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     async def worker():
         await asyncio.sleep(0.5)
@@ -38,7 +38,7 @@ async def main():
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, "Expected at least one internal_metadata.json file"
 
     found_positive = False
@@ -68,13 +68,13 @@ def test_asyncio_task_count_survives_run_teardown():
     loop was live, so the uploaded profile reported asyncio_task_count: 0 when it should have been the peak.
     """
     import asyncio
-    import glob
     import json
     import os
     import time
 
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     NUM_WORKERS = 10
     EXPECTED_PEAK = NUM_WORKERS + 1
@@ -98,7 +98,7 @@ async def main():
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, "Expected internal_metadata.json file"
 
     peak = 0

tests/profiling/collector/test_copy_memory_stats.py

@@ -10,21 +10,21 @@
 )
 def test_copy_memory_error_count_present():
     """copy_memory_error_count is always emitted (even when 0) and is non-negative."""
-    import glob
     import json
     import os
     import time
 
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     p = profiler.Profiler(tracer=tracer)
     p.start()
     time.sleep(3)
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, "Expected at least one internal_metadata.json file"
 
     for f in files:
@@ -51,21 +51,21 @@ def test_copy_memory_error_count_present():
 )
 def test_fast_copy_memory_disabled():
     """fast_copy_memory_enabled is False when _DD_PROFILING_STACK_FAST_COPY=false."""
-    import glob
     import json
     import os
     import time
 
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     p = profiler.Profiler(tracer=tracer)
     p.start()
     time.sleep(3)
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, "Expected at least one internal_metadata.json file"
 
     for i, f in enumerate(files):
@@ -91,7 +91,6 @@ def test_fast_copy_memory_disabled():
 )
 def test_fast_copy_memory_enabled() -> None:
     """Sampler runs on the syscall copy during warmup, then upgrades to safe_memcpy (PROF-14568)."""
-    import glob
     import json
     import os
     import time
@@ -100,6 +99,7 @@ def test_fast_copy_memory_enabled() -> None:
     from ddtrace.internal.datadog.profiling.stack import _stack
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     _stack._set_fast_copy_warmup_seconds(1.0)
 
@@ -121,18 +121,31 @@ def test_fast_copy_memory_enabled() -> None:
             break
         time.sleep(0.05)
 
+    assert saw_warmup, "Expected the sampler to run on the syscall copy during the warmup window"
+    assert saw_upgrade, "Expected the sampler to upgrade to safe_memcpy after warmup"
+
+    # The upgrade flips the flag at the start of a sampling cycle, but the stats it
+    # feeds are only written at the end of that cycle, so stopping here could flush a
+    # window that never saw the upgrade. Wait out an upload interval instead.
+    time.sleep(2)
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, "Expected at least one internal_metadata.json file"
 
-    with open(files[-1]) as fp:
-        metadata = json.load(fp)
+    # A window with no completed sampling cycle inherits the previous window's fast-copy
+    # state, so it says nothing about what the sampler is running on.
+    metadata = None
+    for f in reversed(files):
+        with open(f) as fp:
+            candidate = json.load(fp)
+        if candidate["sampling_event_count"] > 0:
+            metadata = candidate
+            break
+    assert metadata is not None, f"Expected an upload window with at least one sampling cycle: {files}"
+
     assert metadata["fast_copy_memory_user_disabled"] is False, metadata
     assert metadata["fast_copy_memory_capable"] is True, metadata
     assert metadata["fast_copy_memory_syscall_fallback"] is False, metadata
     assert metadata["fast_copy_memory_enabled"] is True, metadata
-
-    assert saw_warmup, "Expected the sampler to run on the syscall copy during the warmup window"
-    assert saw_upgrade, "Expected the sampler to upgrade to safe_memcpy after warmup"

tests/profiling/collector/test_greenlet_count.py

@@ -26,7 +26,6 @@ def test_greenlet_count_present():
 
     monkey.patch_all()
 
-    import glob
     import json
     import os
     import time
@@ -35,6 +34,7 @@ def test_greenlet_count_present():
 
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     stop = False
 
@@ -53,7 +53,7 @@ def worker():
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, "Expected at least one internal_metadata.json file"
 
     found_positive = False

tests/profiling/collector/test_heap_tracker_count.py

@@ -11,13 +11,13 @@
 )
 def test_heap_tracker_count_present():
     """heap_tracker_count is present and non-zero when memory profiling is enabled."""
-    import glob
     import json
     import os
     import time
 
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     p = profiler.Profiler(tracer=tracer)
     p.start()
@@ -33,7 +33,7 @@ def test_heap_tracker_count_present():
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, "Expected at least one internal_metadata.json file"
 
     for f in files:

tests/profiling/collector/test_internal_adaptive_sampling.py

@@ -13,7 +13,6 @@
 )
 def test_internal_adaptive_sampling():
     import asyncio
-    import glob
     import json
     import os
     import time
@@ -22,6 +21,7 @@ def test_internal_adaptive_sampling():
     from ddtrace import ext
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     sleep_time = 0.2
     loop_run_time = 4
@@ -52,7 +52,7 @@ async def hello():
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
 
     # With adaptive sampling enabled, the sampling interval can grow up to 1 second
     # (g_max_sampling_period_us). Since the upload interval is also 1 second, the

tests/profiling/collector/test_sample_count.py

@@ -10,7 +10,6 @@
 )
 def test_sample_count():
     import asyncio
-    import glob
     import json
     import os
     import time
@@ -19,6 +18,7 @@ def test_sample_count():
     from ddtrace import ext
     from ddtrace.profiling import profiler
     from ddtrace.trace import tracer
+    from tests.profiling.collector import pprof_utils
 
     sleep_time = 0.2
     loop_run_time = 2
@@ -49,7 +49,7 @@ async def hello():
     p.stop()
 
     output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
 
     found_at_least_one_with_more_samples_than_sampling_events = False
     for i, f in enumerate(files):

tests/profiling/collector/test_thread_subsampling.py

@@ -19,14 +19,14 @@
 )
 def test_thread_subsampling_cap_respected() -> None:
     """With max_threads=1, at most 1 thread is sampled per cycle, even with many threads alive."""
-    import glob
     import json
     import os
     import threading
     import time
 
     from ddtrace.internal.datadog.profiling import ddup
     from ddtrace.profiling.collector import stack
+    from tests.profiling.collector import pprof_utils
 
     N_THREADS = 10
     max_threads = 1
@@ -59,7 +59,7 @@ def worker() -> None:
     ddup.upload()
 
     output_filename = pprof_prefix + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, f"No internal metadata files found at {output_filename}.*"
 
     for f in files:
@@ -93,14 +93,14 @@ def test_thread_subsampling_all_threads_sampled_without_cap() -> None:
     With N_THREADS additional threads running, sample_count should be
     significantly greater than sampling_event_count.
     """
-    import glob
     import json
     import os
     import threading
     import time
 
     from ddtrace.internal.datadog.profiling import ddup
     from ddtrace.profiling.collector import stack
+    from tests.profiling.collector import pprof_utils
 
     N_THREADS = 10
 
@@ -132,7 +132,7 @@ def worker() -> None:
     ddup.upload()
 
     output_filename = pprof_prefix + "." + str(os.getpid())
-    files = sorted(glob.glob(output_filename + ".*.internal_metadata.json"))
+    files = pprof_utils.get_internal_metadata_files(output_filename)
     assert files, f"No internal metadata files found at {output_filename}.*"
 
     total_events: int = 0