feat: add prometheus metrics and dashboard

pkg/broker/database_instance.go

@@ -13,6 +13,7 @@ import (
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/config"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/install"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/models"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/queue"
 )
@@ -36,6 +37,7 @@ type DatabaseInstance struct {
 	shutdownMu   sync.RWMutex
 	jobsHandled  int64
 	startTime    time.Time
+	metrics      *metrics.Metrics
 
 	// sessionConn holds the pg_try_advisory_lock acquired by
 	// registerInstance. The lock is scoped to this one physical connection,
@@ -45,12 +47,16 @@ type DatabaseInstance struct {
 }
 
 // NewDatabaseInstance creates a new database instance
-func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db adapter.DBAdapter, logger adapter.Logger, version string, parentCtx context.Context) (*DatabaseInstance, error) {
+func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db adapter.DBAdapter, logger adapter.Logger, version string, parentCtx context.Context, brokerMetrics ...*metrics.Metrics) (*DatabaseInstance, error) {
 	hostname, err := os.Hostname()
 	if err != nil {
 		hostname = "unknown"
 	}
 
+	var instanceMetrics *metrics.Metrics
+	if len(brokerMetrics) > 0 {
+		instanceMetrics = brokerMetrics[0]
+	}
 	instance := &DatabaseInstance{
 		Name:         fmt.Sprintf("%s-%s", cfg.Broker.Name, dbCfg.Name),
 		DatabaseName: dbCfg.Name,
@@ -64,6 +70,7 @@ func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db ad
 		queues:       make(map[int]*queue.Queue),
 		ctx:          parentCtx,
 		startTime:    time.Now(),
+		metrics:      instanceMetrics,
 	}
 
 	return instance, nil
@@ -109,9 +116,54 @@ func (i *DatabaseInstance) Start() error {
 	adapter.SupervisedGo(i.logger, "stale-job-recovery-routine", i.staleJobRecoveryRoutine)
 
 	i.logger.Info("database instance started successfully")
+	if i.metrics != nil {
+		adapter.SupervisedGo(i.logger, "metrics-queue-depth-routine", i.queueDepthRoutine)
+	}
 	return nil
 }
 
+// queueDepthRoutine periodically exports pending jobs grouped by queue.
+func (i *DatabaseInstance) queueDepthRoutine() {
+	interval := time.Duration(i.config.Broker.QueueDepthPollSec) * time.Second
+	if interval <= 0 {
+		interval = 15 * time.Second
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	i.updateQueueDepthMetrics()
+	for {
+		select {
+		case <-ticker.C:
+			i.updateQueueDepthMetrics()
+		case <-i.ctx.Done():
+			return
+		}
+	}
+}
+
+func (i *DatabaseInstance) updateQueueDepthMetrics() {
+	rows, err := i.db.Query(i.ctx, "SELECT job_queue, COUNT(*) FROM broker.broker_jobs WHERE complete_status = 0 GROUP BY job_queue")
+	if err != nil {
+		i.logger.Warn("failed to collect queue depth metrics", "error", err)
+		return
+	}
+	defer rows.Close()
+	for queueNumber := 1; queueNumber <= i.dbConfig.QueueCount; queueNumber++ {
+		i.metrics.SetJobsQueued(i.DatabaseName, queueNumber, 0)
+	}
+	for rows.Next() {
+		var queueNumber, count int
+		if err := rows.Scan(&queueNumber, &count); err != nil {
+			i.logger.Warn("failed to scan queue depth metric", "error", err)
+			return
+		}
+		i.metrics.SetJobsQueued(i.DatabaseName, queueNumber, float64(count))
+	}
+	if err := rows.Err(); err != nil {
+		i.logger.Warn("failed to read queue depth metrics", "error", err)
+	}
+}
+
 // ensureSchema checks the embedded migration set against the database and,
 // depending on dbConfig.AutoMigrate, either applies pending migrations or
 // fails startup fast rather than running against a stale/missing schema.
@@ -261,6 +313,8 @@ func (i *DatabaseInstance) startQueues() error {
 			FetchSize:    i.config.Broker.FetchQueryQueSize,
 			TenantID:     i.dbConfig.TenantID,
 			LeaseSeconds: leaseSeconds,
+			Metrics:      i.metrics,
+			DatabaseName: i.DatabaseName,
 		}
 
 		q := queue.New(queueCfg)
@@ -271,6 +325,7 @@ func (i *DatabaseInstance) startQueues() error {
 		i.queues[queueNum] = q
 		i.logger.Info("queue started", "number", queueNum)
 	}
+	i.metrics.SetQueueCount(i.DatabaseName, len(i.queues))
 
 	return nil
 }