[+] define sync metric operations types as constants (#781)

0xgouda · pashagolub · web-flow · commit a33e8487c555 · 2025-06-04T13:27:02.000+02:00
* define sync metric operations types as constants

this adds AddOp DeleteOp constant operation types
and invalidOp as unexported testing-only operation
modifies rpc_test to be at the sinks package not sinks_test

* add `SyncOp` to `api`

---------

Co-authored-by: Pavlo Golub &lt;pavlo.golub@gmail.com&gt;
diff --git a/api/types.go b/api/types.go
@@ -12,5 +12,6 @@ type (
 	// with metadata such as metric name, source type, etc.
 	MeasurementEnvelope = metrics.MeasurementEnvelope
 	// RPCSyncRequest represents a request to sync metrics with the remote RPC sink
+	OpType         = sinks.SyncOp
 	RPCSyncRequest = sinks.SyncReq
 )
diff --git a/internal/reaper/database.go b/internal/reaper/database.go
@@ -11,6 +11,7 @@ import (
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/db"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/log"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
+	"github.com/cybertec-postgresql/pgwatch/v3/internal/sinks"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/sources"
 	"github.com/jackc/pgx/v5"
 )
@@ -543,14 +544,14 @@ func (r *Reaper) CloseResourcesForRemovedMonitoredDBs(shutDownDueToRoleChange ma
 	for _, prevDB := range r.prevLoopMonitoredDBs {
 		if r.monitoredSources.GetMonitoredDatabase(prevDB.Name) == nil { // removed from config
 			prevDB.Conn.Close()
-			_ = r.SinksWriter.SyncMetric(prevDB.Name, "", "remove")
+			_ = r.SinksWriter.SyncMetric(prevDB.Name, "", sinks.DeleteOp)
 		}
 	}
 
 	for roleChangedDB := range shutDownDueToRoleChange {
 		if db := r.monitoredSources.GetMonitoredDatabase(roleChangedDB); db != nil {
 			db.Conn.Close()
 		}
-		_ = r.SinksWriter.SyncMetric(roleChangedDB, "", "remove")
+		_ = r.SinksWriter.SyncMetric(roleChangedDB, "", sinks.DeleteOp)
 	}
 }
diff --git a/internal/reaper/reaper.go b/internal/reaper/reaper.go
@@ -14,6 +14,7 @@ import (
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/cmdopts"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/log"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
+	"github.com/cybertec-postgresql/pgwatch/v3/internal/sinks"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/sources"
 )
 
@@ -170,7 +171,7 @@ func (r *Reaper) Reap(ctx context.Context) {
 							metricNameForStorage = mvp.StorageName
 						}
 
-						if err := r.SinksWriter.SyncMetric(monitoredSource.Name, metricNameForStorage, "add"); err != nil {
+						if err := r.SinksWriter.SyncMetric(monitoredSource.Name, metricNameForStorage, sinks.AddOp); err != nil {
 							srcL.Error(err)
 						}
 
@@ -267,7 +268,7 @@ func (r *Reaper) ShutdownOldWorkers(ctx context.Context, hostsToShutDownDueToRol
 			logger.WithField("source", db).WithField("metric", metric).Info("stopping gatherer...")
 			cancelFunc()
 			delete(r.cancelFuncs, dbMetric)
-			if err := r.SinksWriter.SyncMetric(db, metric, "remove"); err != nil {
+			if err := r.SinksWriter.SyncMetric(db, metric, sinks.DeleteOp); err != nil {
 				logger.Error(err)
 			}
 		}
diff --git a/internal/sinks/json.go b/internal/sinks/json.go
@@ -64,7 +64,7 @@ func (jw *JSONWriter) watchCtx() {
 	jw.lw.Close()
 }
 
-func (jw *JSONWriter) SyncMetric(_, _, _ string) error {
+func (jw *JSONWriter) SyncMetric(_, _ string, _ SyncOp) error {
 	if jw.ctx.Err() != nil {
 		return jw.ctx.Err()
 	}
diff --git a/internal/sinks/json_test.go b/internal/sinks/json_test.go
@@ -60,11 +60,11 @@ func TestJSONWriter_SyncMetric(t *testing.T) {
 	assert.NoError(t, err)
 
 	// Call the function being tested
-	err = jw.SyncMetric("", "", "")
+	err = jw.SyncMetric("", "", invalidOp)
 	assert.NoError(t, err)
 
 	cancel()
-	err = jw.SyncMetric("", "", "")
+	err = jw.SyncMetric("", "", invalidOp)
 	assert.Error(t, err, "context canceled")
 
 }
diff --git a/internal/sinks/multiwriter.go b/internal/sinks/multiwriter.go
@@ -12,7 +12,7 @@ import (
 
 // Writer is an interface that writes metrics values
 type Writer interface {
-	SyncMetric(dbUnique, metricName, op string) error
+	SyncMetric(dbUnique, metricName string, op SyncOp) error
 	Write(msgs metrics.MeasurementEnvelope) error
 }
 
@@ -62,7 +62,7 @@ func (mw *MultiWriter) AddWriter(w Writer) {
 	mw.Unlock()
 }
 
-func (mw *MultiWriter) SyncMetric(dbUnique, metricName, op string) (err error) {
+func (mw *MultiWriter) SyncMetric(dbUnique, metricName string, op SyncOp) (err error) {
 	for _, w := range mw.writers {
 		err = errors.Join(err, w.SyncMetric(dbUnique, metricName, op))
 	}
diff --git a/internal/sinks/multiwriter_test.go b/internal/sinks/multiwriter_test.go
@@ -10,7 +10,7 @@ import (
 
 type MockWriter struct{}
 
-func (mw *MockWriter) SyncMetric(_, _, _ string) error {
+func (mw *MockWriter) SyncMetric(_, _ string, _ SyncOp) error {
 	return nil
 }
 
@@ -74,7 +74,7 @@ func TestSyncMetrics(t *testing.T) {
 	mw := &MultiWriter{}
 	mockWriter := &MockWriter{}
 	mw.AddWriter(mockWriter)
-	err := mw.SyncMetric("db", "metric", "op")
+	err := mw.SyncMetric("db", "metric", invalidOp)
 	assert.NoError(t, err)
 }
 
diff --git a/internal/sinks/postgres.go b/internal/sinks/postgres.go
@@ -150,8 +150,8 @@ var (
 )
 
 // SyncMetric ensures that tables exist for newly added metrics and/or sources
-func (pgw *PostgresWriter) SyncMetric(dbUnique, metricName, op string) error {
-	if op == "add" {
+func (pgw *PostgresWriter) SyncMetric(dbUnique, metricName string, op SyncOp) error {
+	if op == AddOp {
 		return errors.Join(
 			pgw.AddDBUniqueMetricToListingTable(dbUnique, metricName),
 			pgw.EnsureMetricDummy(metricName), // ensure that there is at least an empty top-level table not to get ugly Grafana notifications
diff --git a/internal/sinks/postgres_test.go b/internal/sinks/postgres_test.go
@@ -61,14 +61,14 @@ func TestSyncMetric(t *testing.T) {
 	}
 	dbUnique := "mydb"
 	metricName := "mymetric"
-	op := "add"
+	op := AddOp 
 	conn.ExpectExec("insert into admin\\.all_distinct_dbname_metrics").WithArgs(dbUnique, metricName).WillReturnResult(pgxmock.NewResult("EXECUTE", 1))
 	conn.ExpectExec("select admin\\.ensure_dummy_metrics_table").WithArgs(metricName).WillReturnResult(pgxmock.NewResult("EXECUTE", 1))
 	err = pgw.SyncMetric(dbUnique, metricName, op)
 	assert.NoError(t, err)
 	assert.NoError(t, conn.ExpectationsWereMet())
 
-	op = "foo"
+	op = invalidOp
 	err = pgw.SyncMetric(dbUnique, metricName, op)
 	assert.NoError(t, err, "ignore unknown operation")
 }
diff --git a/internal/sinks/prometheus.go b/internal/sinks/prometheus.go
@@ -117,11 +117,11 @@ func (promw *PrometheusWriter) PurgeMetricsFromPromAsyncCacheIfAny(dbUnique, met
 	}
 }
 
-func (promw *PrometheusWriter) SyncMetric(dbUnique, metricName, op string) error {
+func (promw *PrometheusWriter) SyncMetric(dbUnique, metricName string, op SyncOp) error {
 	switch op {
-	case "remove":
+	case DeleteOp:
 		promw.PurgeMetricsFromPromAsyncCacheIfAny(dbUnique, metricName)
-	case "add":
+	case AddOp:
 		promw.PromAsyncCacheInitIfRequired(dbUnique, metricName)
 	}
 	return nil
diff --git a/internal/sinks/rpc.go b/internal/sinks/rpc.go
@@ -8,16 +8,6 @@ import (
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
 )
 
-// RPCWriter is a sink that sends metric measurements to a remote server using the RPC protocol.
-// Remote server should implement the Receiver interface. It's up to the implementer to define the
-// behavior of the server. It can be a simple logger, external storage, alerting system,
-// or an analytics system.
-type RPCWriter struct {
-	ctx     context.Context
-	address string
-	client  *rpc.Client
-}
-
 func NewRPCWriter(ctx context.Context, address string) (*RPCWriter, error) {
 	client, err := rpc.DialHTTP("tcp", address)
 	if err != nil {
@@ -49,13 +39,7 @@ func (rw *RPCWriter) Write(msg metrics.MeasurementEnvelope) error {
 	return nil
 }
 
-type SyncReq struct {
-	DbName     string
-	MetricName string
-	Operation  string
-}
-
-func (rw *RPCWriter) SyncMetric(dbUnique string, metricName string, op string) error {
+func (rw *RPCWriter) SyncMetric(dbUnique, metricName string, op SyncOp) error {
 	var logMsg string
 	if err := rw.client.Call("Receiver.SyncMetric", &SyncReq{
 		Operation:  op,
diff --git a/internal/sinks/rpc_test.go b/internal/sinks/rpc_test.go
@@ -1,4 +1,4 @@
-package sinks_test
+package sinks
 
 import (
 	"context"
@@ -10,7 +10,6 @@ import (
 	"testing"
 
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
-	"github.com/cybertec-postgresql/pgwatch/v3/internal/sinks"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -30,11 +29,11 @@ func (receiver *Receiver) UpdateMeasurements(msg *metrics.MeasurementEnvelope, l
 	return nil
 }
 
-func (receiver *Receiver) SyncMetric(syncReq *sinks.SyncReq, logMsg *string) error {
+func (receiver *Receiver) SyncMetric(syncReq *SyncReq, logMsg *string) error {
 	if syncReq == nil {
 		return errors.New("msgs is nil")
 	}
-	if syncReq.Operation == "invalid" {
+	if syncReq.Operation == invalidOp {
 		return errors.New("invalid message")
 	}
 	*logMsg = fmt.Sprintf("Received: %+v", *syncReq)
@@ -59,13 +58,13 @@ func init() {
 // Test begin from here ---------------------------------------------------------
 func TestNewRPCWriter(t *testing.T) {
 	a := assert.New(t)
-	_, err := sinks.NewRPCWriter(ctxt, "foo")
+	_, err := NewRPCWriter(ctxt, "foo")
 	a.Error(err)
 }
 
 func TestRPCWrite(t *testing.T) {
 	a := assert.New(t)
-	rw, err := sinks.NewRPCWriter(ctxt, "0.0.0.0:5050")
+	rw, err := NewRPCWriter(ctxt, "0.0.0.0:5050")
 	a.NoError(err)
 
 	// no error for valid messages
@@ -89,7 +88,7 @@ func TestRPCWrite(t *testing.T) {
 
 	// error for cancelled context
 	ctx, cancel := context.WithCancel(ctxt)
-	rw, err = sinks.NewRPCWriter(ctx, "0.0.0.0:5050")
+	rw, err = NewRPCWriter(ctx, "0.0.0.0:5050")
 	a.NoError(err)
 	cancel()
 	err = rw.Write(msgs)
@@ -99,24 +98,24 @@ func TestRPCWrite(t *testing.T) {
 func TestRPCSyncMetric(t *testing.T) {
 	port := 5050
 	a := assert.New(t)
-	rw, err := sinks.NewRPCWriter(ctxt, "0.0.0.0:"+fmt.Sprint(port))
+	rw, err := NewRPCWriter(ctxt, "0.0.0.0:"+fmt.Sprint(port))
 	if err != nil {
 		t.Error("Unable to send sync metric signal")
 	}
 
 	// no error for valid messages
-	err = rw.SyncMetric("Test-DB", "DB-Metric", "Add")
+	err = rw.SyncMetric("Test-DB", "DB-Metric", AddOp)
 	a.NoError(err)
 
 	// error for invalid messages
-	err = rw.SyncMetric("", "", "invalid")
+	err = rw.SyncMetric("", "", invalidOp)
 	a.Error(err)
 
 	// error for cancelled context
 	ctx, cancel := context.WithCancel(ctxt)
-	rw, err = sinks.NewRPCWriter(ctx, "0.0.0.0:5050")
+	rw, err = NewRPCWriter(ctx, "0.0.0.0:5050")
 	a.NoError(err)
 	cancel()
-	err = rw.SyncMetric("Test-DB", "DB-Metric", "Add")
+	err = rw.SyncMetric("Test-DB", "DB-Metric", AddOp)
 	a.Error(err)
 }
diff --git a/internal/sinks/types.go b/internal/sinks/types.go
@@ -0,0 +1,30 @@
+package sinks
+
+import (
+	"context"
+	"net/rpc"
+)
+
+// RPCWriter is a sink that sends metric measurements to a remote server using the RPC protocol.
+// Remote server should implement the Receiver interface. It's up to the implementer to define the
+// behavior of the server. It can be a simple logger, external storage, alerting system,
+// or an analytics system.
+type RPCWriter struct {
+	ctx     context.Context
+	address string
+	client  *rpc.Client
+}
+
+type SyncOp int
+
+const (
+	AddOp SyncOp = iota
+	DeleteOp
+	invalidOp
+)
+
+type SyncReq struct {
+	DbName     string
+	MetricName string
+	Operation  SyncOp
+}

Original file line number	Diff line number	Diff line change
`@@ -12,5 +12,6 @@ type (`
`12`	`12`	`// with metadata such as metric name, source type, etc.`
`13`	`13`	`MeasurementEnvelope = metrics.MeasurementEnvelope`
`14`	`14`	`// RPCSyncRequest represents a request to sync metrics with the remote RPC sink`
	`15`	`+ OpType = sinks.SyncOp`
`15`	`16`	`RPCSyncRequest = sinks.SyncReq`
`16`	`17`	`)`
Original file line number	Diff line number	Diff line change
`@@ -11,6 +11,7 @@ import (`
`11`	`11`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/db"`
`12`	`12`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/log"`
`13`	`13`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"`
	`14`	`+ "github.com/cybertec-postgresql/pgwatch/v3/internal/sinks"`
`14`	`15`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/sources"`
`15`	`16`	`"github.com/jackc/pgx/v5"`
`16`	`17`	`)`
`@@ -543,14 +544,14 @@ func (r *Reaper) CloseResourcesForRemovedMonitoredDBs(shutDownDueToRoleChange ma`
`543`	`544`	`for _, prevDB := range r.prevLoopMonitoredDBs {`
`544`	`545`	`if r.monitoredSources.GetMonitoredDatabase(prevDB.Name) == nil { // removed from config`
`545`	`546`	`prevDB.Conn.Close()`
`546`		`- _ = r.SinksWriter.SyncMetric(prevDB.Name, "", "remove")`
	`547`	`+ _ = r.SinksWriter.SyncMetric(prevDB.Name, "", sinks.DeleteOp)`
`547`	`548`	`}`
`548`	`549`	`}`
`549`	`550`
`550`	`551`	`for roleChangedDB := range shutDownDueToRoleChange {`
`551`	`552`	`if db := r.monitoredSources.GetMonitoredDatabase(roleChangedDB); db != nil {`
`552`	`553`	`db.Conn.Close()`
`553`	`554`	`}`
`554`		`- _ = r.SinksWriter.SyncMetric(roleChangedDB, "", "remove")`
	`555`	`+ _ = r.SinksWriter.SyncMetric(roleChangedDB, "", sinks.DeleteOp)`
`555`	`556`	`}`
`556`	`557`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,7 @@ import (`
`14`	`14`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/cmdopts"`
`15`	`15`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/log"`
`16`	`16`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"`
	`17`	`+ "github.com/cybertec-postgresql/pgwatch/v3/internal/sinks"`
`17`	`18`	`"github.com/cybertec-postgresql/pgwatch/v3/internal/sources"`
`18`	`19`	`)`
`19`	`20`
`@@ -170,7 +171,7 @@ func (r *Reaper) Reap(ctx context.Context) {`
`170`	`171`	`metricNameForStorage = mvp.StorageName`
`171`	`172`	`}`
`172`	`173`
`173`		`- if err := r.SinksWriter.SyncMetric(monitoredSource.Name, metricNameForStorage, "add"); err != nil {`
	`174`	`+ if err := r.SinksWriter.SyncMetric(monitoredSource.Name, metricNameForStorage, sinks.AddOp); err != nil {`
`174`	`175`	`srcL.Error(err)`
`175`	`176`	`}`
`176`	`177`
`@@ -267,7 +268,7 @@ func (r *Reaper) ShutdownOldWorkers(ctx context.Context, hostsToShutDownDueToRol`
`267`	`268`	`logger.WithField("source", db).WithField("metric", metric).Info("stopping gatherer...")`
`268`	`269`	`cancelFunc()`
`269`	`270`	`delete(r.cancelFuncs, dbMetric)`
`270`		`- if err := r.SinksWriter.SyncMetric(db, metric, "remove"); err != nil {`
	`271`	`+ if err := r.SinksWriter.SyncMetric(db, metric, sinks.DeleteOp); err != nil {`
`271`	`272`	`logger.Error(err)`
`272`	`273`	`}`
`273`	`274`	`}`
Original file line number	Diff line number	Diff line change
`@@ -64,7 +64,7 @@ func (jw *JSONWriter) watchCtx() {`
`64`	`64`	`jw.lw.Close()`
`65`	`65`	`}`
`66`	`66`
`67`		`-func (jw *JSONWriter) SyncMetric(_, _, _ string) error {`
	`67`	`+func (jw *JSONWriter) SyncMetric(_, _ string, _ SyncOp) error {`
`68`	`68`	`if jw.ctx.Err() != nil {`
`69`	`69`	`return jw.ctx.Err()`
`70`	`70`	`}`
Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,7 @@ import (`
`12`	`12`
`13`	`13`	`// Writer is an interface that writes metrics values`
`14`	`14`	`type Writer interface {`
`15`		`- SyncMetric(dbUnique, metricName, op string) error`
	`15`	`+ SyncMetric(dbUnique, metricName string, op SyncOp) error`
`16`	`16`	`Write(msgs metrics.MeasurementEnvelope) error`
`17`	`17`	`}`
`18`	`18`
`@@ -62,7 +62,7 @@ func (mw *MultiWriter) AddWriter(w Writer) {`
`62`	`62`	`mw.Unlock()`
`63`	`63`	`}`
`64`	`64`
`65`		`-func (mw *MultiWriter) SyncMetric(dbUnique, metricName, op string) (err error) {`
	`65`	`+func (mw *MultiWriter) SyncMetric(dbUnique, metricName string, op SyncOp) (err error) {`
`66`	`66`	`for _, w := range mw.writers {`
`67`	`67`	`err = errors.Join(err, w.SyncMetric(dbUnique, metricName, op))`
`68`	`68`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@ import (`
`10`	`10`
`11`	`11`	`type MockWriter struct{}`
`12`	`12`
`13`		`-func (mw *MockWriter) SyncMetric(_, _, _ string) error {`
	`13`	`+func (mw *MockWriter) SyncMetric(_, _ string, _ SyncOp) error {`
`14`	`14`	`return nil`
`15`	`15`	`}`
`16`	`16`
`@@ -74,7 +74,7 @@ func TestSyncMetrics(t *testing.T) {`
`74`	`74`	`mw := &MultiWriter{}`
`75`	`75`	`mockWriter := &MockWriter{}`
`76`	`76`	`mw.AddWriter(mockWriter)`
`77`		`- err := mw.SyncMetric("db", "metric", "op")`
	`77`	`+ err := mw.SyncMetric("db", "metric", invalidOp)`
`78`	`78`	`assert.NoError(t, err)`
`79`	`79`	`}`
`80`	`80`
Original file line number	Diff line number	Diff line change
`@@ -117,11 +117,11 @@ func (promw *PrometheusWriter) PurgeMetricsFromPromAsyncCacheIfAny(dbUnique, met`
`117`	`117`	`}`
`118`	`118`	`}`
`119`	`119`
`120`		`-func (promw *PrometheusWriter) SyncMetric(dbUnique, metricName, op string) error {`
	`120`	`+func (promw *PrometheusWriter) SyncMetric(dbUnique, metricName string, op SyncOp) error {`
`121`	`121`	`switch op {`
`122`		`- case "remove":`
	`122`	`+ case DeleteOp:`
`123`	`123`	`promw.PurgeMetricsFromPromAsyncCacheIfAny(dbUnique, metricName)`
`124`		`- case "add":`
	`124`	`+ case AddOp:`
`125`	`125`	`promw.PromAsyncCacheInitIfRequired(dbUnique, metricName)`
`126`	`126`	`}`
`127`	`127`	`return nil`