test_stack.py

import _thread
import os
from pathlib import Path
import sys
import threading
import time
from typing import TYPE_CHECKING
from typing import Generator
from unittest.mock import patch
import uuid

import pytest
from pytest import FixtureRequest
from pytest import MonkeyPatch

from ddtrace import ext
from ddtrace.internal.datadog.profiling import ddup
from ddtrace.profiling.collector import stack
from ddtrace.trace import Tracer
from tests.conftest import get_original_test_name
from tests.profiling.collector import pprof_utils
from tests.profiling.collector import test_collector


if TYPE_CHECKING:
    from tests.profiling.collector.pprof_pb2 import Sample  # pyright: ignore[reportMissingModuleSource]


# Python 3.11.9 is not compatible with gevent, https://github.com/gevent/gevent/issues/2040
# https://github.com/python/cpython/issues/117983
# The fix was not backported to 3.11. The fix was first released in 3.12.5 for
# Python 3.12. Tested with Python 3.11.8 and 3.12.5 to confirm the issue.
GEVENT_COMPATIBLE_WITH_PYTHON_VERSION = os.getenv("DD_PROFILE_TEST_GEVENT", False) and (
    sys.version_info < (3, 11, 9) or sys.version_info >= (3, 12, 5)
)


def _main_thread_has_native_id() -> bool:
    """True if main thread has native_id (False for _DummyThread which lacks _native_id)."""
    return getattr(threading.main_thread(), "native_id", None) is not None


def func1() -> None:
    return func2()


def func2() -> None:
    return func3()


def func3() -> None:
    return func4()


def func4() -> None:
    return func5()


def func5() -> None:
    return time.sleep(1)


# Use subprocess as ddup config persists across tests.
@pytest.mark.subprocess(
    env=dict(
        DD_PROFILING_MAX_FRAMES="5",
        DD_PROFILING_OUTPUT_PPROF="/tmp/test_collect_truncate",
    )
)
def test_collect_truncate() -> None:
    import os

    from ddtrace.profiling import profiler
    from tests.profiling.collector import pprof_utils
    from tests.profiling.collector.test_stack import func1

    pprof_prefix = os.environ["DD_PROFILING_OUTPUT_PPROF"]
    output_filename = pprof_prefix + "." + str(os.getpid())

    max_nframes = int(os.environ["DD_PROFILING_MAX_FRAMES"])

    p = profiler.Profiler()
    p.start()

    func1()

    p.stop()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_value_type(profile, "wall-time")
    assert len(samples) > 0
    for sample in samples:
        # stack adds one extra frame for "%d frames omitted" message
        assert len(sample.location_id) <= max_nframes + 1, len(sample.location_id)


def test_stack_locations(tmp_path: Path) -> None:
    test_name = "test_stack_locations"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    def baz() -> None:
        time.sleep(0.1)

    def bar() -> None:
        baz()

    def foo() -> None:
        bar()

    with stack.StackCollector():
        for _ in range(10):
            foo()

    ddup.upload()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_value_type(profile, "wall-time")
    assert len(samples) > 0

    # thread_name correlation is unreliable when main thread is _DummyThread (no native_id)
    expected_thread_name = "MainThread" if _main_thread_has_native_id() else None
    expected_sample = pprof_utils.StackEvent(
        thread_id=_thread.get_ident(),
        thread_name=expected_thread_name,
        locations=[
            pprof_utils.StackLocation(
                function_name="baz",
                filename="test_stack.py",
                line_no=baz.__code__.co_firstlineno + 1,
            ),
            pprof_utils.StackLocation(
                function_name="bar",
                filename="test_stack.py",
                line_no=bar.__code__.co_firstlineno + 1,
            ),
            pprof_utils.StackLocation(
                function_name="foo",
                filename="test_stack.py",
                line_no=foo.__code__.co_firstlineno + 1,
            ),
        ],
    )

    pprof_utils.assert_profile_has_sample(profile, samples=samples, expected_sample=expected_sample)


def test_push_span(tmp_path: Path, tracer: Tracer) -> None:
    test_name = "test_push_span"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    tracer._endpoint_call_counter_span_processor.enable()

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    resource = str(uuid.uuid4())
    span_type = ext.SpanTypes.WEB

    with stack.StackCollector(
        tracer=tracer,
    ):
        with tracer.trace("foobar", resource=resource, span_type=span_type) as span:
            span_id = span.span_id
            local_root_span_id = span._local_root.span_id
            for _ in range(10):
                time.sleep(0.1)
    ddup.upload(tracer=tracer)

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples_with_span_id = pprof_utils.get_samples_with_label_key(profile, "span id")

    samples: list[Sample] = []
    for sample in samples_with_span_id:
        locations = [pprof_utils.get_location_from_id(profile, location_id) for location_id in sample.location_id]
        if any(location.filename.endswith("test_stack.py") for location in locations):
            samples.append(sample)

    assert samples, "No sample found with locations in test_stack.py"

    pprof_utils.assert_profile_has_sample(
        profile,
        samples=samples,
        expected_sample=pprof_utils.StackEvent(
            span_id=span_id,
            local_root_span_id=local_root_span_id,
            trace_type=span_type,
            trace_endpoint=resource,
        ),
        print_samples_on_failure=True,
    )


