OPS-5392 add datadog event support (#70399)

## Description
<!-- Describe your PR here. -->
I'd like to add support for [Datadog
Events](https://docs.datadoghq.com/service_management/events/guides/dogstatsd/).
Pasting context from
[OPS-5392](https://getsentry.atlassian.net/browse/OPS-5392):
> Currently back pressure sends metrics to Datadog
([https://app.datadoghq.com/dashboard/qnx-z63-3di/backpressure-monitoring](https://app.datadoghq.com/dashboard/qnx-z63-3di/backpressure-monitoring?fromUser=false&refresh_mode=paused&view=spans&from_ts=1709132829007&to_ts=1709137896990&live=false)).
We should send datadog events instead (or as well):
https://docs.datadoghq.com/service_management/events/
This would enable us to do things like:
>- Trigger DD monitors for alerting purposes (eg. notifying slack /
PagerDuty)
>- overlay events on top of dashboard widgets
>- add tags to events to support analysis / grouping

## (Potential) To Do
Added a unit test; please let me know if/how I can test this better.

[OPS-5392]:
https://getsentry.atlassian.net/browse/OPS-5392?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

---------

Co-authored-by: Riccardo Busetti <riccardo.busetti@sentry.io>

src/sentry/metrics/base.py

@@ -78,3 +78,17 @@ class MetricsBackend(local):
         stacklevel: int = 0,
     ) -> None:
         raise NotImplementedError
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        raise NotImplementedError

src/sentry/metrics/composite_experimental.py

@@ -130,3 +130,27 @@ class CompositeExperimentalMetricsBackend(MetricsBackend):
                 unit,
                 stacklevel=stacklevel + 1,
             )
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        self._primary_backend.event(
+            title,
+            message,
+            alert_type,
+            aggregation_key,
+            source_type_name,
+            priority,
+            instance,
+            tags,
+            stacklevel + 1,
+        )

src/sentry/metrics/datadog.py

@@ -111,3 +111,34 @@ class DatadogMetricsBackend(MetricsBackend):
     ) -> None:
         # We keep the same implementation for Datadog.
         self.timing(key, value, instance, tags, sample_rate)
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        tags = dict(tags or ())
+
+        if self.tags:
+            tags.update(self.tags)
+        if instance:
+            tags["instance"] = instance
+
+        tags_list = [f"{k}:{v}" for k, v in tags.items()]
+        self.stats.event(
+            title=title,
+            message=message,
+            alert_type=alert_type,
+            aggregation_key=aggregation_key,
+            source_type_name=source_type_name,
+            priority=priority,
+            tags=tags_list,
+            hostname=self.host,
+        )

src/sentry/metrics/dogstatsd.py

@@ -112,3 +112,34 @@ class DogStatsdMetricsBackend(MetricsBackend):
     ) -> None:
         # We keep the same implementation for Datadog.
         self.timing(key, value, instance, tags, sample_rate)
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        tags = dict(tags or ())
+
+        if self.tags:
+            tags.update(self.tags)
+        if instance:
+            tags["instance"] = instance
+
+        tags_list = [f"{k}:{v}" for k, v in tags.items()]
+        statsd.event(
+            title=title,
+            message=message,
+            alert_type=alert_type,
+            aggregation_key=aggregation_key,
+            source_type_name=source_type_name,
+            priority=priority,
+            tags=tags_list,
+            hostname=self.host,
+        )

src/sentry/metrics/dummy.py

@@ -50,3 +50,17 @@ class DummyMetricsBackend(MetricsBackend):
         stacklevel: int = 0,
     ) -> None:
         pass
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        pass

src/sentry/metrics/logging.py

@@ -54,3 +54,17 @@ class LoggingBackend(MetricsBackend):
         stacklevel: int = 0,
     ) -> None:
         logger.debug("%r: %+g", key, value, extra={"instance": instance, "tags": tags or {}})
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        logger.debug("%r: %+g", title, message, extra={"instance": instance, "tags": tags or {}})

src/sentry/metrics/middleware.py

@@ -189,3 +189,31 @@ class MiddlewareWrapper(MetricsBackend):
         return self.inner.distribution(
             key, value, instance, current_tags, sample_rate, unit, stacklevel + 1
         )
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        current_tags = get_current_global_tags()
+        if tags is not None:
+            current_tags.update(tags)
+
+        return self.inner.event(
+            title,
+            message,
+            alert_type,
+            aggregation_key,
+            source_type_name,
+            priority,
+            instance,
+            current_tags,
+            stacklevel + 1,
+        )

src/sentry/metrics/minimetrics.py

@@ -211,3 +211,17 @@ class MiniMetricsMetricsBackend(MetricsBackend):
                 unit=self._to_minimetrics_unit(unit=unit),
                 stacklevel=stacklevel + 1,
             )
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        pass

src/sentry/metrics/statsd.py

@@ -63,3 +63,17 @@ class StatsdMetricsBackend(MetricsBackend):
         stacklevel: int = 0,
     ) -> None:
         self.timing(key, value, instance, tags, sample_rate)
+
+    def event(
+        self,
+        title: str,
+        message: str,
+        alert_type: str | None = None,
+        aggregation_key: str | None = None,
+        source_type_name: str | None = None,
+        priority: str | None = None,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        pass

src/sentry/utils/metrics.py

@@ -231,3 +231,31 @@ def wraps(
         return inner  # type: ignore[return-value]
 
     return wrapper
+
+
+def event(
+    title: str,
+    message: str,
+    alert_type: str | None = None,
+    aggregation_key: str | None = None,
+    source_type_name: str | None = None,
+    priority: str | None = None,
+    instance: str | None = None,
+    tags: Tags | None = None,
+    stacklevel: int = 0,
+) -> None:
+    try:
+        backend.event(
+            title,
+            message,
+            alert_type,
+            aggregation_key,
+            source_type_name,
+            priority,
+            instance,
+            tags,
+            stacklevel + 1,
+        )
+    except Exception:
+        logger = logging.getLogger("sentry.errors")
+        logger.exception("Unable to record backend metric")