diff --git a/docs/sources/reference/components/database_observability/database_observability.postgres.md b/docs/sources/reference/components/database_observability/database_observability.postgres.md index 3417853e4fe..37de3681116 100644 --- a/docs/sources/reference/components/database_observability/database_observability.postgres.md +++ b/docs/sources/reference/components/database_observability/database_observability.postgres.md @@ -56,6 +56,8 @@ The following collectors are configurable: | Name | Description | Enabled by default | |------------------|-----------------------------------------------------------------------|--------------------| | `explain_plans` | Collect query explain plans. | yes | +| `table_stats` | Collect table-level scan statistics. | no | +| `index_stats` | Collect per-index usage statistics. | no | | `query_details` | Collect queries information. | yes | | `query_samples` | Collect query samples and wait events information. | yes | | `schema_details` | Collect schemas, tables, and columns from PostgreSQL system catalogs. | yes | diff --git a/internal/component/database_observability/postgres/collector/dsn.go b/internal/component/database_observability/postgres/collector/dsn.go index e391582c8de..e689aa45e9a 100644 --- a/internal/component/database_observability/postgres/collector/dsn.go +++ b/internal/component/database_observability/postgres/collector/dsn.go @@ -1,8 +1,10 @@ package collector import ( + "context" "database/sql" "errors" + "fmt" "regexp" ) @@ -13,6 +15,14 @@ var defaultDbConnectionFactory = func(dsn string) (*sql.DB, error) { return sql.Open("postgres", dsn) } +// selectAllDatabases makes use of the initial DB connection to discover other databases on the same Postgres instance +const selectAllDatabases = ` + SELECT datname + FROM pg_database + WHERE datistemplate = false + AND has_database_privilege(datname, 'CONNECT') + AND datname NOT IN %s` + // replaceDatabaseNameInDSN safely replaces the database name in a PostgreSQL DSN // using regex to ensure only the database name portion is replaced, not other occurrences func replaceDatabaseNameInDSN(dsn, newDatabaseName string) (string, error) { @@ -30,3 +40,74 @@ func replaceDatabaseNameInDSN(dsn, newDatabaseName string) (string, error) { newDSN := matches[1] + newDatabaseName + matches[3] return newDSN, nil } + +// databaseNameFromDSN extracts the database name a DSN already points to, so +// callers fanning out per-database can tell whether a target database is the +// one an existing connection already uses. +func databaseNameFromDSN(dsn string) (string, error) { + matches := dsnParseRegex.FindStringSubmatch(dsn) + if len(matches) < 4 { + return "", errors.New("failed to parse DSN for database name") + } + return matches[2], nil +} + +// discoverDatabases lists databases the current connection can reach, via +// pg_database (readable from any single connection) -- used to fan out +// per-database connections, since most stat views only report on the +// database a connection is actually established to. +func discoverDatabases(ctx context.Context, conn *sql.DB, excludeDatabases []string) ([]string, error) { + query := fmt.Sprintf(selectAllDatabases, buildExcludedDatabasesClause(excludeDatabases)) + rows, err := conn.QueryContext(ctx, query) + if err != nil { + return nil, fmt.Errorf("failed to discover databases: %w", err) + } + defer rows.Close() + + var databases []string + for rows.Next() { + var datname string + if err := rows.Scan(&datname); err != nil { + return nil, fmt.Errorf("failed to scan database name: %w", err) + } + databases = append(databases, datname) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating database rows: %w", err) + } + + return databases, nil +} + +// connectToDatabase opens a connection to dbName by rewriting dsn. If dbName +// is already what dsn (and so initial) points to, it reuses initial instead +// of opening a redundant connection -- sql.Open never returns something +// pointer-equal to an existing *sql.DB, so this has to be checked by name up +// front, not via "conn != initial" after the fact. closeFn closes the +// connection unless it's initial. +func connectToDatabase(dsn, dbName string, factory databaseConnectionFactory, initial *sql.DB) (conn *sql.DB, closeFn func(), err error) { + noopClose := func() {} + + if currentDBName, err := databaseNameFromDSN(dsn); err == nil && currentDBName == dbName { + return initial, noopClose, nil + } + + databaseDSN, err := replaceDatabaseNameInDSN(dsn, dbName) + if err != nil { + return nil, nil, fmt.Errorf("failed to create DSN for database %s: %w", dbName, err) + } + + conn, err = factory(databaseDSN) + if err != nil { + return nil, nil, fmt.Errorf("failed to create connection to database %s: %w", dbName, err) + } + + closeFn = func() { + if conn != initial { + conn.Close() + } + } + + return conn, closeFn, nil +} diff --git a/internal/component/database_observability/postgres/collector/dsn_test.go b/internal/component/database_observability/postgres/collector/dsn_test.go index b65c7586fad..2bcdf752a24 100644 --- a/internal/component/database_observability/postgres/collector/dsn_test.go +++ b/internal/component/database_observability/postgres/collector/dsn_test.go @@ -1,8 +1,10 @@ package collector import ( + "database/sql" "testing" + sqlmock "github.com/DATA-DOG/go-sqlmock" "github.com/stretchr/testify/require" ) @@ -96,3 +98,47 @@ func TestReplaceDatabaseNameInDSN(t *testing.T) { }) } } + +// TestConnectToDatabaseReusesInitialConnection guards against connectToDatabase +// opening a redundant connection for the database initial already points to; +// see the comment on connectToDatabase for why a bare "conn != initial" check +// can't catch this. +func TestConnectToDatabaseReusesInitialConnection(t *testing.T) { + initial, _, err := sqlmock.New() + require.NoError(t, err) + defer initial.Close() + + newDB, _, err := sqlmock.New() + require.NoError(t, err) + defer newDB.Close() + + t.Run("same database as the DSN: reuses initial, never calls factory", func(t *testing.T) { + factoryCalls := 0 + factory := func(dsn string) (*sql.DB, error) { + factoryCalls++ + return newDB, nil + } + + conn, closeFn, err := connectToDatabase("postgres://user:pass@localhost:5432/books_store", "books_store", factory, initial) + require.NoError(t, err) + require.Same(t, initial, conn) + require.Equal(t, 0, factoryCalls) + closeFn() // must not close initial + + require.NoError(t, initial.PingContext(t.Context())) // still usable + }) + + t.Run("different database: opens a new connection via factory", func(t *testing.T) { + factoryCalls := 0 + factory := func(dsn string) (*sql.DB, error) { + factoryCalls++ + return newDB, nil + } + + conn, closeFn, err := connectToDatabase("postgres://user:pass@localhost:5432/postgres", "books_store", factory, initial) + require.NoError(t, err) + require.Same(t, newDB, conn) + require.Equal(t, 1, factoryCalls) + closeFn() + }) +} diff --git a/internal/component/database_observability/postgres/collector/index_stats.go b/internal/component/database_observability/postgres/collector/index_stats.go new file mode 100644 index 00000000000..5ec55849235 --- /dev/null +++ b/internal/component/database_observability/postgres/collector/index_stats.go @@ -0,0 +1,164 @@ +package collector + +import ( + "context" + "database/sql" + "log/slog" + "strconv" + + "github.com/prometheus/client_golang/prometheus" + "go.uber.org/atomic" +) + +// IndexStatsCollector emits per-index usage counters from pg_stat_user_indexes, +// scoped to every database the connection can reach rather than only the one +// named in the DSN. +const IndexStatsCollector = "index_stats" + +const selectIndexUsageStats = ` + SELECT + s.schemaname, + s.relname, + s.indexrelname, + s.idx_scan, + i.indisprimary, + i.indisunique, + i.indpred IS NOT NULL AS is_partial, + pg_relation_size(s.indexrelid) AS index_size_bytes + FROM pg_stat_user_indexes s + JOIN pg_index i ON i.indexrelid = s.indexrelid` + +var indexLabels = []string{labelDatname, "schemaname", "relname", "indexrelname"} +var indexSizeLabels = append(append([]string{}, indexLabels...), "is_primary", "is_unique", "is_partial") + +var ( + indexUsageIdxScanTotalDesc = prometheus.NewDesc( + prometheus.BuildFQName("database_observability", "pg_index_stats", "idx_scan_total"), + "Number of index scans initiated on this index", + indexLabels, nil, + ) + indexSizeBytesDesc = prometheus.NewDesc( + prometheus.BuildFQName("database_observability", "pg_index_stats", "size_bytes"), + "Total disk space used by this index, in bytes, labeled with whether it backs the primary key or a unique constraint, or is partial", + indexSizeLabels, nil, + ) +) + +type IndexStatsArguments struct { + DB *sql.DB + DSN string + ExcludeDatabases []string + Registry *prometheus.Registry + + Logger *slog.Logger + + dbConnectionFactory databaseConnectionFactory +} + +type IndexStats struct { + initialConnection *sql.DB + dbDSN string + dbConnectionFactory databaseConnectionFactory + excludeDatabases []string + registry *prometheus.Registry + + logger *slog.Logger + running *atomic.Bool +} + +func NewIndexStats(args IndexStatsArguments) (*IndexStats, error) { + factory := args.dbConnectionFactory + if factory == nil { + factory = defaultDbConnectionFactory + } + + return &IndexStats{ + initialConnection: args.DB, + dbDSN: args.DSN, + dbConnectionFactory: factory, + excludeDatabases: args.ExcludeDatabases, + registry: args.Registry, + logger: args.Logger.With("collector", IndexStatsCollector), + running: &atomic.Bool{}, + }, nil +} + +func (c *IndexStats) Name() string { + return IndexStatsCollector +} + +func (c *IndexStats) Start(_ context.Context) error { + if err := c.registry.Register(c); err != nil { + return err + } + c.running.Store(true) + return nil +} + +func (c *IndexStats) Stopped() bool { + return !c.running.Load() +} + +func (c *IndexStats) Stop() { + c.registry.Unregister(c) + c.running.Store(false) +} + +// Describe implements prometheus.Collector. +func (c *IndexStats) Describe(ch chan<- *prometheus.Desc) { + ch <- indexUsageIdxScanTotalDesc + ch <- indexSizeBytesDesc +} + +// Collect implements prometheus.Collector. It runs synchronously at scrape +// time, fanning out to every database the connection can reach. +func (c *IndexStats) Collect(ch chan<- prometheus.Metric) { + ctx := context.Background() + + databases, err := discoverDatabases(ctx, c.initialConnection, c.excludeDatabases) + if err != nil { + c.logger.Error("failed to discover databases", "err", err) + return + } + + for _, dbName := range databases { + conn, closeConn, err := connectToDatabase(c.dbDSN, dbName, c.dbConnectionFactory, c.initialConnection) + if err != nil { + c.logger.Error("failed to connect to database", "datname", dbName, "err", err) + continue + } + + c.collectIndexUsageStats(ctx, dbName, conn, ch) + + closeConn() + } +} + +func (c *IndexStats) collectIndexUsageStats(ctx context.Context, dbName string, conn *sql.DB, ch chan<- prometheus.Metric) { + rows, err := conn.QueryContext(ctx, selectIndexUsageStats) + if err != nil { + c.logger.Error("failed to query pg_stat_user_indexes", "datname", dbName, "err", err) + return + } + defer rows.Close() + + for rows.Next() { + var schemaname, relname, indexrelname string + var idxScan, indexSizeBytes sql.NullInt64 + var isPrimary, isUnique, isPartial bool + + if err := rows.Scan(&schemaname, &relname, &indexrelname, &idxScan, &isPrimary, &isUnique, &isPartial, &indexSizeBytes); err != nil { + c.logger.Error("failed to scan pg_stat_user_indexes row", "datname", dbName, "err", err) + return + } + + ch <- prometheus.MustNewConstMetric(indexUsageIdxScanTotalDesc, prometheus.CounterValue, float64(idxScan.Int64), dbName, schemaname, relname, indexrelname) + ch <- prometheus.MustNewConstMetric(indexSizeBytesDesc, prometheus.GaugeValue, float64(indexSizeBytes.Int64), + dbName, schemaname, relname, indexrelname, + strconv.FormatBool(isPrimary), strconv.FormatBool(isUnique), strconv.FormatBool(isPartial)) + } + + if err := rows.Err(); err != nil { + c.logger.Error("error iterating pg_stat_user_indexes rows", "datname", dbName, "err", err) + } +} diff --git a/internal/component/database_observability/postgres/collector/index_stats_test.go b/internal/component/database_observability/postgres/collector/index_stats_test.go new file mode 100644 index 00000000000..9642550e35a --- /dev/null +++ b/internal/component/database_observability/postgres/collector/index_stats_test.go @@ -0,0 +1,62 @@ +package collector + +import ( + "database/sql" + "fmt" + "strings" + "testing" + + sqlmock "github.com/DATA-DOG/go-sqlmock" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + + "github.com/grafana/alloy/internal/util" +) + +func TestIndexStats(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual)) + require.NoError(t, err) + defer db.Close() + + registry := prometheus.NewRegistry() + + c, err := NewIndexStats(IndexStatsArguments{ + DB: db, + DSN: "postgres://user:pass@localhost:5432/books_store", + ExcludeDatabases: nil, + Registry: registry, + Logger: util.TestAlloyLogger(t).Slog(), + dbConnectionFactory: func(dsn string) (*sql.DB, error) { + return db, nil + }, + }) + require.NoError(t, err) + + require.NoError(t, c.Start(t.Context())) + defer c.Stop() + + mock.ExpectQuery(fmt.Sprintf(selectAllDatabases, exclusionClause)).WithoutArgs().RowsWillBeClosed(). + WillReturnRows(sqlmock.NewRows([]string{"datname"}).AddRow("books_store")) + + mock.ExpectQuery(selectIndexUsageStats).WithoutArgs().RowsWillBeClosed(). + WillReturnRows( + sqlmock.NewRows([]string{"schemaname", "relname", "indexrelname", "idx_scan", "indisprimary", "indisunique", "is_partial", "index_size_bytes"}). + AddRow("public", "books", "books_pkey", 184000000, true, true, false, 65536). + AddRow("public", "books", "idx_books_title", 0, false, false, true, 32768), + ) + + expected := ` + # HELP database_observability_pg_index_stats_idx_scan_total Number of index scans initiated on this index + # TYPE database_observability_pg_index_stats_idx_scan_total counter + database_observability_pg_index_stats_idx_scan_total{datname="books_store",indexrelname="books_pkey",relname="books",schemaname="public"} 1.84e+08 + database_observability_pg_index_stats_idx_scan_total{datname="books_store",indexrelname="idx_books_title",relname="books",schemaname="public"} 0 + # HELP database_observability_pg_index_stats_size_bytes Total disk space used by this index, in bytes, labeled with whether it backs the primary key or a unique constraint, or is partial + # TYPE database_observability_pg_index_stats_size_bytes gauge + database_observability_pg_index_stats_size_bytes{datname="books_store",indexrelname="books_pkey",is_partial="false",is_primary="true",is_unique="true",relname="books",schemaname="public"} 65536 + database_observability_pg_index_stats_size_bytes{datname="books_store",indexrelname="idx_books_title",is_partial="true",is_primary="false",is_unique="false",relname="books",schemaname="public"} 32768 +` + + require.NoError(t, testutil.CollectAndCompare(registry, strings.NewReader(expected))) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/internal/component/database_observability/postgres/collector/logs.go b/internal/component/database_observability/postgres/collector/logs.go index 0d06190a87a..d5bd998b7b5 100644 --- a/internal/component/database_observability/postgres/collector/logs.go +++ b/internal/component/database_observability/postgres/collector/logs.go @@ -180,7 +180,7 @@ func (l *Logs) initMetrics() { Name: "pg_errors_total", Help: "Number of log lines with errors by severity and sql state code", }, - []string{"severity", "sqlstate", "sqlstate_class", "datname", "user"}, + []string{"severity", "sqlstate", "sqlstate_class", labelDatname, "user"}, ) l.parseErrors = prometheus.NewCounter( diff --git a/internal/component/database_observability/postgres/collector/schema_details.go b/internal/component/database_observability/postgres/collector/schema_details.go index 9571b415ffa..77dc8da5273 100644 --- a/internal/component/database_observability/postgres/collector/schema_details.go +++ b/internal/component/database_observability/postgres/collector/schema_details.go @@ -26,14 +26,6 @@ const ( ) const ( - // selectAllDatabases makes use of the initial DB connection to discover other databases on the same Postgres instance - selectAllDatabases = ` - SELECT datname - FROM pg_database - WHERE datistemplate = false - AND has_database_privilege(datname, 'CONNECT') - AND datname NOT IN %s` - // selectSchemaNames gets all user-defined schemas, excluding system schemas selectSchemaNames = ` SELECT @@ -433,29 +425,11 @@ func (c *SchemaDetails) Stop() { } func (c *SchemaDetails) getAllDatabases(ctx context.Context) ([]string, error) { - query := fmt.Sprintf(selectAllDatabases, buildExcludedDatabasesClause(c.excludeDatabases)) - rows, err := c.initialConnection.QueryContext(ctx, query) + databases, err := discoverDatabases(ctx, c.initialConnection, c.excludeDatabases) if err != nil { c.logger.Error("failed to discover databases", "err", err) - return nil, fmt.Errorf("failed to discover databases: %w", err) - } - defer rows.Close() - - var databases []string - for rows.Next() { - var datname string - if err := rows.Scan(&datname); err != nil { - c.logger.Error("failed to scan database name", "err", err) - continue - } - databases = append(databases, datname) - } - - if err := rows.Err(); err != nil { - c.logger.Error("error iterating database rows", "err", err) - return nil, fmt.Errorf("error iterating database rows: %w", err) + return nil, err } - return databases, nil } @@ -615,31 +589,19 @@ func (c *SchemaDetails) extractNames(ctx context.Context) error { } for _, dbName := range databases { - databaseDSN, err := replaceDatabaseNameInDSN(c.dbDSN, dbName) - if err != nil { - c.logger.Error("failed to create DSN for database", "datname", dbName, "err", err) - continue - } - - conn, err := c.dbConnectionFactory(databaseDSN) + conn, closeConn, err := connectToDatabase(c.dbDSN, dbName, c.dbConnectionFactory, c.initialConnection) if err != nil { - c.logger.Error("failed to create connection to database", "datname", dbName, "err", err) + c.logger.Error("failed to connect to database", "datname", dbName, "err", err) continue } if err := c.extractSchemas(ctx, dbName, conn); err != nil { c.logger.Error("failed to collect schema from database", "datname", dbName, "err", err) - if conn != c.initialConnection { - conn.Close() - } + closeConn() continue } - if conn != c.initialConnection { - if err := conn.Close(); err != nil { - c.logger.Warn("failed to close database connection", "datname", dbName, "err", err) - } - } + closeConn() } // Drop throttle entries for databases that getAllDatabases no longer diff --git a/internal/component/database_observability/postgres/collector/table_stats.go b/internal/component/database_observability/postgres/collector/table_stats.go new file mode 100644 index 00000000000..c4d7a65b766 --- /dev/null +++ b/internal/component/database_observability/postgres/collector/table_stats.go @@ -0,0 +1,164 @@ +package collector + +import ( + "context" + "database/sql" + "log/slog" + + "github.com/prometheus/client_golang/prometheus" + "go.uber.org/atomic" +) + +// TableStatsCollector emits table-level scan counters from pg_stat_user_tables, +// scoped to every database the connection can reach rather than only the one +// named in the DSN. +const TableStatsCollector = "table_stats" + +const selectTableScanStats = ` + SELECT + schemaname, + relname, + seq_scan, + idx_scan, + n_live_tup + FROM pg_stat_user_tables` + +const labelDatname = "datname" + +var tableLabels = []string{labelDatname, "schemaname", "relname"} + +var ( + tableScanStatsSeqScanDesc = prometheus.NewDesc( + prometheus.BuildFQName("database_observability", "pg_table_stats", "seq_scan_total"), + "Number of sequential scans initiated on this table", + tableLabels, nil, + ) + tableScanStatsIdxScanDesc = prometheus.NewDesc( + prometheus.BuildFQName("database_observability", "pg_table_stats", "idx_scan_total"), + "Number of index scans initiated on this table", + tableLabels, nil, + ) + tableStatsRowCountDesc = prometheus.NewDesc( + prometheus.BuildFQName("database_observability", "pg_table_stats", "row_count"), + "Estimated number of live rows in this table", + tableLabels, nil, + ) +) + +type TableStatsArguments struct { + DB *sql.DB + DSN string + ExcludeDatabases []string + Registry *prometheus.Registry + + Logger *slog.Logger + + dbConnectionFactory databaseConnectionFactory +} + +type TableStats struct { + initialConnection *sql.DB + dbDSN string + dbConnectionFactory databaseConnectionFactory + excludeDatabases []string + registry *prometheus.Registry + + logger *slog.Logger + running *atomic.Bool +} + +func NewTableStats(args TableStatsArguments) (*TableStats, error) { + factory := args.dbConnectionFactory + if factory == nil { + factory = defaultDbConnectionFactory + } + + return &TableStats{ + initialConnection: args.DB, + dbDSN: args.DSN, + dbConnectionFactory: factory, + excludeDatabases: args.ExcludeDatabases, + registry: args.Registry, + logger: args.Logger.With("collector", TableStatsCollector), + running: &atomic.Bool{}, + }, nil +} + +func (c *TableStats) Name() string { + return TableStatsCollector +} + +func (c *TableStats) Start(_ context.Context) error { + if err := c.registry.Register(c); err != nil { + return err + } + c.running.Store(true) + return nil +} + +func (c *TableStats) Stopped() bool { + return !c.running.Load() +} + +func (c *TableStats) Stop() { + c.registry.Unregister(c) + c.running.Store(false) +} + +// Describe implements prometheus.Collector. +func (c *TableStats) Describe(ch chan<- *prometheus.Desc) { + ch <- tableScanStatsSeqScanDesc + ch <- tableScanStatsIdxScanDesc + ch <- tableStatsRowCountDesc +} + +// Collect implements prometheus.Collector. It runs synchronously at scrape +// time, fanning out to every database the connection can reach. +func (c *TableStats) Collect(ch chan<- prometheus.Metric) { + ctx := context.Background() + + databases, err := discoverDatabases(ctx, c.initialConnection, c.excludeDatabases) + if err != nil { + c.logger.Error("failed to discover databases", "err", err) + return + } + + for _, dbName := range databases { + conn, closeConn, err := connectToDatabase(c.dbDSN, dbName, c.dbConnectionFactory, c.initialConnection) + if err != nil { + c.logger.Error("failed to connect to database", "datname", dbName, "err", err) + continue + } + + c.collectTableScanStats(ctx, dbName, conn, ch) + + closeConn() + } +} + +func (c *TableStats) collectTableScanStats(ctx context.Context, dbName string, conn *sql.DB, ch chan<- prometheus.Metric) { + rows, err := conn.QueryContext(ctx, selectTableScanStats) + if err != nil { + c.logger.Error("failed to query pg_stat_user_tables", "datname", dbName, "err", err) + return + } + defer rows.Close() + + for rows.Next() { + var schemaname, relname string + var seqScan, idxScan, rowCount sql.NullInt64 + + if err := rows.Scan(&schemaname, &relname, &seqScan, &idxScan, &rowCount); err != nil { + c.logger.Error("failed to scan pg_stat_user_tables row", "datname", dbName, "err", err) + return + } + + ch <- prometheus.MustNewConstMetric(tableScanStatsSeqScanDesc, prometheus.CounterValue, float64(seqScan.Int64), dbName, schemaname, relname) + ch <- prometheus.MustNewConstMetric(tableScanStatsIdxScanDesc, prometheus.CounterValue, float64(idxScan.Int64), dbName, schemaname, relname) + ch <- prometheus.MustNewConstMetric(tableStatsRowCountDesc, prometheus.GaugeValue, float64(rowCount.Int64), dbName, schemaname, relname) + } + + if err := rows.Err(); err != nil { + c.logger.Error("error iterating pg_stat_user_tables rows", "datname", dbName, "err", err) + } +} diff --git a/internal/component/database_observability/postgres/collector/table_stats_test.go b/internal/component/database_observability/postgres/collector/table_stats_test.go new file mode 100644 index 00000000000..3c318163fdd --- /dev/null +++ b/internal/component/database_observability/postgres/collector/table_stats_test.go @@ -0,0 +1,62 @@ +package collector + +import ( + "database/sql" + "fmt" + "strings" + "testing" + + sqlmock "github.com/DATA-DOG/go-sqlmock" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + + "github.com/grafana/alloy/internal/util" +) + +func TestTableStats(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual)) + require.NoError(t, err) + defer db.Close() + + registry := prometheus.NewRegistry() + + c, err := NewTableStats(TableStatsArguments{ + DB: db, + DSN: "postgres://user:pass@localhost:5432/books_store", + ExcludeDatabases: nil, + Registry: registry, + Logger: util.TestAlloyLogger(t).Slog(), + dbConnectionFactory: func(dsn string) (*sql.DB, error) { + return db, nil + }, + }) + require.NoError(t, err) + + require.NoError(t, c.Start(t.Context())) + defer c.Stop() + + mock.ExpectQuery(fmt.Sprintf(selectAllDatabases, exclusionClause)).WithoutArgs().RowsWillBeClosed(). + WillReturnRows(sqlmock.NewRows([]string{"datname"}).AddRow("books_store")) + + mock.ExpectQuery(selectTableScanStats).WithoutArgs().RowsWillBeClosed(). + WillReturnRows( + sqlmock.NewRows([]string{"schemaname", "relname", "seq_scan", "idx_scan", "n_live_tup"}). + AddRow("public", "gen_adjectives", 37154, 0, 500), + ) + + expected := ` + # HELP database_observability_pg_table_stats_idx_scan_total Number of index scans initiated on this table + # TYPE database_observability_pg_table_stats_idx_scan_total counter + database_observability_pg_table_stats_idx_scan_total{datname="books_store",relname="gen_adjectives",schemaname="public"} 0 + # HELP database_observability_pg_table_stats_row_count Estimated number of live rows in this table + # TYPE database_observability_pg_table_stats_row_count gauge + database_observability_pg_table_stats_row_count{datname="books_store",relname="gen_adjectives",schemaname="public"} 500 + # HELP database_observability_pg_table_stats_seq_scan_total Number of sequential scans initiated on this table + # TYPE database_observability_pg_table_stats_seq_scan_total counter + database_observability_pg_table_stats_seq_scan_total{datname="books_store",relname="gen_adjectives",schemaname="public"} 37154 +` + + require.NoError(t, testutil.CollectAndCompare(registry, strings.NewReader(expected))) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/internal/component/database_observability/postgres/component.go b/internal/component/database_observability/postgres/component.go index 7ea6a854c78..fb0307567fd 100644 --- a/internal/component/database_observability/postgres/component.go +++ b/internal/component/database_observability/postgres/component.go @@ -522,6 +522,8 @@ func enableOrDisableCollectors(a Arguments) map[string]bool { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, } for _, disabled := range a.DisableCollectors { @@ -674,6 +676,42 @@ func (c *Component) startCollectors(systemID string, engineVersion string, cloud c.instance.collectors = append(c.instance.collectors, epCollector) } + if collectors[collector.TableStatsCollector] { + tsCollector, err := collector.NewTableStats(collector.TableStatsArguments{ + DB: c.instance.dbConnection, + DSN: string(c.args.DataSourceName), + ExcludeDatabases: c.args.ExcludeDatabases, + Registry: c.instance.registry, + Logger: c.opts.Logger, + }) + if err != nil { + logStartError(collector.TableStatsCollector, "create", err) + } else { + if err := tsCollector.Start(context.Background()); err != nil { + logStartError(collector.TableStatsCollector, "start", err) + } + c.instance.collectors = append(c.instance.collectors, tsCollector) + } + } + + if collectors[collector.IndexStatsCollector] { + isCollector, err := collector.NewIndexStats(collector.IndexStatsArguments{ + DB: c.instance.dbConnection, + DSN: string(c.args.DataSourceName), + ExcludeDatabases: c.args.ExcludeDatabases, + Registry: c.instance.registry, + Logger: c.opts.Logger, + }) + if err != nil { + logStartError(collector.IndexStatsCollector, "create", err) + } else { + if err := isCollector.Start(context.Background()); err != nil { + logStartError(collector.IndexStatsCollector, "start", err) + } + c.instance.collectors = append(c.instance.collectors, isCollector) + } + } + // HealthCheck collector is always enabled hcCollector, err := collector.NewHealthCheck(collector.HealthCheckArguments{ DB: c.instance.dbConnection, diff --git a/internal/component/database_observability/postgres/component_test.go b/internal/component/database_observability/postgres/component_test.go index e736d8b5e7b..c818ff61251 100644 --- a/internal/component/database_observability/postgres/component_test.go +++ b/internal/component/database_observability/postgres/component_test.go @@ -117,6 +117,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -139,6 +141,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -161,6 +165,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -184,6 +190,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -207,6 +215,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -229,6 +239,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -251,6 +263,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -273,6 +287,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: true, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) @@ -295,6 +311,8 @@ func Test_enableOrDisableCollectors(t *testing.T) { collector.QuerySamplesCollector: false, collector.SchemaDetailsCollector: true, collector.ExplainPlanCollector: true, + collector.TableStatsCollector: false, + collector.IndexStatsCollector: false, }, actualCollectors) }) }