[+] decrease memory allocations for metrics.MeasurementEnvelope (#768)

Allocating additional slices increases a memory usage. Instead we can
send metrics.MeasurementEnvelope one by one immediately after receiving.
There is no need to collect them beforehand.
For 12 postgres instances monitored with full preset this change saves
us ~4-9% of allocations.

Author: pashagolub

internal/metrics/logparse.go

@@ -84,7 +84,7 @@ func getFileWithNextModTimestamp(logsGlobPath, currentFile string) (string, erro
 }
 
 // 1. add zero counts for severity levels that didn't have any occurrences in the log
-func eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal map[string]int64, mdb *sources.SourceConn) []MeasurementEnvelope {
+func eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal map[string]int64, mdb *sources.SourceConn) MeasurementEnvelope {
 	allSeverityCounts := NewMeasurement(time.Now().UnixNano())
 	for _, s := range PgSeverities {
 		parsedCount, ok := eventCounts[s]
@@ -100,16 +100,16 @@ func eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal map[string]i
 			allSeverityCounts[strings.ToLower(s)+"_total"] = 0
 		}
 	}
-	return []MeasurementEnvelope{{
+	return MeasurementEnvelope{
 		DBName:     mdb.Name,
 		SourceType: string(mdb.Kind),
 		MetricName: specialMetricServerLogEventCounts,
 		Data:       Measurements{allSeverityCounts},
 		CustomTags: mdb.CustomTags,
-	}}
+	}
 }
 
-func ParseLogs(ctx context.Context, mdb *sources.SourceConn, realDbname string, interval float64, storeCh chan<- []MeasurementEnvelope) {
+func ParseLogs(ctx context.Context, mdb *sources.SourceConn, realDbname string, interval float64, storeCh chan<- MeasurementEnvelope) {
 
 	var latest, previous, serverMessagesLang string
 	var latestHandle *os.File
@@ -305,8 +305,7 @@ func ParseLogs(ctx context.Context, mdb *sources.SourceConn, realDbname string,
 		send_to_storage_if_needed:
 			if lastSendTime.IsZero() || lastSendTime.Before(time.Now().Add(-1*time.Second*time.Duration(interval))) {
 				logger.Debugf("Sending log event counts for last interval to storage channel. Local eventcounts: %+v, global eventcounts: %+v", eventCounts, eventCountsTotal)
-				metricStoreMessages := eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal, mdb)
-				storeCh <- metricStoreMessages
+				storeCh <- eventCountsToMetricStoreMessages(eventCounts, eventCountsTotal, mdb)
 				zeroEventCounts(eventCounts)
 				zeroEventCounts(eventCountsTotal)
 				lastSendTime = time.Now()

internal/reaper/database.go

@@ -50,7 +50,7 @@ func QueryMeasurements(ctx context.Context, dbUnique string, sql string, args ..
 	return nil, err
 }
 
-func DetectSprocChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
+func DetectSprocChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
 	detectedChanges := make(metrics.Measurements, 0)
 	var firstRun bool
 	var changeCounts ChangeDetectionResults
@@ -122,13 +122,18 @@ func DetectSprocChanges(ctx context.Context, md *sources.SourceConn, storageCh c
 	}
 	log.GetLogger(ctx).Debugf("[%s][%s] detected %d sproc changes", md.Name, specialMetricChangeEvents, len(detectedChanges))
 	if len(detectedChanges) > 0 {
-		storageCh <- []metrics.MeasurementEnvelope{{DBName: md.Name, MetricName: "sproc_changes", Data: detectedChanges, CustomTags: md.CustomTags}}
+		storageCh <- metrics.MeasurementEnvelope{
+			DBName:     md.Name,
+			MetricName: "sproc_changes",
+			Data:       detectedChanges,
+			CustomTags: md.CustomTags,
+		}
 	}
 
 	return changeCounts
 }
 
-func DetectTableChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
+func DetectTableChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
 	detectedChanges := make(metrics.Measurements, 0)
 	var firstRun bool
 	var changeCounts ChangeDetectionResults
@@ -200,13 +205,18 @@ func DetectTableChanges(ctx context.Context, md *sources.SourceConn, storageCh c
 
 	log.GetLogger(ctx).Debugf("[%s][%s] detected %d table changes", md.Name, specialMetricChangeEvents, len(detectedChanges))
 	if len(detectedChanges) > 0 {
-		storageCh <- []metrics.MeasurementEnvelope{{DBName: md.Name, MetricName: "table_changes", Data: detectedChanges, CustomTags: md.CustomTags}}
+		storageCh <- metrics.MeasurementEnvelope{
+			DBName:     md.Name,
+			MetricName: "table_changes",
+			Data:       detectedChanges,
+			CustomTags: md.CustomTags,
+		}
 	}
 
 	return changeCounts
 }
 
-func DetectIndexChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
+func DetectIndexChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
 	detectedChanges := make(metrics.Measurements, 0)
 	var firstRun bool
 	var changeCounts ChangeDetectionResults
@@ -276,13 +286,18 @@ func DetectIndexChanges(ctx context.Context, md *sources.SourceConn, storageCh c
 	}
 	log.GetLogger(ctx).Debugf("[%s][%s] detected %d index changes", md.Name, specialMetricChangeEvents, len(detectedChanges))
 	if len(detectedChanges) > 0 {
-		storageCh <- []metrics.MeasurementEnvelope{{DBName: md.Name, MetricName: "index_changes", Data: detectedChanges, CustomTags: md.CustomTags}}
+		storageCh <- metrics.MeasurementEnvelope{
+			DBName:     md.Name,
+			MetricName: "index_changes",
+			Data:       detectedChanges,
+			CustomTags: md.CustomTags,
+		}
 	}
 
 	return changeCounts
 }
 
-func DetectPrivilegeChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
+func DetectPrivilegeChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
 	detectedChanges := make(metrics.Measurements, 0)
 	var firstRun bool
 	var changeCounts ChangeDetectionResults
@@ -346,19 +361,18 @@ func DetectPrivilegeChanges(ctx context.Context, md *sources.SourceConn, storage
 
 	log.GetLogger(ctx).Debugf("[%s][%s] detected %d object privilege changes...", md.Name, specialMetricChangeEvents, len(detectedChanges))
 	if len(detectedChanges) > 0 {
-		storageCh <- []metrics.MeasurementEnvelope{
-			{
-				DBName:     md.Name,
-				MetricName: "privilege_changes",
-				Data:       detectedChanges,
-				CustomTags: md.CustomTags,
-			}}
+		storageCh <- metrics.MeasurementEnvelope{
+			DBName:     md.Name,
+			MetricName: "privilege_changes",
+			Data:       detectedChanges,
+			CustomTags: md.CustomTags,
+		}
 	}
 
 	return changeCounts
 }
 
-func DetectConfigurationChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- []metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
+func DetectConfigurationChanges(ctx context.Context, md *sources.SourceConn, storageCh chan<- metrics.MeasurementEnvelope, hostState map[string]map[string]string) ChangeDetectionResults {
 	detectedChanges := make(metrics.Measurements, 0)
 	var firstRun bool
 	var changeCounts ChangeDetectionResults
@@ -410,12 +424,12 @@ func DetectConfigurationChanges(ctx context.Context, md *sources.SourceConn, sto
 
 	log.GetLogger(ctx).Debugf("[%s][%s] detected %d configuration changes", md.Name, specialMetricChangeEvents, len(detectedChanges))
 	if len(detectedChanges) > 0 {
-		storageCh <- []metrics.MeasurementEnvelope{{
+		storageCh <- metrics.MeasurementEnvelope{
 			DBName:     md.Name,
 			MetricName: "configuration_changes",
 			Data:       detectedChanges,
 			CustomTags: md.CustomTags,
-		}}
+		}
 	}
 
 	return changeCounts
@@ -455,12 +469,13 @@ func (r *Reaper) CheckForPGObjectChangesAndStore(ctx context.Context, dbUnique s
 		influxEntry["details"] = message
 		detectedChangesSummary = append(detectedChangesSummary, influxEntry)
 		md, _ := GetMonitoredDatabaseByUniqueName(ctx, dbUnique)
-		storageCh <- []metrics.MeasurementEnvelope{{DBName: dbUnique,
+		storageCh <- metrics.MeasurementEnvelope{
+			DBName:     dbUnique,
 			SourceType: string(md.Kind),
 			MetricName: "object_changes",
 			Data:       detectedChangesSummary,
 			CustomTags: md.CustomTags,
-		}}
+		}
 
 	}
 }

internal/reaper/reaper.go

@@ -24,7 +24,7 @@ var metricDefs = NewConcurrentMetricDefs()
 type Reaper struct {
 	*cmdopts.Options
 	ready                atomic.Bool
-	measurementCh        chan []metrics.MeasurementEnvelope
+	measurementCh        chan metrics.MeasurementEnvelope
 	measurementCache     *InstanceMetricCache
 	logger               log.Logger
 	monitoredSources     sources.SourceConns
@@ -36,7 +36,7 @@ type Reaper struct {
 func NewReaper(ctx context.Context, opts *cmdopts.Options) (r *Reaper) {
 	return &Reaper{
 		Options:              opts,
-		measurementCh:        make(chan []metrics.MeasurementEnvelope, 10000),
+		measurementCh:        make(chan metrics.MeasurementEnvelope, 10000),
 		measurementCache:     NewInstanceMetricCache(),
 		logger:               log.GetLogger(ctx),
 		monitoredSources:     make(sources.SourceConns, 0),
@@ -270,7 +270,6 @@ func (r *Reaper) reapMetricMeasurements(ctx context.Context, md *sources.SourceC
 	var lastErrorNotificationTime time.Time
 	var err error
 	var ok bool
-	var envelopes []metrics.MeasurementEnvelope
 
 	failedFetches := 0
 	lastDBVersionFetchTime := time.Unix(0, 0) // check DB ver. ev. 5 min
@@ -321,7 +320,7 @@ func (r *Reaper) reapMetricMeasurements(ctx context.Context, md *sources.SourceC
 				lastErrorNotificationTime = time.Now()
 			}
 		} else if metricStoreMessages != nil && len(metricStoreMessages.Data) > 0 {
-			envelopes = []metrics.MeasurementEnvelope{*metricStoreMessages}
+			r.measurementCh <- *metricStoreMessages
 			// pick up "server restarted" events here to avoid doing extra selects from CheckForPGObjectChangesAndStore code
 			if metricName == "db_stats" {
 				postmasterUptimeS, ok := (metricStoreMessages.Data)[0]["postmaster_uptime_s"]
@@ -334,20 +333,18 @@ func (r *Reaper) reapMetricMeasurements(ctx context.Context, md *sources.SourceC
 							entry := metrics.NewMeasurement(metricStoreMessages.Data.GetEpoch())
 							entry["details"] = message
 							detectedChangesSummary = append(detectedChangesSummary, entry)
-							envelopes = append(envelopes,
-								metrics.MeasurementEnvelope{
-									DBName:     md.Name,
-									SourceType: string(md.Kind),
-									MetricName: "object_changes",
-									Data:       detectedChangesSummary,
-									CustomTags: metricStoreMessages.CustomTags,
-								})
+							r.measurementCh <- metrics.MeasurementEnvelope{
+								DBName:     md.Name,
+								SourceType: string(md.Kind),
+								MetricName: "object_changes",
+								Data:       detectedChangesSummary,
+								CustomTags: metricStoreMessages.CustomTags,
+							}
 						}
 					}
 					lastUptimeS = postmasterUptimeS.(int64)
 				}
 			}
-			r.measurementCh <- envelopes
 		}
 
 		select {
@@ -378,9 +375,7 @@ func (r *Reaper) LoadSources() (err error) {
 func (r *Reaper) WriteMonitoredSources(ctx context.Context) {
 	for {
 		if len(monitoredDbCache) > 0 {
-			msms := make([]metrics.MeasurementEnvelope, len(monitoredDbCache))
 			now := time.Now().UnixNano()
-
 			monitoredDbCacheLock.RLock()
 			for _, mdb := range monitoredDbCache {
 				db := metrics.NewMeasurement(now)
@@ -389,14 +384,13 @@ func (r *Reaper) WriteMonitoredSources(ctx context.Context) {
 				for k, v := range mdb.CustomTags {
 					db[metrics.TagPrefix+k] = v
 				}
-				msms = append(msms, metrics.MeasurementEnvelope{
+				r.measurementCh <- metrics.MeasurementEnvelope{
 					DBName:     mdb.Name,
 					MetricName: monitoredDbsDatastoreSyncMetricName,
 					Data:       metrics.Measurements{db},
-				})
+				}
 			}
 			monitoredDbCacheLock.RUnlock()
-			r.measurementCh <- msms
 		}
 		select {
 		case <-time.After(time.Second * monitoredDbsDatastoreSyncIntervalSeconds):

internal/sinks/json.go

@@ -30,28 +30,28 @@ func NewJSONWriter(ctx context.Context, fname string) (*JSONWriter, error) {
 	return jw, nil
 }
 
-func (jw *JSONWriter) Write(msgs []metrics.MeasurementEnvelope) error {
+func (jw *JSONWriter) Write(msg metrics.MeasurementEnvelope) error {
 	if jw.ctx.Err() != nil {
 		return jw.ctx.Err()
 	}
-	if len(msgs) == 0 {
+	if len(msg.Data) == 0 {
 		return nil
 	}
 	enc := json.NewEncoder(jw.lw)
 	t1 := time.Now()
 	written := 0
-	for _, msg := range msgs {
-		dataRow := map[string]any{
-			"metric":      msg.MetricName,
-			"data":        msg.Data,
-			"dbname":      msg.DBName,
-			"custom_tags": msg.CustomTags,
-		}
-		if err := enc.Encode(dataRow); err != nil {
-			return err
-		}
-		written += len(msg.Data)
+
+	dataRow := map[string]any{
+		"metric":      msg.MetricName,
+		"data":        msg.Data,
+		"dbname":      msg.DBName,
+		"custom_tags": msg.CustomTags,
+	}
+	if err := enc.Encode(dataRow); err != nil {
+		return err
 	}
+	written += len(msg.Data)
+
 	diff := time.Since(t1)
 	log.GetLogger(jw.ctx).WithField("rows", written).WithField("elapsed", diff).Info("measurements written")
 	return nil

internal/sinks/json_test.go

@@ -8,10 +8,12 @@ import (
 
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestJSONWriter_Write(t *testing.T) {
 	a := assert.New(t)
+	r := require.New(t)
 	// Define test data
 	msg := metrics.MeasurementEnvelope{
 		MetricName: "test_metric",
@@ -25,23 +27,23 @@ func TestJSONWriter_Write(t *testing.T) {
 	tempFile := t.TempDir() + "/test.json"
 	ctx, cancel := context.WithCancel(context.Background())
 	jw, err := NewJSONWriter(ctx, tempFile)
-	a.NoError(err)
+	r.NoError(err)
 
-	err = jw.Write([]metrics.MeasurementEnvelope{msg})
+	err = jw.Write(msg)
 	a.NoError(err, "write successful")
-	err = jw.Write([]metrics.MeasurementEnvelope{})
-	a.NoError(err, "empty write successful")
+	err = jw.Write(metrics.MeasurementEnvelope{})
+	r.NoError(err, "empty write successful")
 
 	cancel()
-	err = jw.Write([]metrics.MeasurementEnvelope{msg})
+	err = jw.Write(msg)
 	a.Error(err, "context canceled")
 
 	// Read the contents of the file
 	var data map[string]any
 	file, err := os.ReadFile(tempFile)
-	a.NoError(err)
+	r.NoError(err)
 	err = json.Unmarshal(file, &data)
-	a.NoError(err)
+	r.NoError(err)
 	a.Equal(msg.MetricName, data["metric"])
 	a.Equal(len(msg.Data), len(data["data"].([]any)))
 	a.Equal(msg.DBName, data["dbname"])

internal/sinks/multiwriter.go

@@ -13,7 +13,7 @@ import (
 // Writer is an interface that writes metrics values
 type Writer interface {
 	SyncMetric(dbUnique, metricName, op string) error
-	Write(msgs []metrics.MeasurementEnvelope) error
+	Write(msgs metrics.MeasurementEnvelope) error
 }
 
 // MultiWriter ensures the simultaneous storage of data in several storages.
@@ -69,9 +69,9 @@ func (mw *MultiWriter) SyncMetric(dbUnique, metricName, op string) (err error) {
 	return
 }
 
-func (mw *MultiWriter) Write(msgs []metrics.MeasurementEnvelope) (err error) {
+func (mw *MultiWriter) Write(msg metrics.MeasurementEnvelope) (err error) {
 	for _, w := range mw.writers {
-		err = errors.Join(err, w.Write(msgs))
+		err = errors.Join(err, w.Write(msg))
 	}
 	return
 }

internal/sinks/multiwriter_test.go

@@ -14,7 +14,7 @@ func (mw *MockWriter) SyncMetric(_, _, _ string) error {
 	return nil
 }
 
-func (mw *MockWriter) Write(_ []metrics.MeasurementEnvelope) error {
+func (mw *MockWriter) Write(_ metrics.MeasurementEnvelope) error {
 	return nil
 }
 
@@ -82,6 +82,6 @@ func TestWriteMeasurements(t *testing.T) {
 	mw := &MultiWriter{}
 	mockWriter := &MockWriter{}
 	mw.AddWriter(mockWriter)
-	err := mw.Write([]metrics.MeasurementEnvelope{{}})
+	err := mw.Write(metrics.MeasurementEnvelope{})
 	assert.NoError(t, err)
 }