def test_push_span_unregister_thread(tmp_path: Path, monkeypatch: MonkeyPatch, tracer: Tracer) -> None:
    with patch("ddtrace.internal.datadog.profiling.stack.unregister_thread") as unregister_thread:
        tracer._endpoint_call_counter_span_processor.enable()

        test_name = "test_push_span_unregister_thread"
        pprof_prefix = str(tmp_path / test_name)
        output_filename = pprof_prefix + "." + str(os.getpid())

        assert ddup.is_available
        ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
        ddup.start()
        ddup.upload()

        resource = str(uuid.uuid4())
        span_type = ext.SpanTypes.WEB

        def target_fun() -> None:
            for _ in range(10):
                time.sleep(0.1)

        with stack.StackCollector(
            tracer=tracer,
        ):
            with tracer.trace("foobar", resource=resource, span_type=span_type) as span:
                span_id = span.span_id
                local_root_span_id = span._local_root.span_id
                t = threading.Thread(target=target_fun)
                t.start()
                t.join()
                thread_id = t.ident
        ddup.upload(tracer=tracer)

        profile = pprof_utils.parse_newest_profile(output_filename)
        samples_with_span_id = pprof_utils.get_samples_with_label_key(profile, "span id")
        samples: list[Sample] = []
        for sample in samples_with_span_id:
            locations = [pprof_utils.get_location_from_id(profile, location_id) for location_id in sample.location_id]
            if any(location.filename.endswith("test_stack.py") for location in locations):
                samples.append(sample)

        assert samples, "No sample found with locations in test_stack.py"
        pprof_utils.assert_profile_has_sample(
            profile,
            samples=samples,
            expected_sample=pprof_utils.StackEvent(
                span_id=span_id,
                local_root_span_id=local_root_span_id,
                trace_type=span_type,
                trace_endpoint=resource,
            ),
            print_samples_on_failure=True,
        )

        unregister_thread.assert_called_with(thread_id)


def test_push_non_web_span(tmp_path: Path, tracer: Tracer) -> None:
    tracer._endpoint_call_counter_span_processor.enable()

    test_name = "test_push_non_web_span"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    resource = str(uuid.uuid4())
    span_type = ext.SpanTypes.SQL

    with stack.StackCollector(
        tracer=tracer,
    ):
        with tracer.trace("foobar", resource=resource, span_type=span_type) as span:
            span_id = span.span_id
            local_root_span_id = span._local_root.span_id
            for _ in range(10):
                time.sleep(0.1)
    ddup.upload(tracer=tracer)

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples_with_span_id = pprof_utils.get_samples_with_label_key(profile, "span id")
    samples: list[Sample] = []
    for sample in samples_with_span_id:
        locations = [pprof_utils.get_location_from_id(profile, location_id) for location_id in sample.location_id]
        if any(location.filename.endswith("test_stack.py") for location in locations):
            samples.append(sample)

    assert samples, "No sample found with locations in test_stack.py"
    pprof_utils.assert_profile_has_sample(
        profile,
        samples=samples,
        expected_sample=pprof_utils.StackEvent(
            span_id=span_id,
            local_root_span_id=local_root_span_id,
            trace_type=span_type,
            # trace_endpoint is not set for non-web spans
        ),
        print_samples_on_failure=True,
    )


def test_push_span_none_span_type(tmp_path: Path, tracer: Tracer) -> None:
    # Test for https://github.com/DataDog/dd-trace-py/issues/11141
    test_name = "test_push_span_none_span_type"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    tracer._endpoint_call_counter_span_processor.enable()

    resource = str(uuid.uuid4())

    with stack.StackCollector(
        tracer=tracer,
    ):
        # Explicitly set None span_type as the default could change in the
        # future.
        with tracer.trace("foobar", resource=resource, span_type=None) as span:
            span_id = span.span_id
            local_root_span_id = span._local_root.span_id
            for _ in range(10):
                time.sleep(0.1)
    ddup.upload(tracer=tracer)

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples_with_span_id = pprof_utils.get_samples_with_label_key(profile, "span id")
    samples: list[Sample] = []
    for sample in samples_with_span_id:
        locations = [pprof_utils.get_location_from_id(profile, location_id) for location_id in sample.location_id]
        if any(location.filename.endswith("test_stack.py") for location in locations):
            samples.append(sample)

    assert samples, "No sample found with locations in test_stack.py"
    pprof_utils.assert_profile_has_sample(
        profile,
        samples=samples,
        expected_sample=pprof_utils.StackEvent(
            span_id=span_id,
            local_root_span_id=local_root_span_id,
            # span_type is None
            # trace_endpoint is not set for non-web spans
        ),
        print_samples_on_failure=True,
    )


