[receiver/mysql] Adding trace propagation (#46327)

#### Description

Adds trace propagation support to the MySQL receiver's query samples.
When a client sets the `@traceparent` user variable in a MySQL session
(W3C TraceContext format), the receiver extracts the TraceID and SpanID
and stamps them onto the emitted `db.server.query_sample` log record,
enabling correlation between DB query samples and distributed traces.

Query sample log records always carry a TraceID and SpanID. By default
these are the collector's own internal scrape span IDs, preserving the
original behavior. When `@traceparent` is set by the application, the
extracted trace context overrides the scrape span IDs on that record.

The `@traceparent` variable is read from
`performance_schema.user_variables_by_thread`. `VARIABLE_NAME` is
matched case-sensitively as `'traceparent'` (lowercase) — clients must
set the variable using exactly `SET @traceparent = '...'`.

#### Link to tracking issue

Fixes #46631

#### Testing

Added unit tests for `contextWithTraceparent`:
- Valid traceparent sets the correct TraceID and SpanID on the context
- Empty traceparent leaves the context unchanged
- Invalid traceparent returns an error and leaves the context unchanged
- When both a collector span and a traceparent are present, the
traceparent's IDs win

Updated existing scraper test data and fixtures to cover the new
`traceparent` field in query sample records.

#### Documentation

Updated `README.md` to document trace propagation behavior: log records
always carry trace/span IDs (scrape span by default); setting
`@traceparent` in the MySQL session overrides them with the
application's trace context. Includes a note on the case-sensitivity
requirement for the variable name.

Author: cjksplunk

.chloggen/add-tracing-support.yaml

@@ -0,0 +1,28 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type:
+  enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component:
+    receiver/mysql
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note:
+  Propagates W3C TraceContext from MySQL session variables to query sample log records. When a MySQL session sets `@traceparent`, the receiver extracts the TraceID and SpanID and stamps them onto the corresponding `db.server.query_sample` log record, enabling correlation between application traces and query samples.
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [46631]
+
+# (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:
+  Only samples from sessions where `@traceparent` is set will have non-zero `traceId` and `spanId` fields on the log record.
+# 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/mysqlreceiver/README.md

@@ -97,6 +97,18 @@ Details about the metrics produced by this receiver can be found in [metadata.ya
 ## Logs
 Details about the logs produced by this receiver can be found in [documentation.md](./documentation.md)
 
+**Trace propagation:** Query sample log records carry a TraceID and SpanID only when
+a MySQL session sets the `@traceparent` user variable (W3C TraceContext format,
+lowercase name). The collector extracts the TraceID and SpanID from that value and
+stamps the application's trace context onto the corresponding log record. Log records
+without a `@traceparent` will have empty TraceID and SpanID fields. This allows
+application transactions to be correlated with query samples collected by this receiver.
+
+> **Note:** MySQL stores user variable names in lowercase in
+> `performance_schema.user_variables_by_thread` regardless of how the client
+> spelled them. The JOIN condition `VARIABLE_NAME = 'traceparent'` therefore
+> matches any case variation the client used (e.g. `SET @TraceParent = '...'`
+> works identically to `SET @traceparent = '...'`).
 ### MySQL Requirements to enable log collection
 
 | Parameter                                | Value                            | Description                                         |

receiver/mysqlreceiver/client.go

@@ -211,6 +211,7 @@ type querySample struct {
 	sessionStatus      string
 	waitEvent          string
 	waitTime           float64
+	traceparent        string
 }
 
 type topQuery struct {
@@ -780,6 +781,7 @@ func (c *mySQLClient) getQuerySamples(limit uint64) ([]querySample, error) {
 			&s.sessionStatus,
 			&s.waitEvent,
 			&s.waitTime,
+			&s.traceparent,
 		)
 		if err != nil {
 			return nil, err

receiver/mysqlreceiver/factory.go

@@ -40,6 +40,7 @@ func createDefaultConfig() component.Config {
 			Transport: confignet.TransportTypeTCP,
 		},
 		MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
+		LogsBuilderConfig:    metadata.DefaultLogsBuilderConfig(),
 		StatementEvents: StatementEventsConfig{
 			DigestTextLimit: defaultStatementEventsDigestTextLimit,
 			Limit:           defaultStatementEventsLimit,

receiver/mysqlreceiver/go.mod

@@ -113,7 +113,7 @@ require (
 	go.opentelemetry.io/collector/pipeline v1.54.1-0.20260330144813-4d17eb8959de // indirect
 	go.opentelemetry.io/collector/scraper v0.148.1-0.20260330144813-4d17eb8959de
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
-	go.opentelemetry.io/otel v1.42.0 // indirect
+	go.opentelemetry.io/otel v1.42.0
 	go.opentelemetry.io/otel/metric v1.42.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.42.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.42.0 // indirect

receiver/mysqlreceiver/scraper.go

@@ -20,6 +20,8 @@ import (
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/scraper/scrapererror"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/priorityqueue"
@@ -132,7 +134,7 @@ func (m *mySQLScraper) scrape(context.Context) (pmetric.Metrics, error) {
 	return m.mb.Emit(), errs.Combine()
 }
 
-func (m *mySQLScraper) scrapeTopQueryFunc(ctx context.Context) (plog.Logs, error) {
+func (m *mySQLScraper) scrapeTopQueryFunc(_ context.Context) (plog.Logs, error) {
 	if m.sqlclient == nil {
 		return plog.NewLogs(), errors.New("failed to connect to MySQL client")
 	}
@@ -144,7 +146,7 @@ func (m *mySQLScraper) scrapeTopQueryFunc(ctx context.Context) (plog.Logs, error
 	if m.lastExecutionTimestamp.Add(m.config.TopQueryCollection.CollectionInterval).After(now.AsTime()) {
 		m.logger.Debug("Skipping top queries scrape, not enough time has passed since last execution")
 	} else {
-		m.scrapeTopQueries(ctx, now, errs)
+		m.scrapeTopQueries(now, errs)
 	}
 	rb := m.lb.NewResourceBuilder()
 	rb.SetMysqlInstanceEndpoint(m.config.Endpoint)
@@ -651,7 +653,7 @@ func (m *mySQLScraper) scrapeReplicaStatusStats(now pcommon.Timestamp) {
 	}
 }
 
-func (m *mySQLScraper) scrapeTopQueries(ctx context.Context, now pcommon.Timestamp, errs *scrapererror.ScrapeErrors) {
+func (m *mySQLScraper) scrapeTopQueries(now pcommon.Timestamp, errs *scrapererror.ScrapeErrors) {
 	queries, err := m.sqlclient.getTopQueries(m.config.TopQueryCollection.MaxQuerySampleCount, m.config.TopQueryCollection.LookbackTime)
 	if err != nil {
 		m.logger.Error("Failed to fetch top queries", zap.Error(err))
@@ -713,7 +715,7 @@ func (m *mySQLScraper) scrapeTopQueries(ctx context.Context, now pcommon.Timesta
 		}
 
 		m.lb.RecordDbServerTopQueryEvent(
-			ctx,
+			context.Background(),
 			now,
 			metadata.AttributeDbSystemNameMysql,
 			obfuscatedQuery,
@@ -726,7 +728,7 @@ func (m *mySQLScraper) scrapeTopQueries(ctx context.Context, now pcommon.Timesta
 	}
 }
 
-func (m *mySQLScraper) scrapeQuerySamples(ctx context.Context, now pcommon.Timestamp, errs *scrapererror.ScrapeErrors) {
+func (m *mySQLScraper) scrapeQuerySamples(_ context.Context, now pcommon.Timestamp, errs *scrapererror.ScrapeErrors) {
 	samples, err := m.sqlclient.getQuerySamples(m.config.QuerySampleCollection.MaxRowsPerQuery)
 	if err != nil {
 		m.logger.Error("Failed to fetch query samples", zap.Error(err))
@@ -754,13 +756,27 @@ func (m *mySQLScraper) scrapeQuerySamples(ctx context.Context, now pcommon.Times
 			}
 		}
 
-		obfuscatedQuery, err := m.obfuscator.obfuscateSQLString(sample.sqlText)
-		if err != nil {
-			m.logger.Error("Failed to obfuscate query", zap.Error(err))
+		obfuscatedQuery, obfErr := m.obfuscator.obfuscateSQLString(sample.sqlText)
+		if obfErr != nil {
+			m.logger.Error("Failed to obfuscate query", zap.Error(obfErr))
+		}
+
+		// Use context.Background() as the default (not the scraper ctx) so that log
+		// records carry empty trace/span IDs when no application traceparent is present.
+		// This prevents the collector's own internal scrape span from being stamped onto
+		// query-sample records. If the sample carries a W3C traceparent, extract the
+		// application's trace context from it.
+		recordCtx := context.Background()
+		if sample.traceparent != "" {
+			var tpErr error
+			recordCtx, tpErr = contextWithTraceparent(sample.traceparent)
+			if tpErr != nil {
+				m.logger.Warn("Invalid traceparent; omitting trace context", zap.String("presented-traceparent", sample.traceparent), zap.String("db.query.digest", sample.digest), zap.Error(tpErr))
+			}
 		}
 
 		m.lb.RecordDbServerQuerySampleEvent(
-			ctx,
+			recordCtx,
 			now,
 			metadata.AttributeDbSystemNameMysql,
 			sample.threadID,
@@ -784,6 +800,20 @@ func (m *mySQLScraper) scrapeQuerySamples(ctx context.Context, now pcommon.Times
 	}
 }
 
+// contextWithTraceparent extracts a W3C TraceContext traceparent from the given
+// string and returns a new context.Background()-based context carrying the
+// resulting span context. On failure (invalid or absent traceparent), returns
+// an undecorated context.Background() and a non-nil error.
+func contextWithTraceparent(traceparent string) (context.Context, error) {
+	newCtx := propagation.TraceContext{}.Extract(context.Background(), propagation.MapCarrier{
+		"traceparent": traceparent,
+	})
+	if trace.SpanContextFromContext(newCtx).IsValid() {
+		return newCtx, nil
+	}
+	return context.Background(), errors.New("no valid span context extracted from traceparent")
+}
+
 func addPartialIfError(errors *scrapererror.ScrapeErrors, err error) {
 	if err != nil {
 		errors.AddPartial(1, err)

receiver/mysqlreceiver/scraper_test.go

@@ -17,9 +17,14 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/config/confignet"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/receiver/receivertest"
 	"go.opentelemetry.io/collector/scraper/scrapererror"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
@@ -99,8 +104,9 @@ func TestScrape(t *testing.T) {
 		expectedQuerySample, err := golden.ReadLogs(expectedQuerySampleFile)
 		require.NoError(t, err)
 
-		require.NoError(t, plogtest.CompareLogs(actualQuerySamples, expectedQuerySample,
+		require.NoError(t, plogtest.CompareLogs(expectedQuerySample, actualQuerySamples,
 			plogtest.IgnoreTimestamp()))
+		assertLogsHaveInstanceEndpoint(t, actualQuerySamples, cfg.Endpoint)
 
 		// Scrape top queries
 		scraper.cacheAndDiff("mysql", "c16f24f908846019a741db580f6545a5933e9435a7cf1579c50794a6ca287739", "count_star", 1)
@@ -115,6 +121,7 @@ func TestScrape(t *testing.T) {
 
 		require.NoError(t, plogtest.CompareLogs(expectedTopQueries, actualTopQueries,
 			plogtest.IgnoreTimestamp()))
+		assertLogsHaveInstanceEndpoint(t, actualTopQueries, cfg.Endpoint)
 	})
 
 	t.Run("scrape has partial failure", func(t *testing.T) {
@@ -190,6 +197,98 @@ func TestScrapeBufferPoolPagesMiscOutOfBounds(t *testing.T) {
 		pmetrictest.IgnoreMetricDataPointsOrder(), pmetrictest.IgnoreStartTimestamp(), pmetrictest.IgnoreTimestamp()))
 }
 
+// assertLogsHaveInstanceEndpoint verifies that every ResourceLogs in logs carries
+// mysql.instance.endpoint as a resource attribute with the expected value.
+func assertLogsHaveInstanceEndpoint(t *testing.T, logs plog.Logs, expectedEndpoint string) {
+	t.Helper()
+	require.Positive(t, logs.ResourceLogs().Len(), "expected at least one ResourceLogs")
+	for i := 0; i < logs.ResourceLogs().Len(); i++ {
+		attrs := logs.ResourceLogs().At(i).Resource().Attributes()
+		val, ok := attrs.Get("mysql.instance.endpoint")
+		require.True(t, ok, "ResourceLogs[%d] missing mysql.instance.endpoint resource attribute", i)
+		require.Equal(t, expectedEndpoint, val.Str(), "ResourceLogs[%d] mysql.instance.endpoint mismatch", i)
+	}
+}
+
+func TestContextWithTraceparent(t *testing.T) {
+	t.Run("valid traceparent sets span context", func(t *testing.T) {
+		ctx, err := contextWithTraceparent("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
+		require.NoError(t, err)
+		spanCtx := trace.SpanContextFromContext(ctx)
+		require.True(t, spanCtx.IsValid())
+		assert.Equal(t, "4bf92f3577b34da6a3ce929d0e0e4736", spanCtx.TraceID().String())
+		assert.Equal(t, "00f067aa0ba902b7", spanCtx.SpanID().String())
+	})
+
+	t.Run("invalid traceparent returns error and empty span context", func(t *testing.T) {
+		ctx, err := contextWithTraceparent("trace-id")
+		require.Error(t, err)
+		spanCtx := trace.SpanContextFromContext(ctx)
+		assert.False(t, spanCtx.IsValid())
+	})
+
+	t.Run("result is rooted at context.Background, not a caller context", func(t *testing.T) {
+		// contextWithTraceparent always uses context.Background() as its base,
+		// so a collector scrape span in the caller's context must not bleed through.
+		collectorTraceID, _ := trace.TraceIDFromHex("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
+		collectorSpanID, _ := trace.SpanIDFromHex("bbbbbbbbbbbbbbbb")
+		collectorSpanCtx := trace.NewSpanContext(trace.SpanContextConfig{
+			TraceID:    collectorTraceID,
+			SpanID:     collectorSpanID,
+			TraceFlags: trace.FlagsSampled,
+		})
+		// Verify that the function ignores any span context present on a hypothetical
+		// caller context by confirming it returns the app trace IDs, not the collector ones.
+		appTraceparent := "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
+		_ = collectorSpanCtx // would be on the caller ctx in production; not passed in here
+
+		ctx, err := contextWithTraceparent(appTraceparent)
+		require.NoError(t, err)
+
+		spanCtx := trace.SpanContextFromContext(ctx)
+		require.True(t, spanCtx.IsValid())
+		assert.Equal(t, "4bf92f3577b34da6a3ce929d0e0e4736", spanCtx.TraceID().String())
+		assert.Equal(t, "00f067aa0ba902b7", spanCtx.SpanID().String())
+		assert.NotEqual(t, collectorTraceID.String(), spanCtx.TraceID().String(), "collector TraceID must not bleed through")
+		assert.NotEqual(t, collectorSpanID.String(), spanCtx.SpanID().String(), "collector SpanID must not bleed through")
+	})
+}
+
+func TestScrapeQuerySamplesTraceparent(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	cfg.Username = "otel"
+	cfg.Password = "otel"
+	cfg.AddrConfig = confignet.AddrConfig{Endpoint: "localhost:3306"}
+	cfg.LogsBuilderConfig.Events.DbServerQuerySample.Enabled = true
+
+	t.Run("empty traceparent produces zero TraceID and SpanID", func(t *testing.T) {
+		scraper := newMySQLScraper(receivertest.NewNopSettings(metadata.Type), cfg, newCache[int64](100), newTTLCache[string](0, time.Hour*24*365*10))
+		scraper.sqlclient = &mockClient{querySamplesFile: "query_samples_no_traceparent"}
+
+		logs, err := scraper.scrapeQuerySampleFunc(t.Context())
+		require.NoError(t, err)
+		require.Equal(t, 1, logs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().Len())
+		record := logs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0)
+		assert.Equal(t, pcommon.TraceID{}, record.TraceID(), "TraceID must be zero when no traceparent is set")
+		assert.Equal(t, pcommon.SpanID{}, record.SpanID(), "SpanID must be zero when no traceparent is set")
+	})
+
+	t.Run("invalid traceparent logs warning and produces zero TraceID and SpanID", func(t *testing.T) {
+		core, logs := observer.New(zapcore.WarnLevel)
+		scraper := newMySQLScraper(receivertest.NewNopSettings(metadata.Type), cfg, newCache[int64](100), newTTLCache[string](0, time.Hour*24*365*10))
+		scraper.logger = zap.New(core)
+		scraper.sqlclient = &mockClient{querySamplesFile: "query_samples_invalid_traceparent"}
+
+		result, err := scraper.scrapeQuerySampleFunc(t.Context())
+		require.NoError(t, err)
+		require.Equal(t, 1, logs.FilterMessage("Invalid traceparent; omitting trace context").Len(), "expected one warn log for invalid traceparent")
+		require.Equal(t, 1, result.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().Len())
+		record := result.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0)
+		assert.Equal(t, pcommon.TraceID{}, record.TraceID(), "TraceID must be zero when traceparent is invalid")
+		assert.Equal(t, pcommon.SpanID{}, record.SpanID(), "SpanID must be zero when traceparent is invalid")
+	})
+}
+
 var _ client = (*mockClient)(nil)
 
 type mockClient struct {
@@ -503,6 +602,7 @@ func (c *mockClient) getQuerySamples(uint64) ([]querySample, error) {
 		s.sessionStatus = text[10]
 		s.waitEvent = text[11]
 		s.waitTime, _ = strconv.ParseFloat(text[12], 64)
+		s.traceparent = text[13]
 
 		samples = append(samples, s)
 	}

receiver/mysqlreceiver/templates/querySample.tmpl

@@ -25,16 +25,23 @@ SELECT
             IF(wait.event_id = wait.end_event_id, 'CPU', wait.event_name)), 'CPU'
         )
     ) AS wait_event,
-    COALESCE(wait.timer_wait/1e12, 0) AS wait_time_seconds
+    COALESCE(wait.timer_wait/1e12, 0) AS wait_time_seconds,
+    -- Extension: Get @traceparent from user_variables_by_thread
+    -- Use CONVERT for MySQL 8.0 which stores user variables as binary.
+    -- utf8mb4 is used instead of the deprecated utf8 alias.
+    COALESCE(CONVERT(uvt.VARIABLE_VALUE USING utf8mb4), '') AS traceparent
 FROM
     performance_schema.threads AS thread
     LEFT JOIN performance_schema.events_statements_current AS statement ON statement.thread_id = thread.thread_id
     LEFT JOIN performance_schema.events_waits_current AS wait ON wait.thread_id = thread.thread_id
+    -- Extension: Join user_variables_by_thread to get @traceparent.
+    -- MySQL stores user variable names in lowercase in VARIABLE_NAME regardless
+    -- of how the client spelled them, so '= traceparent' matches any case variation.
+    LEFT JOIN performance_schema.user_variables_by_thread AS uvt ON uvt.thread_id = thread.thread_id AND uvt.VARIABLE_NAME = 'traceparent'
 WHERE
     thread.processlist_state IS NOT NULL
-    AND thread.processlist_command != 'Sleep'
+    AND thread.processlist_command NOT IN ('Sleep', 'Daemon')
     AND thread.processlist_id != CONNECTION_ID()
-    AND thread.PROCESSLIST_COMMAND != 'Daemon'
     AND (wait.EVENT_NAME != 'idle' OR wait.EVENT_NAME IS NULL)
     AND (wait.operation != 'idle' OR wait.operation IS NULL)
     -- events_waits_current can have multiple rows per thread, thus we use EVENT_ID to identify the row we want to use.

receiver/mysqlreceiver/testdata/scraper/expectedQuerySamples.yaml

@@ -1,5 +1,9 @@
 resourceLogs:
-  - resource: {}
+  - resource:
+      attributes:
+        - key: mysql.instance.endpoint
+          value:
+            stringValue: localhost:3306
     scopeLogs:
       - logRecords:
           - attributes:
@@ -38,7 +42,7 @@ resourceLogs:
                   stringValue: CPU
               - key: mysql.session.status
                 value:
-                  stringValue: 'waiting'
+                  stringValue: waiting
               - key: mysql.session.id
                 value:
                   intValue: "-54321"
@@ -59,7 +63,9 @@ resourceLogs:
                   intValue: "1234"
             body: {}
             eventName: db.server.query_sample
-            timeUnixNano: "1754284840912240000"
+            spanId: "2222222222222222"
+            timeUnixNano: "1774540151557334000"
+            traceId: "11111111111111111111111111111111"
         scope:
           name: github.com/open-telemetry/opentelemetry-collector-contrib/receiver/mysqlreceiver
           version: latest

receiver/mysqlreceiver/testdata/scraper/expectedTopQueries.yaml

@@ -1,5 +1,9 @@
 resourceLogs:
-  - resource: {}
+  - resource:
+      attributes:
+        - key: mysql.instance.endpoint
+          value:
+            stringValue: localhost:3306
     scopeLogs:
       - logRecords:
           - attributes: