feat: dynamic instrumentation public interface (#4052) (#4216)

## Description

This change exposes the public API for the dynamic instrumentation service and updates the documentation with the new configuration interface.

## Checklist
- [x] Title must conform to [conventional commit](https://github.com/conventional-changelog/commitlint/tree/master/%40commitlint/config-conventional).
- [x] Add additional sections for `feat` and `fix` pull requests.
- [x] Ensure tests are passing for affected code.
- [x] [Library documentation](https://github.com/DataDog/dd-trace-py/tree/1.x/docs) and/or [Datadog's documentation site](https://github.com/DataDog/documentation/) is updated. Link to doc PR in description.

## Motivation

The dynamic instrumentation service is now fully merged into the public repository and it needs a public API to be enabled and configured.

## Design

The design follows the pattern of similar services already exposed by the library, e.g. `RuntimeMetrics`. The public API exposes the `DynamicInstrumentation` class, which has the public methods `enable` and `disable`. When using `ddtrace-run`, enablement of the new service can be controlled via the environment, as per configuration documentation.

## Testing strategy

The enablement API is tested for both the environment approach via `ddtrace-run` using subprocess tests, as well as via the programmatic interface, using the `DynamicInstrumentation.enable` and `DynamicInstrumentation.disable` class methods. An additional test to check the expected behaviour during fork is also included.

## Reviewer Checklist
- [x] Title is accurate.
- [x] Description motivates each change.
- [x] No unnecessary changes were introduced in this PR.
- [x] PR cannot be broken up into smaller PRs.
- [x] Avoid breaking [API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces) changes unless absolutely necessary.
- [x] Tests provided or description of manual testing performed is included in the code or PR.
- [x] Release note has been added for fixes and features, or else `changelog/no-changelog` label added.
- [x] All relevant GitHub issues are correctly linked.
- [x] Backports are identified and tagged with Mergifyio.
- [x] Add to milestone.

(cherry picked from commit c55a28c)

Co-authored-by: Gabriele N. Tornetta <[email protected]>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

Author: mergify[bot]

ddtrace/bootstrap/sitecustomize.py

@@ -19,6 +19,7 @@
 
 from ddtrace import config  # noqa
 from ddtrace import constants
+from ddtrace.debugging._config import config as debugger_config
 from ddtrace.internal.logger import get_logger  # noqa
 from ddtrace.internal.runtime.runtime_metrics import RuntimeWorker
 from ddtrace.internal.utils.formats import asbool  # noqa
@@ -84,6 +85,11 @@ def update_patched_modules():
         log.debug("profiler enabled via environment variable")
         import ddtrace.profiling.auto  # noqa: F401
 
+    if debugger_config.enabled:
+        from ddtrace.debugging import DynamicInstrumentation
+
+        DynamicInstrumentation.enable()
+
     if asbool(os.getenv("DD_RUNTIME_METRICS_ENABLED")):
         RuntimeWorker.enable()
 

ddtrace/debugging/__init__.py

@@ -0,0 +1,25 @@
+"""
+Dynamic Instrumentation
+=======================
+
+Configuration
+-------------
+
+When using ``ddtrace-run``, dynamic instrumentation can be enabled by setting
+the ``DD_DYNAMIC_INSTRUMENTATION_ENABLED`` variable, or programmatically with::
+
+    from ddtrace.debugging import DynamicInstrumentation
+
+    # Enable dynamic instrumentation
+    DynamicInstrumentation.enable()
+
+    ...
+
+    # Disable the debugger
+    DynamicInstrumentation.disable()
+"""
+
+from ddtrace.debugging._debugger import Debugger as DynamicInstrumentation
+
+
+__all__ = ["DynamicInstrumentation"]

ddtrace/debugging/_config.py

@@ -1,86 +1,9 @@
-import os
-from typing import Dict
-from typing import Optional
-
-from ddtrace import config as tracer_config
-from ddtrace.internal.agent import get_trace_url
-from ddtrace.internal.constants import DEFAULT_SERVICE_NAME
 from ddtrace.internal.logger import get_logger
-from ddtrace.internal.utils.config import get_application_name
-from ddtrace.internal.utils.formats import asbool
-from ddtrace.version import get_version
+from ddtrace.settings.dynamic_instrumentation import DynamicInstrumentationConfig
 
 
 log = get_logger(__name__)
 
+config = DynamicInstrumentationConfig()
 
-DEFAULT_DEBUGGER_PORT = 8126
-DEFAULT_MAX_PROBES = 100
-DEFAULT_METRICS = True
-DEFAULT_GLOBAL_RATE_LIMIT = 100.0
-DEFAULT_MAX_PAYLOAD_SIZE = 1 << 20  # 1 MB
-DEFAULT_CONFIG_TIMEOUT = 30  # s
-DEFAULT_UPLOAD_TIMEOUT = 30  # seconds
-DEFAULT_UPLOAD_FLUSH_INTERVAL = 1.0  # seconds
-DEFAULT_PROBE_POLL_INTERVAL = 1.0  # seconds
-DEFAULT_DIAGNOSTIC_INTERVAL = 3600  # 1 hour
-
-
-class DebuggerConfig(object):
-    """Debugger configuration."""
-
-    service_name = DEFAULT_SERVICE_NAME
-    probe_url = get_trace_url()
-    snapshot_intake_url = get_trace_url()
-    max_probes = DEFAULT_MAX_PROBES
-    metrics = DEFAULT_METRICS
-    global_rate_limit = DEFAULT_GLOBAL_RATE_LIMIT
-    max_payload_size = DEFAULT_MAX_PAYLOAD_SIZE
-    config_timeout = DEFAULT_CONFIG_TIMEOUT
-    upload_timeout = DEFAULT_UPLOAD_TIMEOUT
-    upload_flush_interval = DEFAULT_UPLOAD_FLUSH_INTERVAL
-    poll_interval = DEFAULT_PROBE_POLL_INTERVAL
-    diagnostic_interval = DEFAULT_DIAGNOSTIC_INTERVAL
-    tags = None  # type: Optional[str]
-    _tags = {}  # type: Dict[str, str]
-    _tags_in_qs = True
-    _snapshot_intake_endpoint = "/v1/input"
-
-    def __init__(self):
-        # type: () -> None
-        try:
-            self.snapshot_intake_url = os.environ["DD_DEBUGGER_SNAPSHOT_INTAKE_URL"]
-            self._tags_in_qs = False
-        except KeyError:
-            self.snapshot_intake_url = get_trace_url()
-            self._snapshot_intake_endpoint = "/debugger" + self._snapshot_intake_endpoint
-
-        self.probe_url = os.getenv("DD_DEBUGGER_PROBE_URL", get_trace_url())
-        self.upload_timeout = int(os.getenv("DD_DEBUGGER_UPLOAD_TIMEOUT", DEFAULT_UPLOAD_TIMEOUT))
-        self.upload_flush_interval = float(
-            os.getenv("DD_DEBUGGER_UPLOAD_FLUSH_INTERVAL", DEFAULT_UPLOAD_FLUSH_INTERVAL)
-        )
-
-        self._tags["env"] = tracer_config.env
-        self._tags["version"] = tracer_config.version
-        self._tags["debugger_version"] = get_version()
-
-        self._tags.update(tracer_config.tags)
-
-        self.tags = ",".join([":".join((k, v)) for (k, v) in self._tags.items() if v is not None])
-
-        self.service_name = tracer_config.service or get_application_name() or DEFAULT_SERVICE_NAME
-        self.metrics = asbool(os.getenv("DD_DEBUGGER_METRICS_ENABLED", DEFAULT_METRICS))
-        self.max_payload_size = int(os.getenv("DD_DEBUGGER_MAX_PAYLOAD_SIZE", DEFAULT_MAX_PAYLOAD_SIZE))
-
-        self.config_timeout = int(os.getenv("DD_DEBUGGER_CONFIG_TIMEOUT", DEFAULT_CONFIG_TIMEOUT))
-        self.poll_interval = int(os.getenv("DD_DEBUGGER_POLL_INTERVAL", DEFAULT_PROBE_POLL_INTERVAL))
-        self.diagnostic_interval = int(os.getenv("DD_DEBUGGER_DIAGNOSTIC_INTERVAL", DEFAULT_DIAGNOSTIC_INTERVAL))
-
-        log.debug(
-            "Debugger configuration: %r",
-            {k: v for k, v in ((k, getattr(self, k)) for k in type(self).__dict__ if not k.startswith("__"))},
-        )
-
-
-config = DebuggerConfig()
+log.debug("Dynamic instrumentation configuration: %r", config.__dict__)

ddtrace/debugging/_probe/remoteconfig.py

@@ -163,7 +163,7 @@ def __init__(self, callback):
 
     def _next_status_update_timestamp(self):
         # type: () -> None
-        self._status_timestamp = time.time() + config.diagnostic_interval
+        self._status_timestamp = time.time() + config.diagnostics_interval
 
     def __call__(self, metadata, config):
         # type: (Any, Any) -> None

ddtrace/debugging/_probe/status.py

@@ -20,8 +20,6 @@
 
 
 class ProbeStatusLogger(object):
-    ENDPOINT = config.snapshot_intake_url
-
     def __init__(self, service, encoder):
         # type: (str, BufferedEncoder) -> None
         self._service = service
@@ -65,7 +63,7 @@ def _write(self, probe, status, message, exc_info=None):
 
             while self._retry_queue:
                 item, ts = self._retry_queue.popleft()
-                if now - ts > config.diagnostic_interval:
+                if now - ts > config.diagnostics_interval:
                     # We discard the expired items as they wouldn't be picked
                     # up by the backend anyway.
                     continue

ddtrace/debugging/_uploader.py

@@ -36,7 +36,7 @@ def __init__(self, encoder, interval=None):
         }
         if config._tags_in_qs and config.tags:
             self.ENDPOINT += "?ddtags=" + config.tags
-        self._connect = connector(config.snapshot_intake_url, timeout=config.upload_timeout)
+        self._connect = connector(config._snapshot_intake_url, timeout=config.upload_timeout)
         self._retry_upload = tenacity.Retrying(
             # Retry RETRY_ATTEMPTS times within the first half of the processing
             # interval, using a Fibonacci policy with jitter
@@ -48,7 +48,7 @@ def __init__(self, encoder, interval=None):
 
         log.debug(
             "Logs intake uploader initialized (url: %s, endpoint: %s, interval: %f)",
-            config.snapshot_intake_url,
+            config._snapshot_intake_url,
             self.ENDPOINT,
             self.interval,
         )

ddtrace/settings/dynamic_instrumentation.py

@@ -0,0 +1,79 @@
+from envier import En
+
+from ddtrace import config
+from ddtrace.internal.agent import get_trace_url
+from ddtrace.internal.constants import DEFAULT_SERVICE_NAME
+from ddtrace.internal.utils.config import get_application_name
+from ddtrace.version import get_version
+
+
+DEFAULT_MAX_PROBES = 100
+DEFAULT_GLOBAL_RATE_LIMIT = 100.0
+
+
+def _derive_tags(c):
+    # type: (En) -> str
+    _tags = dict(env=config.env, version=config.version, debugger_version=get_version())
+    _tags.update(config.tags)
+
+    return ",".join([":".join((k, v)) for (k, v) in _tags.items() if v is not None])
+
+
+class DynamicInstrumentationConfig(En):
+    __prefix__ = "dd.dynamic_instrumentation"
+
+    service_name = En.d(str, lambda _: config.service or get_application_name() or DEFAULT_SERVICE_NAME)
+    _snapshot_intake_url = En.d(str, lambda _: get_trace_url())
+    max_probes = En.d(int, lambda _: DEFAULT_MAX_PROBES)
+    global_rate_limit = En.d(float, lambda _: DEFAULT_GLOBAL_RATE_LIMIT)
+    _tags_in_qs = En.d(bool, lambda _: True)
+    _snapshot_intake_endpoint = En.d(str, lambda _: "/debugger/v1/input")
+    tags = En.d(str, _derive_tags)
+
+    enabled = En.v(
+        bool,
+        "enabled",
+        default=False,
+        help_type="Boolean",
+        help="Enable the debugger",
+    )
+
+    metrics = En.v(
+        bool,
+        "metrics.enabled",
+        default=True,
+        help_type="Boolean",
+        help="Enable diagnostic metrics",
+    )
+
+    max_payload_size = En.v(
+        int,
+        "max_payload_size",
+        default=1 << 20,  # 1 MB
+        help_type="Integer",
+        help="Maximum size in bytes of a single configuration payload that can be handled per request",
+    )
+
+    upload_timeout = En.v(
+        int,
+        "upload.timeout",
+        default=30,  # seconds
+        help_type="Integer",
+        help="Timeout in seconds for uploading a snapshot",
+    )
+
+    upload_flush_interval = En.v(
+        float,
+        "upload.flush_interval",
+        default=1.0,  # seconds
+        help_type="Float",
+        help="Interval in seconds for flushing the snapshot upload queue.",
+    )
+
+    diagnostics_interval = En.v(
+        int,
+        "diagnostics.interval",
+        default=3600,  # 1 hour
+        help_type="Integer",
+        help="Interval in seconds for periodically sending probe diagnostic messages",
+    )

docs/conf.py

@@ -65,7 +65,13 @@ def _skip(self, word):
 # Add any Sphinx extension module names here, as strings. They can be
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
-extensions = ["sphinx.ext.autodoc", "sphinx.ext.extlinks", "reno.sphinxext", "sphinxcontrib.spelling"]
+extensions = [
+    "sphinx.ext.autodoc",
+    "sphinx.ext.extlinks",
+    "reno.sphinxext",
+    "sphinxcontrib.spelling",
+    "envier.sphinx",
+]
 
 # Add filters for sphinxcontrib.spelling
 spelling_filters = [VersionTagFilter]

docs/configuration.rst

@@ -353,3 +353,9 @@ below:
      - Sensitive parameter value regexp for obfuscation.
 
 .. _Unified Service Tagging: https://docs.datadoghq.com/getting_started/tagging/unified_service_tagging/
+
+
+Dynamic Instrumentation
+-----------------------
+
+.. envier:: ddtrace.settings.dynamic_instrumentation:DynamicInstrumentationConfig

mypy.ini

@@ -11,6 +11,7 @@ warn_unused_configs = true
 no_implicit_optional = true
 ignore_missing_imports = true
 namespace_packages = true
+plugins = envier.mypy
 
 [mypy-ddtrace.contrib.*]
 ignore_errors = true