def test_collect_once_with_class(tmp_path: Path) -> None:
    class SomeClass(object):
        @classmethod
        def sleep_class(cls) -> None:
            return cls().sleep_instance()

        def sleep_instance(self) -> None:
            for _ in range(10):
                time.sleep(0.1)

    test_name = "test_collect_once_with_class"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    with stack.StackCollector():
        SomeClass.sleep_class()

    ddup.upload()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_value_type(profile, "wall-time")
    assert len(samples) > 0

    pprof_utils.assert_profile_has_sample(
        profile,
        samples=samples,
        expected_sample=pprof_utils.StackEvent(
            thread_id=_thread.get_ident(),
            thread_name="MainThread",
            locations=[
                pprof_utils.StackLocation(
                    function_name="sleep_instance",
                    filename="test_stack.py",
                    line_no=SomeClass.sleep_instance.__code__.co_firstlineno + 2,
                ),
                pprof_utils.StackLocation(
                    function_name="sleep_class",
                    filename="test_stack.py",
                    line_no=SomeClass.sleep_class.__code__.co_firstlineno + 2,
                ),
                pprof_utils.StackLocation(
                    function_name="test_collect_once_with_class",
                    filename="test_stack.py",
                    line_no=test_collect_once_with_class.__code__.co_firstlineno + 20,
                ),
            ],
        ),
        print_samples_on_failure=True,
    )


def test_collect_once_with_class_not_right_type(tmp_path: Path) -> None:
    """Test that the stack collector profiles methods with non-conventional parameter names.

    Verifies the profiler handles methods where parameters don't follow standard conventions
    (e.g., using 'foobar' instead of 'self' or 'cls').
    """

    class SomeClass(object):
        @classmethod
        def sleep_class(foobar, cls) -> None:  # pyright: ignore[reportSelfClsParameterName]
            return foobar().sleep_instance(cls)

        def sleep_instance(foobar, self) -> None:  # pyright: ignore[reportUnusedParameter, reportSelfClsParameterName]
            for _ in range(10):
                time.sleep(0.1)

    test_name = "test_collect_once_with_class"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    with stack.StackCollector():
        SomeClass.sleep_class(123)

    ddup.upload()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_value_type(profile, "wall-time")
    assert len(samples) > 0

    pprof_utils.assert_profile_has_sample(
        profile,
        samples=samples,
        expected_sample=pprof_utils.StackEvent(
            thread_id=_thread.get_ident(),
            thread_name="MainThread",
            locations=[
                pprof_utils.StackLocation(
                    function_name="sleep_instance",
                    filename="test_stack.py",
                    line_no=SomeClass.sleep_instance.__code__.co_firstlineno + 2,
                ),
                pprof_utils.StackLocation(
                    function_name="sleep_class",
                    filename="test_stack.py",
                    line_no=SomeClass.sleep_class.__code__.co_firstlineno + 2,
                ),
                pprof_utils.StackLocation(
                    function_name="test_collect_once_with_class_not_right_type",
                    filename="test_stack.py",
                    line_no=test_collect_once_with_class_not_right_type.__code__.co_firstlineno + 26,
                ),
            ],
        ),
        print_samples_on_failure=True,
    )


def _fib(n: int) -> int:
    if n == 1:
        return 1
    elif n == 0:
        return 0
    else:
        return _fib(n - 1) + _fib(n - 2)


@pytest.mark.skipif(
    not GEVENT_COMPATIBLE_WITH_PYTHON_VERSION,
    reason=f"gevent is not compatible with Python {'.'.join(map(str, tuple(sys.version_info)[:3]))}",
)
@pytest.mark.subprocess(ddtrace_run=True)
def test_collect_gevent_thread_task() -> None:
    from gevent import monkey

    monkey.patch_all()

    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
    from tests.profiling.collector.test_stack import _fib
    from tests.profiling.collector.test_stack import _main_thread_has_native_id

    test_name = "test_collect_gevent_thread_task"
    pprof_prefix = "/tmp/" + test_name
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    # Start some (green)threads
    def _do_fib() -> None:
        for _ in range(5):
            # spend some time in CPU so the profiler can catch something
            # On a Mac w/ Apple M3 MAX with Python 3.11 it takes about 200ms to calculate _fib(32)
            # And _fib() is called 5 times so it should take about 1 second
            # We use 5 threads below so it should take about 5 seconds
            _fib(32)
            # Just make sure gevent switches threads/greenlets
            time.sleep(0)

    threads = []

    with stack.StackCollector():
        for i in range(5):
            t = threading.Thread(target=_do_fib, name=f"TestThread {i}")
            t.start()
            threads.append(t)
        for t in threads:
            t.join()

    ddup.upload()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_label_key(profile, "task name")
    assert len(samples) > 0

    # thread_name correlation is unreliable when main thread is _DummyThread (no native_id)
    expected_thread_name = "MainThread" if _main_thread_has_native_id() else None
    pprof_utils.assert_profile_has_sample(
        profile,
        samples,
        expected_sample=pprof_utils.StackEvent(
            thread_name=expected_thread_name,
            task_name=r"Greenlet-\d+$",
            locations=[
                # Since we're using recursive function _fib(), we expect to have
                # multiple locations for _fib(n) = _fib(n-1) + _fib(n-2)
                pprof_utils.StackLocation(
                    filename="test_stack.py",
                    function_name="_fib",
                    line_no=_fib.__code__.co_firstlineno + 6,
                ),
                pprof_utils.StackLocation(
                    filename="test_stack.py",
                    function_name="_fib",
                    line_no=_fib.__code__.co_firstlineno + 6,
                ),
                pprof_utils.StackLocation(
                    filename="test_stack.py",
                    function_name="_fib",
                    line_no=_fib.__code__.co_firstlineno + 6,
                ),
            ],
        ),
        print_samples_on_failure=True,
    )


