add a functions summaries dataset

snuba/clusters/storage_sets.py

@@ -21,6 +21,7 @@
  "GROUP_ATTRIBUTES": "group_attributes",
  "METRICS_SUMMARIES": "metrics_summaries",
  "PROFILE_CHUNKS": "profile_chunks",
+ "FUNCTIONS_SUMMARIES": "functions_summaries",
 }
 
 

snuba/migrations/group_loader.py

@@ -181,3 +181,8 @@ def __init__(self) -> None:
 class ProfileChunksLoader(DirectoryLoader):
  def __init__(self) -> None:
  super().__init__("snuba.snuba_migrations.profile_chunks")
+
+
+class FunctionsSummariesLoader(DirectoryLoader):
+ def __init__(self) -> None:
+ super().__init__("snuba.snuba_migrations.functions_summaries")

snuba/migrations/groups.py

@@ -8,6 +8,7 @@
  EventsAnalyticsPlatformLoader,
  EventsLoader,
  FunctionsLoader,
+ FunctionsSummariesLoader,
  GenericMetricsLoader,
  GroupAttributesLoader,
  GroupLoader,
@@ -47,6 +48,7 @@ class MigrationGroup(Enum):
  GROUP_ATTRIBUTES = "group_attributes"
  METRICS_SUMMARIES = "metrics_summaries"
  PROFILE_CHUNKS = "profile_chunks"
+ FUNCTIONS_SUMMARIES = "functions_summaries"
 
 
 # Migration groups are mandatory by default. Specific groups can
@@ -64,6 +66,7 @@ class MigrationGroup(Enum):
  MigrationGroup.GROUP_ATTRIBUTES,
  MigrationGroup.METRICS_SUMMARIES,
  MigrationGroup.PROFILE_CHUNKS,
+ MigrationGroup.FUNCTIONS_SUMMARIES,
 }
 
 
@@ -186,6 +189,11 @@ def __init__(
  storage_sets_keys={StorageSetKey.PROFILE_CHUNKS},
  readiness_state=ReadinessState.PARTIAL,
  ),
+ MigrationGroup.FUNCTIONS_SUMMARIES: _MigrationGroup(
+ loader=FunctionsSummariesLoader(),
+ storage_sets_keys={StorageSetKey.FUNCTIONS_SUMMARIES},
+ readiness_state=ReadinessState.EXPERIMENTAL,
+ ),
 }
 
 

snuba/settings/__init__.py

@@ -116,6 +116,7 @@
  "generic_metrics_gauges",
  "metrics_summaries",
  "profile_chunks",
+ "functions_summaries",
  },
  "single_node": True,
  },

snuba/settings/settings_distributed.py

@@ -36,6 +36,7 @@
  "generic_metrics_gauges",
  "metrics_summaries",
  "profile_chunks",
+ "functions_summaries",
  },
  "single_node": False,
  "cluster_name": "cluster_one_sh",

snuba/snuba_migrations/functions_summaries/0001_functions_summaries_create_table.py

@@ -0,0 +1,78 @@
+from typing import List, Sequence
+
+from snuba.clickhouse.columns import UUID, Column, DateTime, 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 Float
+
+storage_set_name = StorageSetKey.FUNCTIONS_SUMMARIES
+local_table_name = "functions_summaries_local"
+dist_table_name = "functions_summaries_dist"
+
+columns: List[Column[Modifiers]] = [
+ Column("project_id", UInt(64)),
+ # profile_id is nullable since this will only be used by transaction-based profiling
+ Column("profile_id", UUID(Modifiers(nullable=True, low_cardinality=False))),
+ # profiler_id is nullable since this will only be used by continuous profiling
+ Column("profiler_id", UUID(Modifiers(nullable=True, low_cardinality=False))),
+ # transaction_name is nullable since this will only be used by transaction-based profiling
+ Column("transaction_name", String(Modifiers(nullable=True, low_cardinality=True))),
+ Column("thread_id", String()),
+ Column("min", Float(64)),
+ Column("max", Float(64)),
+ Column("sum", Float(64)),
+ Column("count", UInt(64)),
+ Column("end_timestamp", DateTime()),
+ Column("platform", String(Modifiers(low_cardinality=True))),
+ Column("environment", String(Modifiers(nullable=True, low_cardinality=True))),
+ Column("release", String(Modifiers(nullable=True, low_cardinality=True))),
+ # snuba internals
+ Column("retention_days", UInt(16)),
+]
+
+
+class Migration(migration.ClickhouseNodeMigration):
+ blocking = False
+
+ def forwards_ops(self) -> Sequence[SqlOperation]:
+ return [
+ operations.CreateTable(
+ storage_set=storage_set_name,
+ table_name=local_table_name,
+ columns=columns,
+ engine=table_engines.MergeTree(
+ order_by="(project_id, end_timestamp)",
+ partition_by="(retention_days, toMonday(end_timestamp))",
+ settings={"index_granularity": "8192"},
+ storage_set=storage_set_name,
+ ttl="end_timestamp + toIntervalDay(retention_days)",
+ ),
+ target=OperationTarget.LOCAL,
+ ),
+ operations.CreateTable(
+ storage_set=storage_set_name,
+ table_name=dist_table_name,
+ columns=columns,
+ engine=table_engines.Distributed(
+ local_table_name=local_table_name,
+ sharding_key=None,
+ ),
+ target=OperationTarget.DISTRIBUTED,
+ ),
+ ]
+
+ def backwards_ops(self) -> Sequence[SqlOperation]:
+ return [
+ operations.DropTable(
+ storage_set=storage_set_name,
+ table_name=local_table_name,
+ target=OperationTarget.LOCAL,
+ ),
+ operations.DropTable(
+ storage_set=storage_set_name,
+ table_name=dist_table_name,
+ target=OperationTarget.DISTRIBUTED,
+ ),
+ ]