feat: monitor warehouse databases (#4619)

go.mod

@@ -83,7 +83,7 @@ require (
 	github.com/rudderlabs/analytics-go v3.3.3+incompatible
 	github.com/rudderlabs/bing-ads-go-sdk v0.2.1
 	github.com/rudderlabs/compose-test v0.1.3
-	github.com/rudderlabs/rudder-go-kit v0.27.0
+	github.com/rudderlabs/rudder-go-kit v0.29.0
 	github.com/rudderlabs/rudder-observability-kit v0.0.3
 	github.com/rudderlabs/rudder-schemas v0.4.0
 	github.com/rudderlabs/sql-tunnels v0.1.6

go.sum

@@ -960,8 +960,8 @@ github.com/rudderlabs/compose-test v0.1.3 h1:uyep6jDCIF737sfv4zIaMsKRQKX95IDz5Xb
 github.com/rudderlabs/compose-test v0.1.3/go.mod h1:tuvS1eQdSfwOYv1qwyVAcpdJxPLQXJgy5xGDd/9XmMg=
 github.com/rudderlabs/parquet-go v0.0.2 h1:ZXRdZdimB0PdJtmxeSSxfI0fDQ3kZjwzBxRi6Ut1J8k=
 github.com/rudderlabs/parquet-go v0.0.2/go.mod h1:g6guum7o8uhj/uNhunnt7bw5Vabu/goI5i21/3fnxWQ=
-github.com/rudderlabs/rudder-go-kit v0.27.0 h1:GWJwhOcohTrWPgMlmhif/jCkMWylOcjesYNVFrMLUo4=
-github.com/rudderlabs/rudder-go-kit v0.27.0/go.mod h1:l+kq8EE3r08GcoBw1JO/AJ8O+tQqJnuwLNirWSnVIrk=
+github.com/rudderlabs/rudder-go-kit v0.29.0 h1:zz70OGuT9lCVSoUkLuBKNfGaR9Sz+QZwdu+T4F8ItdY=
+github.com/rudderlabs/rudder-go-kit v0.29.0/go.mod h1:dSJQHi1yR7wWps5BVQFpL56eRXk1bwCRDbuFDeplYSM=
 github.com/rudderlabs/rudder-observability-kit v0.0.3 h1:vZtuZRkGX+6rjaeKtxxFE2YYP6QlmAcVcgecTOjvz+Q=
 github.com/rudderlabs/rudder-observability-kit v0.0.3/go.mod h1:6UjAh3H6rkE0fFLh7z8ZGQEQbKtUkRfhWOf/OUhfqW8=
 github.com/rudderlabs/rudder-schemas v0.4.0 h1:CkpmO8NI0OFMcFyD6WAGN/gIsmqyrC+Bd/htb9eBhpo=

services/notifier/notifier.go

@@ -11,6 +11,8 @@ import (
 
 	"github.com/lib/pq"
 
+	"github.com/rudderlabs/rudder-go-kit/sqlutil"
+
 	"golang.org/x/sync/errgroup"
 
 	"github.com/cenkalti/backoff"
@@ -22,6 +24,7 @@ import (
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
+
 	migrator "github.com/rudderlabs/rudder-server/services/sql-migrator"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	sqlmw "github.com/rudderlabs/rudder-server/warehouse/integrations/middleware/sqlquerywrapper"
@@ -531,6 +534,16 @@ func (n *Notifier) UpdateClaim(
 	}
 }
 
+func (n *Notifier) Monitor(ctx context.Context) {
+	sqlutil.MonitorDatabase(
+		ctx,
+		n.conf,
+		n.statsFactory,
+		n.db.DB,
+		"notifier",
+	)
+}
+
 // RunMaintenance re-triggers zombie jobs which were left behind by dead workers in executing state
 // Since it's a blocking call, it should be run in a separate goroutine
 func (n *Notifier) RunMaintenance(ctx context.Context) error {

warehouse/app.go

@@ -13,6 +13,8 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
+	"github.com/rudderlabs/rudder-go-kit/sqlutil"
+
 	"github.com/cenkalti/backoff/v4"
 
 	"github.com/rudderlabs/rudder-go-kit/config"
@@ -316,6 +318,16 @@ func (a *App) Run(ctx context.Context) error {
 		a.bcManager.Start(gCtx)
 		return nil
 	})
+	g.Go(func() error {
+		sqlutil.MonitorDatabase(
+			gCtx,
+			a.conf,
+			a.statsFactory,
+			a.db.DB,
+			"warehouse",
+		)
+		return nil
+	})
 
 	if mode.IsDegraded(a.config.runningMode) {
 		a.logger.Info("Running warehouse service in degraded mode...")
@@ -410,6 +422,10 @@ func (a *App) Run(ctx context.Context) error {
 	g.Go(func() error {
 		return a.api.Start(gCtx)
 	})
+	g.Go(func() error {
+		a.notifier.Monitor(gCtx)
+		return nil
+	})
 	g.Go(func() error {
 		<-gCtx.Done()
 		return a.notifier.Shutdown()