Merge branch 'main' into mn/mysql-health-check-exclude-schemas

Author: matthewnolf

internal/component/database_observability/mysql/collector/query_samples.go

@@ -27,9 +27,6 @@ const (
 	cpuTimeField              = `, statements.CPU_TIME`
 	maxControlledMemoryField  = `, statements.MAX_CONTROLLED_MEMORY`
 	maxTotalMemoryField       = `, statements.MAX_TOTAL_MEMORY`
-	sqlTextField              = `, statements.SQL_TEXT`
-	sqlTextNotNullClause      = ` AND statements.SQL_TEXT IS NOT NULL`
-	digestTextNotNullClause   = ` AND statements.DIGEST_TEXT IS NOT NULL`
 	endOfTimeline             = ` AND statements.TIMER_END > ? AND statements.TIMER_END <= ?`
 	beginningAndEndOfTimeline = ` AND statements.TIMER_END > ? OR statements.TIMER_END <= ?`
 )
@@ -49,6 +46,7 @@ SELECT
 	statements.EVENT_ID,
 	statements.END_EVENT_ID,
 	statements.DIGEST,
+	statements.SQL_TEXT,
 	statements.TIMER_END,
 	statements.TIMER_WAIT,
 	statements.ROWS_EXAMINED,
@@ -76,8 +74,9 @@ LEFT JOIN
 	ON statements.THREAD_ID = threads.THREAD_ID
 WHERE
 	statements.DIGEST IS NOT NULL
+	AND statements.SQL_TEXT IS NOT NULL
 	AND statements.CURRENT_SCHEMA NOT IN %s
