opensource/alloy
1
0
Fork 0
mirror of https://github.com/grafana/alloy.git synced 2025-11-05 13:28:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Use the database/sql rows API correctly

Browse Source
This commit is contained in:
Ryan Geyer
2025-11-04 10:52:00 -08:00
parent 66c3a7c596
commit a25791fd68
2 changed files with 17 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
internal/component/database_observability/mysql/collector/explain_plans.go
View File

@@ -506,11 +506,6 @@ func (c *ExplainPlans) populateQueryCache(ctx context.Context) error {
// Populate cache
for rs.Next() {
if err := rs.Err(); err != nil {
level.Error(c.logger).Log("msg", "failed to iterate rs digests for explain plans", "err", err)
return err
}
var schemaName, digest, queryText string
var ls time.Time
if err = rs.Scan(&schemaName, &digest, &queryText, &ls); err != nil {
@@ -532,6 +527,12 @@ func (c *ExplainPlans) populateQueryCache(ctx context.Context) error {
c.lastSeen = ls
}
}
if err := rs.Err(); err != nil {
level.Error(c.logger).Log("msg", "failed to iterate digest rows for explain plans", "err", err)
return err
}
// Calculate batch size based on current cache size
c.currentBatchSize = int(math.Ceil(float64(len(c.queryCache)) * c.perScrapeRatio))
level.Debug(c.logger).Log("msg", "populated query cache", "count", len(c.queryCache), "batch_size", c.currentBatchSize)

29
internal/component/database_observability/postgres/collector/explain_plan.go
View File

@@ -316,17 +316,12 @@ func (c *ExplainPlan) populateQueryCache(ctx context.Context) error {
if version17Plus {
selectStatement = fmt.Sprintf(selectQueriesForExplainPlanTemplate, "s.stats_since")
} else {
statReset, err := c.dbConnection.QueryContext(ctx, "SELECT stats_reset FROM pg_stat_statements_info")
if err != nil {
level.Error(c.logger).Log("msg", "failed to fetch stats reset time for explain plans", "err", err)
return err
statReset := c.dbConnection.QueryRowContext(ctx, "SELECT stats_reset FROM pg_stat_statements_info")
if err := statReset.Err(); err != nil {
return fmt.Errorf("failed to fetch stats reset time for explain plans: %w", err)
}
defer statReset.Close()
if statReset.Next() {
if err := statReset.Scan(&resetTS); err != nil {
level.Error(c.logger).Log("msg", "failed to scan stats reset time for explain plans", "err", err)
return err
}
if err := statReset.Scan(&resetTS); err != nil {
return fmt.Errorf("failed to scan stats reset time for explain plans: %w", err)
}
selectStatement = fmt.Sprintf(selectQueriesForExplainPlanTemplate, "NOW() AT TIME ZONE 'UTC' AS stats_since")
}
@@ -334,22 +329,16 @@ func (c *ExplainPlan) populateQueryCache(ctx context.Context) error {
rs, err := c.dbConnection.QueryContext(ctx, selectStatement)
if err != nil {
level.Error(c.logger).Log("msg", "failed to fetch digests for explain plans", "err", err)
return err
return fmt.Errorf("failed to fetch digests for explain plans: %w", err)
}
defer rs.Close()
for rs.Next() {
if err := rs.Err(); err != nil {
level.Error(c.logger).Log("msg", "failed to iterate rs digests for explain plans", "err", err)
return err
}
var datname, queryId, query string
var calls int64
var ls time.Time
if err := rs.Scan(&datname, &queryId, &query, &calls, &ls); err != nil {
level.Error(c.logger).Log("msg", "failed to scan query for explain plan", "err", err)
return err
return fmt.Errorf("failed to scan query for explain plan: %w", err)
}
if slices.ContainsFunc(c.excludeSchemas, func(schema string) bool {
@@ -378,6 +367,10 @@ func (c *ExplainPlan) populateQueryCache(ctx context.Context) error {
}
}
if err := rs.Err(); err != nil {
return fmt.Errorf("failed to iterate query rows for explain plans: %w", err)
}
c.currentBatchSize = int(math.Ceil(float64(len(c.queryCache)) * c.perScrapeRatio))
level.Debug(c.logger).Log("msg", "populated query cache", "count", len(c.queryCache), "batch_size", c.currentBatchSize)
return nil