feat(chart-data): emit StatsD timers for each phase of the chart data API

docs/admin_docs/configuration/event-logging.mdx

@@ -81,3 +81,51 @@ STATS_LOGGER = StatsdStatsLogger(
 
 Note that it’s also possible to implement your own logger by deriving
 `superset.stats_logger.BaseStatsLogger`.
+
+### Chart data latency metrics
+
+The chart data API (`/api/v1/chart/data` and `/api/v1/chart/<pk>/data/`) emits
+timers (in milliseconds) and counters that break a request down by phase, so
+dashboard latency can be attributed to Superset processing, the results cache,
+or the database. Keys are static; the only dimension encoded in a key is the
+database engine name, a bounded set.
+
+The endpoint as a whole is timed by the generic REST API timers
+`ChartDataRestApi.data.time` and `ChartDataRestApi.get_data.time`. Phases nest,
+so durations from different levels overlap and should not be summed across
+levels.
+
+| Metric | Type | Measures |
+| --- | --- | --- |
+| `chart_data.auth` | timer | Authentication and permission checks (`@protect`) before the handler runs |
+| `chart_data.chart_lookup` | timer | Loading the saved chart (`GET /chart/<pk>/data/` only) |
+| `chart_data.dashboard_filters` | timer | Resolving dashboard filter context (`filters_dashboard_id` only) |
+| `chart_data.query_context.build` | timer | Parsing the request into a query context, including loading the datasource |
+| `chart_data.query_context.validate` | timer | Object-level access checks on the query context |
+| `chart_data.async.submit` | timer | Scheduling async query tasks (when the request returns 202) |
+| `chart_data.command.execute` | timer | Running every query of the request (synchronous path) |
+| `chart_data.totals` | timer | Contribution totals query, when a chart needs one |
+| `chart_data.query.total` | timer | One query from validation through payload assembly |
+| `chart_data.query.planning` | timer | Query validation, cache key and force resolution |
+| `chart_data.cache.key` | timer | Computing the results cache key (includes template rendering for extra cache keys) |
+| `chart_data.cache.get` | timer | Reading the results cache backend, including deserialization |
+| `chart_data.cache.hit` / `.miss` / `.bypass` | counter | Results cache outcome; `bypass` is a forced refresh or a disabled cache |
+| `chart_data.query.data_acquisition` | timer | Fetching data from the datasource on a cache miss or bypass |
+| `chart_data.query.annotations` | timer | Loading annotation layer data on a cache miss or bypass |
+| `chart_data.cache.set` | timer | Serializing and writing a fresh result to the results cache |
+| `chart_data.query.payload_assembly` | timer | Building the per-query response payload |
+| `chart_data.query.materialize` | timer | Converting the acquired result into the response payload (records, CSV/XLSX, dtypes, AUTO currency detection) |
+| `chart_data.response.serialize` | timer | JSON-encoding the response body |
+| `datasource.query.sql_generation` | timer | Building and rendering the SQL for a query |
+| `datasource.query.normalize` | timer | Normalizing temporal and metric columns of the result |
+| `datasource.query.time_offsets` | timer | Time comparison queries and joins |
+| `datasource.query.post_processing` | timer | Pandas post-processing operations |
+| `database.connect.<engine>` | timer | Opening a database connection |
+| `database.execute.<engine>` | timer | Driver `execute` call for one statement; with drivers whose `execute` blocks until the query completes, this includes database queueing and execution |
+| `database.fetch.<engine>` | timer | Fetching result rows from the driver |
+| `database.dataframe.<engine>` | timer | Building the pandas DataFrame from fetched rows |
+
+The `datasource.*` and `database.*` timers are emitted by shared query
+execution code, so they also cover other features that query datasources
+through the same path. Async chart queries emit the `chart_data.query.*`,
+`chart_data.cache.*`, `datasource.*` and `database.*` metrics from the worker.

superset/charts/data/api.py

@@ -73,6 +73,11 @@
     redact_sensitive_query_fields,
     sanitize_error_message,
 )
+from superset.utils.timing_metrics import (
+    emit_pre_handler_timer,
+    start_pre_handler_timer,
+    timed,
+)
 from superset.views.base import CsvResponse, generate_download_headers, XlsxResponse
 from superset.views.base_api import statsd_metrics
 
