use event type

Author: mabdinur

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-shutdown"
+    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], ...]]
@@ -46,14 +45,14 @@ cdef class MetricNamespace:
 
         now = int(time.time())
         data = {
-            TELEMETRY_TYPE_GENERATE_METRICS: {},
-            TELEMETRY_TYPE_DISTRIBUTION: {},
+            TELEMETRY_EVENT_TYPE.METRICS: {},
+            TELEMETRY_EVENT_TYPE.DISTRIBUTIONS: {},
         }
         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 []
             if metric_type is MetricType.DISTRIBUTION:
-                data[TELEMETRY_TYPE_DISTRIBUTION].setdefault(namespace, []).append({
+                data[TELEMETRY_EVENT_TYPE.DISTRIBUTIONS].setdefault(namespace, []).append({
                     "metric": name,
                     "points": value,
                     "tags": tags,
@@ -70,7 +69,7 @@ cdef class MetricNamespace:
                 }
                 if metric_type in (MetricType.RATE, MetricType.GAUGE):
                     metric["interval"] = _interval
-                data[TELEMETRY_TYPE_GENERATE_METRICS].setdefault(namespace, []).append(metric)
+                data[TELEMETRY_EVENT_TYPE.METRICS].setdefault(namespace, []).append(metric)
 
         return data
 

ddtrace/internal/telemetry/writer.py

@@ -31,9 +31,9 @@
 from ..utils.version import version as tracer_version
 from . import modules
 from .constants import TELEMETRY_APM_PRODUCT
+from .constants import TELEMETRY_EVENT_TYPE
 from .constants import TELEMETRY_LOG_LEVEL
 from .constants import TELEMETRY_NAMESPACE
-from .constants import TELEMETRY_TYPE_LOGS
 from .data import get_application
 from .data import get_host_info
 from .data import get_python_config_vars
@@ -216,7 +216,7 @@ def __init__(self, is_periodic=True, agentless=None):
             self.enable()
             # Force app started for unit tests
             if config.FORCE_START and (app_started := self._app_started_payload()):
-                self._events_queue.append({"payload": app_started, "request_type": "app-started"})
+                self._events_queue.append({"payload": app_started, "request_type": TELEMETRY_EVENT_TYPE.STARTED})
             get_logger("ddtrace").addHandler(DDTelemetryErrorHandler(self))
 
     def enable(self):
@@ -635,7 +635,7 @@ def periodic(self, force_flush=False, shutting_down=False):
                         )
 
         if logs := self._report_logs():
-            events.append({"payload": {"logs": list(logs)}, "request_type": TELEMETRY_TYPE_LOGS})
+            events.append({"payload": {"logs": list(logs)}, "request_type": TELEMETRY_EVENT_TYPE.LOGS})
 
         # Queue metrics if not at heartbeat interval
         if self._is_periodic and force_flush is False:
@@ -649,34 +649,39 @@ def periodic(self, force_flush=False, shutting_down=False):
         # At heartbeat interval, collect and send all telemetry data
         if app_started_payload := self._app_started_payload():
             # app-started should be the first event in the batch
-            events = [{"payload": app_started_payload, "request_type": "app-started"}] + events
+            events = [{"payload": app_started_payload, "request_type": TELEMETRY_EVENT_TYPE.STARTED}] + events
 
         if products := self._report_app_products():
-            events.append({"payload": {"products": products}, "request_type": "app-product-change"})
+            events.append({"payload": {"products": products}, "request_type": TELEMETRY_EVENT_TYPE.PRODUCT_CHANGE})
 
         if ints := self._report_integrations():
-            events.append({"payload": {"integrations": ints}, "request_type": "app-integrations-change"})
+            events.append({"payload": {"integrations": ints}, "request_type": TELEMETRY_EVENT_TYPE.INTEGRATIONS_CHANGE})
 
         if endpoints := self._report_app_endpoints():
-            events.append({"payload": endpoints, "request_type": "app-endpoints"})
+            events.append({"payload": endpoints, "request_type": TELEMETRY_EVENT_TYPE.ENDPOINTS})
 
         if configs := self._report_configuration_queue():
-            events.append({"payload": {"configuration": configs}, "request_type": "app-client-configuration-change"})
+            events.append(
+                {
+                    "payload": {"configuration": configs},
+                    "request_type": TELEMETRY_EVENT_TYPE.CLIENT_CONFIGURATION_CHANGE,
+                }
+            )
 
         if deps := self._report_app_dependencies():
-            events.append({"payload": {"dependencies": deps}, "request_type": "app-dependencies-loaded"})
+            events.append({"payload": {"dependencies": deps}, "request_type": TELEMETRY_EVENT_TYPE.DEPENDENCIES_LOADED})
 
         if shutting_down and not self._forked:
-            events.append({"payload": {}, "request_type": "app-closing"})
+            events.append({"payload": {}, "request_type": TELEMETRY_EVENT_TYPE.SHUTDOWN})
 
         # Always include a heartbeat to keep RC connections alive
         # Extended heartbeat should be queued after app-dependencies-loaded event. This
         # ensures that that imported dependencies are accurately reported.
         if heartbeat_payload := self._app_heartbeat_payload():
             # Extended heartbeat report dependencies while regular heartbeats report empty payloads
-            events.append({"payload": heartbeat_payload, "request_type": "app-extended-heartbeat"})
+            events.append({"payload": heartbeat_payload, "request_type": TELEMETRY_EVENT_TYPE.EXTENDED_HEARTBEAT})
         else:
-            events.append({"payload": {}, "request_type": "app-heartbeat"})
+            events.append({"payload": {}, "request_type": TELEMETRY_EVENT_TYPE.HEARTBEAT})
 
         # Get any queued events and combine with current batch
         if queued_events := self._flush_events_queue():
@@ -781,7 +786,7 @@ def _telemetry_excepthook(self, tp, value, root_traceback):
                     self.add_integration(integration_name, True, error_msg=error_msg)
 
             if app_started := self._app_started_payload(False):
-                self._events_queue.append({"payload": app_started, "request_type": "app-started"})
+                self._events_queue.append({"payload": app_started, "request_type": TELEMETRY_EVENT_TYPE.STARTED})
 
             self.app_shutdown()
 

tests/appsec/appsec/test_telemetry.py

@@ -15,9 +15,8 @@
 from ddtrace.constants import APPSEC_ENV
 from ddtrace.contrib.internal.trace_utils import set_http_meta
 from ddtrace.ext import SpanTypes
+from ddtrace.internal.telemetry.constants import TELEMETRY_EVENT_TYPE
 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.trace import tracer
 import tests.appsec.rules as rules
 from tests.appsec.utils import asm_context
@@ -38,7 +37,7 @@ def _assert_generate_metrics(metrics_result, is_rule_triggered=False, is_blocked
     # this function and make the tests flaky. That's why we exclude the "enabled" metric from this assert
     generate_metrics = [
         m
-        for m in metrics_result[TELEMETRY_TYPE_GENERATE_METRICS][TELEMETRY_NAMESPACE.APPSEC.value]
+        for m in metrics_result[TELEMETRY_EVENT_TYPE.METRICS][TELEMETRY_NAMESPACE.APPSEC.value]
         if m["metric"] != "enabled"
     ]
     assert (
@@ -76,7 +75,7 @@ def _assert_generate_metrics(metrics_result, is_rule_triggered=False, is_blocked
 
 
 def _assert_distributions_metrics(metrics_result, is_rule_triggered=False, is_blocked_request=False):
-    distributions_metrics = metrics_result[TELEMETRY_TYPE_DISTRIBUTION][TELEMETRY_NAMESPACE.APPSEC.value]
+    distributions_metrics = metrics_result[TELEMETRY_EVENT_TYPE.DISTRIBUTIONS][TELEMETRY_NAMESPACE.APPSEC.value]
 
     assert len(distributions_metrics) == 2, "Expected 2 distributions_metrics"
     for metric in distributions_metrics:
@@ -101,8 +100,8 @@ def test_metrics_when_appsec_doesnt_runs(telemetry_writer, tracer):
                 rules.Config(),
             )
     metrics_data = telemetry_writer._namespace.flush()
-    assert len(metrics_data[TELEMETRY_TYPE_GENERATE_METRICS]) == 0
-    assert len(metrics_data[TELEMETRY_TYPE_DISTRIBUTION]) == 0
+    assert len(metrics_data[TELEMETRY_EVENT_TYPE.METRICS]) == 0
+    assert len(metrics_data[TELEMETRY_EVENT_TYPE.DISTRIBUTIONS]) == 0
 
 
 def test_metrics_when_appsec_runs(telemetry_writer, tracer):
@@ -185,7 +184,7 @@ def test_log_metric_error_ddwaf_timeout(telemetry_writer, tracer):
     list_metrics_logs = list(telemetry_writer._logs)
     assert len(list_metrics_logs) == 0
 
-    generate_metrics = telemetry_writer._namespace.flush()[TELEMETRY_TYPE_GENERATE_METRICS][
+    generate_metrics = telemetry_writer._namespace.flush()[TELEMETRY_EVENT_TYPE.METRICS][
         TELEMETRY_NAMESPACE.APPSEC.value
     ]
 
@@ -228,7 +227,7 @@ def test_log_metric_error_ddwaf_internal_error(telemetry_writer):
             assert len(list_telemetry_logs) == 0
             assert span.get_tag("_dd.appsec.waf.error") == "-3"
             metrics_result = telemetry_writer._namespace.flush()
-            list_telemetry_metrics = metrics_result.get(TELEMETRY_TYPE_GENERATE_METRICS, {}).get(
+            list_telemetry_metrics = metrics_result.get(TELEMETRY_EVENT_TYPE.METRICS, {}).get(
                 TELEMETRY_NAMESPACE.APPSEC.value, {}
             )
             error_metrics = [m for m in list_telemetry_metrics if m["metric"] == "waf.error"]

tests/appsec/iast/test_telemetry.py

@@ -21,8 +21,8 @@
 from ddtrace.appsec._iast.taint_sinks.header_injection import patch as header_injection_patch
 from ddtrace.appsec._iast.taint_sinks.weak_hash import patch as weak_hash_patch
 from ddtrace.ext import SpanTypes
+from ddtrace.internal.telemetry.constants import TELEMETRY_EVENT_TYPE
 from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE
-from ddtrace.internal.telemetry.constants import TELEMETRY_TYPE_GENERATE_METRICS
 from tests.appsec.iast.iast_utils import _iast_patched_module
 from tests.appsec.utils import asm_context
 from tests.utils import DummyTracer
@@ -31,7 +31,7 @@
 
 def _assert_instrumented_sink(telemetry_writer, vuln_type):
     metrics_result = telemetry_writer._namespace.flush()
-    generate_metrics = metrics_result[TELEMETRY_TYPE_GENERATE_METRICS][TELEMETRY_NAMESPACE.IAST.value]
+    generate_metrics = metrics_result[TELEMETRY_EVENT_TYPE.METRICS][TELEMETRY_NAMESPACE.IAST.value]
     assert len(generate_metrics) == 1, "Expected 1 generate_metrics"
     assert [metric["metric"] for metric in generate_metrics] == ["instrumented.sink"]
     assert [metric["tags"] for metric in generate_metrics] == [[f"vulnerability_type:{vuln_type.lower()}"]]
@@ -97,7 +97,7 @@ def test_metric_executed_sink(
         metrics_result = telemetry_writer._namespace.flush()
         _testing_unpatch_iast()
 
-    generate_metrics = metrics_result[TELEMETRY_TYPE_GENERATE_METRICS][TELEMETRY_NAMESPACE.IAST.value]
+    generate_metrics = metrics_result[TELEMETRY_EVENT_TYPE.METRICS][TELEMETRY_NAMESPACE.IAST.value]
     assert len(generate_metrics) == 1
     # Remove potential sinks from internal usage of the lib (like http.client, used to communicate with
     # the agent)
@@ -142,7 +142,7 @@ def test_metric_instrumented_propagation(no_request_sampling, telemetry_writer):
         _iast_patched_module("benchmarks.bm.iast_fixtures.str_methods")
 
     metrics_result = telemetry_writer._namespace.flush()
-    generate_metrics = metrics_result[TELEMETRY_TYPE_GENERATE_METRICS][TELEMETRY_NAMESPACE.IAST.value]
+    generate_metrics = metrics_result[TELEMETRY_EVENT_TYPE.METRICS][TELEMETRY_NAMESPACE.IAST.value]
     # Remove potential sinks from internal usage of the lib (like http.client, used to communicate with
     # the agent)
     filtered_metrics = [
@@ -170,7 +170,7 @@ def test_metric_request_tainted(no_request_sampling, telemetry_writer):
 
     metrics_result = telemetry_writer._namespace.flush()
 
-    generate_metrics = metrics_result[TELEMETRY_TYPE_GENERATE_METRICS][TELEMETRY_NAMESPACE.IAST.value]
+    generate_metrics = metrics_result[TELEMETRY_EVENT_TYPE.METRICS][TELEMETRY_NAMESPACE.IAST.value]
     # Remove potential sinks from internal usage of the lib (like http.client, used to communicate with
     # the agent)
     filtered_metrics = [metric["metric"] for metric in generate_metrics if metric["metric"] != "executed.sink"]

tests/telemetry/test_telemetry_metrics.py

@@ -3,18 +3,17 @@
 
 from mock.mock import ANY
 
+from ddtrace.internal.telemetry.constants import TELEMETRY_EVENT_TYPE
 from ddtrace.internal.telemetry.constants import TELEMETRY_LOG_LEVEL
 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 tests.utils import override_global_config
 
 
 def _assert_metric(
     test_agent,
     expected_metrics,
     namespace=TELEMETRY_NAMESPACE.TRACERS,
-    type_paypload=TELEMETRY_TYPE_GENERATE_METRICS,
+    type_paypload=TELEMETRY_EVENT_TYPE.METRICS,
 ):
     assert len(expected_metrics) > 0, "expected_metrics should not be empty"
     test_agent.telemetry_writer.periodic(force_flush=True)
@@ -291,7 +290,7 @@ def test_send_appsec_distributions_metric(telemetry_writer, test_agent_session,
         test_agent_session,
         expected_series,
         namespace=TELEMETRY_NAMESPACE.APPSEC,
-        type_paypload=TELEMETRY_TYPE_DISTRIBUTION,
+        type_paypload=TELEMETRY_EVENT_TYPE.DISTRIBUTIONS,
     )
 
 
@@ -312,7 +311,7 @@ def test_send_metric_flush_and_distributions_series_is_restarted(telemetry_write
         test_agent_session,
         expected_series,
         namespace=TELEMETRY_NAMESPACE.APPSEC,
-        type_paypload=TELEMETRY_TYPE_DISTRIBUTION,
+        type_paypload=TELEMETRY_EVENT_TYPE.DISTRIBUTIONS,
     )
 
     expected_series = [
@@ -329,7 +328,7 @@ def test_send_metric_flush_and_distributions_series_is_restarted(telemetry_write
         test_agent_session,
         expected_series,
         namespace=TELEMETRY_NAMESPACE.APPSEC,
-        type_paypload=TELEMETRY_TYPE_DISTRIBUTION,
+        type_paypload=TELEMETRY_EVENT_TYPE.DISTRIBUTIONS,
     )