storage: ensure that spans of reproposals of applied commands are not used

Andrew Werner · nvb · commit 5bd37e8eb58c · 2019-08-01T16:14:51.000-04:00
This commit deals with the case where more than one reproposal of a command
exists at the same MaxAppliedIndex. Code that already exist prevents the
command from being finished more than once but leaves open the possibility of
writing events into a potentially closed span.

Release note: None
diff --git a/pkg/storage/replica_application.go b/pkg/storage/replica_application.go
@@ -889,6 +889,11 @@ func (r *Replica) applyCmdAppBatch(
 	}
 	for ok := it.init(&b.cmdBuf); ok; ok = it.next() {
 		cmd := it.cur()
+		// Reset the context for already applied commands to ensure that
+		// reproposals at the same MaxLeaseIndex do not record into closed spans.
+		if cmd.proposedLocally() && cmd.proposal.applied {
+			cmd.ctx = ctx
+		}
 		for _, sc := range cmd.replicatedResult().SuggestedCompactions {
 			r.store.compactor.Suggest(cmd.ctx, sc)
 		}
diff --git a/pkg/storage/replica_test.go b/pkg/storage/replica_test.go
@@ -58,6 +58,7 @@ import (
 	"github.com/cockroachdb/logtags"
 	"github.com/gogo/protobuf/proto"
 	"github.com/kr/pretty"
+	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -11718,6 +11719,108 @@ func TestHighestMaxLeaseIndexReproposalFinishesCommand(t *testing.T) {
 	}
 }
 
+// TestLaterReproposalsDoNotReuseContext ensures that when commands are
+// reproposed more than once at the same MaxLeaseIndex and the first command
+// applies that the later reproposals do not log into the proposal's context
+// as its underlying trace span may already be finished.
+func TestLaterReproposalsDoNotReuseContext(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	// Set the trace infrastructure to log if a span is used after being finished.
+	defer enableTraceDebugUseAfterFree()()
+
+	tc := testContext{}
+	ctx := context.Background()
+
+	// Set logging up to a test specific directory.
+	scope := log.Scope(t)
+	defer scope.Close(t)
+
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+	cfg := TestStoreConfig(hlc.NewClock(hlc.UnixNano, time.Nanosecond))
+	// Set up tracing.
+	tracer := tracing.NewTracer()
+	tracer.Configure(&cfg.Settings.SV)
+	tracer.AlwaysTrace()
+	cfg.AmbientCtx.Tracer = tracer
+	tc.StartWithStoreConfig(t, stopper, cfg)
+	key := roachpb.Key("a")
+	lease, _ := tc.repl.GetLease()
+	txn := newTransaction("test", key, roachpb.NormalUserPriority, tc.Clock())
+	ba := roachpb.BatchRequest{
+		Header: roachpb.Header{
+			RangeID: tc.repl.RangeID,
+			Txn:     txn,
+		},
+	}
+	ba.Timestamp = txn.OrigTimestamp
+	ba.Add(&roachpb.PutRequest{
+		RequestHeader: roachpb.RequestHeader{
+			Key: key,
+		},
+		Value: roachpb.MakeValueFromBytes([]byte("val")),
+	})
+
+	// Hold the RaftLock to encourage the reproposals to occur in the same batch.
+	tc.repl.RaftLock()
+	sp := tracer.StartRootSpan("replica send", logtags.FromContext(ctx), tracing.RecordableSpan)
+	tracedCtx := opentracing.ContextWithSpan(ctx, sp)
+	// Go out of our way to enable recording so that expensive logging is enabled
+	// for this context.
+	tracing.StartRecording(sp, tracing.SingleNodeRecording)
+	ch, _, _, pErr := tc.repl.evalAndPropose(tracedCtx, lease, &ba, &allSpans, endCmds{})
+	if pErr != nil {
+		t.Fatal(pErr)
+	}
+	// Launch a goroutine to finish the span as soon as a result has been sent.
+	errCh := make(chan *roachpb.Error)
+	go func() {
+		res := <-ch
+		sp.Finish()
+		errCh <- res.Err
+	}()
+
+	// Flush the proposal and then repropose it twice.
+	// This test verifies that these later reproposals don't record into the
+	// tracedCtx after its span has been finished.
+	func() {
+		tc.repl.mu.Lock()
+		defer tc.repl.mu.Unlock()
+		if err := tc.repl.mu.proposalBuf.flushLocked(); err != nil {
+			t.Fatal(err)
+		}
+		tc.repl.refreshProposalsLocked(0, reasonNewLeaderOrConfigChange)
+		if err := tc.repl.mu.proposalBuf.flushLocked(); err != nil {
+			t.Fatal(err)
+		}
+		tc.repl.refreshProposalsLocked(0, reasonNewLeaderOrConfigChange)
+	}()
+	tc.repl.RaftUnlock()
+
+	if pErr = <-errCh; pErr != nil {
+		t.Fatal(pErr)
+	}
+	// Round trip another proposal through the replica to ensure that previously
+	// committed entries have been applied.
+	_, pErr = tc.repl.sendWithRangeID(ctx, tc.repl.RangeID, ba)
+	if pErr != nil {
+		t.Fatal(pErr)
+	}
+
+	stopper.Quiesce(ctx)
+	// Check and see if the trace package logged an error.
+	log.Flush()
+	entries, err := log.FetchEntriesFromFiles(0, math.MaxInt64, 1,
+		regexp.MustCompile("net/trace"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(entries) > 0 {
+		t.Fatalf("reused span after free: %v", entries)
+	}
+}
+
 func enableTraceDebugUseAfterFree() (restore func()) {
 	prev := trace.DebugUseAfterFinish
 	trace.DebugUseAfterFinish = true

Original file line number	Diff line number	Diff line change
`@@ -889,6 +889,11 @@ func (r *Replica) applyCmdAppBatch(`
`889`	`889`	`}`
`890`	`890`	`for ok := it.init(&b.cmdBuf); ok; ok = it.next() {`
`891`	`891`	`cmd := it.cur()`
	`892`	`+ // Reset the context for already applied commands to ensure that`
	`893`	`+ // reproposals at the same MaxLeaseIndex do not record into closed spans.`
	`894`	`+ if cmd.proposedLocally() && cmd.proposal.applied {`
	`895`	`+ cmd.ctx = ctx`
	`896`	`+ }`
`892`	`897`	`for _, sc := range cmd.replicatedResult().SuggestedCompactions {`
`893`	`898`	`r.store.compactor.Suggest(cmd.ctx, sc)`
`894`	`899`	`}`