ref: Remove dead metrics processor code (#5340)

* ref: Remove dead code

Remove unused metrics processor code

* can i delete all this?

* this is not a writable storage

* cleanup test

* update test

* workaround for multiple-insertion issue in metrics API tests (#5424)

* test

---------

Co-authored-by: Oliver Newland <oliver.newland@sentry.io>

snuba/datasets/configuration/metrics/storages/counters.yaml

@@ -1,5 +1,5 @@
 version: v1
-kind: writable_storage
+kind: readable_storage
 name: metrics_counters
 storage:
   key: metrics_counters
@@ -48,7 +48,3 @@ query_processors:
     args:
       column_name: tags
   - processor: TableRateLimit
-stream_loader:
-  processor: CounterAggregateProcessor
-  default_topic: snuba-metrics
-  dlq_topic: snuba-dead-letter-metrics-counters

snuba/datasets/configuration/metrics/storages/distributions.yaml

@@ -1,5 +1,5 @@
 version: v1
-kind: writable_storage
+kind: readable_storage
 name: metrics_distributions
 storage:
   key: metrics_distributions
@@ -77,7 +77,3 @@ query_processors:
     args:
       column_name: tags
   - processor: TableRateLimit
-stream_loader:
-  processor: DistributionsAggregateProcessor
-  default_topic: snuba-metrics
-  dlq_topic: snuba-dead-letter-metrics-distributions

snuba/datasets/configuration/metrics/storages/sets.yaml

@@ -1,5 +1,5 @@
 version: v1
-kind: writable_storage
+kind: readable_storage
 name: metrics_sets
 storage:
   key: metrics_sets
@@ -52,7 +52,3 @@ query_processors:
     args:
       column_name: tags
   - processor: TableRateLimit
-stream_loader:
-  processor: SetsAggregateProcessor
-  default_topic: snuba-metrics
-  dlq_topic: snuba-dead-letter-metrics-sets

snuba/datasets/processors/metrics_bucket_processor.py

@@ -3,24 +3,20 @@
 from datetime import datetime
 from typing import Any, Iterable, Mapping, Optional, Tuple, Union
 
-from snuba import settings
 from snuba.consumers.types import KafkaMessageMetadata
 from snuba.datasets.events_format import EventTooOld, enforce_retention
 from snuba.datasets.metrics_messages import (
-    ILLEGAL_VALUE_IN_DIST,
-    ILLEGAL_VALUE_IN_SET,
-    INT_EXPECTED,
     INT_FLOAT_EXPECTED,
     InputType,
     OutputType,
-    is_set_message,
     values_for_distribution_message,
     values_for_set_message,
 )
 from snuba.datasets.processors import DatasetMessageProcessor
 from snuba.processor import InsertBatch, ProcessedMessage, _ensure_valid_date
 
-DISABLED_MATERIALIZATION_VERSION = 1
+ENABLED_MATERIALIZATION_VERSION = 4
+
 ILLEGAL_VALUE_FOR_COUNTER = "Illegal value for counter value."
 
 
@@ -88,11 +84,7 @@ def process_message(
             keys.append(int(key))
             values.append(value)
 
-        mat_version = (
-            DISABLED_MATERIALIZATION_VERSION
-            if settings.WRITE_METRICS_AGG_DIRECTLY
-            else settings.ENABLED_MATERIALIZATION_VERSION
-        )
+        mat_version = ENABLED_MATERIALIZATION_VERSION
 
         try:
             retention_days = enforce_retention(message["retention_days"], timestamp)
@@ -125,44 +117,6 @@ def process_message(
         )
 
 
-class SetsMetricsProcessor(MetricsBucketProcessor):
-    def _should_process(self, message: Mapping[str, Any]) -> bool:
-        return is_set_message(message)
-
-    def _process_values(self, message: Mapping[str, Any]) -> Mapping[str, Any]:
-        values = message["value"]
-        for value in values:
-            assert isinstance(
-                value, int
-            ), f"{ILLEGAL_VALUE_IN_SET} {INT_EXPECTED}: {value}"
-        return {"set_values": values}
-
-
-class CounterMetricsProcessor(MetricsBucketProcessor):
-    def _should_process(self, message: Mapping[str, Any]) -> bool:
-        return message["type"] is not None and message["type"] == "c"
-
-    def _process_values(self, message: Mapping[str, Any]) -> Mapping[str, Any]:
-        value = message["value"]
-        assert isinstance(
-            value, (int, float)
-        ), f"{ILLEGAL_VALUE_FOR_COUNTER} {INT_FLOAT_EXPECTED}: {value}"
-        return {"value": value}
-
-
-class DistributionsMetricsProcessor(MetricsBucketProcessor):
-    def _should_process(self, message: Mapping[str, Any]) -> bool:
-        return message["type"] is not None and message["type"] == "d"
-
-    def _process_values(self, message: Mapping[str, Any]) -> Mapping[str, Any]:
-        values = message["value"]
-        for value in values:
-            assert isinstance(
-                value, (int, float)
-            ), f"{ILLEGAL_VALUE_IN_DIST} {INT_FLOAT_EXPECTED}: {value}"
-        return {"values": values}
-
-
 class PolymorphicMetricsProcessor(MetricsBucketProcessor):
     def _should_process(self, message: Mapping[str, Any]) -> bool:
         return message["type"] in {
@@ -182,3 +136,9 @@ def _process_values(self, message: Mapping[str, Any]) -> Mapping[str, Any]:
             return {"metric_type": OutputType.COUNTER.value, "count_value": value}
         else:  # message["type"] == InputType.DISTRIBUTION.value
             return values_for_distribution_message(message)
+
+
+def timestamp_to_bucket(timestamp: datetime, interval_seconds: int) -> datetime:
+    time_seconds = timestamp.timestamp()
+    out_seconds = interval_seconds * (time_seconds // interval_seconds)
+    return datetime.fromtimestamp(out_seconds, timestamp.tzinfo)

snuba/settings/__init__.py

@@ -338,11 +338,6 @@ class RedisClusters(TypedDict):
 # (entity name, buffer size)
 SUBSCRIPTIONS_ENTITY_BUFFER_SIZE: Mapping[str, int] = {}
 
-# Used for migrating to/from writing metrics directly to aggregate tables
-# rather than using materialized views
-WRITE_METRICS_AGG_DIRECTLY = False
-ENABLED_MATERIALIZATION_VERSION = 4
-
 # Enable profiles ingestion
 ENABLE_PROFILES_CONSUMER = os.environ.get("ENABLE_PROFILES_CONSUMER", False)
 

tests/admin/dead_letter_queue/test_dlq.py

@@ -2,4 +2,4 @@
 
 
 def test_dlq() -> None:
-    assert len(get_dlq_topics()) == 8
+    assert len(get_dlq_topics()) == 5