kvfollowerreadsccl: maybe deflake TestBoundedStalenessDataDriven

This test determines what events occur by parsing the trace. In some cases, the
parsing it was using to determine a "local read followed by remote leaseholder
read" didn't account for changes in the potential trace messages encountered
when leader leases are enabled.

Here, I widen the scope of the trace parsing. Locally under stress this
elliminated the previously encountered failure:

```
    datadriven.go:357: ... SNIP ... boundedstaleness/single_row:24: still running after 10.000889738s
    ... SNIP ...
    boundedstaleness_test.go:405: condition failed to evaluate within 45s: from boundedstaleness_test.go:436: not yet a match, output:
        1
        events (1 found):
         * event 1: colbatchscan trace on node_idx 2: local read
    datadriven.go:343:
```

Fixes #154710
Release note: None

Author: stevendanna

pkg/ccl/kvccl/kvfollowerreadsccl/boundedstaleness_test.go

@@ -45,6 +45,10 @@ var (
 	)
 )
 
+// fullTraceDebug is a flag that controls whether full traces are printed in the
+// case of some errors.
+const fullTraceDebug = false
+
 func TestBoundedStalenessEnterpriseLicense(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -148,8 +152,10 @@ type boundedStalenessEvents struct {
 	// A mutex is needed as the event handlers (onStmtTrace) can race.
 	mu struct {
 		syncutil.Mutex
-		stmt   string
-		events []boundedStalenessDataDrivenEvent
+		stmt string
+		// Only populated if fullTraceDebug constant is true.
+		traceForDebugging string
+		events            []boundedStalenessDataDrivenEvent
 	}
 }
 
@@ -167,6 +173,13 @@ func (bse *boundedStalenessEvents) clearEvents() {
 	bse.mu.events = nil
 }
 
+func (bse *boundedStalenessEvents) fullTrace() string {
+	bse.mu.Lock()
+	defer bse.mu.Unlock()
+
+	return bse.mu.traceForDebugging
+}
+
 func (bse *boundedStalenessEvents) setStmt(s string) {
 	bse.mu.Lock()
 	defer bse.mu.Unlock()
@@ -240,17 +253,23 @@ func (bse *boundedStalenessEvents) onStmtTrace(nodeIdx int, rec tracingpb.Record
 	defer bse.mu.Unlock()
 
 	if bse.mu.stmt != "" && bse.mu.stmt == stmt {
+		if fullTraceDebug {
+			bse.mu.traceForDebugging = rec.String()
+		}
+
 		spans := make(map[tracingpb.SpanID]tracingpb.RecordedSpan)
 		for _, sp := range rec {
 			spans[sp.SpanID] = sp
+			notLeaseHolderError := tracing.LogsContainMsg(sp, "[NotLeaseHolderError] lease held by different store;")
+			notLeaseHolderError = notLeaseHolderError || tracing.LogsContainMsg(sp, "[NotLeaseHolderError] leader lease is not held locally, cannot determine validity;")
+
 			if sp.Operation == "dist sender send" && spans[sp.ParentSpanID].Operation == "colbatchscan" {
 				bse.mu.events = append(bse.mu.events, &boundedStalenessTraceEvent{
-					operation:    spans[sp.ParentSpanID].Operation,
-					nodeIdx:      nodeIdx,
-					localRead:    tracing.LogsContainMsg(sp, kvbase.RoutingRequestLocallyMsg),
-					followerRead: kvtestutils.OnlyFollowerReads(rec),
-					remoteLeaseholderRead: tracing.LogsContainMsg(sp, "[NotLeaseHolderError] lease held by different store;") &&
-						tracing.LogsContainMsg(sp, "trying next peer"),
+					operation:             spans[sp.ParentSpanID].Operation,
+					nodeIdx:               nodeIdx,
+					localRead:             tracing.LogsContainMsg(sp, kvbase.RoutingRequestLocallyMsg),
+					followerRead:          kvtestutils.OnlyFollowerReads(rec),
+					remoteLeaseholderRead: notLeaseHolderError && tracing.LogsContainMsg(sp, "trying next peer"),
 				})
 			}
 		}
@@ -261,10 +280,7 @@ func TestBoundedStalenessDataDriven(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	const msg = "1μs staleness reads may actually succeed due to the slow environment"
-	skip.UnderStress(t, msg)
-	skip.UnderRace(t, msg)
-	skip.UnderDeadlock(t, msg)
+	skip.UnderDuress(t, "1μs staleness reads may actually succeed due to the slow environment")
 	defer ccl.TestingEnableEnterprise()()
 
 	ctx := context.Background()
@@ -426,14 +442,22 @@ func TestBoundedStalenessDataDriven(t *testing.T) {
 						}
 					}()
 					if !followerRead {
+						var trace string
+						if fullTraceDebug {
+							trace = fmt.Sprintf("\nfull_trace:\n%s", bse.fullTrace())
+						}
 						bse.clearEvents()
-						return errors.AssertionFailedf("not follower reads found:\n%s", bse.String())
+						return errors.AssertionFailedf("not follower reads found:\n%s%s", bse.String(), trace)
 					}
 				}
 				if waitUntilMatch {
 					if d.Expected != ret {
+						var trace string
+						if fullTraceDebug {
+							trace = fmt.Sprintf("\nfull_trace:\n%s", bse.fullTrace())
+						}
 						bse.clearEvents()
-						return errors.AssertionFailedf("not yet a match, output:\n%s\n", ret)
+						return errors.AssertionFailedf("not yet a match, output:\n%s%s", ret, trace)
 					}
 				}
 				return nil