[-] remove passwords from sink log messages, fixes #612 (#614)

Author: pashagolub

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/jackc/pgx/v5 v5.7.2
 	github.com/jessevdk/go-flags v1.6.1
-	github.com/pashagolub/pgxmock/v4 v4.3.0
+	github.com/pashagolub/pgxmock/v4 v4.4.0
 	github.com/prometheus/client_golang v1.20.5
 	github.com/rifflock/lfshook v0.0.0-20180920164130-b9218ef580f5
 	github.com/sethvargo/go-retry v0.3.0
@@ -87,10 +87,10 @@ require (
 	go.opentelemetry.io/otel/sdk v1.21.0 // indirect
 	go.opentelemetry.io/otel/trace v1.31.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/crypto v0.32.0 // indirect
 	golang.org/x/net v0.29.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect
-	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/sys v0.29.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1 // indirect

go.sum

@@ -116,6 +116,8 @@ github.com/pashagolub/pgxmock/v3 v3.4.0 h1:87VMr2q7m2+6VzXo4Tsp9kMklGlj6mMN19Hp/
 github.com/pashagolub/pgxmock/v3 v3.4.0/go.mod h1:FvCl7xqPbLLI3XohihJ1NzXnikjM3q/NWSixg4t9hrU=
 github.com/pashagolub/pgxmock/v4 v4.3.0 h1:DqT7fk0OCK6H0GvqtcMsLpv8cIwWqdxWgfZNLeHCb/s=
 github.com/pashagolub/pgxmock/v4 v4.3.0/go.mod h1:9VoVHXwS3XR/yPtKGzwQvwZX1kzGB9sM8SviDcHDa3A=
+github.com/pashagolub/pgxmock/v4 v4.4.0 h1:zrZHBzqlzIFrq5Iw6nQpmpEd77eLqGIC2ol4ZTeojz0=
+github.com/pashagolub/pgxmock/v4 v4.4.0/go.mod h1:9VoVHXwS3XR/yPtKGzwQvwZX1kzGB9sM8SviDcHDa3A=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -201,6 +203,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
+golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
 golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa h1:ELnwvuAXPNtPk1TJRuGkI9fDTwym6AYBu0qzT8AcHdI=
 golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa/go.mod h1:akd2r19cwCdwSwWeIdzYQGa/EZZyqcOdwWiwj5L5eKQ=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -226,8 +230,11 @@ golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
 golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
+golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.27.0 h1:WP60Sv1nlK1T6SupCHbXzSaN0b9wUmsPoRS9b61A23Q=
 golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
+golang.org/x/term v0.28.0 h1:/Ts8HFuMR2E6IP/jlo7QVLZHggjKQbhu/7H0LJFr3Gg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=

internal/log/formatter.go

@@ -13,8 +13,11 @@ import (
 
 func newFormatter(noColors bool) *Formatter {
 	return &Formatter{
-		HideKeys:        false,
-		FieldsOrder:     []string{"source", "sink", "metric", "rows", "sql", "params"},
+		HideKeys: false,
+		FieldsOrder: []string{
+			"source", "metric",
+			"sink", "db", "filename", "address",
+			"rows", "sql", "params"},
 		TimestampFormat: "2006-01-02 15:04:05.000",
 		ShowFullLevel:   true,
 		NoColors:        noColors,

internal/sinks/json.go

@@ -3,7 +3,9 @@ package sinks
 import (
 	"context"
 	"encoding/json"
+	"time"
 
+	"github.com/cybertec-postgresql/pgwatch/v3/internal/log"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
 	"gopkg.in/natefinch/lumberjack.v2"
 )
@@ -18,6 +20,8 @@ type JSONWriter struct {
 }
 
 func NewJSONWriter(ctx context.Context, fname string) (*JSONWriter, error) {
+	l := log.GetLogger(ctx).WithField("sink", "jsonfile").WithField("filename", fname)
+	ctx = log.WithLogger(ctx, l)
 	jw := &JSONWriter{
 		ctx: ctx,
 		lw:  &lumberjack.Logger{Filename: fname, Compress: true},
@@ -34,6 +38,7 @@ func (jw *JSONWriter) Write(msgs []metrics.MeasurementEnvelope) error {
 		return nil
 	}
 	enc := json.NewEncoder(jw.lw)
+	t1 := time.Now()
 	for _, msg := range msgs {
 		dataRow := map[string]any{
 			"metric":      msg.MetricName,
@@ -45,6 +50,8 @@ func (jw *JSONWriter) Write(msgs []metrics.MeasurementEnvelope) error {
 			return err
 		}
 	}
+	diff := time.Since(t1)
+	log.GetLogger(jw.ctx).WithField("rows", len(msgs)).WithField("elapsed", diff).Info("measurements written")
 	return nil
 }
 

internal/sinks/multiwriter.go

@@ -26,11 +26,8 @@ type MultiWriter struct {
 // NewMultiWriter creates and returns new instance of MultiWriter struct.
 func NewMultiWriter(ctx context.Context, opts *CmdOpts, metricDefs *metrics.Metrics) (mw *MultiWriter, err error) {
 	var w Writer
-	logger := log.GetLogger(ctx)
 	mw = &MultiWriter{}
 	for _, s := range opts.Sinks {
-		l := logger.WithField("sink", s)
-		ctx = log.WithLogger(ctx, l)
 		scheme, path, found := strings.Cut(s, "://")
 		if !found || scheme == "" || path == "" {
 			return nil, fmt.Errorf("malformed sink URI %s", s)
@@ -51,9 +48,7 @@ func NewMultiWriter(ctx context.Context, opts *CmdOpts, metricDefs *metrics.Metr
 			return nil, err
 		}
 		mw.AddWriter(w)
-		l.Info(`Measurements sink activated`)
 	}
-
 	if len(mw.writers) == 0 {
 		return nil, errors.New("no sinks specified for measurements")
 	}

internal/sinks/postgres.go

@@ -31,6 +31,8 @@ func NewPostgresWriter(ctx context.Context, connstr string, opts *CmdOpts, metri
 }
 
 func NewWriterFromPostgresConn(ctx context.Context, conn db.PgxPoolIface, opts *CmdOpts, metricDefs *metrics.Metrics) (pgw *PostgresWriter, err error) {
+	l := log.GetLogger(ctx).WithField("sink", "postgres").WithField("db", conn.Config().ConnConfig.Database)
+	ctx = log.WithLogger(ctx, l)
 	pgw = &PostgresWriter{
 		ctx:        ctx,
 		metricDefs: metricDefs,
@@ -40,7 +42,7 @@ func NewWriterFromPostgresConn(ctx context.Context, conn db.PgxPoolIface, opts *
 		sinkDb:     conn,
 	}
 	if err = db.Init(ctx, pgw.sinkDb, func(ctx context.Context, conn db.PgxIface) error {
-		log.GetLogger(ctx).Info("Initialising measurements database...")
+		l.Info("initialising measurements database...")
 		exists, err := db.DoesSchemaExist(ctx, conn, "admin")
 		if err != nil || exists {
 			return err
@@ -63,6 +65,7 @@ func NewWriterFromPostgresConn(ctx context.Context, conn db.PgxPoolIface, opts *
 	go pgw.deleteOldPartitions(deleterDelay)
 	go pgw.maintainUniqueSources()
 	go pgw.poll()
+	l.Info(`measurements sink is activated`)
 	return
 }
 
@@ -178,7 +181,7 @@ func (pgw *PostgresWriter) EnsureMetricDummy(metric string) (err error) {
 	return
 }
 
-// Write send the measurements to the cache channel
+// Write sends the measurements to the cache channel
 func (pgw *PostgresWriter) Write(msgs []metrics.MeasurementEnvelope) error {
 	if pgw.ctx.Err() != nil {
 		return pgw.ctx.Err()
@@ -232,9 +235,7 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 	if len(msgs) == 0 {
 		return
 	}
-	logger := log.GetLogger(pgw.ctx).
-		WithField("sink", "postgres").
-		WithField("db", pgw.sinkDb.Config().ConnConfig.Database)
+	logger := log.GetLogger(pgw.ctx)
 	tsWarningPrinted := false
 	metricsToStorePerMetric := make(map[string][]MeasurementMessagePostgres)
 	rowsBatched := 0
@@ -247,7 +248,7 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 		if len(msg.Data) == 0 {
 			continue
 		}
-		logger.WithField("data", msg.Data).WithField("len", len(msg.Data)).Debug("Sending To Postgres")
+		logger.WithField("data", msg.Data).WithField("len", len(msg.Data)).Debug("sending to postgres")
 
 		for _, dataRow := range msg.Data {
 			var epochTime time.Time
@@ -337,7 +338,7 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 	} else if pgw.metricSchema == DbStorageSchemaTimescale {
 		err = pgw.EnsureMetricTimescale(pgPartBounds, forceRecreatePGMetricPartitions)
 	} else {
-		logger.Fatal("should never happen...")
+		logger.Fatal("unknown storage schema...")
 	}
 	if forceRecreatePGMetricPartitions {
 		forceRecreatePGMetricPartitions = false

internal/sinks/prometheus.go

@@ -36,6 +36,8 @@ func NewPrometheusWriter(ctx context.Context, connstr string) (promw *Prometheus
 	if !found {
 		namespace = "pgwatch"
 	}
+	l := log.GetLogger(ctx).WithField("sink", "prometheus").WithField("address", addr)
+	ctx = log.WithLogger(ctx, l)
 	promw = &PrometheusWriter{
 		ctx:                 ctx,
 		PrometheusNamespace: namespace,
@@ -179,7 +181,7 @@ func (promw *PrometheusWriter) setInstanceUpDownState(ch chan<- prometheus.Metri
 	if len(pm) > 0 {
 		ch <- pm[0]
 	} else {
-		logger.Errorf("Could not formulate an instance state report - should not happen")
+		logger.Error("Could not formulate an instance state report - should not happen")
 	}
 }
 

internal/sinks/rpc.go

@@ -23,7 +23,8 @@ func NewRPCWriter(ctx context.Context, address string) (*RPCWriter, error) {
 	if err != nil {
 		return nil, err
 	}
-
+	l := log.GetLogger(ctx).WithField("sink", "rpc").WithField("address", address)
+	ctx = log.WithLogger(ctx, l)
 	rw := &RPCWriter{
 		ctx:     ctx,
 		address: address,
@@ -41,16 +42,13 @@ func (rw *RPCWriter) Write(msgs []metrics.MeasurementEnvelope) error {
 	if len(msgs) == 0 {
 		return nil
 	}
-	l := log.GetLogger(rw.ctx).
-		WithField("sink", "rpc").
-		WithField("address", rw.address)
 	for _, msg := range msgs {
 		var logMsg string
 		if err := rw.client.Call("Receiver.UpdateMeasurements", &msg, &logMsg); err != nil {
 			return err
 		}
 		if len(logMsg) > 0 {
-			l.Info(logMsg)
+			log.GetLogger(rw.ctx).Info(logMsg)
 		}
 	}
 	return nil
@@ -72,9 +70,7 @@ func (rw *RPCWriter) SyncMetric(dbUnique string, metricName string, op string) e
 		return err
 	}
 	if len(logMsg) > 0 {
-		log.GetLogger(rw.ctx).
-			WithField("sink", "rpc").
-			WithField("address", rw.address).Info(logMsg)
+		log.GetLogger(rw.ctx).Info(logMsg)
 	}
 	return nil
 }