feat(generic-metrics): Add support for subscriptions to gauges

snuba/cli/devserver.py

@@ -319,6 +319,19 @@ def devserver(*, bootstrap: bool, workers: bool) -> None:
                             "--schedule-ttl=10",
                         ],
                     ),
+                    (
+                        "subscriptions-scheduler-generic-metrics-gauges",
+                        [
+                            "snuba",
+                            "subscriptions-scheduler",
+                            "--entity=generic_metrics_gauges",
+                            "--consumer-group=snuba-generic-metrics-gauges-subscriptions-schedulers",
+                            "--followed-consumer-group=snuba-generic-metrics-gauges-consumers",
+                            "--auto-offset-reset=latest",
+                            "--log-level=debug",
+                            "--schedule-ttl=10",
+                        ],
+                    ),
                     (
                         "subscriptions-executor-metrics",
                         [

snuba/datasets/configuration/generic_metrics/entities/gauges.yaml

@@ -26,11 +26,7 @@ schema:
     {
       name: tags.raw_value,
       type: Array,
-      args:
-        {
-          schema_modifiers: [readonly],
-          inner_type: { type: String },
-        },
+      args: { schema_modifiers: [readonly], inner_type: { type: String } },
     },
     {
       name: min,
@@ -259,3 +255,14 @@ validators:
       required_filter_columns: ["org_id", "project_id"]
 required_time_column: timestamp
 partition_key_column_name: org_id
+subscription_processors:
+  - processor: AddColumnCondition
+    args:
+      extra_condition_data_key: organization
+      extra_condition_column: org_id
+subscription_validators:
+  - validator: AggregationValidator
+    args:
+      max_allowed_aggregations: 3
+      disallowed_aggregations: [having, orderby]
+      required_time_column: timestamp

snuba/datasets/configuration/generic_metrics/storages/gauges_bucket.yaml

@@ -78,4 +78,10 @@ stream_loader:
     args:
       header_key: metric_type
       header_value: g
+  commit_log_topic: snuba-generic-metrics-gauges-commit-log
+  subscription_scheduler_mode: global
+  subscription_synchronization_timestamp: orig_message_ts
+  subscription_scheduled_topic: scheduled-subscriptions-generic-metrics-gauges
+  subscription_result_topic: generic-metrics-subscription-results
+  subscription_delay_seconds: 60
   dlq_topic: snuba-dead-letter-generic-metrics

snuba/utils/streams/topics.py

@@ -31,6 +31,9 @@ class Topic(Enum):
     SUBSCRIPTION_SCHEDULED_GENERIC_METRICS_COUNTERS = (
         "scheduled-subscriptions-generic-metrics-counters"
     )
+    SUBSCRIPTION_SCHEDULED_GENERIC_METRICS_GAUGES = (
+        "scheduled-subscriptions-generic-metrics-gauges"
+    )
 
     SUBSCRIPTION_RESULTS_EVENTS = "events-subscription-results"
     SUBSCRIPTION_RESULTS_TRANSACTIONS = "transactions-subscription-results"
@@ -49,6 +52,7 @@ class Topic(Enum):
         "snuba-generic-metrics-distributions-commit-log"
     )
     GENERIC_METRICS_COUNTERS_COMMIT_LOG = "snuba-generic-metrics-counters-commit-log"
+    GENERIC_METRICS_GAUGES_COMMIT_LOG = "snuba-generic-metrics-gauges-commit-log"
     GENERIC_EVENTS = "generic-events"
     GENERIC_EVENTS_COMMIT_LOG = "snuba-generic-events-commit-log"
     GROUP_ATTRIBUTES = "group-attributes"