@pytest.mark.skipif(
    not GEVENT_COMPATIBLE_WITH_PYTHON_VERSION,
    reason=f"gevent is not compatible with Python {'.'.join(map(str, tuple(sys.version_info)[:3]))}",
)
@pytest.mark.subprocess(ddtrace_run=True)
def test_gevent_greenlets_emit_task_labels() -> None:
    """Gevent greenlet wall-time samples must carry task_name and task_id labels."""
    import os

    import gevent
    from gevent import monkey

    monkey.patch_all()

    from ddtrace.internal.datadog.profiling import ddup
    from ddtrace.profiling.collector import stack
    from tests.profiling.collector import pprof_utils
    from tests.profiling.collector.test_stack import _fib
    from tests.profiling.collector.test_stack import _main_thread_has_native_id

    test_name = "test_gevent_greenlets_emit_task_labels"
    pprof_prefix = "/tmp/" + test_name
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    def cpu_work() -> None:
        for _ in range(3):
            _fib(28)
            gevent.sleep(0)

    with stack.StackCollector():
        named_greenlets: list[gevent.Greenlet] = []
        for i in range(3):
            g = gevent.spawn(cpu_work)
            g.name = f"Greenlet-{i}"
            named_greenlets.append(g)
        gevent.joinall(named_greenlets, timeout=30)

    ddup.upload()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_label_key(profile, "task name")
    assert len(samples) > 0

    st = profile.string_table
    # check Greenlet-X is in the task name for each sample
    greenlet_samples_with_task_name = [s for s in samples if any("Greenlet" in st[label.str] for label in s.label)]
    assert len(greenlet_samples_with_task_name) > 0, "No greenlet samples with task name found"

    # check sample with task info also has task id
    for sample in greenlet_samples_with_task_name:
        assert any(st[label.key] == "task id" for label in sample.label), (
            f"No task id found in greenlet sample, labels: {sample.label}"
        )

    # Greenlet samples must still carry the real OS thread identity (MainThread),
    expected_thread_name = "MainThread" if _main_thread_has_native_id() else None
    pprof_utils.assert_profile_has_sample(
        profile,
        samples,
        expected_sample=pprof_utils.StackEvent(
            thread_name=expected_thread_name,
            task_name=r"Greenlet-\d+$",
            locations=[
                pprof_utils.StackLocation(
                    filename="test_stack.py",
                    function_name="_fib",
                    line_no=_fib.__code__.co_firstlineno + 6,
                ),
            ],
        ),
        print_samples_on_failure=True,
    )


@pytest.mark.skipif(
    not GEVENT_COMPATIBLE_WITH_PYTHON_VERSION,
    reason=f"gevent is not compatible with Python {'.'.join(map(str, tuple(sys.version_info)[:3]))}",
)
@pytest.mark.subprocess(
    env=dict(
        DD_PROFILING_OUTPUT_PPROF="/tmp/test_collect_gevent_task_started_before_profiler",
    ),
    err=None,
)
def test_collect_gevent_task_started_before_profiler() -> None:
    from gevent import monkey

    monkey.patch_all()

    import os
    import threading
    import time

    import gevent

    should_stop = threading.Event()

    def pre_started_greenlet_task() -> None:
        # Keep this greenlet alive long enough to be sampled after profiler start.
        while not should_stop.is_set():
            start = time.time()
            while time.time() - start < 0.01:
                pass
            gevent.sleep(0)

    # Start a named greenlet before profiler startup (remote-enable scenario).
    pre_started_greenlet_name = "pre-started-before-profiler"
    pre_started_greenlet = gevent.spawn(pre_started_greenlet_task)
    pre_started_greenlet.name = pre_started_greenlet_name
    gevent.sleep(0.05)

    # Import profiler modules after gevent patching and greenlet creation
    from ddtrace.profiling import profiler
    from tests.profiling.collector import pprof_utils
    from tests.profiling.collector.test_stack import _main_thread_has_native_id

    p = profiler.Profiler()
    p.start()

    try:
        gevent.sleep(1.0)
    finally:
        should_stop.set()
        pre_started_greenlet.join(timeout=2)
        p.stop()

    output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_label_key(profile, "task name")
    assert len(samples) > 0

    # thread_name correlation is unreliable when main thread is _DummyThread (no native_id)
    expected_thread_name = "MainThread" if _main_thread_has_native_id() else None
    pprof_utils.assert_profile_has_sample(
        profile,
        samples,
        expected_sample=pprof_utils.StackEvent(
            thread_name=expected_thread_name,
            task_name=pre_started_greenlet_name,
            locations=[
                pprof_utils.StackLocation(
                    function_name=pre_started_greenlet_task.__name__,
                    filename="test_stack.py",
                    line_no=-1,
                )
            ],
        ),
        print_samples_on_failure=True,
    )


