feat(spans): Add downsampled_retention_days and sample rates to items (#97790)

These fields were added to Relay already.

getsentry/relay#5059

Author: phacops

devservices/config.yml

@@ -143,8 +143,6 @@ x-sentry-service-config:
     post-process-forwarder-issue-platform:
       description: Post-process forwarder for issue platform events
     # Subscription results consumers
-    eap-spans-subscription-results:
-      description: Kafka consumer for processing subscription results for spans
     subscription-results-eap-items:
       description: Kafka consumer for processing subscription results for eap items
     metrics-subscription-results:
@@ -184,7 +182,6 @@ x-sentry-service-config:
         post-process-forwarder-errors,
         post-process-forwarder-transactions,
         post-process-forwarder-issue-platform,
-        eap-spans-subscription-results,
         subscription-results-eap-items,
         metrics-subscription-results,
         generic-metrics-subscription-results,
@@ -339,8 +336,6 @@ x-programs:
     command: sentry run consumer post-process-forwarder-issue-platform --consumer-group=sentry-consumer --auto-offset-reset=latest --no-strict-offset-reset
   ingest-feedback-events:
     command: sentry run consumer ingest-feedback-events --consumer-group=sentry-consumer --auto-offset-reset=latest --no-strict-offset-reset
-  eap-spans-subscription-results:
-    command: sentry run consumer eap-spans-subscription-results --consumer-group=sentry-consumer --auto-offset-reset=latest --no-strict-offset-reset
   subscription-results-eap-items:
     command: sentry run consumer subscription-results-eap-items --consumer-group=sentry-consumer --auto-offset-reset=latest --no-strict-offset-reset
   metrics-subscription-results:

pyproject.toml

@@ -78,9 +78,9 @@ dependencies = [
     # [end] jsonschema format validators
     "sentry-arroyo>=2.29.0",
     "sentry-forked-email-reply-parser>=0.5.12.post1",
-    "sentry-kafka-schemas>=1.3.18",
+    "sentry-kafka-schemas>=2.1.0",
     "sentry-ophio>=1.1.3",
-    "sentry-protos>=0.3.1",
+    "sentry-protos>=0.3.4",
     "sentry-redis-tools>=0.5.0",
     "sentry-relay>=0.9.11",
     "sentry-sdk[http2]>=2.29.1",
@@ -212,36 +212,36 @@ addopts = "--tb=short -p no:celery --nomigrations"
 # TODO: --import-mode=importlib will become the default soon,
 # currently we have a few relative imports that don't work with that.
 markers = [
-  "snuba: test requires access to snuba",
-  "snuba_ci: test is run in snuba ci",
-  "sentry_metrics: test requires access to sentry metrics",
-  "symbolicator: test requires access to symbolicator",
-  "querybuilder: smoke tests for QueryBuilders",
-  "thread_leak_allowlist: temporarily allow known-issue thread leaks",
+    "snuba: test requires access to snuba",
+    "snuba_ci: test is run in snuba ci",
+    "sentry_metrics: test requires access to sentry metrics",
+    "symbolicator: test requires access to symbolicator",
+    "querybuilder: smoke tests for QueryBuilders",
+    "thread_leak_allowlist: temporarily allow known-issue thread leaks",
 ]
 filterwarnings = [
-  # Consider all warnings to be errors other than the ignored ones.
-  "error",
+    # Consider all warnings to be errors other than the ignored ones.
+    "error",
 
-  # TODO: we should fix these, but for now there's a lot
-  "ignore:datetime.datetime.utcfromtimestamp\\(\\) is deprecated.*",
-  "ignore:datetime.datetime.utcnow\\(\\) is deprecated.*",
+    # TODO: we should fix these, but for now there's a lot
+    "ignore:datetime.datetime.utcfromtimestamp\\(\\) is deprecated.*",
+    "ignore:datetime.datetime.utcnow\\(\\) is deprecated.*",
 
-  # The following warning filters are for pytest only.
-  "ignore:.*sentry.digests.backends.dummy.DummyBackend.*:sentry.utils.warnings.UnsupportedBackend",
+    # The following warning filters are for pytest only.
+    "ignore:.*sentry.digests.backends.dummy.DummyBackend.*:sentry.utils.warnings.UnsupportedBackend",
 
-  # pytest has not yet implemented the replacement for this yet
-  "ignore:The --looponfail command line argument.*",
+    # pytest has not yet implemented the replacement for this yet
+    "ignore:The --looponfail command line argument.*",
 ]
 looponfailroots = ["src", "tests"]
 
 [tool.mypy]
 python_version = "3.13"
 mypy_path = ["fixtures/stubs-for-mypy"]
 plugins = [
-  "pydantic.mypy",
-  "mypy_django_plugin.main",
-  "tools.mypy_helpers.plugin",
+    "pydantic.mypy",
+    "mypy_django_plugin.main",
+    "tools.mypy_helpers.plugin",
 ]
 files = ["."]
 exclude = ["^.venv/", "^venv/", "^self-hosted/"]
@@ -254,16 +254,16 @@ warn_unused_configs = true
 warn_unused_ignores = true
 warn_redundant_casts = true
 enable_error_code = ["ignore-without-code", "redundant-self"]
-local_partial_types = true  # compat with dmypy
+local_partial_types = true                                    # compat with dmypy
 
 [tool.django-stubs]
 django_settings_module = "sentry.conf.server_mypy"
 
 # these have py.typed but incorrect types
 [[tool.mypy.overrides]]
 module = [
-  # TODO: these cause type errors when followed
-  "snuba_sdk.*",
+    # TODO: these cause type errors when followed
+    "snuba_sdk.*",
 ]
 follow_imports = "skip"
 

src/sentry/conf/server.py

@@ -900,12 +900,24 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
 
 
 CELERY_QUEUES_CONTROL = [
-    Queue("app_platform.control", routing_key="app_platform.control", exchange=control_exchange),
+    Queue(
+        "app_platform.control",
+        routing_key="app_platform.control",
+        exchange=control_exchange,
+    ),
     Queue("auth.control", routing_key="auth.control", exchange=control_exchange),
     Queue("cleanup.control", routing_key="cleanup.control", exchange=control_exchange),
     Queue("email.control", routing_key="email.control", exchange=control_exchange),
-    Queue("integrations.control", routing_key="integrations.control", exchange=control_exchange),
-    Queue("files.delete.control", routing_key="files.delete.control", exchange=control_exchange),
+    Queue(
+        "integrations.control",
+        routing_key="integrations.control",
+        exchange=control_exchange,
+    ),
+    Queue(
+        "files.delete.control",
+        routing_key="files.delete.control",
+        exchange=control_exchange,
+    ),
     Queue(
         "hybrid_cloud.control_repair",
         routing_key="hybrid_cloud.control_repair",
@@ -914,7 +926,11 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     Queue("options.control", routing_key="options.control", exchange=control_exchange),
     Queue("outbox.control", routing_key="outbox.control", exchange=control_exchange),
     Queue("webhook.control", routing_key="webhook.control", exchange=control_exchange),
-    Queue("relocation.control", routing_key="relocation.control", exchange=control_exchange),
+    Queue(
+        "relocation.control",
+        routing_key="relocation.control",
+        exchange=control_exchange,
+    ),
     Queue(
         "release_registry.control",
         routing_key="release_registry.control",
@@ -943,7 +959,8 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     Queue("default", routing_key="default"),
     Queue("delayed_rules", routing_key="delayed_rules"),
     Queue(
-        "delete_seer_grouping_records_by_hash", routing_key="delete_seer_grouping_records_by_hash"
+        "delete_seer_grouping_records_by_hash",
+        routing_key="delete_seer_grouping_records_by_hash",
     ),
     Queue("digests.delivery", routing_key="digests.delivery"),
     Queue("digests.scheduling", routing_key="digests.scheduling"),
@@ -952,11 +969,16 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     Queue("events.preprocess_event", routing_key="events.preprocess_event"),
     Queue("events.process_event", routing_key="events.process_event"),
     Queue(
-        "events.reprocessing.preprocess_event", routing_key="events.reprocessing.preprocess_event"
+        "events.reprocessing.preprocess_event",
+        routing_key="events.reprocessing.preprocess_event",
     ),
-    Queue("events.reprocessing.process_event", routing_key="events.reprocessing.process_event"),
     Queue(
-        "events.reprocessing.symbolicate_event", routing_key="events.reprocessing.symbolicate_event"
+        "events.reprocessing.process_event",
+        routing_key="events.reprocessing.process_event",
+    ),
+    Queue(
+        "events.reprocessing.symbolicate_event",
+        routing_key="events.reprocessing.symbolicate_event",
     ),
     Queue("events.save_event", routing_key="events.save_event"),
     Queue("events.save_event_highcpu", routing_key="events.save_event_highcpu"),
@@ -969,9 +991,13 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     Queue("files.copy", routing_key="files.copy"),
     Queue("files.delete", routing_key="files.delete"),
     Queue(
-        "group_owners.process_suspect_commits", routing_key="group_owners.process_suspect_commits"
+        "group_owners.process_suspect_commits",
+        routing_key="group_owners.process_suspect_commits",
+    ),
+    Queue(
+        "group_owners.process_commit_context",
+        routing_key="group_owners.process_commit_context",
     ),
-    Queue("group_owners.process_commit_context", routing_key="group_owners.process_commit_context"),
     Queue("integrations", routing_key="integrations"),
     Queue(
         "releasemonitor",
@@ -1016,18 +1042,27 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     Queue("auto_enable_codecov", routing_key="auto_enable_codecov"),
     Queue("weekly_escalating_forecast", routing_key="weekly_escalating_forecast"),
     Queue("relocation", routing_key="relocation"),
-    Queue("performance.statistical_detector", routing_key="performance.statistical_detector"),
+    Queue(
+        "performance.statistical_detector",
+        routing_key="performance.statistical_detector",
+    ),
     Queue("profiling.statistical_detector", routing_key="profiling.statistical_detector"),
     CELERY_ISSUE_STATES_QUEUE,
     Queue("nudge.invite_missing_org_members", routing_key="invite_missing_org_members"),
     Queue("auto_resolve_issues", routing_key="auto_resolve_issues"),
     Queue("on_demand_metrics", routing_key="on_demand_metrics"),
     Queue("check_new_issue_threshold_met", routing_key="check_new_issue_threshold_met"),
-    Queue("integrations_slack_activity_notify", routing_key="integrations_slack_activity_notify"),
+    Queue(
+        "integrations_slack_activity_notify",
+        routing_key="integrations_slack_activity_notify",
+    ),
     Queue("demo_mode", routing_key="demo_mode"),
     Queue("release_registry", routing_key="release_registry"),
     Queue("seer.seer_automation", routing_key="seer.seer_automation"),
-    Queue("workflow_engine.process_workflows", routing_key="workflow_engine.process_workflows"),
+    Queue(
+        "workflow_engine.process_workflows",
+        routing_key="workflow_engine.process_workflows",
+    ),
     Queue("workflow_engine.trigger_action", routing_key="workflow_engine.trigger_action"),
 ]
 
@@ -1400,7 +1435,10 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
 }
 
 BGTASKS: dict[str, BgTaskConfig] = {
-    "sentry.bgtasks.clean_dsymcache:clean_dsymcache": {"interval": 5 * 60, "roles": ["worker"]},
+    "sentry.bgtasks.clean_dsymcache:clean_dsymcache": {
+        "interval": 5 * 60,
+        "roles": ["worker"],
+    },
 }
 
 #######################
@@ -1757,7 +1795,10 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
 elif SILO_MODE == "REGION":
     TASKWORKER_SCHEDULES = TASKWORKER_REGION_SCHEDULES
 else:
-    TASKWORKER_SCHEDULES = {**TASKWORKER_CONTROL_SCHEDULES, **TASKWORKER_REGION_SCHEDULES}
+    TASKWORKER_SCHEDULES = {
+        **TASKWORKER_CONTROL_SCHEDULES,
+        **TASKWORKER_REGION_SCHEDULES,
+    }
 
 TASKWORKER_HIGH_THROUGHPUT_NAMESPACES = {
     "ingest.profiling",
@@ -1783,7 +1824,10 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
         # This `internal` logger is separate from the `Logging` integration in the SDK. Since
         # we have this to record events, in `sdk.py` we set the integration's `event_level` to
         # None, so that it records breadcrumbs for all log calls but doesn't send any events.
-        "internal": {"level": "ERROR", "class": "sentry_sdk.integrations.logging.EventHandler"},
+        "internal": {
+            "level": "ERROR",
+            "class": "sentry_sdk.integrations.logging.EventHandler",
+        },
         "metrics": {
             "level": "WARNING",
             "filters": ["important_django_request"],
@@ -1816,7 +1860,11 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
         # This only needs to go to Sentry for now.
         "sentry.similarity": {"handlers": ["internal"], "propagate": False},
         "sentry.errors": {"handlers": ["console"], "propagate": False},
-        "sentry_sdk.errors": {"handlers": ["console"], "level": "INFO", "propagate": False},
+        "sentry_sdk.errors": {
+            "handlers": ["console"],
+            "level": "INFO",
+            "propagate": False,
+        },
         "sentry.rules": {"handlers": ["console"], "propagate": False},
         "sentry.profiles": {"level": "INFO"},
         "multiprocessing": {
@@ -1836,7 +1884,11 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
             "propagate": False,
         },
         "toronado": {"level": "ERROR", "handlers": ["null"], "propagate": False},
-        "urllib3.connectionpool": {"level": "ERROR", "handlers": ["console"], "propagate": False},
+        "urllib3.connectionpool": {
+            "level": "ERROR",
+            "handlers": ["console"],
+            "propagate": False,
+        },
         "boto3": {"level": "WARNING", "handlers": ["console"], "propagate": False},
         "botocore": {"level": "WARNING", "handlers": ["console"], "propagate": False},
     },
@@ -1887,7 +1939,10 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
         # We override the default behavior to skip adding the choice name to the bullet point if
         # it's identical to the choice value by monkey patching build_choice_description_list.
         "ENUM_GENERATE_CHOICE_DESCRIPTION": True,
-        "LICENSE": {"name": "Apache 2.0", "url": "http://www.apache.org/licenses/LICENSE-2.0.html"},
+        "LICENSE": {
+            "name": "Apache 2.0",
+            "url": "http://www.apache.org/licenses/LICENSE-2.0.html",
+        },
         "PARSER_WHITELIST": ["rest_framework.parsers.JSONParser"],
         "POSTPROCESSING_HOOKS": [
             "sentry.apidocs.hooks.custom_postprocessing_hook",
@@ -2371,7 +2426,12 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
         ("org:write", "Read and write access to organization details."),
         ("org:read", "Read access to organization details."),
     ),
-    (("org:integrations", "Read, write, and admin access to organization integrations."),),
+    (
+        (
+            "org:integrations",
+            "Read, write, and admin access to organization integrations.",
+        ),
+    ),
     (
         ("member:admin", "Read, write, and admin access to organization members."),
         ("member:write", "Read and write access to organization members."),
@@ -2737,7 +2797,10 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
         {
             "image": f"ghcr.io/getsentry/image-mirror-library-postgres:{PG_VERSION}-alpine",
             "ports": {"5432/tcp": 5432},
-            "environment": {"POSTGRES_DB": "sentry", "POSTGRES_HOST_AUTH_METHOD": "trust"},
+            "environment": {
+                "POSTGRES_DB": "sentry",
+                "POSTGRES_HOST_AUTH_METHOD": "trust",
+            },
             "volumes": {
                 "postgres": {"bind": "/var/lib/postgresql/data"},
                 "wal2json": {"bind": "/wal2json"},
@@ -3373,7 +3436,6 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
     "transactions-subscription-results": "default",
     "generic-metrics-subscription-results": "default",
     "metrics-subscription-results": "default",
-    "eap-spans-subscription-results": "default",
     "subscription-results-eap-items": "default",
     "ingest-events": "default",
     "ingest-feedback-events": "default",
@@ -3407,12 +3469,10 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
     "generic-events": "default",
     "snuba-generic-events-commit-log": "default",
     "group-attributes": "default",
-    "snuba-spans": "default",
     "snuba-items": "default",
     "shared-resources-usage": "default",
     "buffered-segments": "default",
     "buffered-segments-dlq": "default",
-    "snuba-ourlogs": "default",
     "preprod-artifact-events": "default",
     # Taskworker topics
     "taskworker": "default",

src/sentry/conf/types/kafka_definition.py

@@ -27,8 +27,6 @@ class Topic(Enum):
     TRANSACTIONS_SUBSCRIPTIONS_RESULTS = "transactions-subscription-results"
     GENERIC_METRICS_SUBSCRIPTIONS_RESULTS = "generic-metrics-subscription-results"
     METRICS_SUBSCRIPTIONS_RESULTS = "metrics-subscription-results"
-    EAP_SPANS_SUBSCRIPTIONS_RESULTS = "eap-spans-subscription-results"
-    EAP_ITEMS_SUBSCRIPTIONS_RESULTS = "subscription-results-eap-items"
     INGEST_EVENTS = "ingest-events"
     INGEST_EVENTS_DLQ = "ingest-events-dlq"
     INGEST_EVENTS_BACKLOG = "ingest-events-backlog"
@@ -65,9 +63,8 @@ class Topic(Enum):
     GENERIC_EVENTS_COMMIT_LOG = "snuba-generic-events-commit-log"
     GROUP_ATTRIBUTES = "group-attributes"
     SHARED_RESOURCES_USAGE = "shared-resources-usage"
-    SNUBA_SPANS = "snuba-spans"
-    SNUBA_OURLOGS = "snuba-ourlogs"
     SNUBA_ITEMS = "snuba-items"
+    EAP_ITEMS_SUBSCRIPTIONS_RESULTS = "subscription-results-eap-items"
     BUFFERED_SEGMENTS = "buffered-segments"
     BUFFERED_SEGMENTS_DLQ = "buffered-segments-dlq"
 
@@ -114,7 +111,6 @@ class Topic(Enum):
 
 
 class ConsumerDefinition(TypedDict, total=False):
-
     # Default topic
     topic: Required[Topic]