[receiver/postgresqlreceiver] add schema attribute to postgresqlrecei…

…ver (open-telemetry#30142)

**Description:** 

Changes the postgresql statistics receiver so that a table or indices
schema is stored in it's own attribute.

Currently the schema is stored as table name for *some* of the
statistics, while others completely ignore what schema a statistic comes
out of.

The code will allow computing things like sequential vs index scans
against a table without having to pick apart the resource attribute.

**Link to tracking Issue:**  open-telemetry#29559 

**Testing:** I've built the docker container for otel contrib and have
been running the result for a few days to observe that the attributes
are properly propagated.

**Documentation:** Documentation for the new field was added to the
metadata.yaml and documentation.md

.chloggen/pgcol-add-schema-attribute.yaml

@@ -0,0 +1,29 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# 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: add schema attribute to postgresqlreceiver
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [ 29559 ]
+
+# (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: |
+ Adds a new resource attribute to the PSQL receiver to store the schema of the table or index
+ Existing table attributes are adjusted to not include the schema, which was inconsistently used
+
+# 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

@@ -264,6 +264,7 @@ func (c *postgreSQLClient) getDatabaseSize(ctx context.Context, databases []stri
 // tableStats contains a result for a row of the getDatabaseTableMetrics result
 type tableStats struct {
  database string
+ schema string
  table string
  live int64
  dead int64
@@ -277,7 +278,7 @@ type tableStats struct {
 }
 
 func (c *postgreSQLClient) getDatabaseTableMetrics(ctx context.Context, db string) (map[tableIdentifier]tableStats, error) {
- query := `SELECT schemaname || '.' || relname AS table,
+ query := `SELECT schemaname as schema, relname AS table,
  n_live_tup AS live,
  n_dead_tup AS dead,
  n_tup_ins AS ins,
@@ -296,15 +297,16 @@ func (c *postgreSQLClient) getDatabaseTableMetrics(ctx context.Context, db strin
  return nil, err
  }
  for rows.Next() {
- var table string
+ var schema, table string
  var live, dead, ins, upd, del, hotUpd, seqScans, tableSize, vacuumCount int64
- err = rows.Scan(&table, &live, &dead, &ins, &upd, &del, &hotUpd, &seqScans, &tableSize, &vacuumCount)
+ err = rows.Scan(&schema, &table, &live, &dead, &ins, &upd, &del, &hotUpd, &seqScans, &tableSize, &vacuumCount)
  if err != nil {
  errors = multierr.Append(errors, err)
  continue
  }
- ts[tableKey(db, table)] = tableStats{
+ ts[tableKey(db, schema, table)] = tableStats{
  database: db,
+ schema: schema,
  table: table,
  live: live,
  inserts: ins,
@@ -321,6 +323,7 @@ func (c *postgreSQLClient) getDatabaseTableMetrics(ctx context.Context, db strin
 
 type tableIOStats struct {
  database string
+ schema string
  table string
  heapRead int64
  heapHit int64
@@ -333,7 +336,7 @@ type tableIOStats struct {
 }
 
 func (c *postgreSQLClient) getBlocksReadByTable(ctx context.Context, db string) (map[tableIdentifier]tableIOStats, error) {
- query := `SELECT schemaname || '.' || relname AS table,
+ query := `SELECT schemaname as schema, relname AS table,
  coalesce(heap_blks_read, 0) AS heap_read,
  coalesce(heap_blks_hit, 0) AS heap_hit,
  coalesce(idx_blks_read, 0) AS idx_read,
@@ -351,15 +354,16 @@ func (c *postgreSQLClient) getBlocksReadByTable(ctx context.Context, db string)
  return nil, err
  }
  for rows.Next() {
- var table string
+ var schema, table string
  var heapRead, heapHit, idxRead, idxHit, toastRead, toastHit, tidxRead, tidxHit int64
- err = rows.Scan(&table, &heapRead, &heapHit, &idxRead, &idxHit, &toastRead, &toastHit, &tidxRead, &tidxHit)
+ err = rows.Scan(&schema, &table, &heapRead, &heapHit, &idxRead, &idxHit, &toastRead, &toastHit, &tidxRead, &tidxHit)
  if err != nil {
  errors = multierr.Append(errors, err)
  continue
  }
- tios[tableKey(db, table)] = tableIOStats{
+ tios[tableKey(db, schema, table)] = tableIOStats{
  database: db,
+ schema: schema,
  table: table,
  heapRead: heapRead,
  heapHit: heapHit,
@@ -377,13 +381,14 @@ func (c *postgreSQLClient) getBlocksReadByTable(ctx context.Context, db string)
 type indexStat struct {
  index string
  table string
+ schema string
  database string
  size int64
  scans int64
 }
 
 func (c *postgreSQLClient) getIndexStats(ctx context.Context, database string) (map[indexIdentifer]indexStat, error) {
- query := `SELECT relname, indexrelname,
+ query := `SELECT schemaname, relname, indexrelname,
  pg_relation_size(indexrelid) AS index_size,
  idx_scan
  FROM pg_stat_user_indexes;`
@@ -399,17 +404,18 @@ func (c *postgreSQLClient) getIndexStats(ctx context.Context, database string) (
  var errs []error
  for rows.Next() {
  var (
- table, index  string
+ schema, table, index string
  indexSize, indexScans int64
  )
- err := rows.Scan(&table, &index, &indexSize, &indexScans)
+ err := rows.Scan(&schema, &table, &index, &indexSize, &indexScans)
  if err != nil {
  errs = append(errs, err)
  continue
  }
- stats[indexKey(database, table, index)] = indexStat{
+ stats[indexKey(database, schema, table, index)] = indexStat{
  index: index,
  table: table,
+ schema: schema,
  database: database,
  size: indexSize,
  scans: indexScans,
@@ -642,10 +648,10 @@ func filterQueryByDatabases(baseQuery string, databases []string, groupBy bool)
  return baseQuery + ";"
 }
 
-func tableKey(database, table string) tableIdentifier {
- return tableIdentifier(fmt.Sprintf("%s|%s", database, table))
+func tableKey(database, schema, table string) tableIdentifier {
+ return tableIdentifier(fmt.Sprintf("%s|%s|%s", database, schema, table))
 }
 
-func indexKey(database, table, index string) indexIdentifer {
- return indexIdentifer(fmt.Sprintf("%s|%s|%s", database, table, index))
+func indexKey(database, schema, table, index string) indexIdentifer {
+ return indexIdentifer(fmt.Sprintf("%s|%s|%s|%s", database, schema, table, index))
 }

receiver/postgresqlreceiver/documentation.md

@@ -317,4 +317,5 @@ This metric requires WAL to be enabled with at least one replica.
 | ---- | ----------- | ------ | ------- |
 | postgresql.database.name | The name of the database. | Any Str | true |
 | postgresql.index.name | The name of the index on a table. | Any Str | true |
-| postgresql.table.name | The schema name followed by the table name. | Any Str | true |
+| postgresql.schema.name | The schema name. | Any Str | true |
+| postgresql.table.name | The table name. | Any Str | true |

receiver/postgresqlreceiver/integration_test.go

@@ -15,18 +15,27 @@ import (
  "github.com/testcontainers/testcontainers-go/wait"
  "go.opentelemetry.io/collector/component"
 
+ "github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
  "github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/scraperinttest"
  "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
 )
 
 const postgresqlPort = "5432"
 
 func TestIntegration(t *testing.T) {
+ defer testutil.SetFeatureGateForTest(t, separateSchemaAttrGate, false)()
  t.Run("single_db", integrationTest("single_db", []string{"otel"}))
  t.Run("multi_db", integrationTest("multi_db", []string{"otel", "otel2"}))
  t.Run("all_db", integrationTest("all_db", []string{}))
 }
 
+func TestIntegrationWithSeparateSchemaAttr(t *testing.T) {
+ defer testutil.SetFeatureGateForTest(t, separateSchemaAttrGate, true)()
+ t.Run("single_db_schemaattr", integrationTest("single_db_schemaattr", []string{"otel"}))
+ t.Run("multi_db_schemaattr", integrationTest("multi_db_schemaattr", []string{"otel", "otel2"}))
+ t.Run("all_db_schemaattr", integrationTest("all_db_schemaattr", []string{}))
+}
+
 func integrationTest(name string, databases []string) func(*testing.T) {
  expectedFile := filepath.Join("testdata", "integration", "expected_"+name+".yaml")
  return scraperinttest.NewIntegrationTest(

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

@@ -60,6 +60,8 @@ all_set:
  enabled: true
  postgresql.index.name:
  enabled: true
+ postgresql.schema.name:
+ enabled: true
  postgresql.table.name:
  enabled: true
 none_set:
@@ -123,5 +125,7 @@ none_set:
  enabled: false
  postgresql.index.name:
  enabled: false
+ postgresql.schema.name:
+ enabled: false
  postgresql.table.name:
  enabled: false

receiver/postgresqlreceiver/metadata.yaml

@@ -13,8 +13,12 @@ resource_attributes:
  description: The name of the database.
  enabled: true
  type: string
+ postgresql.schema.name:
+ description: The schema name.
+ enabled: true
+ type: string
  postgresql.table.name:
- description: The schema name followed by the table name.
+ description: The table name.
  enabled: true
  type: string
  postgresql.index.name: