feat(kafka-control-plane): bump sentry-kafka-schemas and add topic te…

…st (#5711)

Validates that all topics are registered in sentry-kafka-schemas

requirements.txt

@@ -23,7 +23,7 @@ python-dateutil==2.8.2
 python-rapidjson==1.8
 redis==4.3.4
 sentry-arroyo==2.17.1
-sentry-kafka-schemas==0.1.65
+sentry-kafka-schemas==0.1.67
 sentry-redis-tools==0.3.0
 sentry-relay==0.8.44
 sentry-sdk==1.40.5

rust_snuba/Cargo.toml

@@ -33,7 +33,7 @@ pyo3 = { version = "0.18.1", features = ["chrono"] }
 reqwest = { version = "0.11.11", features = ["stream"] }
 rust_arroyo = { version = "*", git = "https://github.com/getsentry/arroyo" }
 sentry = { version = "0.32.0", features = ["anyhow", "tracing"] }
-sentry-kafka-schemas = "0.1.60"
+sentry-kafka-schemas = "0.1.67"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0" }
 thiserror = "1.0"

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@snuba-queries-QuerylogProcessor-snuba-queries__1__snuba-queries-empty-trace-id.json.snap

@@ -0,0 +1,102 @@
+---
+source: src/processors/mod.rs
+description: "{\n \"request\": {\n \"id\": \"24a78d10a0134f2aa6367ba2a393b504\",\n \"body\": {\n \"legacy\": true,\n \"query\": \"MATCH (events) SELECT count() AS `count`, min(timestamp) AS `first_seen`, max(timestamp) AS `last_seen` BY tags_key, tags_value WHERE timestamp >= toDateTime('2023-02-08T21:07:12.769001') AND timestamp < toDateTime('2023-02-08T21:12:39.015094') AND project_id IN tuple(1) AND project_id IN tuple(1) AND group_id IN tuple(5) ORDER BY count DESC LIMIT 4 BY tags_key\",\n \"dataset\": \"events\",\n \"app_id\": \"legacy\",\n \"parent_api\": \"/api/0/issues|groups/{issue_id}/tags/\"\n },\n \"referrer\": \"tagstore.__get_tag_keys_and_top_values\",\n \"team\": \"<unknown>\",\n \"feature\": \"<unknown>\",\n \"app_id\": \"legacy\"\n },\n \"dataset\": \"events\",\n \"entity\": \"events\",\n \"start_timestamp\": 1675919232,\n \"end_timestamp\": 1675919559,\n \"query_list\": [\n {\n \"sql\": \"SELECT (tupleElement((arrayJoin(arrayMap((x, y -> (x, y)), tags.key, tags.value)) AS snuba_all_tags), 1) AS _snuba_tags_key), (tupleElement(snuba_all_tags, 2) AS _snuba_tags_value), (count() AS _snuba_count), (min((timestamp AS _snuba_timestamp)) AS _snuba_first_seen), (max(_snuba_timestamp) AS _snuba_last_seen) FROM errors_local PREWHERE in((group_id AS _snuba_group_id), tuple(5)) WHERE equals(deleted, 0) AND greaterOrEquals(_snuba_timestamp, toDateTime('2023-02-08T21:07:12', 'Universal')) AND less(_snuba_timestamp, toDateTime('2023-02-08T21:12:39', 'Universal')) AND in((project_id AS _snuba_project_id), tuple(1)) AND in(_snuba_project_id, tuple(1)) GROUP BY _snuba_tags_key, _snuba_tags_value ORDER BY _snuba_count DESC LIMIT 4 BY _snuba_tags_key LIMIT 1000 OFFSET 0\",\n \"sql_anonymized\": \"SELECT (tupleElement((arrayJoin(arrayMap((x, y -> (x, y)), tags.key, tags.value)) AS snuba_all_tags), -1337) AS _snuba_tags_key), (tupleElement(snuba_all_tags, -1337) AS _snuba_tags_value), (count() AS _snuba_count), (min((timestamp AS _snuba_timestamp)) AS _snuba_first_seen), (max(_snuba_timestamp) AS _snuba_last_seen) FROM errors_local PREWHERE in((group_id AS _snuba_group_id), tuple(-1337)) WHERE equals(deleted, -1337) AND greaterOrEquals(_snuba_timestamp, toDateTime('2023-02-08T21:07:12', 'Universal')) AND less(_snuba_timestamp, toDateTime('2023-02-08T21:12:39', 'Universal')) AND in((project_id AS _snuba_project_id), tuple(-1337)) AND in(_snuba_project_id, tuple(-1337)) GROUP BY _snuba_tags_key, _snuba_tags_value ORDER BY _snuba_count DESC LIMIT 4 BY _snuba_tags_key LIMIT 1000 OFFSET 0\",\n \"start_timestamp\": 1675919232,\n \"end_timestamp\": 1675919559,\n \"stats\": {\n \"clickhouse_table\": \"errors_local\",\n \"final\": false,\n \"referrer\": \"tagstore.__get_tag_keys_and_top_values\",\n \"sample\": null,\n \"table_rate\": 0.6,\n \"table_concurrent\": 1,\n \"project_rate\": 0.6333333333333333,\n \"project_concurrent\": 1,\n \"consistent\": false,\n \"result_rows\": 22,\n \"result_cols\": 5,\n \"query_id\": \"9079915acbacff0804ed45c72b865024\"\n },\n \"status\": \"success\",\n \"trace_id\": \"\",\n \"profile\": {\n \"time_range\": null,\n \"table\": \"errors_local\",\n \"all_columns\": [\n \"errors_local.deleted\",\n \"errors_local.group_id\",\n \"errors_local.project_id\",\n \"errors_local.tags.key\",\n \"errors_local.tags.value\",\n \"errors_local.timestamp\"\n ],\n \"multi_level_condition\": false,\n \"where_profile\": {\n \"columns\": [\n \"errors_local.deleted\",\n \"errors_local.project_id\",\n \"errors_local.timestamp\"\n ],\n \"mapping_cols\": []\n },\n \"groupby_cols\": [\"errors_local.tags.key\", \"errors_local.tags.value\"],\n \"array_join_cols\": [\"errors_local.tags.key\", \"errors_local.tags.value\"]\n },\n \"result_profile\": {\n \"bytes\": 1305,\n \"blocks\": 1,\n \"rows\": 22,\n \"elapsed\": 0.009863138198852539\n },\n \"request_status\": \"success\",\n \"slo\": \"for\"\n }\n ],\n \"status\": \"success\",\n \"request_status\": \"success\",\n \"slo\": \"for\",\n \"timing\": {\n \"timestamp\": 1675890758,\n \"duration_ms\": 55,\n \"marks_ms\": {\n \"cache_get\": 2,\n \"cache_set\": 6,\n \"execute\": 10,\n \"get_configs\": 0,\n \"prepare_query\": 15,\n \"rate_limit\": 5,\n \"validate_schema\": 15\n },\n \"tags\": {}\n },\n \"projects\": [1],\n \"snql_anonymized\": \"MATCH Entity(events) SELECT tags_key, tags_value, (count() AS count), (min(timestamp) AS first_seen), (max(timestamp) AS last_seen) GROUP BY tags_key, tags_value WHERE greaterOrEquals(timestamp, toDateTime('$S')) AND less(timestamp, toDateTime('$S')) AND in(project_id, tuple(-1337)) AND in(project_id, tuple(-1337)) AND in(group_id, tuple(-1337)) ORDER BY count DESC LIMIT 4 BY tags_key LIMIT 1000 OFFSET 0\"\n}\n"
+expression: snapshot_payload
+---
+[
+ {
+ "clickhouse_queries.all_columns": [
+ [
+ "errors_local.deleted",
+ "errors_local.group_id",
+ "errors_local.project_id",
+ "errors_local.tags.key",
+ "errors_local.tags.value",
+ "errors_local.timestamp"
+ ]
+ ],
+ "clickhouse_queries.array_join_columns": [
+ [
+ "errors_local.tags.key",
+ "errors_local.tags.value"
+ ]
+ ],
+ "clickhouse_queries.bytes_scanned": [
+ 1305
+ ],
+ "clickhouse_queries.cache_hit": [
+ 0
+ ],
+ "clickhouse_queries.clickhouse_table": [
+ "errors_local"
+ ],
+ "clickhouse_queries.consistent": [
+ 0
+ ],
+ "clickhouse_queries.duration_ms": [
+ 9
+ ],
+ "clickhouse_queries.final": [
+ 0
+ ],
+ "clickhouse_queries.groupby_columns": [
+ [
+ "errors_local.tags.key",
+ "errors_local.tags.value"
+ ]
+ ],
+ "clickhouse_queries.is_duplicate": [
+ 0
+ ],
+ "clickhouse_queries.max_threads": [
+ 0
+ ],
+ "clickhouse_queries.num_days": [
+ 0
+ ],
+ "clickhouse_queries.or_conditions": [
+ 0
+ ],
+ "clickhouse_queries.query_id": [
+ "9079915acbacff0804ed45c72b865024"
+ ],
+ "clickhouse_queries.sample": [
+ 0.0
+ ],
+ "clickhouse_queries.sql": [
+ "SELECT (tupleElement((arrayJoin(arrayMap((x, y -> (x, y)), tags.key, tags.value)) AS snuba_all_tags), 1) AS _snuba_tags_key), (tupleElement(snuba_all_tags, 2) AS _snuba_tags_value), (count() AS _snuba_count), (min((timestamp AS _snuba_timestamp)) AS _snuba_first_seen), (max(_snuba_timestamp) AS _snuba_last_seen) FROM errors_local PREWHERE in((group_id AS _snuba_group_id), tuple(5)) WHERE equals(deleted, 0) AND greaterOrEquals(_snuba_timestamp, toDateTime('2023-02-08T21:07:12', 'Universal')) AND less(_snuba_timestamp, toDateTime('2023-02-08T21:12:39', 'Universal')) AND in((project_id AS _snuba_project_id), tuple(1)) AND in(_snuba_project_id, tuple(1)) GROUP BY _snuba_tags_key, _snuba_tags_value ORDER BY _snuba_count DESC LIMIT 4 BY _snuba_tags_key LIMIT 1000 OFFSET 0"
+ ],
+ "clickhouse_queries.stats": [
+ "{\"clickhouse_table\":\"errors_local\",\"consistent\":false,\"final\":false,\"project_concurrent\":1,\"project_rate\":0.6333333333333333,\"query_id\":\"9079915acbacff0804ed45c72b865024\",\"referrer\":\"tagstore.__get_tag_keys_and_top_values\",\"result_cols\":5,\"result_rows\":22,\"sample\":null,\"table_concurrent\":1,\"table_rate\":0.6}"
+ ],
+ "clickhouse_queries.status": [
+ "success"
+ ],
+ "clickhouse_queries.trace_id": [
+ "00000000-0000-0000-0000-000000000000"
+ ],
+ "clickhouse_queries.where_columns": [
+ [
+ "errors_local.deleted",
+ "errors_local.project_id",
+ "errors_local.timestamp"
+ ]
+ ],
+ "clickhouse_queries.where_mapping_columns": [
+ []
+ ],
+ "dataset": "events",
+ "duration_ms": 55,
+ "offset": 1,
+ "organization": null,
+ "partition": 0,
+ "projects": [
+ 1
+ ],
+ "referrer": "tagstore.__get_tag_keys_and_top_values",
+ "request_body": "{\"app_id\":\"legacy\",\"dataset\":\"events\",\"legacy\":true,\"parent_api\":\"/api/0/issues|groups/{issue_id}/tags/\",\"query\":\"MATCH (events) SELECT count() AS `count`, min(timestamp) AS `first_seen`, max(timestamp) AS `last_seen` BY tags_key, tags_value WHERE timestamp >= toDateTime('2023-02-08T21:07:12.769001') AND timestamp < toDateTime('2023-02-08T21:12:39.015094') AND project_id IN tuple(1) AND project_id IN tuple(1) AND group_id IN tuple(5) ORDER BY count DESC LIMIT 4 BY tags_key\"}",
+ "request_id": "24a78d10-a013-4f2a-a636-7ba2a393b504",
+ "status": "success",
+ "timestamp": 1675890758
+ }
+]

tests/utils/streams/test_topics.py

@@ -0,0 +1,30 @@
+import sentry_kafka_schemas
+
+from snuba.utils.streams.topics import Topic
+
+
+def test_valid_topics() -> None:
+ # Ensures that Snuba's topic list matches those registered in sentry-kafka-schemas
+ for topic in Topic:
+ try:
+ sentry_kafka_schemas.get_topic(
+ topic.value
+ ) # Throws an exception if topic not defined
+ except sentry_kafka_schemas.SchemaNotFound:
+ # These topics are not in use but have not yet been removed from snuba's codebase
+ deprecated_topics = (
+ Topic.CDC,
+ Topic.SESSIONS,
+ Topic.SESSIONS_COMMIT_LOG,
+ Topic.SUBSCRIPTION_SCHEDULED_SESSIONS,
+ Topic.DEAD_LETTER_SESSIONS,
+ )
+
+ # TODO: Remove this once these missing topics are added to sentry-kafka-schemas
+ not_yet_defined = (
+ Topic.EVENT_REPLACEMENTS,
+ Topic.ATTRIBUTION,
+ )
+
+ if topic not in deprecated_topics and topic not in not_yet_defined:
+ raise