Add alerts for database check metrics.

Signed-off-by: Harkishen-Singh <[email protected]>

Author: Harkishen-Singh

CHANGELOG.md

@@ -15,7 +15,8 @@ We use the following categories for changes:
 ## [Unreleased]
 
 ## Added
-- Alerting rules for Promscale. You can find them at [here](docs/promscale_alerting.md) [#1181]
+- Alerting rules for Promscale. You can find them [here](docs/promscale_alerting.md) [#1181, #1185]
+- Add database status and request metrics [#1185]
 
 ### Fixed
 - Register `promscale_ingest_channel_len_bucket` metric and make it a gauge [#1177]

docs/mixin/alerts/alerts.yaml

@@ -1,3 +1,5 @@
+# Note: Alert thresholds are experimental. Feel free to change them or suggest back at
+# Promscale channel in TimescaleDB slack.
 groups:
 - name: promscale-general
   rules:
@@ -61,7 +63,7 @@ groups:
       severity: warning
     annotations:
       summary: Slow Promscale ingestion
-      description: "90% of ingestion batch took {{ $value }} seconds to ingest."
+      description: "Slowest 10% of ingestion batch took more than {{ $value }} seconds to ingest."
   - alert: PromscaleIngestHighLatency
     expr: |
       (
@@ -81,7 +83,7 @@ groups:
       severity: critical
     annotations:
       summary: Slow Promscale ingestion
-      description: "90% of ingestion batch took {{ $value }} seconds to ingest."
+      description: "Slowest 10% of ingestion batch took more than {{ $value }} seconds to ingest."
 - name: promscale-query
   rules:
   - alert: PromscaleQueryHighErrorRate
@@ -135,7 +137,7 @@ groups:
       severity: warning
     annotations:
       summary: Slow Promscale querying
-      description: "90% of the queries took {{ $value }} seconds to evaluate."
+      description: "Slowest 10% of the queries took more than {{ $value }} seconds to evaluate."
   - alert: PromscaleQueryHighLatency
     expr: |
       (
@@ -155,7 +157,7 @@ groups:
       severity: critical
     annotations:
       summary: Slow Promscale querying
-      description: "90% of the queries took {{ $value }} seconds to evaluate."
+      description: "Slowest 10% of the queries took {{ $value }} seconds to evaluate."
 - name: promscale-cache
   rules:
   - alert: PromscaleCacheHighNumberOfEvictions
@@ -189,4 +191,41 @@ groups:
       severity: warning
     annotations:
       summary: High cache eviction in Promscale
-      description: "Promscale {{ $labels.name }} has a hit ratio of {{ $value }}."
+      description: "Promscale {{ $labels.name }} has a hit ratio of {{ $value | humanizePercentage }}."
+- name: promscale-database-connection
+  rules:
+  - alert: PromscaleStorageHighErrorRate
+    expr: |
+      (
+        sum by (job) (
+          # Error counter exists for query, query_row & exec, and not for send_batch.
+          rate(promscale_database_request_errors_total{method=~"query.*|exec"}[5m])
+        )
+      /
+        sum by (job) (
+          rate(promscale_database_requests_total{method=~"query.*|exec"}[5m])
+        )
+      ) > 0.05
+    labels:
+      severity: warning
+    annotations:
+      summary: Promscale experiences a high error rate when connecting to the database
+      description: "Promscale connection with the database has an error of {{ $value | humanizePercentage }}."
+  - alert: PromscaleStorageHighLatency
+    expr: |
+      (
+        histogram_quantile(0.9,
+          sum by (le, job, type) (
+            rate(promscale_database_requests_duration_seconds_bucket[5m])
+          )
+        ) > 5
+      and
+        sum by (job, type) (
+          rate(promscale_database_requests_duration_seconds_count[5m])
+        ) > 0
+      )
+    labels:
+      severity: warning
+    annotations:
+      summary: Slow database response
+      description: "Slowest 10% of database requests are taking more than {{ $value }} seconds to respond."

pkg/pgclient/client.go

@@ -8,13 +8,10 @@ import (
 	"context"
 	"fmt"
 	"net/url"
-	"os"
-	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v4"
 	"github.com/jackc/pgx/v4/pgxpool"
-	"github.com/prometheus/client_golang/prometheus"
 	"go.opentelemetry.io/collector/model/pdata"
 
 	"github.com/timescale/promscale/pkg/ha"
@@ -30,7 +27,6 @@ import (
 	"github.com/timescale/promscale/pkg/query"
 	"github.com/timescale/promscale/pkg/telemetry"
 	"github.com/timescale/promscale/pkg/tenancy"
-	"github.com/timescale/promscale/pkg/util"
 )
 
 var PromscaleID uuid.UUID
@@ -60,7 +56,6 @@ type Client struct {
 	sigClose          chan struct{}
 	haService         *ha.Service
 	TelemetryEngine   telemetry.Engine
-	stopHealthChecker context.CancelFunc
 }
 
 // NewClient creates a new PostgreSQL client
@@ -185,9 +180,6 @@ func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt t
 		}
 	}
 
-	healthCheckerCtx, stopHealthChecker := context.WithCancel(context.Background())
-	healthCheckRoutine(healthCheckerCtx, dbConn)
-
 	client := &Client{
 		Connection:        dbConn,
 		QuerierConnection: dbQuerierConn,
@@ -200,7 +192,6 @@ func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt t
 		seriesCache:       seriesCache,
 		sigClose:          sigClose,
 		TelemetryEngine:   telemetryEngine,
-		stopHealthChecker: stopHealthChecker,
 	}
 
 	InitClientMetrics(client)
@@ -210,9 +201,6 @@ func NewClientWithPool(cfg *Config, numCopiers int, connPool *pgxpool.Pool, mt t
 // Close closes the client and performs cleanup
 func (c *Client) Close() {
 	log.Info("msg", "Shutting down Client")
-	if c.stopHealthChecker != nil {
-		c.stopHealthChecker()
-	}
 	if c.TelemetryEngine != nil {
 		c.TelemetryEngine.Stop()
 	}
@@ -309,43 +297,3 @@ func observeStatementCacheState(conn *pgx.Conn) bool {
 	statementCacheLen.Observe(float64(statementCacheSize))
 	return true
 }
-
-func healthCheckRoutine(ctx context.Context, conn pgxconn.PgxConn) {
-	r := prometheus.DefaultRegisterer
-	if env := os.Getenv("IS_TEST"); env == "true" {
-		r = prometheus.NewRegistry()
-	}
-	dbHealthChecks := prometheus.NewCounter(
-		prometheus.CounterOpts{
-			Namespace: util.PromNamespace,
-			Subsystem: "database",
-			Name:      "health_checks_total",
-			Help:      "Total number of database health checks performed.",
-		},
-	)
-	dbHealthErrors := prometheus.NewCounter(
-		prometheus.CounterOpts{
-			Namespace: util.PromNamespace,
-			Subsystem: "database",
-			Name:      "health_check_errors_total",
-			Help:      "Total number of database health check errors.",
-		},
-	)
-	r.MustRegister(dbHealthChecks, dbHealthErrors)
-	go func() {
-		check := time.NewTicker(time.Minute)
-		defer check.Stop()
-		connection := health.NewHealthChecker(conn)
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-check.C:
-			}
-			dbHealthChecks.Inc()
-			if err := connection(); err != nil {
-				dbHealthErrors.Inc()
-			}
-		}
-	}()
-}

pkg/pgxconn/implement.go

@@ -0,0 +1,99 @@
+package pgxconn
+
+import (
+	"time"
+
+	"github.com/jackc/pgconn"
+	"github.com/jackc/pgproto3/v2"
+	"github.com/jackc/pgx/v4"
+)
+
+// rowWithTelemetry wraps the row returned by QueryRow() for metrics telemetry.
+type rowWithTelemetry struct {
+	row pgx.Row
+}
+
+func (w rowWithTelemetry) Scan(dest ...interface{}) error {
+	err := w.row.Scan(dest...)
+	if err != nil && err != pgx.ErrNoRows {
+		errorsTotal.With(promMethodLabel("query_row")).Inc()
+	}
+	return err
+}
+
+// rowsWithDuration wraps the Query() function with duration metric taken to complete the entire execution.
+type rowsWithDuration struct {
+	rows  pgx.Rows
+	start time.Time
+}
+
+func newRowsWithDuration(rows pgx.Rows, start time.Time) rowsWithDuration {
+	return rowsWithDuration{rows, start}
+}
+
+func (r rowsWithDuration) Next() bool {
+	return r.rows.Next()
+}
+
+func (r rowsWithDuration) Scan(dest ...interface{}) error {
+	return r.rows.Scan(dest...)
+}
+
+func (r rowsWithDuration) Err() error {
+	return r.rows.Err()
+}
+
+func (r rowsWithDuration) Close() {
+	defer func() { duration.With(promMethodLabel("query")).Observe(time.Since(r.start).Seconds()) }()
+	r.rows.Close()
+}
+
+func (r rowsWithDuration) CommandTag() pgconn.CommandTag {
+	return r.rows.CommandTag()
+}
+
+func (r rowsWithDuration) FieldDescriptions() []pgproto3.FieldDescription {
+	return r.rows.FieldDescriptions()
+}
+
+func (r rowsWithDuration) Values() ([]interface{}, error) {
+	return r.rows.Values()
+}
+
+func (r rowsWithDuration) RawValues() [][]byte {
+	return r.rows.RawValues()
+}
+
+// batchResultsWithDuration wraps the SendBatch and records the duration of the batch till it closed.
+type batchResultsWithDuration struct {
+	batch pgx.BatchResults
+	start time.Time
+}
+
+func newBatchResultsWithDuration(b pgx.BatchResults, start time.Time) batchResultsWithDuration {
+	return batchResultsWithDuration{
+		batch: b,
+		start: start,
+	}
+}
+
+func (w batchResultsWithDuration) Close() error {
+	defer func() { duration.With(promMethodLabel("send_batch")).Observe(time.Since(w.start).Seconds()) }()
+	return w.batch.Close()
+}
+
+func (w batchResultsWithDuration) Exec() (pgconn.CommandTag, error) {
+	return w.batch.Exec()
+}
+
+func (w batchResultsWithDuration) Query() (pgx.Rows, error) {
+	return w.batch.Query()
+}
+
+func (w batchResultsWithDuration) QueryRow() pgx.Row {
+	return rowWithTelemetry{w.batch.QueryRow()}
+}
+
+func (w batchResultsWithDuration) QueryFunc(scans []interface{}, f func(pgx.QueryFuncRow) error) (pgconn.CommandTag, error) {
+	return w.batch.QueryFunc(scans, f)
+}