Metric/postgresql connections (#62)

receiver/postgresqlreceiver/client.go

@@ -55,6 +55,7 @@ type client interface {
 	getLatestWalAgeSeconds(ctx context.Context) (int64, error)
 	getMaxConnections(ctx context.Context) (int64, error)
 	getIndexStats(ctx context.Context, database string) (map[indexIdentifer]indexStat, error)
+	getActiveConnections(ctx context.Context) (int64, error)
 	listDatabases(ctx context.Context) ([]string, error)
 }
 
@@ -488,6 +489,14 @@ func (c *postgreSQLClient) getMaxConnections(ctx context.Context) (int64, error)
 	return maxConns, err
 }
 
+func (c *postgreSQLClient) getActiveConnections(ctx context.Context) (int64, error) {
+	query := `SELECT COUNT(*) FROM pg_stat_activity WHERE state = 'active';`
+	row := c.client.QueryRowContext(ctx, query)
+	var activeConns int64
+	err := row.Scan(&activeConns)
+	return activeConns, err
+}
+
 type replicationStats struct {
 	clientAddr   string
 	pendingBytes int64

receiver/postgresqlreceiver/documentation.md

@@ -100,6 +100,14 @@ The number of commits.
 | ---- | ----------- | ---------- | ----------------------- | --------- |
 | 1 | Sum | Int | Cumulative | true |
 
+### postgresql.connection.count
+
+The number of active connections to this database. If DBM is enabled, this metric is tagged with state, app, db and user
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {connection} | Gauge | Int |
+
 ### postgresql.connection.max
 
 Configured maximum number of client connections allowed
@@ -229,7 +237,7 @@ This metric requires WAL to be enabled with at least one replica.
 
 Time between flushing recent WAL locally and receiving notification that the standby server has completed an operation with it.
 
-This metric requires WAL to be enabled with at least one replica.
+This metric requires WAL to be enabled with at least one replica. 
 
 
 | Unit | Metric Type | Value Type |

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

@@ -17,6 +17,8 @@ all_set:
       enabled: true
     postgresql.commits:
       enabled: true
+    postgresql.connection.count:
+      enabled: true
     postgresql.connection.max:
       enabled: true
     postgresql.database.count:
@@ -82,6 +84,8 @@ none_set:
       enabled: false
     postgresql.commits:
       enabled: false
+    postgresql.connection.count:
+      enabled: false
     postgresql.connection.max:
       enabled: false
     postgresql.database.count:

receiver/postgresqlreceiver/metadata.yaml

@@ -12,8 +12,7 @@ status:
 resource_attributes:
   postgresql.database.name:
     description: The name of the database.
-    enabled: true
-    type: string
+    enabled: true type: string
   postgresql.schema.name:
     description: The schema name.
     enabled: true
@@ -311,6 +310,15 @@ metrics:
       value_type: double
     extended_documentation: |
       This metric requires WAL to be enabled with at least one replica.
-  
+
 tests:
   config:
+
+      This metric requires WAL to be enabled with at least one replica.
+  postgresql.connection.count:
+    enabled: true
+    description: The number of active connections to this database. If DBM is enabled,
+      this metric is tagged with state, app, db and user
+    unit: '{connection}'
+    gauge:
+      value_type: int

receiver/postgresqlreceiver/scraper.go

@@ -163,6 +163,8 @@ func (p *postgreSQLScraper) scrape(ctx context.Context) (pmetric.Metrics, error)
 	p.collectDatabaseLocks(ctx, now, listClient, &errs)
 
 
+	p.collectActiveConnections(ctx, now, listClient, &errs)
+
 	rb := p.mb.NewResourceBuilder()
 	rb.SetPostgresqlDatabaseName("N/A")
 	p.mb.EmitForResource(metadata.WithResource(rb.Emit()))
@@ -348,6 +350,20 @@ func (p *postgreSQLScraper) collectMaxConnections(
 	p.mb.RecordPostgresqlConnectionMaxDataPoint(now, mc)
 }
 
+func (p *postgreSQLScraper) collectActiveConnections(
+	ctx context.Context,
+	now pcommon.Timestamp,
+	client client,
+	errs *errsMux,
+) {
+	ac, err := client.getActiveConnections(ctx)
+	if err != nil {
+		errs.addPartial(err)
+		return
+	}
+	p.mb.RecordPostgresqlConnectionCountDataPoint(now, ac)
+}
+
 func (p *postgreSQLScraper) collectReplicationStats(
 	ctx context.Context,
 	now pcommon.Timestamp,