Fix mysql query metric collection not recovering after database resta…

…rts (#10811)

* Fix mysql query metric collection not recovering after database restarts

* Revert "Fix mysql query metric collection not recovering after database restarts"

This reverts commit 2e46b8d.

* Let Exceptions Bubble up and Restart Query Metrics Job

mysql/datadog_checks/mysql/statements.py

@@ -109,29 +109,26 @@ def run_job(self):
         self.collect_per_statement_metrics()
 
     def collect_per_statement_metrics(self):
-        try:
-            rows = self._collect_per_statement_metrics()
-            if not rows:
-                return
-
-            for event in self._rows_to_fqt_events(rows):
-                self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
-
-            # truncate query text to the maximum length supported by metrics tags
-            for row in rows:
-                row['digest_text'] = row['digest_text'][0:200] if row['digest_text'] is not None else None
-
-            payload = {
-                'host': self._check.resolved_hostname,
-                'timestamp': time.time() * 1000,
-                'ddagentversion': datadog_agent.get_version(),
-                'min_collection_interval': self._metric_collection_interval,
-                'tags': self._tags,
-                'mysql_rows': rows,
-            }
-            self._check.database_monitoring_query_metrics(json.dumps(payload, default=default_json_event_encoding))
-        except Exception:
-            self.log.exception('Unable to collect statement metrics due to an error')
+        rows = self._collect_per_statement_metrics()
+        if not rows:
+            return
+
+        for event in self._rows_to_fqt_events(rows):
+            self._check.database_monitoring_query_sample(json.dumps(event, default=default_json_event_encoding))
+
+        # truncate query text to the maximum length supported by metrics tags
+        for row in rows:
+            row['digest_text'] = row['digest_text'][0:200] if row['digest_text'] is not None else None
+
+        payload = {
+            'host': self._check.resolved_hostname,
+            'timestamp': time.time() * 1000,
+            'ddagentversion': datadog_agent.get_version(),
+            'min_collection_interval': self._metric_collection_interval,
+            'tags': self._tags,
+            'mysql_rows': rows,
+        }
+        self._check.database_monitoring_query_metrics(json.dumps(payload, default=default_json_event_encoding))
 
     def _collect_per_statement_metrics(self):
         # type: () -> List[PyMysqlRow]
@@ -169,15 +166,10 @@ def _query_summary_per_statement(self):
             ORDER BY `count_star` DESC
             LIMIT 10000"""
 
-        rows = []  # type: List[PyMysqlRow]
-
-        try:
-            with closing(self._get_db_connection().cursor(pymysql.cursors.DictCursor)) as cursor:
-                cursor.execute(sql_statement_summary)
+        with closing(self._get_db_connection().cursor(pymysql.cursors.DictCursor)) as cursor:
+            cursor.execute(sql_statement_summary)
 
-                rows = cursor.fetchall() or []  # type: ignore
-        except (pymysql.err.InternalError, pymysql.err.OperationalError) as e:
-            self.log.warning("Statement summary metrics are unavailable at this time: %s", e)
+            rows = cursor.fetchall() or []  # type: ignore
 
         return rows