-	%s %s %s`
+	%s %s`
 
 const updateSetupConsumers = `
 	UPDATE performance_schema.setup_consumers
@@ -246,15 +245,6 @@ func (c *QuerySamples) fetchQuerySamples(ctx context.Context) error {
 
 	timerClause, limit := c.determineTimerClauseAndLimit(uptime)
 
-	var textField, textNotNullClause string
-	if c.disableQueryRedaction {
-		textField = sqlTextField
-		textNotNullClause = sqlTextNotNullClause
-	} else {
-		textField = ""
-		textNotNullClause = digestTextNotNullClause
-	}
-
 	excludedSchemasClause := buildExcludedSchemasClause(c.excludeSchemas)
 
 	var waitDurationClause string
@@ -269,13 +259,12 @@ func (c *QuerySamples) fetchQuerySamples(ctx context.Context) error {
 
 	query := ""
 	if semver.MustParseRange("<8.0.28")(c.engineVersion) {
-		query = fmt.Sprintf(selectQuerySamples, textField, waitDurationClause, excludedSchemasClause, textNotNullClause, sampleDurationClause, timerClause)
+		query = fmt.Sprintf(selectQuerySamples, "", waitDurationClause, excludedSchemasClause, sampleDurationClause, timerClause)
 	} else if semver.MustParseRange("<8.0.31")(c.engineVersion) {
-		additionalFields := cpuTimeField + textField
-		query = fmt.Sprintf(selectQuerySamples, additionalFields, waitDurationClause, excludedSchemasClause, textNotNullClause, sampleDurationClause, timerClause)
+		query = fmt.Sprintf(selectQuerySamples, cpuTimeField, waitDurationClause, excludedSchemasClause, sampleDurationClause, timerClause)
 	} else {
-		additionalFields := cpuTimeField + maxControlledMemoryField + maxTotalMemoryField + textField
-		query = fmt.Sprintf(selectQuerySamples, additionalFields, waitDurationClause, excludedSchemasClause, textNotNullClause, sampleDurationClause, timerClause)
+		additionalFields := cpuTimeField + maxControlledMemoryField + maxTotalMemoryField
+		query = fmt.Sprintf(selectQuerySamples, additionalFields, waitDurationClause, excludedSchemasClause, sampleDurationClause, timerClause)
 	}
 
 	rs, err := c.dbConnection.QueryContext(ctx, query, c.timerBookmark, limit)
@@ -336,6 +325,7 @@ func (c *QuerySamples) fetchQuerySamples(ctx context.Context) error {
 			&row.StatementEventID,
 			&row.StatementEndEventID,
 			&row.Digest,
+			&row.SQLText,
 			&row.TimerEndPicoseconds,
 			&row.ElapsedTimePicoseconds,
 			&row.RowsExamined,
@@ -360,10 +350,6 @@ func (c *QuerySamples) fetchQuerySamples(ctx context.Context) error {
 			scanArgs = append(scanArgs, &row.MaxTotalMemory)
 		}
 
-		if c.disableQueryRedaction {
-			scanArgs = append(scanArgs, &row.SQLText)
-		}
-
 		err := rs.Scan(scanArgs...)
 		if err != nil {
 			level.Error(c.logger).Log("msg", "failed to scan history table samples", "err", err)
@@ -379,6 +365,7 @@ func (c *QuerySamples) fetchQuerySamples(ctx context.Context) error {
 		row.TimestampMilliseconds = calculateWallTime(serverStartTime, row.TimerEndPicoseconds.Float64, uptime)
 		cpuTime := picosecondsToMilliseconds(row.CPUTime)
 		elapsedTime := picosecondsToMilliseconds(row.ElapsedTimePicoseconds.Float64)
+		traceParent := tryExtractTraceParent(row.SQLText.String)
 
 		logMessage := fmt.Sprintf(
 			`schema="%s" user="%s" client_host="%s" thread_id="%s" event_id="%s" end_event_id="%s" digest="%s" rows_examined="%d" rows_sent="%d" rows_affected="%d" errors="%d" max_controlled_memory="%db" max_total_memory="%db" cpu_time="%fms" elapsed_time="%fms" elapsed_time_ms="%fms"`,
@@ -395,6 +382,9 @@ func (c *QuerySamples) fetchQuerySamples(ctx context.Context) error {
 			elapsedTime,
 			elapsedTime,
 		)
+		if traceParent != "" {
+			logMessage += fmt.Sprintf(` traceparent="%s"`, traceParent)
+		}
 		if c.disableQueryRedaction && row.SQLText.Valid {
 			logMessage += fmt.Sprintf(` sql_text="%s"`, row.SQLText.String)
 		}
@@ -526,3 +516,54 @@ func classifyMySQLWaitEventType(waitEventName string) string {
 	}
 	return "Other Wait"
 }
+
+// tryExtractTraceParent attempts to extract a W3C traceparent value added at the end of SQL text as a trailing
+// block comment, e.g. "/*traceparent='00-<traceid>-<spanid>-<flags>'*/".
+// It returns the traceparent string when matched, otherwise an empty string.
+func tryExtractTraceParent(sqlText string) string {
+	if strings.HasSuffix(sqlText, "...") {
+		return ""
+	}
+
+	// Find the last comment: strip out /* and */
+	start := strings.LastIndex(sqlText, "/*")
+	if start < 0 {
+		return ""
+	}
+	body := sqlText[start+2:]
+	end := strings.Index(body, "*/")
+	if end < 0 {
+		return ""
+	}
+
+	body = body[:end]
+	body = strings.TrimSpace(body)
+	if body == "" {
+		return ""
+	}
+
+	// Split the comment by comma into key value pairs
+	pairs := strings.Split(body, ",")
+	for _, pair := range pairs {
+		pair = strings.TrimSpace(pair)
+		key, val, ok := strings.Cut(pair, "=")
+		if !ok {
+			continue
+		}
+
+		if !strings.EqualFold(strings.TrimSpace(key), "traceparent") {
+			continue
+		}
+
+		// SQL unescape: trim ' or " at beginning and end of value
+		if strings.HasPrefix(val, "'") || strings.HasPrefix(val, `"`) {
+			quote := string(val[0])
+			val = strings.TrimPrefix(val, quote)
+			val = strings.TrimSuffix(val, quote)
+		}
+
+		return strings.TrimSpace(val)
+	}
+
+	return ""
+}