@@ -87,7 +92,9 @@ class ChartDataRestApi(ChartRestApi):
     include_route_methods = {"get_data", "data"}
 
     @expose("/<int:pk>/data/", methods=("GET",))
+    @start_pre_handler_timer
     @protect()
+    @emit_pre_handler_timer("chart_data.auth")
     @statsd_metrics
     @event_logger.log_this_with_context(
         action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.data",
@@ -165,7 +172,8 @@ def get_data(  # noqa: C901
             500:
               $ref: '#/components/responses/500'
         """
-        chart = self.datamodel.get(pk, self._base_filters)
+        with timed("chart_data.chart_lookup"):
+            chart = self.datamodel.get(pk, self._base_filters)
         if not chart:
             return self.response_404()
 
@@ -203,10 +211,11 @@ def get_data(  # noqa: C901
 
         if filters_dashboard_id is not None:
             try:
-                dashboard_filter_context = get_dashboard_filter_context(
-                    dashboard_id=filters_dashboard_id,
-                    chart_id=pk,
-                )
+                with timed("chart_data.dashboard_filters"):
+                    dashboard_filter_context = get_dashboard_filter_context(
+                        dashboard_id=filters_dashboard_id,
+                        chart_id=pk,
+                    )
             except ValueError as error:
                 return self.response_400(message=str(error))
             except SupersetSecurityException:
@@ -225,9 +234,11 @@ def get_data(  # noqa: C901
         set_form_data(json_body)
 
         try:
-            query_context = self._create_query_context_from_form(json_body)
+            with timed("chart_data.query_context.build"):
+                query_context = self._create_query_context_from_form(json_body)
             command = ChartDataCommand(query_context)
-            command.validate()
+            with timed("chart_data.query_context.validate"):
+                command.validate()
         except SemanticLayersDisabledError as error:
             return self.response(404, message=str(error))
         except DatasourceNotFound:
@@ -269,7 +280,9 @@ def get_data(  # noqa: C901
         )
 
     @expose("/data", methods=("POST",))
+    @start_pre_handler_timer
     @protect()
+    @emit_pre_handler_timer("chart_data.auth")
     @statsd_metrics
     @event_logger.log_this_with_context(
         action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.data",
@@ -330,9 +343,11 @@ def data(  # noqa: C901
             return self.response_400(message=_("Request is not JSON"))
 
         try:
-            query_context = self._create_query_context_from_form(json_body)
+            with timed("chart_data.query_context.build"):
+                query_context = self._create_query_context_from_form(json_body)
             command = ChartDataCommand(query_context)
-            command.validate()
+            with timed("chart_data.query_context.validate"):
+                command.validate()
         except SemanticLayersDisabledError as error:
             return self.response(404, message=str(error))
         except DatasourceNotFound:
@@ -453,7 +468,8 @@ def _run_async(
         # chart query. The client polls /api/v1/task/status_changes, aggregates the
         # tasks' statuses, and on success re-issues this same request — now served
         # synchronously from the per-query DATA cache the tasks populated.
-        job = submit_chart_data_query_tasks(command.query_context, get_user_id())
+        with timed("chart_data.async.submit"):
+            job = submit_chart_data_query_tasks(command.query_context, get_user_id())
         return self.response(202, **job)
 
     def _send_chart_response(  # noqa: C901
@@ -575,7 +591,10 @@ def _process_data(query_data: Any) -> Any:
             if dashboard_filter_context is not None:
                 payload["dashboard_filters"] = dashboard_filter_context.to_dict()
 
-            with event_logger.log_context(f"{self.__class__.__name__}.json_dumps"):
+            with (
+                timed("chart_data.response.serialize"),
+                event_logger.log_context(f"{self.__class__.__name__}.json_dumps"),
+            ):
                 response_data = json.dumps(
                     payload,
                     default=json.json_int_dttm_ser,
@@ -690,7 +709,8 @@ def _get_data_response(
                     slice_=slice_ or query_context.slice_,
                 )
 
-            result = command.execute(force_cached=force_cached)
+            with timed("chart_data.command.execute"):
+                result = command.execute(force_cached=force_cached)
         except ChartDataCacheLoadError as exc:
             return self.response_422(message=sanitize_error_message(exc.message))
         except (

superset/common/query_actions.py

@@ -43,6 +43,7 @@
     detect_currency_from_df,
     has_auto_currency_in_column_config,
 )
+from superset.utils.timing_metrics import timed
 
 if TYPE_CHECKING:
     from superset.common.query_context import QueryContext
@@ -214,7 +215,8 @@ def _get_full_with_timing(
 ) -> tuple[dict[str, Any], QueryAcquisitionTiming, int]:
     acquired = query_context.get_df_payload_result(query_obj, force_cached=force_cached)
     payload_assembly_start_ns = time.perf_counter_ns()
-    payload = _materialize_full_payload(query_context, query_obj, acquired.payload)
+    with timed("chart_data.query.materialize"):
+        payload = _materialize_full_payload(query_context, query_obj, acquired.payload)
     payload_assembly_ns = max(0, time.perf_counter_ns() - payload_assembly_start_ns)
     return payload, acquired.timing, payload_assembly_ns
 

superset/common/query_context_processor.py

@@ -86,6 +86,14 @@
     is_adhoc_metric,
 )
 from superset.utils.pandas_postprocessing.utils import unescape_separator
+from superset.utils.timing_metrics import (
+    elapsed_ms,
+    emit_incr,
+    emit_timing,
+    NANOSECONDS_PER_MILLISECOND,
+    now_ns,
+    timed,
+)
 
 if TYPE_CHECKING:
     from superset.common.query_context import QueryContext
@@ -269,13 +277,33 @@ def get_df_payload_result(
         self, query_obj: QueryObject, force_cached: bool | None = False
     ) -> QueryAcquisitionResult:
         """Acquire a dataframe and return timing as a typed sidecar."""
+        metrics_start_ns = now_ns()
+        phases: dict[str, int] = {}
+        try:
+            result = self._get_df_payload_result(query_obj, force_cached, phases)
+        finally:
+            self._emit_acquisition_metrics(phases, metrics_start_ns)
+        return result
+
+    def _get_df_payload_result(
+        self,
+        query_obj: QueryObject,
+        force_cached: bool | None,
+        phases: dict[str, int],
+    ) -> QueryAcquisitionResult:
+        """
+        Acquire a dataframe, recording phase starts and durations in ``phases``
+        so the caller can emit them even when a phase raises.
+        """
         query_planning_start_ns = time.perf_counter_ns()
+        phases["planning_start_ns"] = query_planning_start_ns
         if query_obj:
             # Always validate the query object before generating cache key
             # This ensures sanitize_clause() is called and extras are normalized
             query_obj.validate()
 
-        cache_key = self.query_cache_key(query_obj)
+        with timed("chart_data.cache.key"):
+            cache_key = self.query_cache_key(query_obj)
         timeout = self.get_cache_timeout()
         force_query = (
             self._resolve_forced_query(query_obj, cache_key)
@@ -288,6 +316,7 @@ def get_df_payload_result(
             query_obj.force_query = force_query
             query_obj.cache_timeout = timeout
         query_planning_ns = max(0, time.perf_counter_ns() - query_planning_start_ns)
+        phases["planning_ns"] = query_planning_ns
 
         cache_resolution_start_ns = time.perf_counter_ns()
         cache = QueryCacheManager.get(
@@ -311,10 +340,20 @@ def get_df_payload_result(
             cache.is_loaded = False
 
         cache_resolution_ns = max(0, time.perf_counter_ns() - cache_resolution_start_ns)
+        self._emit_cache_outcome(
+            bypassed=(
+                not cache_key
+                or force_query
+                or not QueryCacheManager.has_backend(CacheRegion.DATA)
+            ),
+            hit=cache.is_loaded,
+            cache_resolution_ns=cache_resolution_ns,
+        )
 
         data_acquisition_ns: int | None = None
         if query_obj and cache_key and not cache.is_loaded:
             data_acquisition_start_ns = time.perf_counter_ns()
+            phases["data_acquisition_start_ns"] = data_acquisition_start_ns
             try:
                 if invalid_columns := [
                     col
@@ -333,7 +372,8 @@ def get_df_payload_result(
                     )
 
                 query_result = self.get_query_result(query_obj)
-                annotation_data = self.get_annotation_data(query_obj)
+                with timed("chart_data.query.annotations"):
+                    annotation_data = self.get_annotation_data(query_obj)
             except SemanticResultCompletenessError:
                 raise
             except QueryObjectValidationError as ex:
@@ -344,6 +384,7 @@ def get_df_payload_result(
                 data_acquisition_ns = max(
                     0, time.perf_counter_ns() - data_acquisition_start_ns
                 )
+                phases["data_acquisition_ns"] = data_acquisition_ns
 
             if cache.status != QueryStatus.FAILED:
                 cache.set_query_result(
@@ -361,6 +402,7 @@ def get_df_payload_result(
                 self._mark_force_executed(query_obj, cache_key, cache.result_persisted)
 
         payload_assembly_start_ns = time.perf_counter_ns()
+        phases["payload_assembly_start_ns"] = payload_assembly_start_ns
         # the N-dimensional DataFrame has converted into flat DataFrame
         # by `flatten operator`, "comma" in the column is escaped by `escape_separator`
         # the result DataFrame columns should be unescaped
@@ -443,16 +485,64 @@ def get_df_payload_result(
             "label_map": label_map,
             "warning": warning,
         }
+        payload_assembly_ns = max(0, time.perf_counter_ns() - payload_assembly_start_ns)
+        phases["payload_assembly_ns"] = payload_assembly_ns
         timing = QueryAcquisitionTiming(
             query_planning_ns=query_planning_ns,
             cache_resolution_ns=cache_resolution_ns,
             data_acquisition_ns=data_acquisition_ns,
-            payload_assembly_ns=max(
-                0, time.perf_counter_ns() - payload_assembly_start_ns
-            ),
+            payload_assembly_ns=payload_assembly_ns,
         )
         return QueryAcquisitionResult(payload=payload, timing=timing)
 
+    @staticmethod
+    def _emit_cache_outcome(
+        bypassed: bool, hit: bool, cache_resolution_ns: int
+    ) -> None:
+        """
+        Count the results cache outcome and time real cache reads.
+
+        A forced or keyless query never reads the cache backend, so its
+        near-zero lookup time is kept out of the cache read distribution.
+        """
+        if bypassed:
+            emit_incr("chart_data.cache.bypass")
+            return
+        emit_timing(
+            "chart_data.cache.get",
+            cache_resolution_ns / NANOSECONDS_PER_MILLISECOND,
+        )
+        emit_incr("chart_data.cache.hit" if hit else "chart_data.cache.miss")
+
+    @staticmethod
+    def _emit_acquisition_metrics(phases: dict[str, int], started_ns: int) -> None:
+        """
+        Emit the per-query acquisition phases as StatsD timers.
+
+        Runs in both the synchronous request and the async worker, so the
+        distributions cover every executed chart-data query. Runs on failure
+        too: a phase that started but did not finish is measured up to the
+        failure. ``chart_data.query.data_acquisition`` is only emitted when the
+        datasource was actually queried (a cache miss or bypass).
+        """
+        now = now_ns()
+
+        def phase_ms(name: str) -> float | None:
+            if (duration := phases.get(f"{name}_ns")) is None:
+                if (start := phases.get(f"{name}_start_ns")) is None:
+                    return None
+                duration = max(0, now - start)
+            return duration / NANOSECONDS_PER_MILLISECOND
+
+        for name, key in (
+            ("planning", "chart_data.query.planning"),
+            ("data_acquisition", "chart_data.query.data_acquisition"),
+            ("payload_assembly", "chart_data.query.payload_assembly"),
+        ):
+            if (value := phase_ms(name)) is not None:
+                emit_timing(key, value)
+        emit_timing("chart_data.query.total", elapsed_ms(started_ns))
+
     def query_cache_key(self, query_obj: QueryObject, **kwargs: Any) -> str | None:
         """
         Returns a QueryObject cache key for objects in self.queries
@@ -767,7 +857,11 @@ def get_payload_result(
         # Skip ensure_totals_available when force_cached=True
         # This prevents recalculating contribution_totals from cached results
         if not force_cached:
-            self.ensure_totals_available(queries_needing_totals, totals_idx)
+            if queries_needing_totals and totals_idx is not None:
+                with timed("chart_data.totals"):
+                    self.ensure_totals_available(queries_needing_totals, totals_idx)
+            else:
+                self.ensure_totals_available(queries_needing_totals, totals_idx)
 
             # Update cache_values to reflect modifications made by
             # ensure_totals_available()