Make replication stats return whole number

.chloggen/replication-stats-conversion-failure.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: postgresqlreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Deprecation of replication lag metrics in favor of more precise 'write_ms','replay_ms' and 'flush_ms'"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [26714]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/postgresqlreceiver/client.go

@@ -15,9 +15,19 @@ import (
 	"github.com/lib/pq"
 	"go.opentelemetry.io/collector/config/confignet"
 	"go.opentelemetry.io/collector/config/configtls"
+	"go.opentelemetry.io/collector/featuregate"
 	"go.uber.org/multierr"
 )
 
+const lagMetricsInSecondsFeatureGateID = "postgresqlreceiver.lagmetricsinseconds"
+
+var lagInSecondsFG = featuregate.GlobalRegistry().MustRegister(
+	lagMetricsInSecondsFeatureGateID,
+	featuregate.StageDeprecated,
+	featuregate.WithRegisterDescription("Metrics 'flush', 'replay' and 'write' are replaced by 'flush_ms','replay_ms' and 'write_ms'."),
+	featuregate.WithRegisterToVersion("0.91.0"),
+)
+
 // databaseName is a name that refers to a database so that it can be uniquely referred to later
 // i.e. database1
 type databaseName string
@@ -64,6 +74,10 @@ type postgreSQLConfig struct {
 	tls      configtls.TLSClientSetting
 }
 
