ref(rust): Remove Python processors in favor of Rust ones

requirements.txt

@@ -23,7 +23,7 @@ python-dateutil==2.8.2
 python-rapidjson==1.8
 redis==4.3.4
 sentry-arroyo==2.15.3
-sentry-kafka-schemas==0.1.38
+sentry-kafka-schemas==0.1.41
 sentry-redis-tools==0.1.7
 sentry-relay==0.8.39
 sentry-sdk==1.28.0

rust_snuba/src/processors/functions.rs

@@ -100,7 +100,6 @@ struct Function<'a> {
     http_method: Option<&'a str>,
     is_application: u8,
     materialization_version: u8,
-    module: &'a str,
     name: &'a str,
     package: &'a str,
     platform: &'a str,

rust_snuba/src/processors/spans.rs

@@ -74,6 +74,7 @@ struct FromSentryTags {
     transaction_method: Option<String>,
     #[serde(rename(deserialize = "transaction.op"))]
     transaction_op: Option<String>,
+    user: Option<String>,
     #[serde(flatten)]
     extra: BTreeMap<String, String>,
 }
@@ -130,6 +131,10 @@ impl FromSentryTags {
             tags.insert("status_code".into(), status_code.into());
         }
 
+        if let Some(user) = &self.user {
+            tags.insert("user".into(), user.into());
+        }
+
         for (key, value) in &self.extra {
             tags.insert(key.into(), value.into());
         }
@@ -203,10 +208,11 @@ impl TryFrom<FromSpanMessage> for Span {
             0
         };
         let status = from.sentry_tags.status.unwrap_or_default() as u8;
+
         let (sentry_tag_keys, sentry_tag_values) = from.sentry_tags.to_keys_values();
         let transaction_op = from.sentry_tags.transaction_op.unwrap_or_default();
 
-        let (mut tag_keys, mut tag_values): (Vec<_>, Vec<_>) =
+        let (tag_keys, tag_values): (Vec<_>, Vec<_>) =
             from.tags.unwrap_or_default().into_iter().unzip();
 
         let (measurement_keys, measurement_values) = from
@@ -216,21 +222,6 @@ impl TryFrom<FromSpanMessage> for Span {
             .map(|(k, v)| (k, v.value))
             .unzip();
 
-        if let Some(http_method) = from.sentry_tags.http_method {
-            tag_keys.push("http.method".into());
-            tag_values.push(http_method);
-        }
-
-        if let Some(status_code) = from.sentry_tags.status_code {
-            tag_keys.push("status_code".into());
-            tag_values.push(status_code);
-        }
-
-        if let Some(transaction_method) = from.sentry_tags.transaction_method {
-            tag_keys.push("transaction.method".into());
-            tag_values.push(transaction_method);
-        }
-
         let metrics_summary = match from._metrics_summary {
             Value::Object(v) => serde_json::to_string(&v).unwrap_or_default(),
             _ => "".into(),
@@ -270,6 +261,7 @@ impl TryFrom<FromSpanMessage> for Span {
             trace_id: from.trace_id,
             transaction_id: from.event_id,
             transaction_op,
+            user: from.sentry_tags.user.unwrap_or_default(),
             ..Default::default()
         })
     }

snuba/datasets/configuration/spans/entities/spans.yaml

@@ -59,28 +59,25 @@ schema:
         },
     },
     {
-      name: sentry_tags,
-      type: Nested,
+      name: _tags_hash_map,
+      type: Array,
       args:
         {
-          subcolumns:
-            [{ name: key, type: String }, { name: value, type: String }],
+          inner_type: { type: UInt, args: { size: 64 } },
+          schema_modifiers: [readonly],
         },
     },
     {
-      name: measurements,
+      name: sentry_tags,
       type: Nested,
       args:
         {
           subcolumns:
-            [
-              { name: key, type: String },
-              { name: value, type: Float, args: { size: 64 } },
-            ],
+            [{ name: key, type: String }, { name: value, type: String }],
         },
     },
     {
-      name: _tags_hash_map,
+      name: _sentry_tags_hash_map,
       type: Array,
       args:
         {
@@ -89,12 +86,15 @@ schema:
         },
     },
     {
-      name: _sentry_tags_hash_map,
-      type: Array,
+      name: measurements,
+      type: Nested,
       args:
         {
-          inner_type: { type: UInt, args: { size: 64 } },
-          schema_modifiers: [readonly],
+          subcolumns:
+            [
+              { name: key, type: String },
+              { name: value, type: Float, args: { size: 64 } },
+            ],
         },
     },
     { name: partition, type: UInt, args: { size: 16 } },