@pytest.mark.skipif(
    not GEVENT_COMPATIBLE_WITH_PYTHON_VERSION,
    reason=f"gevent is not compatible with Python {'.'.join(map(str, tuple(sys.version_info)[:3]))}",
)
@pytest.mark.subprocess(
    env=dict(
        DD_PROFILING_OUTPUT_PPROF="/tmp/test_gevent_cpu_time_total_accuracy",
        _DD_PROFILING_STACK_ADAPTIVE_SAMPLING_ENABLED="0",
    ),
    err=None,
)
def test_gevent_cpu_time_total_accuracy() -> None:
    """Verify that the sum of cpu-time across all profile samples roughly matches
    the actual process CPU time consumed by gevent workers.

    Without the on-CPU swap in unwind_greenlets, when the running greenlet is
    not the first entry in current_greenlets, render_task_begin starts a new
    sample for it and pushes thread_state.cpu_time_ns a second time (the first
    push already happened on the sample inherited from render_thread_begin).
    For M=4 workers, the over-count converges to ~1.5-1.6x of the real CPU
    consumed by the process. With the swap in place, the on-CPU greenlet
    always reuses the first sample and the duplicate push is avoided.

    Adaptive sampling is disabled to mirror the experimental measurement cell
    that gave the cleanest 1.6x signal; this also keeps the sample interval
    deterministic across the timed region.
    """
    from gevent import monkey

    monkey.patch_all()

    import os
    import time

    import gevent

    from ddtrace.profiling import profiler
    from tests.profiling.collector import pprof_utils

    DURATION_S = 3.0
    CPU_BURN_S = 0.01
    SLEEP_S = 0.05
    NUM_WORKERS = 4

    def worker() -> None:
        deadline = time.monotonic() + DURATION_S
        while time.monotonic() < deadline:
            burn_until = time.process_time_ns() + int(CPU_BURN_S * 1e9)
            while time.process_time_ns() < burn_until:
                pass
            gevent.sleep(SLEEP_S)

    p = profiler.Profiler()
    p.start()
    try:
        # Stagger workers by CPU_BURN_S so their CPU bursts interleave with
        # each other's sleep, which spreads the on-CPU greenlet across the
        # unordered_map iteration order. Without staggering, the same greenlet
        # tends to win the on-CPU position each sample, masking the bug.
        workers = []
        cpu_start_ns = time.process_time_ns()
        for i in range(NUM_WORKERS):
            if i > 0:
                gevent.sleep(CPU_BURN_S)
            workers.append(gevent.spawn(worker))
        gevent.joinall(workers, timeout=DURATION_S + 10)
        cpu_end_ns = time.process_time_ns()
    finally:
        p.stop()

    assert all(g.dead for g in workers), "gevent workers did not finish within timeout"

    actual_cpu_ns = cpu_end_ns - cpu_start_ns
    assert actual_cpu_ns > 0, "process_time_ns did not advance"

    output_filename = os.environ["DD_PROFILING_OUTPUT_PPROF"] + "." + str(os.getpid())
    profile = pprof_utils.parse_newest_profile(output_filename)
    cpu_time_index = pprof_utils.get_sample_type_index(profile, "cpu-time")
    profile_cpu_ns = sum(sample.value[cpu_time_index] for sample in profile.sample)

    ratio = profile_cpu_ns / actual_cpu_ns

    # Bounds picked from measured ratios across local Mac (n=10) and CI Linux
    # py3.9-3.14 (n=6). With the fix applied: observed range was 0.975 - 1.058.
    # Without the fix (bug): observed range was 1.61 - 1.63 (CI, n=5).
    #
    # Upper bound 1.20:
    #   - ~14% above the worst observed fixed value (1.058)
    #   - ~25% below the smallest observed bug value (1.61)
    # Lower bound 0.85:
    #   - ~13% below the worst observed fixed value (0.975)
    #   - rejects a regression that drops real CPU (e.g. reintroducing the
    #     is_running() gate removed in PR #16273) which would push the ratio
    #     toward 0.
    #
    # These bounds exist to catch regressions while staying flake-safe across
    # CI runners. They are NOT a claim about the profiler's CPU attribution
    # accuracy.
    assert 0.85 <= ratio <= 1.20, (
        f"profile cpu-time total ({profile_cpu_ns / 1e9:.3f}s) does not match "
        f"actual process CPU time ({actual_cpu_ns / 1e9:.3f}s); ratio={ratio:.2f} "
        f"(expected ~1.0, bug produces ~1.6)"
    )


