Skip to content

fix(database_observability.postgres): Use of defer in explain_plans collector #6819

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cristiangreco wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(database_observability.postgres): Use of defer in explain_plans collector #6819

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a4dd649
fix(database_observability.postgres): explain_plans collector for pos…
cristiangreco Aug 3, 2026
c764b14
review feedback
cristiangreco Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
203 changes: 101 additions & 102 deletions internal/component/database_observability/postgres/collector/explain_plans.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -497,8 +497,6 @@ func (c *ExplainPlans) fetchExplainPlans(ctx context.Context) error {

processedCount := 0
for _, qi := range c.queryCache {
generatedAt := time.Now().Format(time.RFC3339)
nonRecoverableFailureOccurred := false
if c.isThrottled(qi.uniqueKey) {
c.markFinished(qi)
delete(c.queryCache, qi.uniqueKey)
Expand All @@ -507,126 +505,127 @@ func (c *ExplainPlans) fetchExplainPlans(ctx context.Context) error {
if processedCount >= c.currentBatchSize {
break
}
logger := c.logger.With("query_id", qi.queryId)

defer func(nonRecoverableFailureOccurred *bool) {
if *nonRecoverableFailureOccurred {
c.queryDenylist[qi.uniqueKey] = struct{}{}
} else {
c.markFinished(qi)
}
delete(c.queryCache, qi.uniqueKey)
processedCount++
}(&nonRecoverableFailureOccurred)

if strings.HasSuffix(qi.queryText, "...") {
err := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultSkipped,
"query is truncated",
nil,
)
if err != nil {
c.logger.Error("failed to send truncated query skip explain plan output", "err", err)
}
continue
if c.processExplainPlan(ctx, qi) {
Comment thread
cristiangreco marked this conversation as resolved.
c.queryDenylist[qi.uniqueKey] = struct{}{}
} else {
c.markFinished(qi)
}
delete(c.queryCache, qi.uniqueKey)
processedCount++
}

containsReservedWord, err := database_observability.ContainsReservedKeywords(qi.queryText, database_observability.ExplainReservedWordDenyList, sqllexer.DBMSPostgres)
return nil
}

// processExplainPlan processes a single query from the cache.
// Returns true if the query encountered a non-recoverable failure and should be denylisted.
func (c *ExplainPlans) processExplainPlan(ctx context.Context, qi *queryInfo) bool {
generatedAt := time.Now().Format(time.RFC3339)
logger := c.logger.With("query_id", qi.queryId)

if strings.HasSuffix(qi.queryText, "...") {
err := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultSkipped,
"query is truncated",
nil,
)
if err != nil {
logger.Error("failed to check for reserved keywords", "err", err)
err := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultError,
fmt.Sprintf("failed to check for reserved keywords: %s", err.Error()),
nil,
)
if err != nil {
c.logger.Error("failed to send reserved keyword check error explain plan output", "err", err)
}
continue
c.logger.Error("failed to send truncated query skip explain plan output", "err", err)
}
return false
}

if containsReservedWord {
err := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultSkipped,
"query contains reserved word",
nil,
)
if err != nil {
c.logger.Error("failed to send reserved keyword check error explain plan output", "err", err)
}
continue
containsReservedWord, err := database_observability.ContainsReservedKeywords(qi.queryText, database_observability.ExplainReservedWordDenyList, sqllexer.DBMSPostgres)
if err != nil {
logger.Error("failed to check for reserved keywords", "err", err)
sendErr := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultError,
fmt.Sprintf("failed to check for reserved keywords: %s", err.Error()),
nil,
)
if sendErr != nil {
logger.Error("failed to send reserved keyword check error explain plan output", "err", sendErr)
}
return false
}

if containsReservedWord {
sendErr := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultSkipped,
"query contains reserved word",
nil,
)
if sendErr != nil {
logger.Error("failed to send reserved keyword skip explain plan output", "err", sendErr)
}
return false
}

logger = logger.With("datname", qi.datname)
logger = logger.With("datname", qi.datname)

