PMM-12392 Collect processlist from performance_schema on newer MySQL

collector/exporter.go

@@ -132,7 +132,7 @@ func (e *Exporter) Collect(ch chan<- prometheus.Metric) {
 func (e *Exporter) scrape(ctx context.Context, ch chan<- prometheus.Metric) float64 {
 	var err error
 	scrapeTime := time.Now()
-	instance, err := newInstance(e.dsn)
+	instance, err := newInstance(ctx, e.dsn)
 	if err != nil {
 		e.logger.Error("Error opening connection to database", "err", err)
 		return 0.0

collector/exporter_test.go

@@ -126,7 +126,7 @@ func TestGetMySQLVersion(t *testing.T) {
 	}
 
 	convey.Convey("Version parsing", t, func() {
-		instance, err := newInstance(dsn)
+		instance, err := newInstance(t.Context(), dsn)
 		convey.So(err, convey.ShouldBeNil)
 
 		convey.So(instance.versionMajorMinor, convey.ShouldBeBetweenOrEqual, 5.7, 11.4)

collector/instance.go

@@ -14,7 +14,9 @@
 package collector
 
 import (
+	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"regexp"
 	"strconv"
@@ -25,9 +27,10 @@ import (
 )
 
 const (
-	FlavorMySQL   = "mysql"
-	FlavorMariaDB = "mariadb"
-	versionQuery  = "SELECT @@version;"
+	FlavorMySQL            = "mysql"
+	FlavorMariaDB          = "mariadb"
+	versionQuery           = "SELECT @@version;"
+	performanceSchemaQuery = "SELECT @@performance_schema;"
 )
 
 var (
@@ -46,13 +49,14 @@ var (
 )
 
 type instance struct {
-	db                *sql.DB
-	flavor            string
-	version           semver.Version
-	versionMajorMinor float64
+	db                         *sql.DB
+	flavor                     string
+	version                    semver.Version
+	versionMajorMinor          float64
+	isPerformanceSchemaEnabled bool
 }
 
-func newInstance(dsn string) (*instance, error) {
+func newInstance(ctx context.Context, dsn string) (*instance, error) {
 	i := &instance{}
 	db, err := sql.Open("mysql", dsn)
 	if err != nil {
@@ -64,7 +68,7 @@ func newInstance(dsn string) (*instance, error) {
 
 	i.db = db
 
-	version, versionString, err := queryVersion(db)
+	version, versionString, err := queryVersion(ctx, db)
 	if err != nil {
 		db.Close()
 		return nil, err
@@ -86,9 +90,29 @@ func newInstance(dsn string) (*instance, error) {
 		i.flavor = FlavorMySQL
 	}
 
+	isPerformanceSchemaEnabled, err := queryPerformanceSchemaEnabled(ctx, db)
+	if err != nil {
+		db.Close()
+		return nil, err
+	}
+	i.isPerformanceSchemaEnabled = isPerformanceSchemaEnabled
+
 	return i, nil
 }
 
+// queryPerformanceSchemaEnabled reports whether performance_schema is enabled in the server
+func queryPerformanceSchemaEnabled(ctx context.Context, db *sql.DB) (bool, error) {
+	var enabled uint8
+	err := db.QueryRowContext(ctx, performanceSchemaQuery).Scan(&enabled)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("failed to query performance_schema status: %w", err)
+	}
+	return enabled == 1, nil
+}
+
 // getDB returns the database connection for the instance.
 func (i *instance) getDB() *sql.DB {
 	return i.db
@@ -115,9 +139,9 @@ func (i *instance) Ping() error {
 // for MySQL: "8.0.36-28.1"
 var versionRegex = regexp.MustCompile(`^((\d+)(\.\d+)(\.\d+))`)
 
-func queryVersion(db *sql.DB) (semver.Version, string, error) {
+func queryVersion(ctx context.Context, db *sql.DB) (semver.Version, string, error) {
 	var version string
-	err := db.QueryRow(versionQuery).Scan(&version)
+	err := db.QueryRowContext(ctx, versionQuery).Scan(&version)
 	if err != nil {
 		return semver.Version{}, version, err
 	}

collector/instance_test.go

@@ -33,32 +33,32 @@ func TestGetMySQLVersion_Percona(t *testing.T) {
 		var semVer semver.Version
 		var err error
 		mock.ExpectQuery(versionQuery).WillReturnRows(sqlmock.NewRows([]string{""}).AddRow(""))
-		semVer, _, err = queryVersion(db)
+		semVer, _, err = queryVersion(t.Context(), db)
 		convey.ShouldBeError(err)
 		convey.So(semVer.String(), convey.ShouldEqual, "0.0.0")
 
 		mock.ExpectQuery(versionQuery).WillReturnRows(sqlmock.NewRows([]string{""}).AddRow("something"))
-		semVer, _, err = queryVersion(db)
+		semVer, _, err = queryVersion(t.Context(), db)
 		convey.ShouldBeNil(err)
 		convey.So(semVer.String(), convey.ShouldEqual, "0.0.0")
 
 		mock.ExpectQuery(versionQuery).WillReturnRows(sqlmock.NewRows([]string{""}).AddRow("10.1.17-MariaDB"))
-		semVer, _, err = queryVersion(db)
+		semVer, _, err = queryVersion(t.Context(), db)
 		convey.ShouldBeNil(err)
 		convey.So(semVer.String(), convey.ShouldEqual, "10.1.17")
 
 		mock.ExpectQuery(versionQuery).WillReturnRows(sqlmock.NewRows([]string{""}).AddRow("5.7.13-6-log"))
-		semVer, _, err = queryVersion(db)
+		semVer, _, err = queryVersion(t.Context(), db)
 		convey.ShouldBeNil(err)
 		convey.So(semVer.String(), convey.ShouldEqual, "5.7.13")
 
 		mock.ExpectQuery(versionQuery).WillReturnRows(sqlmock.NewRows([]string{""}).AddRow("5.6.30-76.3-56-log"))
-		semVer, _, err = queryVersion(db)
+		semVer, _, err = queryVersion(t.Context(), db)
 		convey.ShouldBeNil(err)
 		convey.So(semVer.String(), convey.ShouldEqual, "5.6.30")
 
 		mock.ExpectQuery(versionQuery).WillReturnRows(sqlmock.NewRows([]string{""}).AddRow("5.5.51-38.1"))
-		semVer, _, err = queryVersion(db)
+		semVer, _, err = queryVersion(t.Context(), db)
 		convey.ShouldBeNil(err)
 		convey.So(semVer.String(), convey.ShouldEqual, "5.5.51")
 	})

collector/percona_info_schema_process_list.go

@@ -22,18 +22,31 @@ import (
 	"strings"
 
 	"github.com/alecthomas/kingpin/v2"
+	"github.com/blang/semver/v4"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+const (
+	processlistInfoSchema = "information_schema"
+	processlistPerfSchema = "performance_schema"
+)
+
 const pInfoSchemaProcesslistQuery = `
 		SELECT COALESCE(command,''),COALESCE(state,''),count(*),sum(time)
-		  FROM information_schema.processlist
+		  FROM %s.processlist
 		  WHERE ID != connection_id()
 		    AND TIME >= %d
 		  GROUP BY command,state
 		  ORDER BY null
 		`
 
+// MySQL version boundaries for querying perf schema
+var (
+	v8_0_22 = semver.MustParse("8.0.22")
+	v5_7_39 = semver.MustParse("5.7.39")
+	v8_0_0  = semver.MustParse("8.0.0")
+)
+
 // Tunable flags.
 var (
 	pProcesslistMinTime = kingpin.Flag(
@@ -185,10 +198,19 @@ func (PScrapeProcesslist) Version() float64 {
 
 // Scrape collects data from database connection and sends it over channel as prometheus metric.
 func (PScrapeProcesslist) Scrape(ctx context.Context, instance *instance, ch chan<- prometheus.Metric, logger *slog.Logger) error {
-	processQuery := fmt.Sprintf(
-		pInfoSchemaProcesslistQuery,
-		*pProcesslistMinTime,
-	)
+	// Prefer querying performance_schema.processlist instead of information_schema.processlist to avoid negative perf consequences
+	// Supported by MySQL >=5.7.39 and >=8.0.22
+	usePerfSchema := instance.flavor == FlavorMySQL &&
+		instance.isPerformanceSchemaEnabled &&
+		(instance.version.GTE(v8_0_22) ||
+			(instance.version.GTE(v5_7_39) && instance.version.LT(v8_0_0)))
+
+	schema := processlistInfoSchema
+	if usePerfSchema {
+		schema = processlistPerfSchema
+	}
+
+	processQuery := fmt.Sprintf(pInfoSchemaProcesslistQuery, schema, *pProcesslistMinTime)
 	db := instance.getDB()
 	processlistRows, err := db.QueryContext(ctx, processQuery)
 	if err != nil {

collector/percona_info_schema_process_list_test.go

@@ -0,0 +1,87 @@
+// Copyright 2018 The Prometheus Authors, 2023 Percona LLC
+// Licensed 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.
+
+package collector
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/alecthomas/kingpin/v2"
+	"github.com/blang/semver/v4"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/promslog"
+)
+
+func TestPScrapeProcesslistQuerySelection(t *testing.T) {
+	if _, err := kingpin.CommandLine.Parse([]string{}); err != nil {
+		t.Fatal(err)
+	}
+
+	cases := []struct {
+		name              string
+		flavor            string
+		version           semver.Version
+		perfSchemaEnabled bool
+		expectedSchema    string
+	}{
+		{"MySQL 8.0.22 + PS on -> perf_schema", FlavorMySQL, semver.MustParse("8.0.22"), true, processlistPerfSchema},
+		{"MySQL 8.0.30 + PS on -> perf_schema", FlavorMySQL, semver.MustParse("8.0.30"), true, processlistPerfSchema},
+		{"MySQL 5.7.39 + PS on -> perf_schema", FlavorMySQL, semver.MustParse("5.7.39"), true, processlistPerfSchema},
+		{"MySQL 8.0.22 + PS off -> info_schema", FlavorMySQL, semver.MustParse("8.0.22"), false, processlistInfoSchema},
+		{"MySQL 5.7.39 + PS off -> info_schema", FlavorMySQL, semver.MustParse("5.7.39"), false, processlistInfoSchema},
+		{"MySQL 8.0.21 -> info_schema", FlavorMySQL, semver.MustParse("8.0.21"), true, processlistInfoSchema},
+		{"MySQL 5.7.38 -> info_schema", FlavorMySQL, semver.MustParse("5.7.38"), true, processlistInfoSchema},
+		{"MySQL 8.0.0 -> info_schema", FlavorMySQL, semver.MustParse("8.0.0"), true, processlistInfoSchema},
+		{"MySQL 5.6.50 -> info_schema", FlavorMySQL, semver.MustParse("5.6.50"), true, processlistInfoSchema},
+		{"MariaDB 10.11.0 -> info_schema", FlavorMariaDB, semver.MustParse("10.11.0"), true, processlistInfoSchema},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			db, mock, err := sqlmock.New()
+			if err != nil {
+				t.Fatalf("error opening a stub database connection: %s", err)
+			}
+			defer db.Close()
+
+			inst := &instance{
+				db:                         db,
+				flavor:                     tc.flavor,
+				version:                    tc.version,
+				isPerformanceSchemaEnabled: tc.perfSchemaEnabled,
+			}
+
+			expectedSQL := fmt.Sprintf(pInfoSchemaProcesslistQuery, tc.expectedSchema, 0)
+			columns := []string{"command", "state", "count", "time"}
+			mock.ExpectQuery(sanitizeQuery(expectedSQL)).
+				WillReturnRows(sqlmock.NewRows(columns))
+
+			ch := make(chan prometheus.Metric)
+			go func() {
+				if err := (PScrapeProcesslist{}).Scrape(context.Background(), inst, ch, promslog.NewNopLogger()); err != nil {
+					t.Errorf("error calling Scrape: %s", err)
+				}
+				close(ch)
+			}()
+			for range ch {
+			}
+
+			if err := mock.ExpectationsWereMet(); err != nil {
+				t.Errorf("unfulfilled expectations: %s", err)
+			}
+		})
+	}
+}

percona/tests/Makefile

@@ -40,9 +40,9 @@ prepare-base-exporter:
 	tar -xf assets/mysqld_exporter_percona.tar.xz -C assets/
 
 start-mysql-db:
-	docker-compose -f assets/mysql-compose.yml up -d --force-recreate --renew-anon-volumes --remove-orphans
+	docker compose -f assets/mysql-compose.yml up -d --force-recreate --renew-anon-volumes --remove-orphans
 
 stop-mysql-db:
-	docker-compose -f assets/mysql-compose.yml down
+	docker compose -f assets/mysql-compose.yml down
 
 prepare-env-from-repo: prepare-exporter-from-repo prepare-base-exporter start-mysql-db

percona/tests/assets/test.exporter-flags.txt

@@ -23,7 +23,6 @@
 --exporter.max-idle-conns=3
 --exporter.max-open-conns=3
 --exporter.conn-max-lifetime=55s
---exporter.global-conn-pool
 --collect.info_schema.innodb_tablespaces
 --collect.auto_increment.columns
 --collect.info_schema.tables

percona/tests/env_prepare_test.go

@@ -51,7 +51,7 @@ func extractExporter(gzipStream io.Reader, fileName string) {
 		case tar.TypeDir:
 			continue
 		case tar.TypeReg:
-			if strings.HasSuffix(header.Name, "postgres_exporter") {
+			if strings.HasSuffix(header.Name, "mysqld_exporter") {
 				outFile, err := os.Create(fileName)
 				if err != nil {
 					log.Fatalf("ExtractTarGz: Create() failed: %s", err.Error())

percona/tests/utils_test.go

@@ -53,21 +53,27 @@ func launchExporter(fileName string) (cmd *exec.Cmd, port int, collectOutput fun
 		return nil, 0, nil, fmt.Errorf("Failed to find free port in range [%d..%d]", portRangeStart, portRangeEnd)
 	}
 
-	linesStr := string(lines)
+	linesStr := strings.TrimSpace(string(lines))
 	linesStr += fmt.Sprintf("\n--web.listen-address=127.0.0.1:%d", port)
+	linesStr += fmt.Sprintf("\n--mysqld.address=%s:%d", mysqlHost, mysqlPort)
+	linesStr += fmt.Sprintf("\n--mysqld.username=%s", mysqlUser)
+	linesStr += "\n--config.my-cnf=/dev/null"
 
 	absolutePath, _ := filepath.Abs("custom-queries")
 	linesStr += fmt.Sprintf("\n--collect.custom_query.hr.directory=%s/high-resolution", absolutePath)
 	linesStr += fmt.Sprintf("\n--collect.custom_query.mr.directory=%s/medium-resolution", absolutePath)
 	linesStr += fmt.Sprintf("\n--collect.custom_query.lr.directory=%s/low-resolution", absolutePath)
 
-	linesArr := strings.Split(linesStr, "\n")
-
-	dsn := fmt.Sprintf("DATA_SOURCE_NAME=%s:%s@(%s:%d)/", mysqlUser, mysqlPassword, mysqlHost, mysqlPort)
+	linesArr := make([]string, 0)
+	for _, l := range strings.Split(linesStr, "\n") {
+		if l = strings.TrimSpace(l); l != "" {
+			linesArr = append(linesArr, l)
+		}
+	}
 
 	cmd = exec.Command(fileName, linesArr...)
 	cmd.Env = os.Environ()
-	cmd.Env = append(cmd.Env, dsn)
+	cmd.Env = append(cmd.Env, "MYSQLD_EXPORTER_PASSWORD="+mysqlPassword)
 
 	var outBuffer, errorBuffer bytes.Buffer
 	cmd.Stdout = &outBuffer