def test_repr() -> None:
    test_collector._test_repr(
        stack.StackCollector,
        "StackCollector(status=<ServiceStatus.STOPPED: 'stopped'>, nframes=64, tracer=None)",
    )


# Tests from tests/profiling/collector/test_stack.py
# Function to use for stress-test of polling
MAX_FN_NUM = 30
FN_TEMPLATE = """def _f{num}():
  return _f{nump1}()"""

for num in range(MAX_FN_NUM):
    exec(FN_TEMPLATE.format(num=num, nump1=num + 1))

exec(
    """def _f{MAX_FN_NUM}():
    try:
      raise ValueError('test')
    except Exception:
      time.sleep(2)""".format(MAX_FN_NUM=MAX_FN_NUM)
)


def test_stress_threads_run_as_thread(tmp_path: Path) -> None:
    test_name = "test_stress_threads_run_as_thread"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    quit_thread = threading.Event()

    def wait_for_quit() -> None:
        quit_thread.wait()

    with stack.StackCollector():
        NB_THREADS = 40

        threads = []
        for _ in range(NB_THREADS):
            t = threading.Thread(target=wait_for_quit)
            t.start()
            threads.append(t)

        time.sleep(3)

        quit_thread.set()
        for t in threads:
            t.join()

    ddup.upload()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_value_type(profile, "wall-time")
    assert len(samples) > 0


# if you don't need to check the output profile, you can use this fixture
@pytest.fixture
def tracer_and_collector(
    tracer: Tracer, request: FixtureRequest, tmp_path: Path
) -> Generator[tuple[Tracer, stack.StackCollector], None, None]:
    test_name = get_original_test_name(request)
    pprof_prefix = str(tmp_path / test_name)

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    c = stack.StackCollector(tracer=tracer)
    c.start()
    try:
        yield tracer, c
    finally:
        c.stop()
        ddup.upload(tracer=tracer)


def test_collect_span_id(tracer: Tracer, tmp_path: Path) -> None:
    test_name = "test_collect_span_id"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    tracer._endpoint_call_counter_span_processor.enable()
    with stack.StackCollector(tracer=tracer):
        resource = str(uuid.uuid4())
        span_type = ext.SpanTypes.WEB
        with tracer.start_span("foobar", activate=True, resource=resource, span_type=span_type) as span:
            for _ in range(10):
                time.sleep(0.1)
            span_id = span.span_id
            local_root_span_id = span._local_root.span_id

    ddup.upload(tracer=tracer)

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_label_key(profile, "trace endpoint")
    pprof_utils.assert_profile_has_sample(
        profile,
        samples,
        expected_sample=pprof_utils.StackEvent(
            thread_id=_thread.get_ident(),
            span_id=span_id,
            trace_type=span_type,
            local_root_span_id=local_root_span_id,
            trace_endpoint=resource,
            locations=[
                pprof_utils.StackLocation(
                    filename=os.path.basename(__file__),
                    function_name=test_name,
                    line_no=test_collect_span_id.__code__.co_firstlineno + 16,
                )
            ],
        ),
        print_samples_on_failure=True,
    )


def test_collect_span_resource_after_finish(tracer: Tracer, tmp_path: Path, request: FixtureRequest) -> None:
    test_name = get_original_test_name(request)
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    tracer._endpoint_call_counter_span_processor.enable()
    with stack.StackCollector(tracer=tracer):
        resource = str(uuid.uuid4())
        span_type = ext.SpanTypes.WEB
        span = tracer.start_span("foobar", activate=True, span_type=span_type, resource=resource)
        for _ in range(10):
            time.sleep(0.1)
    ddup.upload(tracer=tracer)
    span.finish()

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = profile.sample
    pprof_utils.assert_profile_has_sample(
        profile,
        samples,
        expected_sample=pprof_utils.StackEvent(
            thread_id=_thread.get_ident(),
            span_id=span.span_id,
            trace_type=span_type,
            # Looks like the endpoint is not collected if the span is not finished
            # trace_endpoint=resource,
            locations=[
                pprof_utils.StackLocation(
                    filename=os.path.basename(__file__),
                    function_name=test_name,
                    line_no=test_collect_span_resource_after_finish.__code__.co_firstlineno + 15,
                )
            ],
        ),
        print_samples_on_failure=True,
    )


def test_resource_not_collected(tmp_path: Path, tracer: Tracer) -> None:
    test_name = "test_resource_not_collected"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    with stack.StackCollector(tracer=tracer):
        # Give the Profiler some time to start
        time.sleep(0.1)

        resource = str(uuid.uuid4())
        span_type = ext.SpanTypes.WEB
        with tracer.start_span("foobar", activate=True, resource=resource, span_type=span_type) as span:
            _fib(35)

    ddup.upload(tracer=tracer)

    profile = pprof_utils.parse_newest_profile(output_filename)
    pprof_utils.assert_profile_has_sample(
        profile,
        profile.sample,
        expected_sample=pprof_utils.StackEvent(
            thread_id=_thread.get_ident(),
            span_id=span.span_id,
            trace_type=span_type,
            locations=[
                pprof_utils.StackLocation(
                    filename=os.path.basename(__file__),
                    function_name=test_name,
                    line_no=test_resource_not_collected.__code__.co_firstlineno + 17,
                )
            ],
        ),
        print_samples_on_failure=True,
    )