@@ -128,7 +128,7 @@ storages:
             from_column_table: null
             from_column_name: sentry_tags
             to_nested_col_table: null
-            to_nested_col_name: tags
+            to_nested_col_name: sentry_tags
             value_subcolumn_name: value
             nullable: false
         - mapper: SubscriptableMapper

snuba/datasets/configuration/spans/storages/spans.yaml

@@ -49,24 +49,33 @@ schema:
           },
       },
       {
-        name: measurements,
+        name: _tags_hash_map,
+        type: Array,
+        args:
+          {
+            inner_type: { type: UInt, args: { size: 64 } },
+            schema_modifiers: [ readonly ],
+          },
+      },
+      {
+        name: sentry_tags,
         type: Nested,
         args:
           {
             subcolumns:
-              [
-                { name: key, type: String },
-                { name: value, type: Float, args: { size: 64 } },
-              ],
+              [ { name: key, type: String }, { name: value, type: String } ],
           },
       },
       {
-        name: _tags_hash_map,
-        type: Array,
+        name: measurements,
+        type: Nested,
         args:
           {
-            inner_type: { type: UInt, args: { size: 64 } },
-            schema_modifiers: [ readonly ],
+            subcolumns:
+              [
+                { name: key, type: String },
+                { name: value, type: Float, args: { size: 64 } },
+              ],
           },
       },
       { name: partition, type: UInt, args: { size: 16 } },
@@ -112,6 +121,9 @@ query_processors:
   - processor: ArrayJoinKeyValueOptimizer
     args:
       column_name: tags
+  - processor: ArrayJoinKeyValueOptimizer
+    args:
+      column_name: sentry_tags
   - processor: ArrayJoinKeyValueOptimizer
     args:
       column_name: measurements

snuba/datasets/processors/functions_processor.py

@@ -1,72 +1,6 @@
-import uuid
-from datetime import datetime
-from typing import Any, Mapping, Optional
+from snuba.datasets.processors.rust_compat_processor import RustCompatProcessor
 
-from sentry_relay.consts import SPAN_STATUS_NAME_TO_CODE
 
-from snuba import environment
-from snuba.consumers.types import KafkaMessageMetadata
-from snuba.datasets.processors import DatasetMessageProcessor
-from snuba.processor import InsertBatch, ProcessedMessage
-from snuba.utils.metrics.wrapper import MetricsWrapper
-
-metrics = MetricsWrapper(environment.metrics, "functions.processor")
-
-UNKNOWN_SPAN_STATUS = 2
-MAX_DEPTH = 1024
-
-
-class FunctionsMessageProcessor(DatasetMessageProcessor):
-    def process_message(
-        self, message: Mapping[str, Any], metadata: KafkaMessageMetadata
-    ) -> Optional[ProcessedMessage]:
-        status = message.get("transaction_status")
-        if status:
-            int_status = SPAN_STATUS_NAME_TO_CODE.get(
-                status,
-                UNKNOWN_SPAN_STATUS,
-            )
-        else:
-            int_status = UNKNOWN_SPAN_STATUS
-        functions_list = [
-            {
-                "project_id": message["project_id"],
-                "transaction_name": message["transaction_name"],
-                "timestamp": int(
-                    message.get("timestamp", datetime.utcnow().timestamp())
-                ),
-                "depth": 0,  # deprecated
-                "parent_fingerprint": 0,  # deprecated
-                "fingerprint": function["fingerprint"],
-                "name": function["function"],  # to be removed
-                "function": function["function"],
-                "package": function.get("package", ""),
-                "module": function.get("module", ""),
-                "path": "",  # deprecated
-                "is_application": 1 if function.get("in_app", True) else 0,
-                "platform": message["platform"],
-                "environment": message.get("environment"),
-                "release": message.get("release"),
-                "dist": message.get("dist"),
-                "transaction_op": message.get("transaction_op", ""),
-                "transaction_status": int_status,
-                "http_method": message.get("http_method"),
-                "browser_name": message.get("browser_name"),
-                "device_classification": message.get("device_class", 0),
-                "os_name": "",  # deprecated
-                "os_version": "",  # deprecated
-                "retention_days": message["retention_days"],
-                "durations": function["self_times_ns"],
-                "profile_id": str(uuid.UUID(message["profile_id"])),
-                "materialization_version": 0,
-            }
-            for function in message["functions"]
-            if function.get("function") and function.get("self_times_ns")
-        ]
-
-        received = message.get("received")
-
-        return InsertBatch(
-            functions_list,
-            datetime.utcfromtimestamp(received) if received else None,
-        )
+class FunctionsMessageProcessor(RustCompatProcessor):
+    def __init__(self) -> None:
+        super().__init__("FunctionsMessageProcessor")