Bug fix: Make collector not fail on null values

collector/pg_database.go

@@ -15,6 +15,7 @@ package collector
 
 import (
 "context"
+"database/sql"
 
 "github.com/go-kit/log"
 "github.com/prometheus/client_golang/prometheus"
@@ -79,32 +80,39 @@ func (c PGDatabaseCollector) Update(ctx context.Context, instance *instance, ch
 var databases []string
 
 for rows.Next() {
-var datname string
+var datname sql.NullString
 if err := rows.Scan(&datname); err != nil {
 return err
 }
 
+if !datname.Valid {
+continue
+}
 // Ignore excluded databases
 // Filtering is done here instead of in the query to avoid
 // a complicated NOT IN query with a variable number of parameters
-if sliceContains(c.excludedDatabases, datname) {
+if sliceContains(c.excludedDatabases, datname.String) {
 continue
 }
 
-databases = append(databases, datname)
+databases = append(databases, datname.String)
 }
 
 // Query the size of the databases
 for _, datname := range databases {
-var size int64
+var size sql.NullFloat64
 err = db.QueryRowContext(ctx, pgDatabaseSizeQuery, datname).Scan(&size)
 if err != nil {
 return err
 }
 
+sizeMetric := 0.0
+if size.Valid {
+sizeMetric = size.Float64
+}
 ch <- prometheus.MustNewConstMetric(
 pgDatabaseSizeDesc,
-prometheus.GaugeValue, float64(size), datname,
+prometheus.GaugeValue, sizeMetric, datname,
 )
 }
 if err := rows.Err(); err != nil {

collector/pg_database_test.go

@@ -59,3 +59,43 @@ func TestPGDatabaseCollector(t *testing.T) {
 t.Errorf("there were unfulfilled exceptions: %s", err)
 }
 }
+
+// TODO add a null db test
+
+func TestPGDatabaseCollectorNullMetric(t *testing.T) {
+db, mock, err := sqlmock.New()
+if err != nil {
+t.Fatalf("Error opening a stub db connection: %s", err)
+}
+defer db.Close()
+
+inst := &instance{db: db}
+
+mock.ExpectQuery(sanitizeQuery(pgDatabaseQuery)).WillReturnRows(sqlmock.NewRows([]string{"datname"}).
+AddRow("postgres"))
+
+mock.ExpectQuery(sanitizeQuery(pgDatabaseSizeQuery)).WithArgs("postgres").WillReturnRows(sqlmock.NewRows([]string{"pg_database_size"}).
+AddRow(nil))
+
+ch := make(chan prometheus.Metric)
+go func() {
+defer close(ch)
+c := PGDatabaseCollector{}
+if err := c.Update(context.Background(), inst, ch); err != nil {
+t.Errorf("Error calling PGDatabaseCollector.Update: %s", err)
+}
+}()
+
+expected := []MetricResult{
+{labels: labelMap{"datname": "postgres"}, value: 0, metricType: dto.MetricType_GAUGE},
+}
+convey.Convey("Metrics comparison", t, func() {
+for _, expect := range expected {
+m := readMetric(<-ch)
+convey.So(expect, convey.ShouldResemble, m)
+}
+})
+if err := mock.ExpectationsWereMet(); err != nil {
+t.Errorf("there were unfulfilled exceptions: %s", err)
+}
+}

collector/pg_postmaster.go

@@ -15,6 +15,7 @@ package collector
 
 import (
 "context"
+"database/sql"
 
 "github.com/prometheus/client_golang/prometheus"
 )
@@ -51,14 +52,18 @@ func (c *PGPostmasterCollector) Update(ctx context.Context, instance *instance,
 row := db.QueryRowContext(ctx,
 pgPostmasterQuery)
 
-var startTimeSeconds float64
+var startTimeSeconds sql.NullFloat64
 err := row.Scan(&startTimeSeconds)
 if err != nil {
 return err
 }
+startTimeSecondsMetric := 0.0
+if startTimeSeconds.Valid {
+startTimeSecondsMetric = startTimeSeconds.Float64
+}
 ch <- prometheus.MustNewConstMetric(
 pgPostMasterStartTimeSeconds,
-prometheus.GaugeValue, startTimeSeconds,
+prometheus.GaugeValue, startTimeSecondsMetric,
 )
 return nil
 }

collector/pg_postmaster_test.go

@@ -57,3 +57,39 @@ func TestPgPostmasterCollector(t *testing.T) {
 t.Errorf("there were unfulfilled exceptions: %s", err)
 }
 }
+
+func TestPgPostmasterCollectorNullTime(t *testing.T) {
+db, mock, err := sqlmock.New()
+if err != nil {
+t.Fatalf("Error opening a stub db connection: %s", err)
+}
+defer db.Close()
+
+inst := &instance{db: db}
+
+mock.ExpectQuery(sanitizeQuery(pgPostmasterQuery)).WillReturnRows(sqlmock.NewRows([]string{"pg_postmaster_start_time"}).
+AddRow(nil))
+
+ch := make(chan prometheus.Metric)
+go func() {
+defer close(ch)
+c := PGPostmasterCollector{}
+
+if err := c.Update(context.Background(), inst, ch); err != nil {
+t.Errorf("Error calling PGPostmasterCollector.Update: %s", err)
+}
+}()
+
+expected := []MetricResult{
+{labels: labelMap{}, value: 0, metricType: dto.MetricType_GAUGE},
+}
+convey.Convey("Metrics comparison", t, func() {
+for _, expect := range expected {
+m := readMetric(<-ch)
+convey.So(expect, convey.ShouldResemble, m)
+}
+})
+if err := mock.ExpectationsWereMet(); err != nil {
+t.Errorf("there were unfulfilled exceptions: %s", err)
+}
+}

collector/pg_process_idle.go

@@ -15,21 +15,23 @@ package collector
 
 import (
 "context"
+"database/sql"
 
 "github.com/go-kit/log"
 "github.com/prometheus/client_golang/prometheus"
 )
 
-const processIdleSubsystem = "process_idle"
-
 func init() {
-registerCollector(processIdleSubsystem, defaultEnabled, NewPGProcessIdleCollector)
+// Making this default disabled because we have no tests for it
+registerCollector(processIdleSubsystem, defaultDisabled, NewPGProcessIdleCollector)
 }
 
 type PGProcessIdleCollector struct {
 log log.Logger
 }
 
+const processIdleSubsystem = "process_idle"
+
 func NewPGProcessIdleCollector(config collectorConfig) (Collector, error) {
 return &PGProcessIdleCollector{log: config.logger}, nil
 }
@@ -41,8 +43,8 @@ var pgProcessIdleSeconds = prometheus.NewDesc(
 prometheus.Labels{},
 )
 
-func (PGProcessIdleCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
-db := instance.getDB()
+func (PGProcessIdleCollector) Update(ctx context.Context, inst *instance, ch chan<- prometheus.Metric) error {
+db := inst.getDB()
 row := db.QueryRowContext(ctx,
 `WITH
 metrics AS (
@@ -79,9 +81,9 @@ func (PGProcessIdleCollector) Update(ctx context.Context, instance *instance, ch
 FROM metrics JOIN buckets USING (application_name)
 GROUP BY 1, 2, 3;`)
 
-var applicationName string
-var secondsSum int64
-var secondsCount uint64
+var applicationName sql.NullString
+var secondsSum sql.NullInt64
+var secondsCount sql.NullInt64
 var seconds []int64
 var secondsBucket []uint64
 
@@ -97,10 +99,24 @@ func (PGProcessIdleCollector) Update(ctx context.Context, instance *instance, ch
 if err != nil {
 return err
 }
+
+applicationNameLabel := "unknown"
+if applicationName.Valid {
+applicationNameLabel = applicationName.String
+}
+
+var secondsCountMetric uint64
+if secondsCount.Valid {
+secondsCountMetric = uint64(secondsCount.Int64)
+}
+secondsSumMetric := 0.0
+if secondsSum.Valid {
+secondsSumMetric = float64(secondsSum.Int64)
+}
 ch <- prometheus.MustNewConstHistogram(
 pgProcessIdleSeconds,
-secondsCount, float64(secondsSum), buckets,
-applicationName,
+secondsCountMetric, secondsSumMetric, buckets,
+applicationNameLabel,
 )
 return nil
 }

collector/pg_replication_slot.go

@@ -15,6 +15,7 @@ package collector
 
 import (
 "context"
+"database/sql"
 
 "github.com/go-kit/log"
 "github.com/prometheus/client_golang/prometheus"
@@ -82,32 +83,44 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, instance *instance
 defer rows.Close()
 
 for rows.Next() {
-var slotName string
-var walLSN int64
-var flushLSN int64
-var isActive bool
+var slotName sql.NullString
+var walLSN sql.NullFloat64
+var flushLSN sql.NullFloat64
+var isActive sql.NullBool
 if err := rows.Scan(&slotName, &walLSN, &flushLSN, &isActive); err != nil {
 return err
 }
 
-isActiveValue := 0
-if isActive {
-isActiveValue = 1
+isActiveValue := 0.0
+if isActive.Valid && isActive.Bool {
+isActiveValue = 1.0
+}
+slotNameLabel := "unknown"
+if slotName.Valid {
+slotNameLabel = slotName.String
 }
 
+var walLSNMetric float64
+if walLSN.Valid {
+walLSNMetric = walLSN.Float64
+}
 ch <- prometheus.MustNewConstMetric(
 pgReplicationSlotCurrentWalDesc,
-prometheus.GaugeValue, float64(walLSN), slotName,
+prometheus.GaugeValue, walLSNMetric, slotNameLabel,
 )
-if isActive {
+if isActive.Valid && isActive.Bool {
+var flushLSNMetric float64
+if flushLSN.Valid {
+flushLSNMetric = flushLSN.Float64
+}
 ch <- prometheus.MustNewConstMetric(
 pgReplicationSlotCurrentFlushDesc,
-prometheus.GaugeValue, float64(flushLSN), slotName,
+prometheus.GaugeValue, flushLSNMetric, slotNameLabel,
 )
 }
 ch <- prometheus.MustNewConstMetric(
 pgReplicationSlotIsActiveDesc,
-prometheus.GaugeValue, float64(isActiveValue), slotName,
+prometheus.GaugeValue, isActiveValue, slotNameLabel,
 )
 }
 if err := rows.Err(); err != nil {

collector/pg_replication_slot_test.go

@@ -103,3 +103,84 @@ func TestPgReplicationSlotCollectorInActive(t *testing.T) {
 }
 
 }
+
+func TestPgReplicationSlotCollectorActiveNil(t *testing.T) {
+db, mock, err := sqlmock.New()
+if err != nil {
+t.Fatalf("Error opening a stub db connection: %s", err)
+}
+defer db.Close()
+
+inst := &instance{db: db}
+
+columns := []string{"slot_name", "current_wal_lsn", "confirmed_flush_lsn", "active"}
+rows := sqlmock.NewRows(columns).
+AddRow("test_slot", 6, 12, nil)
+mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)
+
+ch := make(chan prometheus.Metric)
+go func() {
+defer close(ch)
+c := PGReplicationSlotCollector{}
+
+if err := c.Update(context.Background(), inst, ch); err != nil {
+t.Errorf("Error calling PGReplicationSlotCollector.Update: %s", err)
+}
+}()
+
+expected := []MetricResult{
+{labels: labelMap{"slot_name": "test_slot"}, value: 6, metricType: dto.MetricType_GAUGE},
+{labels: labelMap{"slot_name": "test_slot"}, value: 0, metricType: dto.MetricType_GAUGE},
+}
+
+convey.Convey("Metrics comparison", t, func() {
+for _, expect := range expected {
+m := readMetric(<-ch)
+convey.So(expect, convey.ShouldResemble, m)
+}
+})
+if err := mock.ExpectationsWereMet(); err != nil {
+t.Errorf("there were unfulfilled exceptions: %s", err)
+}
+}
+
+func TestPgReplicationSlotCollectorTestNilValues(t *testing.T) {
+db, mock, err := sqlmock.New()
+if err != nil {
+t.Fatalf("Error opening a stub db connection: %s", err)
+}
+defer db.Close()
+
+inst := &instance{db: db}
+
+columns := []string{"slot_name", "current_wal_lsn", "confirmed_flush_lsn", "active"}
+rows := sqlmock.NewRows(columns).
+AddRow(nil, nil, nil, true)
+mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)
+
+ch := make(chan prometheus.Metric)
+go func() {
+defer close(ch)
+c := PGReplicationSlotCollector{}
+
+if err := c.Update(context.Background(), inst, ch); err != nil {
+t.Errorf("Error calling PGReplicationSlotCollector.Update: %s", err)
+}
+}()
+
+expected := []MetricResult{
+{labels: labelMap{"slot_name": "unknown"}, value: 0, metricType: dto.MetricType_GAUGE},
+{labels: labelMap{"slot_name": "unknown"}, value: 0, metricType: dto.MetricType_GAUGE},
+{labels: labelMap{"slot_name": "unknown"}, value: 1, metricType: dto.MetricType_GAUGE},
+}
+
+convey.Convey("Metrics comparison", t, func() {
+for _, expect := range expected {
+m := readMetric(<-ch)
+convey.So(expect, convey.ShouldResemble, m)
+}
+})
+if err := mock.ExpectationsWereMet(); err != nil {
+t.Errorf("there were unfulfilled exceptions: %s", err)
+}
+}