def test_collect_nested_span_id(tmp_path: Path, tracer: Tracer, request: FixtureRequest) -> None:
    test_name = get_original_test_name(request)
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    tracer._endpoint_call_counter_span_processor.enable()
    with stack.StackCollector(tracer=tracer):
        resource = str(uuid.uuid4())
        span_type = ext.SpanTypes.WEB
        with tracer.start_span("foobar", activate=True, resource=resource, span_type=span_type):
            with tracer.start_span("foobar", activate=True, resource=resource, span_type=span_type) as child_span:
                for _ in range(10):
                    time.sleep(0.1)
    ddup.upload(tracer=tracer)

    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_label_key(profile, "span id")
    pprof_utils.assert_profile_has_sample(
        profile,
        samples,
        expected_sample=pprof_utils.StackEvent(
            thread_id=_thread.get_ident(),
            span_id=child_span.span_id,
            trace_type=span_type,
            local_root_span_id=child_span._local_root.span_id,
            trace_endpoint=resource,
            locations=[
                pprof_utils.StackLocation(
                    filename=os.path.basename(__file__),
                    function_name=test_name,
                    line_no=test_collect_nested_span_id.__code__.co_firstlineno + 17,
                )
            ],
        ),
        print_samples_on_failure=True,
    )


@pytest.mark.skipif(
    not GEVENT_COMPATIBLE_WITH_PYTHON_VERSION,
    reason=f"gevent is not compatible with Python {'.'.join(map(str, tuple(sys.version_info)[:3]))}",
)
@pytest.mark.subprocess(
    env=dict(
        DD_PROFILING_OUTPUT_PPROF="/tmp/test_gevent_greenlet_switch_not_blocked_by_profiler",
    ),
    out=None,
    err=None,
)
def test_gevent_greenlet_switch_not_blocked_by_profiler() -> None:
    """Verify that profiler sampling does not block greenlet switches as the
    number of tracked greenlets grows.

    Before the fix, unwind_greenlets() held greenlet_info_map_lock for the
    entire stack unwinding of ALL tracked greenlets.  Every greenlet switch
    calls update_greenlet_switch() under the same lock, so more tracked
    greenlets meant longer lock hold and more switch blocking.

    This test measures greenlet-switch wall time with zero vs many idle
    tracked greenlets (each with deep stacks) while the profiler samples
    aggressively.  The ratio (many / zero) isolates lock-contention scaling
    from general profiler overhead.

    With the fix the lock is only held for a fast snapshot, so the ratio
    stays close to 1.  Without the fix it grows with N.
    """
    from gevent import monkey

    monkey.patch_all()

    import time

    import gevent

    from ddtrace.internal.datadog.profiling import stack
    from ddtrace.profiling import profiler

    N_ACTIVE = 20
    SWITCHES = 200
    N_IDLE_HIGH = 2000
    STACK_DEPTH = 50
    MAX_SCALING_RATIO = 3.0
    MEASURE_TIMEOUT = 30  # generous timeout to prevent CI hangs

    def active_worker() -> None:
        for _ in range(SWITCHES):
            gevent.sleep(0)

    def _idle_deep(depth: int) -> None:
        if depth > 0:
            _idle_deep(depth - 1)
        else:
            gevent.sleep(1000)

    def idle_greenlet() -> None:
        _idle_deep(STACK_DEPTH)

    def measure(n_idle: int) -> float:
        idles = [gevent.spawn(idle_greenlet) for _ in range(n_idle)]
        gevent.sleep(0.1)  # let them start and get tracked

        actives = [gevent.spawn(active_worker) for _ in range(N_ACTIVE)]
        t0 = time.monotonic()
        try:
            gevent.joinall(actives, timeout=MEASURE_TIMEOUT)
            elapsed = time.monotonic() - t0
            assert all(g.dead for g in actives), "active workers did not finish within timeout"
        finally:
            gevent.killall(actives, timeout=5)
            gevent.killall(idles, timeout=5)
            gevent.sleep(0)  # let the hub process kills before next measurement
        return elapsed

    p = profiler.Profiler()
    p.start()
    stack.set_interval(0.005)  # 5ms (minimum allowed) for aggressive sampling
    stack.set_adaptive_sampling(False)
    try:
        measure(0)  # warm up
        t_low = min(measure(0) for _ in range(3))
        t_high = min(measure(N_IDLE_HIGH) for _ in range(3))
    finally:
        p.stop()

    ratio = t_high / t_low if t_low > 0 else 1.0

    assert ratio < MAX_SCALING_RATIO, (
        f"Greenlet switch time scaled {ratio:.1f}x when adding {N_IDLE_HIGH} "
        f"tracked greenlets (low={t_low:.4f}s, high={t_high:.4f}s). "
        f"This indicates greenlet_info_map_lock contention during sampling."
    )