byteExplainPlanJSON, err := c.fetchExplainPlanJSON(ctx, *qi)
if err != nil {
logger.Debug("failed to fetch explain plan json bytes", "err", err)
for _, code := range unrecoverablePostgresSQLErrors {
if strings.Contains(err.Error(), code) {
nonRecoverableFailureOccurred = true
break
}
byteExplainPlanJSON, err := c.fetchExplainPlanJSON(ctx, *qi)
if err != nil {
logger.Debug("failed to fetch explain plan json bytes", "err", err)
for _, code := range unrecoverablePostgresSQLErrors {
if strings.Contains(err.Error(), code) {
return true
}
continue
}
return false
}

if len(byteExplainPlanJSON) == 0 {
logger.Error("explain plan json bytes is empty")
nonRecoverableFailureOccurred = true
continue
}
if len(byteExplainPlanJSON) == 0 {
logger.Error("explain plan json bytes is empty")
return true
}

if !utf8.Valid(byteExplainPlanJSON) {
logger.Error("explain plan json bytes is not valid UTF-8")
nonRecoverableFailureOccurred = true
continue
}
if !utf8.Valid(byteExplainPlanJSON) {
logger.Error("explain plan json bytes is not valid UTF-8")
return true
}

redactedByteExplainPlanJSON := database_observability.RedactSql(string(byteExplainPlanJSON))
redactedByteExplainPlanJSON := database_observability.RedactSql(string(byteExplainPlanJSON))

logger.Debug("db native explain plan", "db_native_explain_plan", base64.StdEncoding.EncodeToString([]byte(redactedByteExplainPlanJSON)))
logger.Debug("db native explain plan", "db_native_explain_plan", base64.StdEncoding.EncodeToString([]byte(redactedByteExplainPlanJSON)))

explainPlanOutput, genErr := newExplainPlanOutput(byteExplainPlanJSON)
explainPlanOutputJSON, err := json.Marshal(explainPlanOutput)
if err != nil {
logger.Error("failed to marshal explain plan output", "err", err)
nonRecoverableFailureOccurred = true
continue
}
explainPlanOutput, genErr := newExplainPlanOutput(byteExplainPlanJSON)
explainPlanOutputJSON, err := json.Marshal(explainPlanOutput)
if err != nil {
logger.Error("failed to marshal explain plan output", "err", err)
return true
}

if genErr != nil {
logger.Error(
"failed to create explain plan output",
"incomplete_explain_plan", base64.StdEncoding.EncodeToString(explainPlanOutputJSON),
"err", genErr,
)
nonRecoverableFailureOccurred = true
continue
}
if genErr != nil {
logger.Error(
"failed to create explain plan output",
"incomplete_explain_plan", base64.StdEncoding.EncodeToString(explainPlanOutputJSON),
"err", genErr,
)
return true
}

if err := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultSuccess,
"",
explainPlanOutput,
); err != nil {
c.logger.Error("failed to send explain plan output", "err", err)
}
if err := c.sendExplainPlansOutput(
qi.datname,
qi.queryId,
generatedAt,
database_observability.ExplainProcessingResultSuccess,
"",
explainPlanOutput,
); err != nil {
c.logger.Error("failed to send explain plan output", "err", err)
}

return nil
return false
}

// postgresPreparedStatementParamCount returns N for EXECUTE, where N is the highest
Expand Down
35 changes: 35 additions & 0 deletions internal/component/database_observability/postgres/collector/explain_plans_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2914,6 +2914,41 @@ func TestPlanNode_ToExplainPlanOutputNode(t *testing.T) {
assert.Equal(t, database_observability.ExplainPlanJoinAlgorithmHash, *result.Details.JoinAlgorithm)
}

func TestExplainPlanBatchSizeLimitsProcessing(t *testing.T) {
lokiClient := loki.NewCollectingHandler()
defer lokiClient.Stop()

c, err := NewExplainPlan(ExplainPlansArguments{
Logger: logging.NewSlogNop(),
ScrapeInterval: time.Second,
PerScrapeRatio: 1,
EntryHandler: lokiClient,
DBVersion: "17.0",
})
require.NoError(t, err)

c.queryCache = map[string]*queryInfo{
explainPlanQueryKey("db", "1"): newQueryInfo("db", "1", "select * from table_1 where ...", 1, time.Now()),
explainPlanQueryKey("db", "2"): newQueryInfo("db", "2", "select * from table_2 where ...", 1, time.Now()),
explainPlanQueryKey("db", "3"): newQueryInfo("db", "3", "select * from table_3 where ...", 1, time.Now()),
explainPlanQueryKey("db", "4"): newQueryInfo("db", "4", "select * from table_4 where ...", 1, time.Now()),
}
c.currentBatchSize = 2

require.NoError(t, c.fetchExplainPlans(t.Context()))
require.Len(t, c.queryCache, 2, "batch size limit should leave unprocessed items in cache")
require.Len(t, c.finishedQueryCache, 2)
require.Empty(t, c.queryDenylist)
require.Eventually(
t,
func() bool { return len(lokiClient.Received()) == 2 },
5*time.Second,
10*time.Millisecond,
"expected exactly 2 Loki entries, got %d",
len(lokiClient.Received()),
)
}

func TestExplainPlanFetchExplainPlans(t *testing.T) {
db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
require.NoError(t, err)
Expand Down
Loading