Send replication slots metrics to Prometheus

.travis.yml

@@ -4,6 +4,10 @@ language: go
 go: "1.10"
 addons:
   postgresql: "9.6"
+before_script:
+  - psql -U travis -c "ALTER SYSTEM SET wal_level TO 'logical';"
+  - psql -U travis -c "ALTER SYSTEM SET max_replication_slots TO '10';"
+  - sudo /etc/init.d/postgresql restart 
 services:
   - docker
   - postgresql

gauges/logical_replication.go

@@ -0,0 +1,95 @@
+package gauges
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/ContaAzul/postgresql_exporter/postgres"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type slots struct {
+	Name     string  `db:"slot_name"`
+	Active   float64 `db:"active"`
+	TotalLag float64 `db:"total_lag"`
+}
+
+// ReplicationSlotStatus returns the state of the replication slots
+func (g *Gauges) ReplicationSlotStatus() *prometheus.GaugeVec {
+	var gauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name:        "postgresql_replication_slot_status",
+			Help:        "Returns 1 if the slot is currently actively being used",
+			ConstLabels: g.labels,
+		},
+		[]string{"slot_name"},
+	)
+	go func() {
+		for {
+			gauge.Reset()
+			var slots []slots
+			if err := g.query(
+				`
+					SELECT
+						slot_name,
+						active::int
+					FROM pg_replication_slots
+					WHERE slot_type = 'logical'
+					  AND "database" = current_database();
+				`,
+				&slots,
+				emptyParams,
+			); err == nil {
+				for _, slot := range slots {
+					gauge.With(prometheus.Labels{
+						"slot_name": slot.Name,
+					}).Set(slot.Active)
+				}
+			}
+			time.Sleep(g.interval)
+		}
+	}()
+	return gauge
+}
+
+// ReplicationSlotLagInBytes returns the total lag in bytes from the replication slots
+func (g *Gauges) ReplicationSlotLagInBytes() *prometheus.GaugeVec {
+	var gauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name:        "postgresql_replication_lag_bytes",
+			Help:        "Total lag of the replication slots in bytes",
+			ConstLabels: g.labels,
+		},
+		[]string{"slot_name"},
+	)
+	go func() {
+		for {
+			gauge.Reset()
+			var slots []slots
+			if err := g.query(
+				fmt.Sprintf(
+					`
+						SELECT
+							slot_name,
+							%s(%s(), confirmed_flush_lsn) AS total_lag
+						FROM pg_replication_slots
+						WHERE slot_type = 'logical'
+						AND "database" = current_database();
+					`,
+					postgres.Version(g.version()).WalLsnDiffFunctionName(),
+					postgres.Version(g.version()).CurrentWalLsnFunctionName(),
+				),
+				&slots,
+				emptyParams,
+			); err == nil {
+				for _, slot := range slots {
+					gauge.With(prometheus.Labels{
+						"slot_name": slot.Name,
+					}).Set(slot.TotalLag)
+				}
+			}
+			time.Sleep(g.interval)
+		}
+	}()
+	return gauge
+}

gauges/logical_replication_test.go

@@ -0,0 +1,43 @@
+package gauges
+
+import (
+	"database/sql"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReplicationSlotStatus(t *testing.T) {
+	var assert = assert.New(t)
+	db, gauges, close := prepare(t)
+	defer close()
+	dropTestLogicalReplicationSlot := createTestLogicalReplicationSlot("test_status", t, db)
+	defer dropTestLogicalReplicationSlot()
+	var metrics = evaluate(t, gauges.ReplicationSlotStatus())
+	assert.Len(metrics, 1)
+	assertEqual(t, 0, metrics[0])
+	assertNoErrs(t, gauges)
+}
+
+func TestReplicationSlotLagInMegabytes(t *testing.T) {
+	var assert = assert.New(t)
+	db, gauges, close := prepare(t)
+	defer close()
+	dropTestLogicalReplicationSlot := createTestLogicalReplicationSlot("test_lag", t, db)
+	defer dropTestLogicalReplicationSlot()
+	var metrics = evaluate(t, gauges.ReplicationDelayInBytes())
+	assert.Len(metrics, 1)
+	assertEqual(t, 0, metrics[0])
+	assertNoErrs(t, gauges)
+}
+
+func createTestLogicalReplicationSlot(slotName string, t *testing.T, db *sql.DB) func() {
+	_, err := db.Exec(fmt.Sprintf("SELECT * FROM pg_create_logical_replication_slot('%s', 'test_decoding');", slotName))
+	require.NoError(t, err)
+	return func() {
+		_, err := db.Exec(fmt.Sprintf("SELECT pg_drop_replication_slot('%s');", slotName))
+		assert.New(t).NoError(err)
+	}
+}

main.go

@@ -120,4 +120,6 @@ func watch(db *sql.DB, reg prometheus.Registerer, name string) {
 	reg.MustRegister(gauges.LastTimeVacuumRan())
 	reg.MustRegister(gauges.LastTimeAutoVacuumRan())
 	reg.MustRegister(gauges.VacuumRunningTotal())
+	reg.MustRegister(gauges.ReplicationSlotStatus())
+	reg.MustRegister(gauges.ReplicationSlotLagInBytes())
 }

postgres/version.go

@@ -47,5 +47,13 @@ func (v Version) LastWalReplayedLsnFunctionName() string {
 		return "pg_last_wal_replay_lsn"
 	}
 	return "pg_last_xlog_replay_location"
+}
 
+// CurrentWalLsnFunctionName returns the name of the function that gets current 
+// write-ahead log write location according to the postgres version
+func (v Version) CurrentWalLsnFunctionName() string {
+	if v.IsEqualOrGreaterThan10() {
+		return "pg_current_wal_lsn"
+	}
+	return "pg_current_xlog_location"
 }