Add collector for performance_schema.processlist

README.md

@@ -136,7 +136,8 @@ collect.perf_schema.replication_applier_status_by_worker     | 5.7           | C
 collect.slave_status                                         | 5.1           | Collect from SHOW SLAVE STATUS (Enabled by default)
 collect.slave_hosts                                          | 5.1           | Collect from SHOW SLAVE HOSTS
 collect.sys.user_summary                                     | 5.7           | Collect metrics from sys.x$user_summary (disabled by default).
-
+collect.perf_schema.processlist                              | 8.0           | Collect thread state counts from performance_schema.processlist.
+collect.perf_schema.processlist.min_time                     | 8.0           | Minimum time a thread must be in each state to be counted. (default: 0)
 
 ### General Flags
 Name                                       | Description

collector/mysql_user.go

@@ -101,7 +101,7 @@ var (
 		labelNames, nil)
 )
 
-// ScrapeUser collects from `information_schema.processlist`.
+// ScrapeUser collects from `information_schema.processlist` or `performance_schema.processlist`.
 type ScrapeUser struct{}
 
 // Name of the Scraper. Should be unique.

collector/perf_schema_processlist.go

@@ -0,0 +1,201 @@
+// Copyright 2018 The Prometheus Authors
+// 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.
+
+// Scrape `performance_schema.processlist`.
+
+package collector
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"maps"
+	"slices"
+	"strings"
+
+	"github.com/alecthomas/kingpin/v2"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const perfSchemaProcesslistQuery = `
+		  SELECT
+		    user,
+		    SUBSTRING_INDEX(host, ':', 1) AS host,
+		    COALESCE(command, '') AS command,
+		    COALESCE(state, '') AS state,
+		    COUNT(*) AS processes,
+		    SUM(time) AS seconds
+		  FROM performance_schema.processlist
+		  WHERE ID != connection_id()
+		    AND TIME >= %d
+		  GROUP BY user, host, command, state
+	`
+
+// Tunable flags.
+var (
+	processlistMinTime = kingpin.Flag(
+		"collect.perf_schema.processlist.min_time",
+		"Minimum time a thread must be in each state to be counted",
+	).Default("0").Int()
+	processesByUserFlag = kingpin.Flag(
+		"collect.perf_schema.processlist.processes_by_user",
+		"Enable collecting the number of processes by user",
+	).Default("true").Bool()
+	processesByHostFlag = kingpin.Flag(
+		"collect.perf_schema.processlist.processes_by_host",
+		"Enable collecting the number of processes by host",
+	).Default("true").Bool()
+)
+
+// Metric descriptors.
+var (
+	processlistCountDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, performanceSchema, "processlist_threads"),
+		"The number of threads split by current state.",
+		[]string{"command", "state"}, nil)
+	processlistTimeDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, performanceSchema, "processlist_seconds"),
+		"The number of seconds threads have used split by current state.",
+		[]string{"command", "state"}, nil)
+	processesByUserDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, performanceSchema, "processlist_processes_by_user"),
+		"The number of processes by user.",
+		[]string{"mysql_user"}, nil)
+	processesByHostDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, performanceSchema, "processlist_processes_by_host"),
+		"The number of processes by host.",
+		[]string{"client_host"}, nil)
+)
+
+// ScrapePerfProcesslist collects from `performance_schema.processlist`.
+type ScrapePerfProcesslist struct{}
+
+// Name of the Scraper. Should be unique.
+func (ScrapePerfProcesslist) Name() string {
+	return performanceSchema + ".processlist"
+}
+
+// Help describes the role of the Scraper.
+func (ScrapePerfProcesslist) Help() string {
+	return "Collect current thread state counts from the performance_schema.processlist"
+}
+
+// Version of MySQL from which scraper is available.
+func (ScrapePerfProcesslist) Version() float64 {
+	return 8.0
+}
+
+// Scrape collects data from database connection and sends it over channel as prometheus metric.
+func (ScrapePerfProcesslist) Scrape(ctx context.Context, instance *instance, ch chan<- prometheus.Metric, logger *slog.Logger) error {
+	processQuery := fmt.Sprintf(
+		perfSchemaProcesslistQuery,
+		*processlistMinTime,
+	)
+	db := instance.getDB()
+	processlistRows, err := db.QueryContext(ctx, processQuery)
+	if err != nil {
+		return err
+	}
+	defer processlistRows.Close()
+
+	var (
+		user    string
+		host    string
+		command string
+		state   string
+		count   uint32
+		time    uint32
+	)
+	// Define maps
+	stateCounts := make(map[string]map[string]uint32)
+	stateTime := make(map[string]map[string]uint32)
+	stateHostCounts := make(map[string]uint32)
+	stateUserCounts := make(map[string]uint32)
+
+	for processlistRows.Next() {
+		err = processlistRows.Scan(&user, &host, &command, &state, &count, &time)
+		if err != nil {
+			return err
+		}
+		command = sanitizeState(command)
+		state = sanitizeState(state)
+		if host == "" {
+			host = "unknown"
+		}
+
+		// Init maps
+		if _, ok := stateCounts[command]; !ok {
+			stateCounts[command] = make(map[string]uint32)
+			stateTime[command] = make(map[string]uint32)
+		}
+		if _, ok := stateCounts[command][state]; !ok {
+			stateCounts[command][state] = 0
+			stateTime[command][state] = 0
+		}
+		if _, ok := stateHostCounts[host]; !ok {
+			stateHostCounts[host] = 0
+		}
+		if _, ok := stateUserCounts[user]; !ok {
+			stateUserCounts[user] = 0
+		}
+
+		stateCounts[command][state] += count
+		stateTime[command][state] += time
+		stateHostCounts[host] += count
+		stateUserCounts[user] += count
+	}
+
+	for _, command := range slices.Sorted(maps.Keys(stateCounts)) {
+		for _, state := range slices.Sorted(maps.Keys(stateCounts[command])) {
+			ch <- prometheus.MustNewConstMetric(processlistCountDesc, prometheus.GaugeValue, float64(stateCounts[command][state]), command, state)
+			ch <- prometheus.MustNewConstMetric(processlistTimeDesc, prometheus.GaugeValue, float64(stateTime[command][state]), command, state)
+		}
+	}
+
+	if *processesByHostFlag {
+		for _, host := range slices.Sorted(maps.Keys(stateHostCounts)) {
+			ch <- prometheus.MustNewConstMetric(processesByHostDesc, prometheus.GaugeValue, float64(stateHostCounts[host]), host)
+		}
+	}
+	if *processesByUserFlag {
+		for _, user := range slices.Sorted(maps.Keys(stateUserCounts)) {
+			ch <- prometheus.MustNewConstMetric(processesByUserDesc, prometheus.GaugeValue, float64(stateUserCounts[user]), user)
+		}
+	}
+
+	return nil
+}
+
+func sanitizeState(state string) string {
+	if state == "" {
+		state = "unknown"
+	}
+	state = strings.ToLower(state)
+	replacements := map[string]string{
+		";": "",
+		",": "",
+		":": "",
+		".": "",
+		"(": "",
+		")": "",
+		" ": "_",
+		"-": "_",
+	}
+	for r := range replacements {
+		state = strings.ReplaceAll(state, r, replacements[r])
+	}
+	return state
+}
+
+// check interface
+var _ Scraper = ScrapePerfProcesslist{}

