feat(rpc): Add more query logging

src/sentry/snuba/rpc_dataset_common.py

@@ -108,18 +108,6 @@ def check_timeseries_has_data(timeseries: SnubaData, y_axes: list[str]):
     return False
 
 
-def log_rpc_request(message: str, rpc_request, rpc_logger: logging.Logger = logger):
-    rpc_debug_json = json.loads(MessageToJson(rpc_request))
-    rpc_logger.info(
-        message,
-        extra={
-            "rpc_query": rpc_debug_json,
-            "referrer": rpc_request.meta.referrer,
-            "trace_item_type": rpc_request.meta.trace_item_type,
-        },
-    )
-
-
 class RPCBase:
     """Utility Methods"""
 
@@ -350,7 +338,6 @@ def _run_table_query(
         """Run the query"""
         table_request = cls.get_table_rpc_request(query)
         rpc_request = table_request.rpc_request
-        log_rpc_request("Running a table query with debug on", rpc_request)
         try:
             rpc_response = snuba_rpc.table_rpc([rpc_request])[0]
         except Exception as e:
@@ -560,7 +547,6 @@ def update_timestamps(
     def _run_timeseries_rpc(
         self, debug: bool, rpc_request: TimeSeriesRequest
     ) -> TimeSeriesResponse:
-        log_rpc_request("Running a timeseries query with debug on", rpc_request)
         try:
             return snuba_rpc.timeseries_rpc([rpc_request])[0]
         except Exception as e:
@@ -832,8 +818,6 @@ def run_top_events_timeseries_query(
             requests.append(other_request)
 
         """Run the query"""
-        for rpc_request in requests:
-            log_rpc_request("Running a top events query with debug on", rpc_request)
         try:
             timeseries_rpc_response = snuba_rpc.timeseries_rpc(requests)
             rpc_response = timeseries_rpc_response[0]

src/sentry/snuba/spans_rpc.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 import sentry_sdk
+from google.protobuf.json_format import MessageToJson
 from sentry_protos.snuba.v1.endpoint_get_trace_pb2 import GetTraceRequest
 from sentry_protos.snuba.v1.endpoint_trace_item_stats_pb2 import (
     AttributeDistributionsRequest,
@@ -31,7 +32,7 @@
 from sentry.snuba import rpc_dataset_common
 from sentry.snuba.discover import zerofill
 from sentry.snuba.rpc_dataset_common import set_debug_meta
-from sentry.utils import snuba_rpc
+from sentry.utils import json, snuba_rpc
 from sentry.utils.snuba import SnubaTSResult
 
 logger = logging.getLogger("sentry.snuba.spans_rpc")
@@ -271,10 +272,14 @@ def run_trace_query(
                     break
                 if MAX_TIMEOUT > 0 and time.time() - start_time > MAX_TIMEOUT:
                     # If timeout is not set then logging this is not helpful
-                    rpc_dataset_common.log_rpc_request(
+                    rpc_debug_json = json.loads(MessageToJson(request))
+                    logger.info(
                         "running a trace query timed out while paginating",
-                        request,
-                        logger,
+                        extra={
+                            "rpc_query": rpc_debug_json,
+                            "referrer": request.meta.referrer,
+                            "trace_item_type": request.meta.trace_item_type,
+                        },
                     )
                     break
                 request.page_token.CopyFrom(response.page_token)

src/sentry/utils/snuba_rpc.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import os
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
@@ -9,6 +10,7 @@
 import sentry_sdk
 import sentry_sdk.scope
 import urllib3
+from google.protobuf.json_format import MessageToJson
 from google.protobuf.message import Message as ProtobufMessage
 from rest_framework.exceptions import NotFound
 from sentry_protos.snuba.v1.endpoint_create_subscription_pb2 import (
@@ -44,8 +46,10 @@
 from sentry_protos.snuba.v1.request_common_pb2 import RequestMeta
 from urllib3.response import BaseHTTPResponse
 
+from sentry.utils import json, metrics
 from sentry.utils.snuba import SnubaError, _snuba_pool
 
+logger = logging.getLogger(__name__)
 RPCResponseType = TypeVar("RPCResponseType", bound=ProtobufMessage)
 
 # Show the snuba query params and the corresponding sql or errors in the server logs
@@ -109,10 +113,23 @@ def _make_rpc_requests(
     timeseries_requests = [] if timeseries_requests is None else timeseries_requests
     requests = table_requests + timeseries_requests
 
-    endpoint_names = [
-        "EndpointTraceItemTable" if isinstance(req, TraceItemTableRequest) else "EndpointTimeSeries"
-        for req in requests
-    ]
+    endpoint_names: list[str] = []
+    for request in requests:
+        endpoint_name = (
+            "EndpointTraceItemTable"
+            if isinstance(request, TraceItemTableRequest)
+            else "EndpointTimeSeries"
+        )
+        endpoint_names.append(endpoint_name)
+        logger.info(
+            f"Running a {endpoint_name} RPC query",  # noqa: LOG011
+            extra={
+                "rpc_query": json.loads(MessageToJson(request)),
+                "referrer": request.meta.referrer,
+                "organization_id": request.meta.organization_id,
+                "trace_item_type": request.meta.trace_item_type,
+            },
+        )
 
     referrers = [req.meta.referrer for req in requests]
     assert (
@@ -150,10 +167,37 @@ def _make_rpc_requests(
             table_response = TraceItemTableResponse()
             table_response.ParseFromString(item.data)
             table_results.append(table_response)
+
+            if len(table_response.column_values) > 0:
+                rpc_rows = len(table_response.column_values[0].results)
+            else:
+                rpc_rows = 0
+            logger.info(
+                "Table RPC query response",
+                extra={
+                    "rpc_rows": rpc_rows,
+                    "page_token": table_response.page_token,
+                    "meta": table_response.meta,
+                },
+            )
+            metrics.distribution("snuba_rpc.table_response.length", rpc_rows)
         elif isinstance(request, TimeSeriesRequest):
             timeseries_response = TimeSeriesResponse()
             timeseries_response.ParseFromString(item.data)
             timeseries_results.append(timeseries_response)
+
+            if len(timeseries_response.result_timeseries) > 0:
+                rpc_rows = len(timeseries_response.result_timeseries[0].data_points)
+            else:
+                rpc_rows = 0
+            logger.info(
+                "Timeseries RPC query response",
+                extra={
+                    "rpc_rows": rpc_rows,
+                    "meta": timeseries_response.meta,
+                },
+            )
+            metrics.distribution("snuba_rpc.timeseries_response.length", rpc_rows)
     return MultiRpcResponse(table_results, timeseries_results)
 
 
@@ -271,8 +315,6 @@ def _make_rpc_request(
         sentry_sdk.get_current_scope() if thread_current_scope is None else thread_current_scope
     )
     if SNUBA_INFO:
-        from google.protobuf.json_format import MessageToJson
-
         log_snuba_info(f"{referrer}.body:\n{MessageToJson(req)}")  # type: ignore[arg-type]
     with sentry_sdk.scope.use_isolation_scope(thread_isolation_scope):
         with sentry_sdk.scope.use_scope(thread_current_scope):