diff --git a/docs/admin_docs/configuration/event-logging.mdx b/docs/admin_docs/configuration/event-logging.mdx index b064d6e6e90a..adf688a65557 100644 --- a/docs/admin_docs/configuration/event-logging.mdx +++ b/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//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//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.` | timer | Opening a database connection | +| `database.execute.` | timer | Driver `execute` call for one statement; with drivers whose `execute` blocks until the query completes, this includes database queueing and execution | +| `database.fetch.` | timer | Fetching result rows from the driver | +| `database.dataframe.` | 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. diff --git a/superset/charts/data/api.py b/superset/charts/data/api.py index be0316d40b6d..1c124a273840 100644 --- a/superset/charts/data/api.py +++ b/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("//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 ( diff --git a/superset/common/query_actions.py b/superset/common/query_actions.py index 71955cc80b1f..bffb3db70755 100644 --- a/superset/common/query_actions.py +++ b/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 diff --git a/superset/common/query_context_processor.py b/superset/common/query_context_processor.py index b27b95134627..9991166ba8b3 100644 --- a/superset/common/query_context_processor.py +++ b/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() diff --git a/superset/common/utils/query_cache_manager.py b/superset/common/utils/query_cache_manager.py index 15f884189637..7f60c8311c31 100644 --- a/superset/common/utils/query_cache_manager.py +++ b/superset/common/utils/query_cache_manager.py @@ -17,15 +17,17 @@ from __future__ import annotations import logging +from contextlib import nullcontext from datetime import datetime, timezone from typing import Any from flask import current_app, g, has_request_context from flask_caching import Cache +from flask_caching.backends import NullCache from pandas import DataFrame from superset.common.db_query_status import QueryStatus -from superset.constants import CacheRegion, SemanticCacheStatus +from superset.constants import CACHE_DISABLED_TIMEOUT, CacheRegion, SemanticCacheStatus from superset.exceptions import CacheLoadError from superset.extensions import cache_manager from superset.models.helpers import QueryResult @@ -33,6 +35,7 @@ from superset.superset_typing import Column from superset.utils.cache import set_and_log_cache from superset.utils.core import error_msg_from_exception, get_stacktrace +from superset.utils.timing_metrics import timed logger = logging.getLogger(__name__) @@ -157,13 +160,19 @@ def set_query_result( "bq_memory_limited_row_count": self.bq_memory_limited_row_count, } if self.is_loaded and key and self.status != QueryStatus.FAILED: - self.result_persisted = self.set( - key=key, - value=value, - timeout=timeout, - datasource_uid=datasource_uid, - region=region, + # Only time writes that reach the backend; set_and_log_cache + # returns early for a null cache or a disabled timeout. + attempts_write = timeout != CACHE_DISABLED_TIMEOUT and not isinstance( + _cache[region].cache, NullCache ) + with timed("chart_data.cache.set") if attempts_write else nullcontext(): + self.result_persisted = self.set( + key=key, + value=value, + timeout=timeout, + datasource_uid=datasource_uid, + region=region, + ) except Exception as ex: # pylint: disable=broad-except logger.exception(ex) if not self.error_message: @@ -171,6 +180,13 @@ def set_query_result( self.status = QueryStatus.FAILED self.stacktrace = get_stacktrace() + @staticmethod + def has_backend(region: CacheRegion = CacheRegion.DEFAULT) -> bool: + """ + Whether a cache backend is configured for the region + """ + return bool(_cache[region]) + @classmethod def get( cls, @@ -183,7 +199,7 @@ def get( Initialize QueryCacheManager by query-cache key """ query_cache = cls() - if not key or not _cache[region] or force_query: + if not key or not cls.has_backend(region) or force_query: return query_cache try: diff --git a/superset/models/core.py b/superset/models/core.py index c6a34ab2f45e..ed64666cc1ef 100755 --- a/superset/models/core.py +++ b/superset/models/core.py @@ -103,6 +103,7 @@ is_oauth2_retry_active, OAuth2ClientConfigSchema, ) +from superset.utils.timing_metrics import metric_suffix, timed metadata = Model.metadata # pylint: disable=no-member logger = logging.getLogger(__name__) @@ -998,6 +999,22 @@ def mutate_sql_based_on_config(self, sql_: str, is_split: bool = False) -> str: ) return sql_ + @contextmanager + def _timed_raw_connection( + self, engine_suffix: str, catalog: str | None, schema: str | None + ) -> Iterator[Any]: + """ + Open a raw connection, timing the attempt even when it fails. + + Only establishing the connection is timed, not the body that uses it. + """ + with ExitStack() as stack: + with timed(f"database.connect.{engine_suffix}"): + conn = stack.enter_context( + self.get_raw_connection(catalog=catalog, schema=schema) + ) + yield conn + def _execute_sql_with_mutation_and_logging( self, sql: str, @@ -1032,7 +1049,10 @@ def _log_query(sql_: str) -> None: security_manager, ) - with self.get_raw_connection(catalog=catalog, schema=schema) as conn: + engine_suffix = metric_suffix(self.db_engine_spec.engine) + with self._timed_raw_connection( + engine_suffix, catalog=catalog, schema=schema + ) as conn: cursor = conn.cursor() rows = None description = None @@ -1069,12 +1089,14 @@ def _log_query(sql_: str) -> None: database=self, object_ref=__name__, ): - self.db_engine_spec.execute(cursor, sql_, self) + with timed(f"database.execute.{engine_suffix}"): + self.db_engine_spec.execute(cursor, sql_, self) query_executed() # Fetch results from last statement if requested if fetch_last_result and i == len(script.statements) - 1: - rows = self.db_engine_spec.fetch_data(cursor) + with timed(f"database.fetch.{engine_suffix}"): + rows = self.db_engine_spec.fetch_data(cursor) # Some asynchronous DB-API drivers expose placeholder metadata # until fetching waits for the operation to finish. description = cursor.description @@ -1123,7 +1145,10 @@ def get_df( df = None if rows is not None: - df = self.load_into_dataframe(description, rows) + with timed( + f"database.dataframe.{metric_suffix(self.db_engine_spec.engine)}" + ): + df = self.load_into_dataframe(description, rows) if mutator: df = mutator(df) diff --git a/superset/models/helpers.py b/superset/models/helpers.py index 6b30f75f53ba..6fe0f0e420fd 100644 --- a/superset/models/helpers.py +++ b/superset/models/helpers.py @@ -192,6 +192,7 @@ ) from superset.utils.dates import datetime_to_epoch from superset.utils.rls import apply_rls, get_predicates_for_table, get_rls_schema +from superset.utils.timing_metrics import timed class ValidationResultDict(TypedDict): @@ -2661,7 +2662,8 @@ def query(self, query_obj: QueryObjectDict) -> QueryResult: datasource types (Query, SqlaTable, etc.). """ qry_start_dttm = datetime.now() - query_str_ext = self.get_query_str_extended(query_obj) + with timed("datasource.query.sql_generation"): + query_str_ext = self.get_query_str_extended(query_obj) sql = query_str_ext.sql self._raise_for_disallowed_sql(sql) @@ -2986,22 +2988,24 @@ def get_query_result(self, query_object: QueryObject) -> QueryResult: df = result.df if not df.empty: # Normalize datetime columns and metrics - df = self.normalize_df( - df, - query_object, - result.sql_shifted_temporal_labels, - keep_exact_integers=dataframe_utils.can_keep_exact_integers( - query_object - ), - ) + with timed("datasource.query.normalize"): + df = self.normalize_df( + df, + query_object, + result.sql_shifted_temporal_labels, + keep_exact_integers=dataframe_utils.can_keep_exact_integers( + query_object + ), + ) # Process time offsets if requested if query_object.time_offsets: # Process time offsets using the datasource's own method # Note: caching is disabled here as we don't have query context - time_offsets = self.processing_time_offsets( - df, query_object, cache_key_fn=None, cache_timeout_fn=None - ) + with timed("datasource.query.time_offsets"): + time_offsets = self.processing_time_offsets( + df, query_object, cache_key_fn=None, cache_timeout_fn=None + ) df = time_offsets["df"] queries = time_offsets["queries"] query += ";\n\n".join(queries) @@ -3009,7 +3013,11 @@ def get_query_result(self, query_object: QueryObject) -> QueryResult: # Execute post-processing operations try: - df = query_object.exec_post_processing(df) + if query_object.post_processing: + with timed("datasource.query.post_processing"): + df = query_object.exec_post_processing(df) + else: + df = query_object.exec_post_processing(df) except InvalidPostProcessingError as ex: raise QueryObjectValidationError(ex.message) from ex except (TypeError, pd.errors.DataError) as ex: diff --git a/superset/utils/timing_metrics.py b/superset/utils/timing_metrics.py new file mode 100644 index 000000000000..561c4652c0af --- /dev/null +++ b/superset/utils/timing_metrics.py @@ -0,0 +1,146 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Best-effort StatsD timers and counters for latency attribution. + +Every helper here measures with a monotonic clock, reports durations in +milliseconds, and swallows metrics-backend failures so instrumentation can +never change the outcome of the code it measures. + +Metric keys are static, dot-separated names. ``BaseStatsLogger`` has no tag +support, so any dimension must be encoded in the key; only bounded values +(such as a DB engine name) may be appended, never identifiers, SQL, or +user-controlled strings. +""" + +from __future__ import annotations + +import functools +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from time import perf_counter_ns +from typing import Any, Callable, TypeVar + +from flask import g, has_request_context + +from superset.extensions import stats_logger_manager + +logger = logging.getLogger(__name__) + +NANOSECONDS_PER_MILLISECOND = 1_000_000 + +F = TypeVar("F", bound=Callable[..., Any]) + +_PRE_HANDLER_START_ATTR = "_timing_metrics_pre_handler_start_ns" + + +def now_ns() -> int: + """ + Read the monotonic clock used by these helpers. + + The clock is bound at import time so that tests patching + ``time.perf_counter_ns`` to drive phase timings elsewhere are not + perturbed by metric emission. + """ + return perf_counter_ns() + + +def elapsed_ms(start_ns: int) -> float: + """Return milliseconds elapsed since a :func:`now_ns` reading.""" + return max(0, now_ns() - start_ns) / NANOSECONDS_PER_MILLISECOND + + +def emit_timing(key: str, value_ms: float) -> None: + """Emit a timer in milliseconds, ignoring metrics-backend failures.""" + try: + stats_logger_manager.instance.timing(key, value_ms) + except Exception: # pylint: disable=broad-except + logger.debug("Failed to emit timing metric %s", key, exc_info=True) + + +def emit_incr(key: str) -> None: + """Increment a counter, ignoring metrics-backend failures.""" + try: + stats_logger_manager.instance.incr(key) + except Exception: # pylint: disable=broad-except + logger.debug("Failed to emit counter metric %s", key, exc_info=True) + + +@contextmanager +def timed(key: str) -> Iterator[None]: + """ + Time the enclosed block and emit it as ``key``. + + The timer is emitted whether the block succeeds or raises, so slow failures + (timeouts, cancelled warehouse queries) stay visible in the distribution. + """ + start_ns = now_ns() + try: + yield + finally: + emit_timing(key, elapsed_ms(start_ns)) + + +def metric_suffix(value: str | None) -> str: + """ + Normalize a bounded dimension (e.g. a DB engine name) into a key segment. + + Dots would split the dimension into extra key levels, so they are replaced. + """ + if not value: + return "unknown" + return "".join(ch if ch.isalnum() or ch in "_-" else "_" for ch in value) + + +def start_pre_handler_timer(func: F) -> F: + """ + Record when a request reaches the route, before auth decorators run. + + Pair with :func:`emit_pre_handler_timer` below ``@protect()`` to time the + authentication and permission checks that run in between. + """ + + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + if has_request_context(): + setattr(g, _PRE_HANDLER_START_ATTR, now_ns()) + return func(*args, **kwargs) + + return wrapper # type: ignore[return-value] + + +def emit_pre_handler_timer(key: str) -> Callable[[F], F]: + """ + Emit the time elapsed since :func:`start_pre_handler_timer` ran. + + Only requests that pass the decorators in between reach this point, so + rejected (401/403) requests are not part of the distribution. + """ + + def decorate(func: F) -> F: + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + if has_request_context(): + start_ns = g.pop(_PRE_HANDLER_START_ATTR, None) + if start_ns is not None: + emit_timing(key, elapsed_ms(start_ns)) + return func(*args, **kwargs) + + return wrapper # type: ignore[return-value] + + return decorate diff --git a/tests/unit_tests/common/test_query_context_processor_timing.py b/tests/unit_tests/common/test_query_context_processor_timing.py index 7f0a4ed15158..dd59926da539 100644 --- a/tests/unit_tests/common/test_query_context_processor_timing.py +++ b/tests/unit_tests/common/test_query_context_processor_timing.py @@ -18,10 +18,13 @@ import pandas as pd import pytest +from flask_caching.backends import NullCache, SimpleCache from superset.common.chart_data_timing import QueryDataResult, QueryTiming from superset.common.db_query_status import QueryStatus from superset.common.query_context_processor import QueryContextProcessor +from superset.common.utils.query_cache_manager import QueryCacheManager +from superset.constants import CacheRegion def _query_timing() -> QueryTiming: @@ -338,3 +341,158 @@ def test_dataframe_payload_result_resolves_force_and_timeout_onto_query_object( assert query_obj.cache_timeout == -1 cache_manager.get.assert_called_once() assert cache_manager.get.call_args.kwargs["force_query"] is True + + +def _cache(is_loaded: bool) -> MagicMock: + cache: MagicMock = MagicMock() + cache.is_loaded = is_loaded + cache.is_cached = is_loaded + cache.df = pd.DataFrame({"col1": [1]}) + cache.cache_dttm = "2026-01-01T00:00:00" if is_loaded else None + cache.queried_dttm = "2026-01-01T00:00:00" + cache.applied_template_filters = [] + cache.applied_filter_columns = [] + cache.rejected_filter_columns = [] + cache.annotation_data = {} + cache.error_message = None + cache.query = "SELECT 1" + cache.status = QueryStatus.SUCCESS + cache.stacktrace = None + cache.sql_rowcount = 1 + cache.bq_memory_limited = False + cache.bq_memory_limited_row_count = 0 + return cache + + +def _emitted(stats_logger: MagicMock) -> tuple[set[str], set[str]]: + timers = {call.args[0] for call in stats_logger.timing.call_args_list} + counters = {call.args[0] for call in stats_logger.incr.call_args_list} + return timers, counters + + +@pytest.mark.parametrize( + ("is_loaded", "timeout", "expected_counter"), + [ + (True, 300, "chart_data.cache.hit"), + (False, 300, "chart_data.cache.miss"), + (False, -1, "chart_data.cache.bypass"), + ], +) +@patch("superset.utils.timing_metrics.stats_logger_manager") +@patch("superset.common.query_context_processor.QueryCacheManager") +def test_dataframe_payload_result_emits_cache_outcome( + cache_manager: MagicMock, + stats_logger_manager: MagicMock, + is_loaded: bool, + timeout: int, + expected_counter: str, +) -> None: + cache_manager.get.return_value = _cache(is_loaded) + processor: QueryContextProcessor = _processor() + with ( + patch.object(processor, "query_cache_key", return_value="key"), + patch.object(processor, "get_cache_timeout", return_value=timeout), + patch.object(processor, "get_query_result", return_value=MagicMock()), + patch.object(processor, "get_annotation_data", return_value={}), + ): + processor.get_df_payload_result(_query_obj()) + + timers, counters = _emitted(stats_logger_manager.instance) + assert counters == {expected_counter} + # A bypass never reads the cache backend, so it has no read timing. + assert ("chart_data.cache.get" in timers) is ( + expected_counter != "chart_data.cache.bypass" + ) + assert { + "chart_data.cache.key", + "chart_data.query.planning", + "chart_data.query.payload_assembly", + "chart_data.query.total", + } <= timers + # Datasource time is only measured when the datasource was queried. + assert ("chart_data.query.data_acquisition" in timers) is (not is_loaded) + assert ("chart_data.query.annotations" in timers) is (not is_loaded) + + +@pytest.mark.parametrize( + ("status", "timeout", "backend", "expect_write"), + [ + (QueryStatus.SUCCESS, 60, "simple", True), + (QueryStatus.FAILED, 60, "simple", False), + (QueryStatus.SUCCESS, 60, "null", False), + (QueryStatus.SUCCESS, -1, "simple", False), + ], +) +@patch("superset.utils.timing_metrics.stats_logger_manager") +@patch.object(QueryCacheManager, "set", return_value=True) +def test_cache_set_timed_only_for_cache_writes( + cache_set: MagicMock, + stats_logger_manager: MagicMock, + status: QueryStatus, + timeout: int, + backend: str, + expect_write: bool, + app_context: None, +) -> None: + """Failed results and writes skipped by the cache layer are not timed.""" + query_result: MagicMock = MagicMock() + query_result.status = status + query_result.df = pd.DataFrame({"col1": [1]}) + region_cache: MagicMock = MagicMock() + region_cache.cache = NullCache() if backend == "null" else SimpleCache() + + with patch.dict( + "superset.common.utils.query_cache_manager._cache", + {CacheRegion.DEFAULT: region_cache}, + ): + QueryCacheManager().set_query_result( + key="key", query_result=query_result, timeout=timeout + ) + + timers, _ = _emitted(stats_logger_manager.instance) + assert ("chart_data.cache.set" in timers) is expect_write + + +@patch("superset.utils.timing_metrics.stats_logger_manager") +@patch("superset.common.query_context_processor.QueryCacheManager") +def test_cache_outcome_is_bypass_without_cache_backend( + cache_manager: MagicMock, stats_logger_manager: MagicMock +) -> None: + cache_manager.has_backend.return_value = False + cache_manager.get.return_value = _cache(False) + processor: QueryContextProcessor = _processor() + with ( + patch.object(processor, "query_cache_key", return_value="key"), + patch.object(processor, "get_cache_timeout", return_value=300), + patch.object(processor, "get_query_result", return_value=MagicMock()), + patch.object(processor, "get_annotation_data", return_value={}), + ): + processor.get_df_payload_result(_query_obj()) + + timers, counters = _emitted(stats_logger_manager.instance) + assert counters == {"chart_data.cache.bypass"} + assert "chart_data.cache.get" not in timers + + +@patch("superset.utils.timing_metrics.stats_logger_manager") +@patch("superset.common.query_context_processor.QueryCacheManager") +def test_query_timers_emitted_when_acquisition_raises( + cache_manager: MagicMock, stats_logger_manager: MagicMock +) -> None: + cache_manager.get.return_value = _cache(False) + processor: QueryContextProcessor = _processor() + with ( + patch.object(processor, "query_cache_key", return_value="key"), + patch.object(processor, "get_cache_timeout", return_value=300), + patch.object(processor, "get_query_result", side_effect=RuntimeError("boom")), + pytest.raises(RuntimeError), + ): + processor.get_df_payload_result(_query_obj()) + + timers, _ = _emitted(stats_logger_manager.instance) + assert { + "chart_data.query.planning", + "chart_data.query.data_acquisition", + "chart_data.query.total", + } <= timers + assert "chart_data.query.payload_assembly" not in timers diff --git a/tests/unit_tests/models/core_test.py b/tests/unit_tests/models/core_test.py index a14b68d50c8d..1f102c9bbd3b 100644 --- a/tests/unit_tests/models/core_test.py +++ b/tests/unit_tests/models/core_test.py @@ -1163,9 +1163,9 @@ def test_prequeries_execute_on_real_connections( with database.get_raw_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE name = 'prequery_marker'") - assert cursor.fetchone() is not None, ( - "prequery did not run on the new connection" - ) + assert ( + cursor.fetchone() is not None + ), "prequery did not run on the new connection" assert _ENGINE_CACHE == {} @@ -2405,6 +2405,75 @@ def fetch_data(_: object) -> list[tuple[int]]: assert description == result_description +def test_get_df_emits_warehouse_phase_timers(mocker: MockerFixture) -> None: + """Connect, execute, fetch and dataframe build are timed per engine.""" + database = Database(database_name="my_db", sqlalchemy_uri="sqlite://") + + cursor = mocker.MagicMock() + cursor.description = [("value", "BIGINT", None, None, None, None, None)] + conn = mocker.MagicMock() + conn.cursor.return_value = cursor + get_raw_connection = mocker.patch.object(database, "get_raw_connection") + get_raw_connection.return_value.__enter__.return_value = conn + mocker.patch.object(database.db_engine_spec, "execute") + mocker.patch.object(database.db_engine_spec, "fetch_data", return_value=[(1,)]) + stats_logger_manager = mocker.patch( + "superset.utils.timing_metrics.stats_logger_manager" + ) + + database.get_df("SELECT 1") + + timers = [ + call.args[0] for call in stats_logger_manager.instance.timing.call_args_list + ] + assert timers == [ + "database.connect.sqlite", + "database.execute.sqlite", + "database.fetch.sqlite", + "database.dataframe.sqlite", + ] + + +def test_get_df_emits_connect_timer_when_connection_fails( + mocker: MockerFixture, +) -> None: + """A failed connection attempt is still timed.""" + database = Database(database_name="my_db", sqlalchemy_uri="sqlite://") + mocker.patch.object(database, "get_raw_connection", side_effect=OSError("down")) + stats_logger_manager = mocker.patch( + "superset.utils.timing_metrics.stats_logger_manager" + ) + + with pytest.raises(OSError, match="down"): + database.get_df("SELECT 1") + + timers = [ + call.args[0] for call in stats_logger_manager.instance.timing.call_args_list + ] + assert timers == ["database.connect.sqlite"] + + +def test_get_df_dataframe_timer_excludes_mutator(mocker: MockerFixture) -> None: + """The dataframe timer covers DataFrame construction only.""" + database = Database(database_name="my_db", sqlalchemy_uri="sqlite://") + mocker.patch.object( + database, + "_execute_sql_with_mutation_and_logging", + return_value=(mocker.MagicMock(), [(1,)], [("value",)]), + ) + mocker.patch.object(database, "load_into_dataframe", return_value=pd.DataFrame()) + timed = mocker.patch("superset.models.core.timed") + seen_during_mutator: list[bool] = [] + + def mutator(df: pd.DataFrame) -> pd.DataFrame: + seen_during_mutator.append(timed.return_value.__exit__.called) + return df + + database.get_df("SELECT 1", mutator=mutator) + + assert seen_during_mutator == [True] + + def test_post_process_df_non_zero_based_index() -> None: """ post_process_df must not raise when the DataFrame index doesn't contain 0 diff --git a/tests/unit_tests/utils/test_timing_metrics.py b/tests/unit_tests/utils/test_timing_metrics.py new file mode 100644 index 000000000000..6e3769835b8a --- /dev/null +++ b/tests/unit_tests/utils/test_timing_metrics.py @@ -0,0 +1,123 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from collections.abc import Callable, Iterator +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from flask import Flask + +from superset.utils.timing_metrics import ( + emit_incr, + emit_pre_handler_timer, + emit_timing, + metric_suffix, + start_pre_handler_timer, + timed, +) + + +@pytest.fixture +def stats_logger() -> Iterator[MagicMock]: + with patch("superset.utils.timing_metrics.stats_logger_manager") as manager: + yield manager.instance + + +def test_timed_emits_milliseconds(stats_logger: MagicMock) -> None: + with patch( + "superset.utils.timing_metrics.perf_counter_ns", + side_effect=[1_000_000, 3_500_000], + ): + with timed("chart_data.cache.get"): + pass + + stats_logger.timing.assert_called_once_with("chart_data.cache.get", 2.5) + + +def test_timed_emits_when_block_raises(stats_logger: MagicMock) -> None: + with pytest.raises(ValueError, match="boom"): + with timed("database.execute.snowflake"): + raise ValueError("boom") + + stats_logger.timing.assert_called_once() + assert stats_logger.timing.call_args.args[0] == "database.execute.snowflake" + + +def test_metrics_backend_failures_are_swallowed(stats_logger: MagicMock) -> None: + stats_logger.timing.side_effect = OSError("statsd down") + stats_logger.incr.side_effect = OSError("statsd down") + + emit_timing("chart_data.query.total", 1.0) + emit_incr("chart_data.cache.hit") + with timed("chart_data.cache.set"): + result = 42 + + assert result == 42 + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("snowflake", "snowflake"), + ("google.bigquery", "google_bigquery"), + ("", "unknown"), + (None, "unknown"), + ], +) +def test_metric_suffix_keeps_single_key_segment( + value: str | None, expected: str +) -> None: + assert metric_suffix(value) == expected + + +def test_pre_handler_timer_spans_wrapped_decorators(stats_logger: MagicMock) -> None: + """The timer covers whatever sits between the two markers (e.g. @protect).""" + + def handler() -> str: + return "ok" + + def fake_protect(func: Callable[..., Any]) -> Callable[..., Any]: + return func + + route = start_pre_handler_timer( + fake_protect(emit_pre_handler_timer("chart_data.auth")(handler)) + ) + + app = Flask(__name__) + with ( + app.test_request_context(), + patch( + "superset.utils.timing_metrics.perf_counter_ns", + side_effect=[1_000_000, 5_000_000], + ), + ): + assert route() == "ok" + + stats_logger.timing.assert_called_once_with("chart_data.auth", 4.0) + assert route.__name__ == "handler" + + +def test_pre_handler_timer_skips_requests_never_started( + stats_logger: MagicMock, +) -> None: + handler = emit_pre_handler_timer("chart_data.auth")(lambda: "ok") + + app = Flask(__name__) + with app.test_request_context(): + assert handler() == "ok" + + stats_logger.timing.assert_not_called()