MySQL blocking queries (#20008)

* blocking joins

* blocking column

* blocking column

* idle subquery

* idle blockers subquery as an option

* changelog

* linter

* fixed variables

* fixed variables

* blocking_thread_id

* idle session bug fix

* config

* cache the query

* specify mysql version in the change log

* validate config

* linter

* linter

* test case

* moved config to query_activity

* validate config

* linter

* fixed idle blocker test

* removed flakey lock assert

* trigger ci

* linter

Author: nenadnoveljic

mysql/assets/configuration/spec.yaml

@@ -542,6 +542,13 @@ files:
                 value:
                   type: number
                   example: 10
+              - name: collect_blocking_queries
+                description: |
+                  Enable collection of blocking queries. Supported only on MySQL 8.0.
+                value:
+                  type: boolean
+                  example: false
+                  display_default: false
           - name: index_metrics
             description: |
               Configure collection of index metrics.

mysql/changelog.d/20008.added

@@ -0,0 +1 @@
+Add blocking queries support for MySQL 8

mysql/datadog_checks/mysql/activity.py

@@ -58,29 +58,53 @@
     waits_a.index_name,
     waits_a.object_type,
     waits_a.source
+    {blocking_columns}
 FROM
     performance_schema.threads AS thread_a
     LEFT JOIN performance_schema.events_waits_current AS waits_a ON waits_a.thread_id = thread_a.thread_id
     LEFT JOIN performance_schema.events_statements_current AS statement ON statement.thread_id = thread_a.thread_id
+    {blocking_joins}
 WHERE
-    thread_a.processlist_state IS NOT NULL
-    AND thread_a.processlist_command != 'Sleep'
-    AND thread_a.processlist_id != CONNECTION_ID()
-    AND thread_a.PROCESSLIST_COMMAND != 'Daemon'
-    AND (waits_a.EVENT_NAME != 'idle' OR waits_a.EVENT_NAME IS NULL)
-    AND (waits_a.operation != 'idle' OR waits_a.operation IS NULL)
-    -- events_waits_current can have multiple rows per thread, thus we use EVENT_ID to identify the row we want to use.
-    -- Additionally, we want the row with the highest EVENT_ID which reflects the most recent and current wait.
-    AND (
-        waits_a.event_id = (
-           SELECT
-              MAX(waits_b.EVENT_ID)
-          FROM  performance_schema.events_waits_current AS waits_b
-          Where waits_b.thread_id = thread_a.thread_id
-    ) OR waits_a.event_id is NULL)
-    -- We ignore rows without SQL text because there will be rows for background operations that do not have
-    -- SQL text associated with it.
-    AND COALESCE(statement.sql_text, thread_a.PROCESSLIST_info) != '';
+    (
+        thread_a.processlist_state IS NOT NULL
+        AND thread_a.processlist_id != CONNECTION_ID()
+        AND thread_a.PROCESSLIST_COMMAND != 'Daemon'
+        AND thread_a.processlist_command != 'Sleep'
+        AND (waits_a.EVENT_NAME != 'idle' OR waits_a.EVENT_NAME IS NULL)
+        AND (waits_a.operation != 'idle' OR waits_a.operation IS NULL)
+        -- events_waits_current can have multiple rows per thread, thus we use EVENT_ID to identify the row
+        -- we want to use. Additionally, we want the row with the highest EVENT_ID which reflects the most recent wait.
+        AND (
+            waits_a.event_id = (
+            SELECT
+                MAX(waits_b.EVENT_ID)
+            FROM  performance_schema.events_waits_current AS waits_b
+            Where waits_b.thread_id = thread_a.thread_id
+        ) OR waits_a.event_id is NULL)
+        -- We ignore rows without SQL text because there will be rows for background operations that do not have
+        -- SQL text associated with it.
+        AND COALESCE(statement.sql_text, thread_a.PROCESSLIST_info) != ''
+    )
+    {idle_blockers_subquery};
+"""
+
+BLOCKING_COLUMNS = """\
+    ,blocking_thread.thread_id AS blocking_thread_id,
+    blocking_thread.processlist_id AS blocking_processlist_id
+"""
+
+BLOCKING_JOINS = """\
+    LEFT JOIN performance_schema.data_lock_waits AS lock_waits ON thread_a.thread_id = lock_waits.requesting_thread_id
+    LEFT JOIN performance_schema.threads AS blocking_thread ON lock_waits.blocking_thread_id = blocking_thread.thread_id
+"""
+
+IDLE_BLOCKERS_SUBQUERY = """\
+        OR
+        -- Include idle sessions that are blocking others
+        thread_a.thread_id IN (
+            SELECT blocking_thread_id
+            FROM performance_schema.data_lock_waits
+        )
 """
 
 
@@ -127,6 +151,7 @@ def __init__(self, check, config, connection_args):
         self._db = None
         self._db_version = None
         self._obfuscator_options = to_native_string(json.dumps(self._config.obfuscator_options))
+        self._activity_query = None
 
     def run_job(self):
         # type: () -> None
@@ -180,11 +205,36 @@ def _collect_activity(self):
                 tags=tags + self._check._get_debug_tags(),
             )
 
+    def _should_collect_blocking_queries(self):
+        # type: () -> bool
+        blocking_queries_configured = self._config.activity_config.get("collect_blocking_queries", False)
+        return (
+            blocking_queries_configured and self._db_version == MySQLVersion.VERSION_80 and not self._check.is_mariadb
+        )
+
+    def _get_activity_query(self):
+        # type: () -> str
+        if self._activity_query:
+            return self._activity_query
+        blocking_columns = ""
+        blocking_joins = ""
+        idle_blockers_subquery = ""
+        if self._should_collect_blocking_queries():
+            blocking_columns = BLOCKING_COLUMNS
+            blocking_joins = BLOCKING_JOINS
+            idle_blockers_subquery = IDLE_BLOCKERS_SUBQUERY
+        return ACTIVITY_QUERY.format(
+            blocking_columns=blocking_columns,
+            blocking_joins=blocking_joins,
+            idle_blockers_subquery=idle_blockers_subquery,
+        )
+
     @tracked_method(agent_check_getter=agent_check_getter, track_result_length=True)
     def _get_activity(self, cursor):
         # type: (pymysql.cursor) -> List[Dict[str]]
-        self._log.debug("Running activity query [%s]", ACTIVITY_QUERY)
-        cursor.execute(ACTIVITY_QUERY)
+        query = self._get_activity_query()
+        self._log.debug("Running activity query [%s]", query)
+        cursor.execute(query)
         return cursor.fetchall()
 
     def _normalize_rows(self, rows):

mysql/datadog_checks/mysql/config.py

@@ -48,6 +48,7 @@ def __init__(self, instance, init_config):
         self.activity_config = instance.get('query_activity', {}) or {}
         self.schemas_config: dict = instance.get('schemas_collection', {}) or {}
         self.index_config: dict = instance.get('index_metrics', {}) or {}
+        self.collect_blocking_queries = is_affirmative(instance.get('collect_blocking_queries', False))
 
         self.cloud_metadata = {}
         aws = instance.get('aws', {})

mysql/datadog_checks/mysql/config_models/instance.py

@@ -130,6 +130,7 @@ class QueryActivity(BaseModel):
         arbitrary_types_allowed=True,
         frozen=True,
     )
+    collect_blocking_queries: Optional[bool] = None
     collection_interval: Optional[float] = None
     enabled: Optional[bool] = None
 

mysql/datadog_checks/mysql/data/conf.yaml.example

@@ -520,6 +520,11 @@ instances:
         #
         # collection_interval: 10
 
+        ## @param collect_blocking_queries - boolean - optional - default: false
+        ## Enable collection of blocking queries. Supported only on MySQL 8.0.
+        #
+        # collect_blocking_queries: false
+
     ## Configure collection of index metrics.
     ## Metrics provided by the options:
     ##   - mysql.index.size (per index)

mysql/tests/test_query_activity.py

@@ -8,7 +8,7 @@
 import time
 from concurrent.futures.thread import ThreadPoolExecutor
 from contextlib import closing
-from copy import copy
+from copy import copy, deepcopy
 from datetime import datetime
 from threading import Event
 
@@ -52,7 +52,7 @@ def dbm_instance(instance_complex):
 @pytest.mark.integration
 @pytest.mark.usefixtures('dd_environment')
 @pytest.mark.parametrize(
-    "query,query_signature,expected_query_truncated",
+    "query,query_signature,expected_query_truncated,collect_blocking_queries",
     [
         (
             'SELECT id, name FROM testdb.users FOR UPDATE',
@@ -62,6 +62,7 @@ def dbm_instance(instance_complex):
                 else 'aca1be410fbadb61'
             ),
             StatementTruncationState.not_truncated.value,
+            True,
         ),
         (
             'SELECT id, {} FROM testdb.users FOR UPDATE'.format(
@@ -73,11 +74,16 @@ def dbm_instance(instance_complex):
                 else ('da7d6b1e9deb88e' if MYSQL_VERSION_PARSED > parse_version('5.7') else '63bd1fd025c7f7fb')
             ),
             StatementTruncationState.truncated.value,
+            False,
         ),
     ],
 )
-def test_activity_collection(aggregator, dbm_instance, dd_run_check, query, query_signature, expected_query_truncated):
-    check = MySql(CHECK_NAME, {}, [dbm_instance])
+def test_activity_collection(
+    aggregator, dbm_instance, dd_run_check, query, query_signature, expected_query_truncated, collect_blocking_queries
+):
+    config = deepcopy(dbm_instance)
+    config['query_activity']['collect_blocking_queries'] = collect_blocking_queries
+    check = MySql(CHECK_NAME, {}, instances=[config])
 
     blocking_query = 'SELECT id FROM testdb.users FOR UPDATE'
 
@@ -148,9 +154,17 @@ def _run_blocking(conn):
     assert blocked_row['wait_timer_end'], "missing wait timer end"
     assert blocked_row['event_timer_start'], "missing event timer start"
     assert blocked_row['event_timer_end'], "missing event timer end"
-    assert blocked_row['lock_time'], "missing lock time"
     assert blocked_row['query_truncated'] == expected_query_truncated
 
+    if check._query_activity._should_collect_blocking_queries():
+        assert len(activity['mysql_activity']) >= 2, "should have collected at least two activity payloads"
+        captured_idle_blocker = False
+        for activity in dbm_activity:
+            for row in activity['mysql_activity']:
+                if row['processlist_user'] == 'bob':
+                    captured_idle_blocker = True
+        assert captured_idle_blocker, "should have captured the idle blocker"
+
 
 @pytest.mark.integration
 @pytest.mark.usefixtures('dd_environment')