perf(rc): improve performance of subprocess service sharing

ddtrace/settings/_config.py

@@ -504,6 +504,7 @@ def __init__(self):
         self.version = _get_config("DD_VERSION", self.tags.get("version"))
         self._http_server = self._HTTPServerConfig()
 
+        self._extra_services_sent = set()  # type: set[str]
         self._extra_services_queue = None
         if self._remote_config_enabled and not in_aws_lambda():
             # lazy load slow import
@@ -667,8 +668,12 @@ def __getattr__(self, name) -> Any:
     def _add_extra_service(self, service_name: str) -> None:
         if self._extra_services_queue is None:
             return
-        if service_name != self.service:
-            self._extra_services_queue.put(service_name)
+
+        if service_name == self.service or service_name in self._extra_services_sent:
+            return
+
+        self._extra_services_queue.put(service_name)
+        self._extra_services_sent.add(service_name)
 
     def _get_extra_services(self):
         # type: () -> set[str]

tests/internal/service_name/test_extra_services_names.py

@@ -1,42 +1,87 @@
-import random
-import re
-import threading
-import time
+import os
+import sys
 
 import pytest
 
+
+@pytest.mark.skipif(sys.platform in ("win32", "cygwin"), reason="Fork not supported on Windows")
+def test_config_extra_service_names_fork(run_python_code_in_subprocess):
+    code = f"""
+import ddtrace.auto
 import ddtrace
 
+import re
+import os
+import sys
+import time
+
+children = []
+for i in range(10):
+    pid = os.fork()
+    if pid == 0:
+        # Child process
+        ddtrace.config._add_extra_service(f"extra_service_{{i}}")
+        time.sleep(0.1)  # Ensure the child has time to save the service
+        sys.exit(0)
+    else:
+        # Parent process
+        children.append(pid)
+
+for pid in children:
+    os.waitpid(pid, 0)
+
+extra_services = ddtrace.config._get_extra_services()
+extra_services.discard("sqlite")  # coverage
+assert len(extra_services) == 10, extra_services
+assert all(re.match(r"extra_service_\\d+_\\d+", service) for service in extra_services)
+"""
 
-MAX_NAMES = 64
+    env = os.environ.copy()
+    env["DD_REMOTE_CONFIGURATION_ENABLED"] = "true"
+    stdout, stderr, status, _ = run_python_code_in_subprocess(code, env=env)
+    assert status == 0, (stdout, stderr, status)
 
 
-@pytest.mark.parametrize("nb_service", [2, 16, 64, 256])
-def test_service_name(nb_service):
-    ddtrace.config._extra_services = set()
+def test_config_extra_service_names_duplicates(run_python_code_in_subprocess):
+    code = """
+import ddtrace.auto
+import ddtrace
+import re
+import os
+import sys
+import time
+
+for _ in range(10):
+    ddtrace.config._add_extra_service("extra_service_1")
 
-    def write_in_subprocess(id_nb):
-        time.sleep(random.random())
-        ddtrace.config._add_extra_service(f"extra_service_{id_nb}")
+extra_services = ddtrace.config._get_extra_services()
+extra_services.discard("sqlite")  # coverage
+assert extra_services == {"extra_service_1"}
+    """
 
-    default_remote_config_enabled = ddtrace.config._remote_config_enabled
-    ddtrace.config._remote_config_enabled = True
-    if ddtrace.config._extra_services_queue is None:
-        import ddtrace.internal._file_queue as file_queue
+    env = os.environ.copy()
+    env["DD_REMOTE_CONFIGURATION_ENABLED"] = "true"
+    stdout, stderr, status, _ = run_python_code_in_subprocess(code, env=env)
+    assert status == 0, (stdout, stderr, status)
 
-        ddtrace.config._extra_services_queue = file_queue.File_Queue()
 
-    threads = [threading.Thread(target=write_in_subprocess, args=(i,)) for i in range(nb_service)]
-    for thread in threads:
-        thread.start()
-    for thread in threads:
-        thread.join()
+def test_config_extra_service_names_rc_disabled(run_python_code_in_subprocess):
+    code = """
+import ddtrace.auto
+import ddtrace
+import re
+import os
+import sys
+import time
+
+for _ in range(10):
+    ddtrace.config._add_extra_service("extra_service_1")
 
-    extra_services = ddtrace.config._get_extra_services()
-    assert len(extra_services) == min(nb_service, MAX_NAMES)
-    assert all(re.match(r"extra_service_\d+", service) for service in extra_services)
+extra_services = ddtrace.config._get_extra_services()
+assert len(extra_services) == 0
+    """
 
-    ddtrace.config._remote_config_enabled = default_remote_config_enabled
-    if not default_remote_config_enabled:
-        ddtrace.config._extra_services_queue = None
-    ddtrace.config._extra_services = set()
+    env = os.environ.copy()
+    env["DD_REMOTE_CONFIGURATION_ENABLED"] = "false"
+    stdout, stderr, status, _ = run_python_code_in_subprocess(code, env=env)
+    assert status == 0, (stdout, stderr, status)