+func init() {
+	featuregate.GlobalRegistry().Set(lagInSecondsFG.ID(), true)
+}
+
 func sslConnectionString(tls configtls.TLSClientSetting) string {
 	if tls.Insecure {
 		return "sslmode='disable'"
@@ -484,18 +498,71 @@ func (c *postgreSQLClient) getMaxConnections(ctx context.Context) (int64, error)
 type replicationStats struct {
 	clientAddr   string
 	pendingBytes int64
-	flushLag     int64
-	replayLag    int64
-	writeLag     int64
+	flushLagInt  int64 // Deprecated
+	replayLagInt int64 // Deprecated
+	writeLagInt  int64 // Deprecated
+	flushLag     float64
+	replayLag    float64
+	writeLag     float64
+}
+
+func (c *postgreSQLClient) getDeprecatedReplicationStats(ctx context.Context) ([]replicationStats, error) {
+	query := `SELECT
+	client_addr,
+	coalesce(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn), -1) AS replication_bytes_pending,
+	extract('epoch' from coalesce(write_lag, '-1 seconds'))::integer,
+	extract('epoch' from coalesce(flush_lag, '-1 seconds'))::integer,
+	extract('epoch' from coalesce(replay_lag, '-1 seconds'))::integer,
+	extract('epoch' from coalesce(write_lag, '-1 seconds'))::decimal AS write_lag_fractional,
+	extract('epoch' from coalesce(flush_lag, '-1 seconds'))::decimal AS flush_lag_fractional,
+	extract('epoch' from coalesce(replay_lag, '-1 seconds'))::decimal AS replay_lag_fractional
+	FROM pg_stat_replication;
+	`
+	rows, err := c.client.QueryContext(ctx, query)
+	if err != nil {
+		return nil, fmt.Errorf("unable to query pg_stat_replication: %w", err)
+	}
+	defer rows.Close()
+	var rs []replicationStats
+	var errors error
+	for rows.Next() {
+		var client string
+		var replicationBytes int64
+		var writeLagInt, flushLagInt, replayLagInt int64
+		var writeLagFractional, flushLagFractional, replayLagFractional float64
+		err = rows.Scan(&client, &replicationBytes,
+			&writeLagInt, &flushLagInt, &replayLagInt,
+			&writeLagFractional, &flushLagFractional, &replayLagFractional)
+		if err != nil {
+			errors = multierr.Append(errors, err)
+			continue
+		}
+		rs = append(rs, replicationStats{
+			clientAddr:   client,
+			pendingBytes: replicationBytes,
+			replayLagInt: replayLagInt,
+			writeLagInt:  writeLagInt,
+			flushLagInt:  flushLagInt,
+			replayLag:    replayLagFractional,
+			writeLag:     writeLagFractional,
+			flushLag:     flushLagFractional,
+		})
+	}
+
+	return rs, errors
 }
 
 func (c *postgreSQLClient) getReplicationStats(ctx context.Context) ([]replicationStats, error) {
+	if lagInSecondsFG.IsEnabled() {
+		return c.getDeprecatedReplicationStats(ctx)
+	}
+
 	query := `SELECT
 	client_addr,
 	coalesce(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn), -1) AS replication_bytes_pending,
-	extract('epoch' from coalesce(write_lag, '-1 seconds')),
-	extract('epoch' from coalesce(flush_lag, '-1 seconds')),
-	extract('epoch' from coalesce(replay_lag, '-1 seconds'))
+	extract('epoch' from coalesce(write_lag, '-1 seconds'))::decimal AS write_lag_fractional,
+	extract('epoch' from coalesce(flush_lag, '-1 seconds'))::decimal AS flush_lag_fractional,
+	extract('epoch' from coalesce(replay_lag, '-1 seconds'))::decimal AS replay_lag_fractional
 	FROM pg_stat_replication;
 	`
 	rows, err := c.client.QueryContext(ctx, query)
@@ -507,7 +574,8 @@ func (c *postgreSQLClient) getReplicationStats(ctx context.Context) ([]replicati
 	var errors error
 	for rows.Next() {
 		var client string
-		var replicationBytes, writeLag, flushLag, replayLag int64
+		var replicationBytes int64
+		var writeLag, flushLag, replayLag float64
 		err = rows.Scan(&client, &replicationBytes, &writeLag, &flushLag, &replayLag)
 		if err != nil {
 			errors = multierr.Append(errors, err)

receiver/postgresqlreceiver/documentation.md

@@ -225,6 +225,24 @@ This metric requires WAL to be enabled with at least one replica.
 | ---- | ----------- | ---------- |
 | s | Gauge | Int |
 
+### postgresql.wal.delay
+
+Time between flushing recent WAL locally and receiving notification that the standby server has completed an operation with it captured in seconds and milliseconds.
+
+This metric requires WAL to be enabled with at least one replica.
+
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| s | Gauge | Double |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| operation | The operation which is responsible for the lag. | Str: ``flush``, ``replay``, ``write`` |
+| replication_client | The IP address of the client connected to this backend. If this field is "unix", it indicates either that the client is connected via a Unix socket. | Any Str |
+
 ### postgresql.wal.lag
 
 Time between flushing recent WAL locally and receiving notification that the standby server has completed an operation with it.

receiver/postgresqlreceiver/internal/metadata/testdata/config.yaml

@@ -51,6 +51,8 @@ all_set:
       enabled: true
     postgresql.wal.age:
       enabled: true
+    postgresql.wal.delay:
+      enabled: true
     postgresql.wal.lag:
       enabled: true
   resource_attributes:
@@ -112,6 +114,8 @@ none_set:
       enabled: false
     postgresql.wal.age:
       enabled: false
+    postgresql.wal.delay:
+      enabled: false
     postgresql.wal.lag:
       enabled: false
   resource_attributes:

receiver/postgresqlreceiver/metadata.yaml

@@ -297,3 +297,13 @@ metrics:
       value_type: int
     extended_documentation: |
       This metric requires WAL to be enabled with at least one replica.
+  postgresql.wal.delay:
+    attributes: [wal_operation_lag, replication_client]
+    description: Time between flushing recent WAL locally and receiving notification that the standby server has completed an operation with it captured in seconds and milliseconds.
+    enabled: true
+    unit: s
+    gauge:
+      value_type: double
+    extended_documentation: |
+      This metric requires WAL to be enabled with at least one replica.
+

receiver/postgresqlreceiver/scraper.go

@@ -313,14 +313,26 @@ func (p *postgreSQLScraper) collectReplicationStats(
 		if rs.pendingBytes >= 0 {
 			p.mb.RecordPostgresqlReplicationDataDelayDataPoint(now, rs.pendingBytes, rs.clientAddr)
 		}
+		if lagInSecondsFG.IsEnabled() {
+			if rs.writeLagInt >= 0 {
+				p.mb.RecordPostgresqlWalLagDataPoint(now, rs.writeLagInt, metadata.AttributeWalOperationLagWrite, rs.clientAddr)
+			}
+			if rs.replayLagInt >= 0 {
+				p.mb.RecordPostgresqlWalLagDataPoint(now, rs.replayLagInt, metadata.AttributeWalOperationLagReplay, rs.clientAddr)
+			}
+			if rs.flushLagInt >= 0 {
+				p.mb.RecordPostgresqlWalLagDataPoint(now, rs.flushLagInt, metadata.AttributeWalOperationLagFlush, rs.clientAddr)
+			}
+		}
+
 		if rs.writeLag >= 0 {
-			p.mb.RecordPostgresqlWalLagDataPoint(now, rs.writeLag, metadata.AttributeWalOperationLagWrite, rs.clientAddr)
+			p.mb.RecordPostgresqlWalDelayDataPoint(now, rs.writeLag, metadata.AttributeWalOperationLagWrite, rs.clientAddr)
 		}
 		if rs.replayLag >= 0 {
-			p.mb.RecordPostgresqlWalLagDataPoint(now, rs.replayLag, metadata.AttributeWalOperationLagReplay, rs.clientAddr)
+			p.mb.RecordPostgresqlWalDelayDataPoint(now, rs.replayLag, metadata.AttributeWalOperationLagReplay, rs.clientAddr)
 		}
 		if rs.flushLag >= 0 {
-			p.mb.RecordPostgresqlWalLagDataPoint(now, rs.flushLag, metadata.AttributeWalOperationLagFlush, rs.clientAddr)
+			p.mb.RecordPostgresqlWalDelayDataPoint(now, rs.flushLag, metadata.AttributeWalOperationLagFlush, rs.clientAddr)
 		}
 	}
 }