def test_greenlet_labels_do_not_depend_on_string_table_cleanup(tmp_path: Path) -> None:
    """Verify that tracking/untracking greenlets across many uploads does not
    crash or lose greenlet names now that greenlet labels are not StringTable entries.
    """
    test_name = "test_greenlet_labels_do_not_depend_on_string_table_cleanup"
    pprof_prefix = str(tmp_path / test_name)
    output_filename = pprof_prefix + "." + str(os.getpid())

    assert ddup.is_available
    ddup.config(env="test", service=test_name, version="my_version", output_filename=pprof_prefix)
    ddup.start()
    ddup.upload()

    from ddtrace.internal.datadog.profiling import stack as native_stack

    with stack.StackCollector():
        # Track a long-lived greenlet that should survive many uploads
        long_lived_id = 0xDEAD0001
        native_stack.track_greenlet(long_lived_id, "long-lived-greenlet", False)

        # Track and untrack short-lived greenlets to exercise erase()
        for i in range(50):
            short_id = 0xBEEF0000 + i
            native_stack.track_greenlet(short_id, f"short-lived-{i}", False)
            native_stack.untrack_greenlet(short_id)

        # Do 30 uploads to exercise repeated profile serialization.
        for _ in range(30):
            ddup.upload()

        # Track more short-lived greenlets after repeated uploads to make sure
        # label handling remains functional
        for i in range(50):
            short_id = 0xCAFE0000 + i
            native_stack.track_greenlet(short_id, f"post-clear-{i}", False)
            native_stack.untrack_greenlet(short_id)

        # Give the sampler a moment to collect a sample with the long-lived greenlet
        time.sleep(0.1)

        native_stack.untrack_greenlet(long_lived_id)

    # Final upload — if label handling is corrupted this would crash
    ddup.upload()

    # Verify we got a valid profile (no crash, no corruption)
    profile = pprof_utils.parse_newest_profile(output_filename)
    samples = pprof_utils.get_samples_with_value_type(profile, "wall-time")
    assert len(samples) > 0


def test_stress_trace_collection(tracer_and_collector: tuple[Tracer, stack.StackCollector]) -> None:
    tracer, _ = tracer_and_collector

    def _trace() -> None:
        for _ in range(5000):
            with tracer.trace("hello"):
                time.sleep(0.001)

    NB_THREADS = 30

    threads = []
    for _ in range(NB_THREADS):
        t = threading.Thread(target=_trace)
        threads.append(t)

    for t in threads:
        t.start()

    for t in threads:
        t.join()


@pytest.mark.skipif(not sys.platform.startswith("linux"), reason="fork test only on linux")
@pytest.mark.subprocess(
    ddtrace_run=True,
    env=dict(
        DD_PROFILING_ENABLED="true",
        DD_PROFILING_OUTPUT_PPROF="/tmp/test_span_id_in_profile_after_fork_ddtrace_run",
    ),
    err=None,
)
def test_span_id_in_profile_after_fork() -> None:
    """Same as test_span_id_in_profile_after_fork but exercising the real user path:
    ddtrace-run with DD_PROFILING_ENABLED=true.  The profiler and tracer are both
    bootstrapped automatically; no manual ddup/StackCollector setup is needed.
    """
    import os
    import time

    import ddtrace
    import ddtrace.profiling.bootstrap as bootstrap
    from tests.profiling.collector import pprof_utils

    tracer = ddtrace.tracer
    pprof_prefix = os.environ["DD_PROFILING_OUTPUT_PPROF"]

    with tracer.trace("job", resource="delancie-job") as root_span:
        span_id = root_span.span_id
        local_root_span_id = root_span._local_root.span_id

        pid = os.fork()
        if pid == 0:
            # child
            # Give some time for the sampler to capture samples
            end = time.monotonic() + 2
            while time.monotonic() < end:
                time.sleep(0.05)

            bootstrap.profiler._profiler._scheduler.flush()  # type: ignore[attr-defined]

            child_output = pprof_prefix + "." + str(os.getpid())
            profile = pprof_utils.parse_newest_profile(child_output)
            samples_with_span_id = pprof_utils.get_samples_with_label_key(profile, "span id")

            try:
                pprof_utils.assert_profile_has_sample(
                    profile,
                    samples=samples_with_span_id,
                    expected_sample=pprof_utils.StackEvent(
                        span_id=span_id,
                        local_root_span_id=local_root_span_id,
                    ),
                    print_samples_on_failure=True,
                )
            except AssertionError as e:
                print(
                    f"FAIL: no profile sample in child carries span_id={span_id} / "
                    f"local_root_span_id={local_root_span_id}.\n"
                    "ThreadSpanLinks was not repopulated after fork.\n"
                    f"AssertionError: {e}"
                )
                os._exit(1)
            os._exit(0)
        else:
            # parent process
            _, status = os.waitpid(pid, 0)
            exit_code = os.waitstatus_to_exitcode(status)
            assert exit_code == 0, (
                "Child process failed: profile samples after fork did not contain the expected span_id. "
                "The Trace to Profile link is broken for forked processes."
            )