collector/perf_schema_processlist_test.go

@@ -0,0 +1,98 @@
+// Copyright 2021 The Prometheus Authors
+// 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/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/promslog"
+	"github.com/smartystreets/goconvey/convey"
+)
+
+func TestScrapePerfProcesslist(t *testing.T) {
+	_, err := kingpin.CommandLine.Parse([]string{
+		"--collect.perf_schema.processlist.processes_by_user",
+		"--collect.perf_schema.processlist.processes_by_host",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	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}
+
+	query := fmt.Sprintf(perfSchemaProcesslistQuery, 0)
+	columns := []string{"user", "host", "command", "state", "processes", "seconds"}
+	rows := sqlmock.NewRows(columns).
+		AddRow("manager", "10.0.7.234", "Sleep", "", 10, 87).
+		AddRow("feedback", "10.0.7.154", "Sleep", "", 8, 842).
+		AddRow("root", "10.0.7.253", "Sleep", "", 1, 20).
+		AddRow("feedback", "10.0.7.179", "Sleep", "", 2, 14).
+		AddRow("system user", "", "Connect", "waiting for handler commit", 1, 7271248).
+		AddRow("manager", "10.0.7.234", "Sleep", "", 4, 62).
+		AddRow("system user", "", "Query", "Slave has read all relay log; waiting for more updates", 1, 7271248).
+		AddRow("event_scheduler", "localhost", "Daemon", "Waiting on empty queue", 1, 7271248)
+	mock.ExpectQuery(sanitizeQuery(query)).WillReturnRows(rows)
+
+	ch := make(chan prometheus.Metric)
+	go func() {
+		if err = (ScrapePerfProcesslist{}).Scrape(context.Background(), inst, ch, promslog.NewNopLogger()); err != nil {
+			t.Errorf("error calling function on test: %s", err)
+		}
+		close(ch)
+	}()
+
+	expected := []MetricResult{
+		{labels: labelMap{"command": "connect", "state": "waiting_for_handler_commit"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"command": "connect", "state": "waiting_for_handler_commit"}, value: 7271248, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"command": "daemon", "state": "waiting_on_empty_queue"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"command": "daemon", "state": "waiting_on_empty_queue"}, value: 7271248, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"command": "query", "state": "slave_has_read_all_relay_log_waiting_for_more_updates"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"command": "query", "state": "slave_has_read_all_relay_log_waiting_for_more_updates"}, value: 7271248, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"command": "sleep", "state": "unknown"}, value: 25, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"command": "sleep", "state": "unknown"}, value: 1025, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"client_host": "10.0.7.154"}, value: 8, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"client_host": "10.0.7.179"}, value: 2, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"client_host": "10.0.7.234"}, value: 14, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"client_host": "10.0.7.253"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"client_host": "localhost"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"client_host": "unknown"}, value: 2, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"mysql_user": "event_scheduler"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"mysql_user": "feedback"}, value: 10, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"mysql_user": "manager"}, value: 14, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"mysql_user": "root"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"mysql_user": "system user"}, value: 2, metricType: dto.MetricType_GAUGE},
+	}
+	convey.Convey("Metrics comparison", t, func() {
+		for _, expect := range expected {
+			got := readMetric(<-ch)
+			convey.So(expect, convey.ShouldResemble, got)
+		}
+	})
+
+	// Ensure all SQL queries were executed
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("there were unfulfilled exceptions: %s", err)
+	}
+}