chore(telemetry): clean up payload generation and typing in the writer (#14615)

- Improves typing for the TelemetryWriter. Converts all python2 type
hints to python3 typing
- Adds typing where it's missing. 
- Adds an `TELEMETRY_EVENT_TYPE` enum to store all telemetry types.
- Removes duplicate methods that simply generate telemetry payloads.
Instead adds a single `_get_event()` method that accepts a `payload` and
a TELEMETRY_EVENT_TYPE.
- Removes `TelemetryWriter.add_event()` and
`TelemetryWriter.add_events()`. These methods are no longer used in
ddtrace. Telemetry writer should not support queuing arbitrary payloads,
only event types in `TELEMETRY_EVENT_TYPE` are supported.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Author: mabdinur

.gitlab/benchmarks/bp-runner.microbenchmarks.fail-on-breach.yml

@@ -1085,7 +1085,7 @@ experiments:
               - max_rss_usage < 34.00 MB
           - name: telemetryaddmetric-1-count-metrics-100-times
             thresholds:
-              - execution_time < 0.25 ms
+              - execution_time < 0.22 ms
               - max_rss_usage < 34.00 MB
           - name: telemetryaddmetric-1-distribution-metric-1-times
             thresholds:
@@ -1113,11 +1113,11 @@ experiments:
               - max_rss_usage < 34.00 MB
           - name: telemetryaddmetric-100-count-metrics-100-times
             thresholds:
-              - execution_time < 23.50 ms
+              - execution_time < 22.0 ms
               - max_rss_usage < 34.00 MB
           - name: telemetryaddmetric-100-distribution-metrics-100-times
             thresholds:
-              - execution_time < 2.25 ms
+              - execution_time < 2.30 ms
               - max_rss_usage < 34.00 MB
           - name: telemetryaddmetric-100-gauge-metrics-100-times
             thresholds:

ddtrace/internal/telemetry/constants.py

@@ -11,9 +11,20 @@ class TELEMETRY_NAMESPACE(Enum):
     PROFILER = "profiler"
 
 
-TELEMETRY_TYPE_GENERATE_METRICS = "generate-metrics"
-TELEMETRY_TYPE_DISTRIBUTION = "distributions"
-TELEMETRY_TYPE_LOGS = "logs"
+class TELEMETRY_EVENT_TYPE(Enum):
+    STARTED = "app-started"
+    SHUTDOWN = "app-closing"
+    HEARTBEAT = "app-heartbeat"
+    EXTENDED_HEARTBEAT = "app-extended-heartbeat"
+    DEPENDENCIES_LOADED = "app-dependencies-loaded"
+    PRODUCT_CHANGE = "app-product-change"
+    INTEGRATIONS_CHANGE = "app-integrations-change"
+    ENDPOINTS = "app-endpoints"
+    CLIENT_CONFIGURATION_CHANGE = "app-client-configuration-change"
+    LOGS = "logs"
+    METRICS = "generate-metrics"
+    DISTRIBUTIONS = "distributions"
+    MESSAGE_BATCH = "message-batch"
 
 
 class TELEMETRY_LOG_LEVEL(Enum):

ddtrace/internal/telemetry/metrics_namespaces.pyx

@@ -6,8 +6,7 @@ from typing import Tuple
 
 from ddtrace.internal import forksafe
 from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE
-from ddtrace.internal.telemetry.constants import TELEMETRY_TYPE_DISTRIBUTION
-from ddtrace.internal.telemetry.constants import TELEMETRY_TYPE_GENERATE_METRICS
+from ddtrace.internal.telemetry.constants import TELEMETRY_EVENT_TYPE
 
 
 MetricTagType = Optional[Tuple[Tuple[str, str], ...]]
@@ -23,10 +22,16 @@ class MetricType(str, enum.Enum):
 cdef class MetricNamespace:
     cdef object _metrics_data_lock
     cdef public dict _metrics_data
+    # Cache enum objects at class level for maximum performance
+    cdef readonly object _metrics_key
+    cdef readonly object _distributions_key
 
     def __cinit__(self):
         self._metrics_data_lock = forksafe.Lock()
         self._metrics_data = {}
+        # Initialize cached enum references
+        self._metrics_key = TELEMETRY_EVENT_TYPE.METRICS
+        self._distributions_key = TELEMETRY_EVENT_TYPE.DISTRIBUTIONS
 
     def flush(self, interval: float = None):
         cdef float _interval = float(interval or 1.0)
@@ -46,19 +51,21 @@ cdef class MetricNamespace:
 
         now = int(time.time())
         data = {
-            TELEMETRY_TYPE_GENERATE_METRICS: {},
-            TELEMETRY_TYPE_DISTRIBUTION: {},
+            self._metrics_key: {},
+            self._distributions_key: {},
         }
         for metric_id, value in namespace_metrics.items():
             name, namespace, _tags, metric_type = metric_id
-            tags = ["{}:{}".format(k, v).lower() for k, v in _tags] if _tags else []
+            tags = [f"{k}:{v}".lower() for k, v in _tags] if _tags else []
             if metric_type is MetricType.DISTRIBUTION:
-                data[TELEMETRY_TYPE_DISTRIBUTION].setdefault(namespace, []).append({
+                payload_type = self._distributions_key
+                metric = {
                     "metric": name,
                     "points": value,
                     "tags": tags,
-                })
+                }
             else:
+                payload_type = self._metrics_key
                 if metric_type is MetricType.RATE:
                     value = value / _interval
                 metric = {
@@ -70,8 +77,12 @@ cdef class MetricNamespace:
                 }
                 if metric_type in (MetricType.RATE, MetricType.GAUGE):
                     metric["interval"] = _interval
-                data[TELEMETRY_TYPE_GENERATE_METRICS].setdefault(namespace, []).append(metric)
 
+            namespace_dict = data[payload_type]
+            if namespace not in namespace_dict:
+                namespace_dict[namespace] = [metric]
+            else:
+                namespace_dict[namespace].append(metric)
         return data
 
     def add_metric(
@@ -86,16 +97,20 @@ cdef class MetricNamespace:
         Adds a new telemetry metric to the internal metrics.
         Telemetry metrics are stored under "dd.instrumentation_telemetry_data.<namespace>.<name>".
         """
-        cdef float v
-        cdef tuple metric_id
         metric_id = (name, namespace.value, tags, metric_type)
-        if metric_type is MetricType.DISTRIBUTION:
+        if metric_type is MetricType.COUNT or metric_type is MetricType.RATE:
             with self._metrics_data_lock:
-                self._metrics_data.setdefault(metric_id, []).append(value)
+                existing = self._metrics_data.get(metric_id)
+                if existing is not None:
+                    self._metrics_data[metric_id] = existing + value
+                else:
+                    self._metrics_data[metric_id] = value
         elif metric_type is MetricType.GAUGE:
-            # Dict writes are atomic, no need to lock
             self._metrics_data[metric_id] = value
-        else:
+        else:  # MetricType.DISTRIBUTION
             with self._metrics_data_lock:
-                v = self._metrics_data.get(metric_id, 0)
-                self._metrics_data[metric_id] = v + value
+                existing = self._metrics_data.get(metric_id)
+                if existing is not None:
+                    existing.append(value)
+                else:
+                    self._metrics_data[metric_id] = [value]