feat(profiling): enable agent upload by default (#1494)

The upcoming Datadog agent release will support uploading profiles. This
enables its usage by default when configuration the profiler with no DD_API_KEY
set in the env.

Author: jd

ddtrace/profiling/exporter/http.py

@@ -7,7 +7,6 @@
 
 import tenacity
 
-from ddtrace.utils import deprecation
 from ddtrace.utils.formats import parse_tags_str
 from ddtrace.vendor import six
 from ddtrace.vendor.six.moves import http_client
@@ -28,50 +27,22 @@
 PYTHON_VERSION = platform.python_version().encode()
 
 
-class InvalidEndpoint(ValueError):
-    pass
-
-
 class UploadFailed(exporter.ExportError):
     """Upload failure."""
 
-    def __init__(self, exception):
+    def __init__(self, exception, msg=None):
         """Create a failed upload error based on raised exceptions."""
         self.exception = exception
-        super(UploadFailed, self).__init__("Unable to upload profile: " + _traceback.format_exception(exception))
-
-
-def _get_api_key():
-    legacy = _attr.from_env("DD_PROFILING_API_KEY", "", str)()
-    if legacy:
-        deprecation.deprecation("DD_PROFILING_API_KEY", "Use DD_API_KEY")
-        return legacy
-    return _attr.from_env("DD_API_KEY", "", str)()
-
-
-ENDPOINT_TEMPLATE = "https://intake.profile.{}/v1/input"
-
-
-def _get_endpoint():
-    legacy = _attr.from_env("DD_PROFILING_API_URL", "", str)()
-    if legacy:
-        deprecation.deprecation("DD_PROFILING_API_URL", "Use DD_SITE")
-        return legacy
-    site = _attr.from_env("DD_SITE", "datadoghq.com", str)()
-    return ENDPOINT_TEMPLATE.format(site)
-
-
-def _validate_enpoint(instance, attribute, value):
-    if not value:
-        raise InvalidEndpoint("Endpoint is empty")
+        msg = _traceback.format_exception(exception) if msg is None else msg
+        super(UploadFailed, self).__init__("Unable to upload profile: " + msg)
 
 
 @attr.s
 class PprofHTTPExporter(pprof.PprofExporter):
     """PProf HTTP exporter."""
 
-    endpoint = attr.ib(factory=_get_endpoint, type=str, validator=_validate_enpoint)
-    api_key = attr.ib(factory=_get_api_key, type=str)
+    endpoint = attr.ib()
+    api_key = attr.ib(default=None)
     timeout = attr.ib(factory=_attr.from_env("DD_PROFILING_API_TIMEOUT", 10, float), type=float)
     service = attr.ib(default=None)
     env = attr.ib(default=None)
@@ -138,16 +109,25 @@ def _get_tags(self, service):
         tags.update({k: six.ensure_binary(v) for k, v in user_tags.items()})
         return tags
 
+    @staticmethod
+    def _retry_on_http_5xx(exception):
+        if isinstance(exception, error.HTTPError):
+            return 500 <= exception.code < 600
+        return True
+
     def export(self, events, start_time_ns, end_time_ns):
         """Export events to an HTTP endpoint.
 
         :param events: The event dictionary from a `ddtrace.profiling.recorder.Recorder`.
         :param start_time_ns: The start time of recording.
         :param end_time_ns: The end time of recording.
         """
-        common_headers = {
-            "DD-API-KEY": self.api_key.encode(),
-        }
+        if self.api_key:
+            headers = {
+                "DD-API-KEY": self.api_key.encode(),
+            }
+        else:
+            headers = {}
 
         profile = super(PprofHTTPExporter, self).export(events, start_time_ns, end_time_ns)
         s = six.BytesIO()
@@ -170,7 +150,6 @@ def export(self, events, start_time_ns, end_time_ns):
         service = self.service or os.path.basename(profile.string_table[profile.mapping[0].filename])
 
         content_type, body = self._encode_multipart_formdata(fields, tags=self._get_tags(service),)
-        headers = common_headers.copy()
         headers["Content-Type"] = content_type
 
         # urllib uses `POST` if `data` is supplied (Python 2 version does not handle `method` kwarg)
@@ -180,12 +159,20 @@ def export(self, events, start_time_ns, end_time_ns):
             # Retry after 1s, 2s, 4s, 8s with some randomness
             wait=tenacity.wait_random_exponential(multiplier=0.5),
             stop=tenacity.stop_after_delay(self.max_retry_delay),
-            retry=tenacity.retry_if_exception_type(
-                (error.HTTPError, error.URLError, http_client.HTTPException, OSError, IOError)
-            ),
+            retry=tenacity.retry_if_exception_type((http_client.HTTPException, OSError, IOError))
+            & tenacity.retry_if_exception(self._retry_on_http_5xx),
         )
 
         try:
             retry(request.urlopen, req, timeout=self.timeout)
         except tenacity.RetryError as e:
             raise UploadFailed(e.last_attempt.exception())
+        except error.HTTPError as e:
+            if e.code == 404 and not self.api_key:
+                msg = (
+                    "Datadog Agent is not accepting profiles. "
+                    "Agent-based profiling deployments require Datadog Agent >= 7.20"
+                )
+            else:
+                msg = None
+            raise UploadFailed(e, msg)

ddtrace/profiling/profiler.py

@@ -4,6 +4,7 @@
 
 from ddtrace.profiling import recorder
 from ddtrace.profiling import scheduler
+from ddtrace.utils import deprecation
 from ddtrace.vendor import attr
 from ddtrace.profiling.collector import exceptions
 from ddtrace.profiling.collector import memory
@@ -16,19 +17,45 @@
 LOG = logging.getLogger(__name__)
 
 
-def _build_default_exporters(service, env, version):
-    exporters = []
-    if "DD_PROFILING_API_KEY" in os.environ or "DD_API_KEY" in os.environ:
-        exporters.append(http.PprofHTTPExporter(service=service, env=env, version=version))
+ENDPOINT_TEMPLATE = "https://intake.profile.{}/v1/input"
+
+
+def _get_endpoint():
+    legacy = os.environ.get("DD_PROFILING_API_URL")
+    if legacy:
+        deprecation.deprecation("DD_PROFILING_API_URL", "Use DD_SITE")
+        return legacy
+    site = os.environ.get("DD_SITE", "datadoghq.com")
+    return ENDPOINT_TEMPLATE.format(site)
+
+
+def _get_api_key():
+    legacy = os.environ.get("DD_PROFILING_API_KEY")
+    if legacy:
+        deprecation.deprecation("DD_PROFILING_API_KEY", "Use DD_API_KEY")
+        return legacy
+    return os.environ.get("DD_API_KEY")
 
+
+def _build_default_exporters(service, env, version):
     _OUTPUT_PPROF = os.environ.get("DD_PROFILING_OUTPUT_PPROF")
     if _OUTPUT_PPROF:
-        exporters.append(file.PprofFileExporter(_OUTPUT_PPROF))
-
-    if not exporters:
-        LOG.warning("No exporters are configured, no profile will be output")
+        return [
+            file.PprofFileExporter(_OUTPUT_PPROF),
+        ]
 
-    return exporters
+    api_key = _get_api_key()
+    if api_key:
+        # Agentless mode
+        endpoint = _get_endpoint()
+    else:
+        hostname = os.environ.get("DD_AGENT_HOST", os.environ.get("DATADOG_TRACE_AGENT_HOSTNAME", "localhost"))
+        port = int(os.environ.get("DD_TRACE_AGENT_PORT", 8126))
+        endpoint = os.environ.get("DD_TRACE_AGENT_URL", "http://%s:%d" % (hostname, port)) + "/profiling/v1/input"
+
+    return [
+        http.PprofHTTPExporter(service=service, env=env, version=version, api_key=api_key, endpoint=endpoint),
+    ]
 
 
 def _get_service_name():