bulk,backupccl: log ExportRequest trace on failure

This change adds logic to fetch the trace for each ExportRequest
once it has returned a response or failed with an error. If the request
failed with an error, we unconditionally log the trace, otherwise
we log it when the appropriate vmodule is set.

This change also fixes a bug where the child context created by
the tracing aggregator would not inherit its parent's RecordingMode.
Today there is no mechanism to toggle a job's RecordingMode from
Structured to Verbose, but when we do add the ability to do so
we expect all operations in the job to collect information in accordance
with that mode.

Fixes: cockroachdb#86047

Release note: None

pkg/ccl/backupccl/backup_processor.go

@@ -249,6 +249,12 @@ type exportedSpan struct {
 	atKeyBoundary  bool
 }
 
+func fetchExportRequestTrace(ctx context.Context) string {
+	sp := tracing.SpanFromContext(ctx)
+	rec := sp.GetConfiguredRecording()
+	return rec.String()
+}
+
 func runBackupProcessor(
 	ctx context.Context,
 	flowCtx *execinfra.FlowCtx,
@@ -440,7 +446,12 @@ func runBackupProcessor(
 							}
 							return nil
 						})
+
+					// Fetch the trace for the ExportRequest even if it failed with an
+					// error.
+					exportReqTrace := fetchExportRequestTrace(ctx)
 					if exportRequestErr != nil {
+						log.Infof(ctx, "ExportRequest for span %s trace:\n%s", span.span.String(), exportReqTrace)
 						if intentErr, ok := pErr.GetDetail().(*roachpb.WriteIntentError); ok {
 							span.lastTried = timeutil.Now()
 							span.attempts++
@@ -471,6 +482,7 @@ func runBackupProcessor(
 						return errors.Wrapf(exportRequestErr, "exporting %s", span.span)
 					}
 
+					log.VEventf(ctx, 2, "ExportRequest for span %s trace:\n%s", span.span.String(), exportReqTrace)
 					resp := rawResp.(*roachpb.ExportResponse)
 
 					// If the reply has a resume span, we process it immediately.

pkg/util/bulk/tracing_aggregator.go

@@ -89,8 +89,10 @@ func MakeTracingAggregatorWithSpan(
 	ctx context.Context, aggregatorName string, tracer *tracing.Tracer,
 ) (context.Context, *TracingAggregator) {
 	agg := &TracingAggregator{}
+
+	sp := tracing.SpanFromContext(ctx)
 	aggCtx, aggSpan := tracing.EnsureChildSpan(ctx, tracer, aggregatorName,
-		tracing.WithEventListeners(agg))
+		tracing.WithEventListeners(agg), tracing.WithRecording(sp.RecordingType()))
 
 	agg.mu.Lock()
 	defer agg.mu.Unlock()