feat(uptime): add initial table migration

scripts/load_uptime_checks.py

@@ -13,14 +13,14 @@
 query = """
 INSERT INTO default.uptime_monitor_checks_local (
     organization_id, project_id, environment, uptime_subscription_id, uptime_check_id,
-    scheduled_check_time, timestamp, duration, region_id, check_status,
+    scheduled_check_time, timestamp, duration_ms, region_slug, check_status,
     check_status_reason, http_status_code, trace_id, retention_days
 ) FORMAT JSONEachRow
 """
 
 total_records = 0
 
-for project_id in range(1, 2):
+for project_id in range(2, 100):
     project_data = []
     for minute in range(24 * 60 * 90):  # 24 hours * 60 minutes * 90 days
         timestamp = base_time + datetime.timedelta(minutes=minute)
@@ -36,10 +36,12 @@
                 "environment": "production",
                 "uptime_subscription_id": random.randint(1, 3) * project_id,
                 "uptime_check_id": str(uuid.uuid4()),
-                "scheduled_check_time": scheduled_time.strftime("%Y-%m-%d %H:%M:%S"),
-                "timestamp": timestamp.strftime("%Y-%m-%d %H:%M:%S"),
-                "duration": random.randint(1, 1000),
-                "region_id": random.randint(1, 3),
+                "scheduled_check_time": scheduled_time.strftime("%Y-%m-%d %H:%M:%S.%f")[
+                    :-3
+                ],
+                "timestamp": timestamp.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
+                "duration_ms": random.randint(1, 1000),
+                "region_slug": f"region-{random.randint(1, 3)}",
                 "check_status": check_status,
                 "check_status_reason": "Timeout error"
                 if check_status == "failure"

snuba/snuba_migrations/uptime_monitor_checks/0001_uptime_monitor_checks.py

@@ -0,0 +1,80 @@
+from typing import List, Sequence
+
+from snuba.clickhouse.columns import UUID, Column, String, UInt
+from snuba.clusters.storage_sets import StorageSetKey
+from snuba.migrations import migration, operations, table_engines
+from snuba.migrations.columns import MigrationModifiers as Modifiers
+from snuba.migrations.operations import OperationTarget, SqlOperation
+from snuba.utils.schemas import DateTime64
+
+storage_set = StorageSetKey.UPTIME_MONITOR_CHECKS
+table_prefix = "uptime_monitor_checks"
+local_table_name = f"{table_prefix}_local"
+dist_table_name = f"{table_prefix}_dist"
+
+## what about all the fancy codecs? do we need those?
+columns: List[Column[Modifiers]] = [
+    Column("organization_id", UInt(64)),
+    Column("project_id", UInt(64)),
+    Column("environment", String(Modifiers(nullable=True, low_cardinality=True))),
+    Column("uptime_subscription_id", UUID()),
+    Column("uptime_check_id", UUID()),
+    Column("scheduled_check_time", DateTime64(3)),  # millisecond precision
+    Column("timestamp", DateTime64(3)),  # millisecond precision
+    Column("duration_ms", UInt(64)),
+    Column("region_slug", String(Modifiers(low_cardinality=True))),
+    Column("check_status", String(Modifiers(low_cardinality=True))),
+    Column(
+        "check_status_reason",
+        String(Modifiers(nullable=True, low_cardinality=True)),
+    ),
+    Column("http_status_code", UInt(16)),
+    Column("trace_id", UUID()),
+    Column("retention_days", UInt(16)),
+]
+
+
+class Migration(migration.ClickhouseNodeMigration):
+    blocking = False
+
+    def forwards_ops(self) -> Sequence[SqlOperation]:
+        return [
+            operations.CreateTable(
+                storage_set=storage_set,
+                table_name=local_table_name,
+                columns=columns,
+                engine=table_engines.ReplacingMergeTree(
+                    primary_key="(organization_id, project_id, toDateTime(timestamp), uptime_check_id, trace_id)",
+                    order_by="(organization_id, project_id, toDateTime(timestamp), uptime_check_id, trace_id)",
+                    partition_by="(retention_days, toMonday(timestamp))",
+                    settings={"index_granularity": "8192"},
+                    storage_set=storage_set,
+                    ttl="toDateTime(timestamp) + toIntervalDay(retention_days)",
+                ),
+                target=OperationTarget.LOCAL,
+            ),
+            operations.CreateTable(
+                storage_set=storage_set,
+                table_name=dist_table_name,
+                columns=columns,
+                engine=table_engines.Distributed(
+                    local_table_name=local_table_name,
+                    sharding_key="cityHash64(reinterpretAsUInt128(trace_id))",
+                ),
+                target=OperationTarget.DISTRIBUTED,
+            ),
+        ]
+
+    def backwards_ops(self) -> Sequence[SqlOperation]:
+        return [
+            operations.DropTable(
+                storage_set=storage_set,
+                table_name=dist_table_name,
+                target=OperationTarget.DISTRIBUTED,
+            ),
+            operations.DropTable(
+                storage_set=storage_set,
+                table_name=local_table_name,
+                target=OperationTarget.LOCAL,
+            ),
+        ]