feat(generic-metrics): Add generic metrics interface

src/sentry/sentry_metrics/kafka_metrics_interface.py

@@ -0,0 +1,42 @@
+from typing import Mapping
+
+from sentry.sentry_metrics.metrics_interface import GenericMetricsBackend
+from sentry.sentry_metrics.use_case_id_registry import UseCaseID
+
+
+class KafkaGenericMetricsBackend(GenericMetricsBackend):
+ def __init__(self) -> None:
+ super().__init__()
+
+ def counter(
+ self,
+ use_case_id: UseCaseID,
+ org_id: int,
+ project_id: int,
+ metric_name: str,
+ value: int,
+ tags: Mapping[str, str],
+ ) -> None:
+ pass
+
+ def set(
+ self,
+ use_case_id: UseCaseID,
+ org_id: int,
+ project_id: int,
+ metric_name: str,
+ value: int,
+ tags: Mapping[str, str],
+ ) -> None:
+ pass
+
+ def distribution(
+ self,
+ use_case_id: UseCaseID,
+ org_id: int,
+ project_id: int,
+ metric_name: str,
+ value: int,
+ tags: Mapping[str, str],
+ ) -> None:
+ pass

src/sentry/sentry_metrics/metrics_interface.py

@@ -0,0 +1,42 @@
+import abc
+from typing import Mapping
+
+from sentry.sentry_metrics.use_case_id_registry import UseCaseID
+
+
+class GenericMetricsBackend(metaclass=abc.ABCMeta):
+ @abc.abstractmethod
+ def counter(
+ self,
+ use_case_id: UseCaseID,
+ org_id: int,
+ project_id: int,
+ metric_name: str,
+ value: int,
+ tags: Mapping[str, str],
+ ) -> None:
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def set(
+ self,
+ use_case_id: UseCaseID,
+ org_id: int,
+ project_id: int,
+ metric_name: str,
+ value: int,
+ tags: Mapping[str, str],
+ ) -> None:
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def distribution(
+ self,
+ use_case_id: UseCaseID,
+ org_id: int,
+ project_id: int,
+ metric_name: str,
+ value: int,
+ tags: Mapping[str, str],
+ ) -> None:
+ raise NotImplementedError()

src/sentry/sentry_metrics/use_case_id_registry.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from typing import Any, Iterator, Mapping, Optional, Set
+
+from sentry.sentry_metrics.configuration import UseCaseKey
+
+_HARDCODED_USE_CASES = {"PERFORMANCE": "performance", "RELEASE_HEALTH": "release-health"}
+
+_REGISTERED_USE_CASES: dict[str, str] = {}
+
+
+class _UseCaseID(type):
+ def __getattr__(self, attr: str) -> UseCaseID:
+ if attr not in _HARDCODED_USE_CASES and attr not in _REGISTERED_USE_CASES:
+ raise AttributeError(attr)
+
+ return UseCaseID(attr.lower())
+
+ def __iter__(self) -> Iterator[UseCaseID]:
+ return iter(
+ UseCaseID(value)
+ for value in {
+ **_HARDCODED_USE_CASES,
+ **_REGISTERED_USE_CASES,
+ }.values()
+ )
+
+
+class UseCaseID(metaclass=_UseCaseID):
+ def __init__(self, value: str):
+ if (
+ value not in _HARDCODED_USE_CASES.values()
+ and value not in _REGISTERED_USE_CASES.values()
+ ):
+ raise ValueError("Passed use case has not been registered")
+
+ self.value = value
+
+ def __hash__(self) -> int:
+ return hash(self.value)
+
+ def __eq__(self, other: Any) -> bool:
+ return isinstance(other, UseCaseID) and other.value == self.value
+
+ def __repr__(self) -> str:
+ return f"UseCaseID.{self.value.upper()}"
+
+
+def register_use_case(key: str) -> UseCaseID:
+ _REGISTERED_USE_CASES[key.upper()] = key.lower()
+ return UseCaseID(key)
+
+
+METRIC_PATH_MAPPING: Mapping[UseCaseKey, Set[UseCaseID]] = {
+ UseCaseKey.RELEASE_HEALTH: {UseCaseID.RELEASE_HEALTH},
+ UseCaseKey.PERFORMANCE: {UseCaseID.PERFORMANCE},
+}
+
+
+def get_metric_path_from_usecase(use_case: UseCaseID) -> Optional[UseCaseKey]:
+ for metric_path_key, use_case_list in METRIC_PATH_MAPPING.items():
+ if use_case in use_case_list:
+ return metric_path_key
+ return None