Merge pull request #183 from roidelapluie/pth

Implement heartbeat metrics

README.md

@@ -66,6 +66,9 @@ collect.perf_schema.indexiowaits                       | 5.6           | Collect
 collect.perf_schema.tableiowaits                       | 5.6           | Collect metrics from performance_schema.table_io_waits_summary_by_table.
 collect.perf_schema.tablelocks                         | 5.6           | Collect metrics from performance_schema.table_lock_waits_summary_by_table.
 collect.slave_status                                   | 5.1           | Collect from SHOW SLAVE STATUS (Enabled by default)
+collect.heartbeat                                      | 5.1           | Collect from [heartbeat](#heartbeat).
+collect.heartbeat.database                             | 5.1           | Database from where to collect heartbeat data. (default: heartbeat)
+collect.heartbeat.table                                | 5.1           | Table from where to collect heartbeat data. (default: heartbeat)
 
 
 ### General Flags
@@ -97,6 +100,14 @@ docker run -d -p 9104:9104 --link=my_mysql_container:bdd  \
         -e DATA_SOURCE_NAME="user:password@(bdd:3306)/database" prom/mysqld-exporter
 ```
 
+## heartbeat
+
+With `collect.heartbeat` enabled, mysqld_exporter will scrape replication delay
+measured by heartbeat mechanisms. [Pt-heartbeat][pth] is the
+reference heartbeat implementation supported.
+
+[pth]:https://www.percona.com/doc/percona-toolkit/2.2/pt-heartbeat.html
+
 ## Example Rules
 
 There are some sample rules available in [example.rules](example.rules)

collector/heartbeat.go

@@ -0,0 +1,89 @@
+// Scrape heartbeat data.
+
+package collector
+
+import (
+	"database/sql"
+	"fmt"
+	"strconv"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	// heartbeat is the Metric subsystem we use.
+	heartbeat = "heartbeat"
+	// heartbeatQuery is the query used to fetch the stored and current
+	// timestamps. %s will be replaced by the database and table name.
+	// The second column allows gets the server timestamp at the exact same
+	// time the query is run.
+	heartbeatQuery = "SELECT UNIX_TIMESTAMP(ts), UNIX_TIMESTAMP(NOW(6)), server_id from `%s`.`%s`"
+)
+
+// Metric descriptors.
+var (
+	HeartbeatStoredDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, heartbeat, "stored_timestamp_seconds"),
+		"Timestamp stored in the heartbeat table.",
+		[]string{"server_id"}, nil,
+	)
+	HeartbeatNowDesc = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, heartbeat, "now_timestamp_seconds"),
+		"Timestamp of the current server.",
+		[]string{"server_id"}, nil,
+	)
+)
+
+// ScrapeHeartbeat scrapes from the heartbeat table.
+// This is mainly targeting pt-heartbeat, but will work with any heartbeat
+// implementation that writes to a table with two columns:
+// CREATE TABLE heartbeat (
+//  ts                    varchar(26) NOT NULL,
+//  server_id             int unsigned NOT NULL PRIMARY KEY,
+// );
+func ScrapeHeartbeat(db *sql.DB, ch chan<- prometheus.Metric, collectDatabase, collectTable *string) error {
+	query := fmt.Sprintf(heartbeatQuery, *collectDatabase, *collectTable)
+	heartbeatRows, err := db.Query(query)
+	if err != nil {
+		return err
+	}
+	defer heartbeatRows.Close()
+
+	var (
+		now, ts  sql.RawBytes
+		serverId int
+	)
+
+	for heartbeatRows.Next() {
+		if err := heartbeatRows.Scan(&ts, &now, &serverId); err != nil {
+			return err
+		}
+
+		tsFloatVal, err := strconv.ParseFloat(string(ts), 64)
+		if err != nil {
+			return err
+		}
+
+		nowFloatVal, err := strconv.ParseFloat(string(now), 64)
+		if err != nil {
+			return err
+		}
+
+		serverId := strconv.Itoa(serverId)
+
+		ch <- prometheus.MustNewConstMetric(
+			HeartbeatNowDesc,
+			prometheus.GaugeValue,
+			nowFloatVal,
+			serverId,
+		)
+		ch <- prometheus.MustNewConstMetric(
+			HeartbeatStoredDesc,
+			prometheus.GaugeValue,
+			tsFloatVal,
+			serverId,
+		)
+	}
+
+	return nil
+}

collector/heartbeat_test.go

@@ -0,0 +1,49 @@
+package collector
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/smartystreets/goconvey/convey"
+	"gopkg.in/DATA-DOG/go-sqlmock.v1"
+)
+
+func TestScrapeHeartbeat(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("error opening a stub database connection: %s", err)
+	}
+	defer db.Close()
+
+	columns := []string{"UNIX_TIMESTAMP(ts)", "UNIX_TIMESTAMP(NOW(6))", "server_id"}
+	rows := sqlmock.NewRows(columns).
+		AddRow("1487597613.001320", "1487598113.448042", 1)
+	mock.ExpectQuery(sanitizeQuery("SELECT UNIX_TIMESTAMP(ts), UNIX_TIMESTAMP(NOW(6)), server_id from `heartbeat`.`heartbeat`")).WillReturnRows(rows)
+
+	ch := make(chan prometheus.Metric)
+	go func() {
+		database := "heartbeat"
+		table := "heartbeat"
+		if err = ScrapeHeartbeat(db, ch, &database, &table); err != nil {
+			t.Errorf("error calling function on test: %s", err)
+		}
+		close(ch)
+	}()
+
+	counterExpected := []MetricResult{
+		{labels: labelMap{"server_id": "1"}, value: 1487598113.448042, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"server_id": "1"}, value: 1487597613.00132, metricType: dto.MetricType_GAUGE},
+	}
+	convey.Convey("Metrics comparison", t, func() {
+		for _, expect := range counterExpected {
+			got := readMetric(<-ch)
+			convey.So(got, convey.ShouldResemble, expect)
+		}
+	})
+
+	// Ensure all SQL queries were executed
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("there were unfulfilled expections: %s", err)
+	}
+}

collector/slave_status.go

@@ -81,7 +81,7 @@ func ScrapeSlaveStatus(db *sql.DB, ch chan<- prometheus.Metric) error {
 
 		masterUUID := columnValue(scanArgs, slaveCols, "Master_UUID")
 		masterHost := columnValue(scanArgs, slaveCols, "Master_Host")
-		channelName := columnValue(scanArgs, slaveCols, "Channel_Name") // MySQL & Percona
+		channelName := columnValue(scanArgs, slaveCols, "Channel_Name")       // MySQL & Percona
 		connectionName := columnValue(scanArgs, slaveCols, "Connection_name") // MariaDB
 
 		for i, col := range slaveCols {

mysqld_exporter.go

@@ -117,6 +117,18 @@ var (
 	collectEngineInnodbStatus = flag.Bool("collect.engine_innodb_status", false,
 		"Collect from SHOW ENGINE INNODB STATUS",
 	)
+	collectHeartbeat = flag.Bool(
+		"collect.heartbeat", false,
+		"Collect from heartbeat",
+	)
+	collectHeartbeatDatabase = flag.String(
+		"collect.heartbeat.database", "heartbeat",
+		"Database from where to collect heartbeat data",
+	)
+	collectHeartbeatTable = flag.String(
+		"collect.heartbeat.table", "heartbeat",
+		"Table from where to collect heartbeat data",
+	)
 )
 
 // Metric name parts.
@@ -392,6 +404,12 @@ func (e *Exporter) scrape(ch chan<- prometheus.Metric) {
 			e.scrapeErrors.WithLabelValues("collect.engine_innodb_status").Inc()
 		}
 	}
+	if *collectHeartbeat {
+		if err = collector.ScrapeHeartbeat(db, ch, collectHeartbeatDatabase, collectHeartbeatTable); err != nil {
+			log.Errorln("Error scraping for collect.heartbeat:", err)
+			e.scrapeErrors.WithLabelValues("collect.heartbeat").Inc()
+		}
+	}
 }
 
 func parseMycnf(config interface{}) (string, error) {