*: wrap ctx.Err() with a stack

It's all to common to get an error like "context canceled" without any
helpful information about which context was canceled and when. This
commit strives to improve the situation by at least including the stack
trace of when the ctx.Err() was retrieved.

Release note: None

pkg/ccl/backupccl/show.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/pkg/errors"
 )
 
 // showBackupPlanHook implements PlanHookFn.
@@ -74,7 +75,7 @@ func showBackupPlanHook(
 		for _, row := range shower.fn(desc) {
 			select {
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			case resultsCh <- row:
 			}
 		}

pkg/ccl/changefeedccl/buffer.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/pkg/errors"
 )
 
 type bufferEntry struct {
@@ -51,7 +52,7 @@ func (b *buffer) addEntry(ctx context.Context, e bufferEntry) error {
 	// TODO(dan): Spill to a temp rocksdb if entriesCh would block.
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	case b.entriesCh <- e:
 		return nil
 	}
@@ -62,7 +63,7 @@ func (b *buffer) addEntry(ctx context.Context, e bufferEntry) error {
 func (b *buffer) Get(ctx context.Context) (bufferEntry, error) {
 	select {
 	case <-ctx.Done():
-		return bufferEntry{}, ctx.Err()
+		return bufferEntry{}, errors.WithStack(ctx.Err())
 	case e := <-b.entriesCh:
 		return e, nil
 	}

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/pkg/errors"
 )
 
 func init() {
@@ -191,7 +192,7 @@ func (w *changefeedResultWriter) AddRow(ctx context.Context, row tree.Datums) er
 
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	case w.rowsCh <- row:
 		return nil
 	}

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -219,7 +219,7 @@ func changefeedPlanHook(
 		}
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		case err := <-errCh:
 			return err
 		case <-startedCh:

pkg/ccl/changefeedccl/poller.go

@@ -144,7 +144,7 @@ func (p *poller) Run(ctx context.Context) error {
 			log.VEventf(ctx, 1, `sleeping for %s`, pollDuration)
 			select {
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			case <-time.After(pollDuration):
 			}
 		}
@@ -219,7 +219,7 @@ func (p *poller) Run(ctx context.Context) error {
 
 			select {
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			case exportsSem <- struct{}{}:
 			}
 
@@ -392,7 +392,7 @@ func (p *poller) runUsingRangefeeds(ctx context.Context) error {
 
 				select {
 				case <-ctx.Done():
-					return ctx.Err()
+					return errors.WithStack(ctx.Err())
 				case exportsSem <- struct{}{}:
 				}
 
@@ -463,7 +463,7 @@ func (p *poller) runUsingRangefeeds(ctx context.Context) error {
 					log.Fatalf(ctx, "unexpected RangeFeedEvent variant %v", t)
 				}
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			}
 		}
 	})

pkg/ccl/changefeedccl/sink.go

@@ -241,7 +241,7 @@ func (s *kafkaSink) Flush(ctx context.Context) error {
 	}
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	case <-flushCh:
 		s.mu.Lock()
 		flushErr := s.mu.flushErr
@@ -259,7 +259,7 @@ func (s *kafkaSink) emitMessage(ctx context.Context, msg *sarama.ProducerMessage
 
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	case s.producer.Input() <- msg:
 	}
 

pkg/ccl/importccl/read_import_csv.go

@@ -73,7 +73,7 @@ func (c *csvInputReader) flushBatch(ctx context.Context, finished bool, progFn p
 	if len(c.batch.r) > 0 {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		case c.recordCh <- c.batch:
 		}
 	}

pkg/ccl/importccl/read_import_proc.go

@@ -85,7 +85,7 @@ func readInputFiles(
 		currentFile++
 		select {
 		case <-done:
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		default:
 		}
 		if err := func() error {
@@ -324,7 +324,7 @@ func (c *rowConverter) sendBatch(ctx context.Context) error {
 	select {
 	case c.kvCh <- c.kvBatch:
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	}
 	c.kvBatch = make(kvBatch, 0, c.batchCap)
 	return nil

pkg/ccl/importccl/sst_writer_proc.go

@@ -356,7 +356,7 @@ func makeSSTs(
 		select {
 		case contentCh <- sc:
 		case <-ctx.Done():
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		}
 		counts.Reset()
 		sst.Close()

pkg/ccl/workloadccl/fixture.go

@@ -197,7 +197,7 @@ func (c *groupCSVWriter) groupWriteCSVs(
 	if err := func() error {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		case c.sem <- struct{}{}:
 		}
 		defer func() { <-c.sem }()

pkg/cmd/roachtest/allocator.go

@@ -221,7 +221,7 @@ func waitForRebalance(ctx context.Context, l *logger, db *gosql.DB, maxStdDev fl
 	for {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		case <-statsTimer.C:
 			statsTimer.Read = true
 			stats, err := allocatorStats(db)

pkg/cmd/roachtest/cdc.go

@@ -157,7 +157,7 @@ func cdcBasicTest(ctx context.Context, t *test, c *cluster, args cdcTestArgs) {
 		for {
 			select {
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			case <-tpccComplete:
 				return nil
 			case <-time.After(time.Second):
@@ -217,7 +217,7 @@ func cdcBasicTest(ctx context.Context, t *test, c *cluster, args cdcTestArgs) {
 				case <-tpccComplete:
 					return nil
 				case <-ctx.Done():
-					return ctx.Err()
+					return errors.WithStack(ctx.Err())
 				case <-t.C:
 				}
 
@@ -227,7 +227,7 @@ func cdcBasicTest(ctx context.Context, t *test, c *cluster, args cdcTestArgs) {
 				case <-tpccComplete:
 					return nil
 				case <-ctx.Done():
-					return ctx.Err()
+					return errors.WithStack(ctx.Err())
 				case <-time.After(downTime):
 				}
 

pkg/cmd/roachtest/chaos.go

@@ -17,6 +17,8 @@ package main
 import (
 	"context"
 	"time"
+
+	"github.com/pkg/errors"
 )
 
 // ChaosTimer configures a chaos schedule.
@@ -66,7 +68,7 @@ func (ch *Chaos) Runner(c *cluster, m *monitor) func(context.Context) error {
 			case <-ch.Stopper:
 				return nil
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			case <-t.C:
 			}
 
@@ -85,7 +87,7 @@ func (ch *Chaos) Runner(c *cluster, m *monitor) func(context.Context) error {
 			case <-ch.Stopper:
 				return nil
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			case <-time.After(downTime):
 			}
 

pkg/cmd/roachtest/clearrange.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/pkg/errors"
 )
 
 func registerClearRange(r *registry) {
@@ -116,7 +117,7 @@ func registerClearRange(r *registry) {
 					select {
 					case <-after:
 					case <-ctx.Done():
-						return ctx.Err()
+						return errors.WithStack(ctx.Err())
 					}
 				}
 				// TODO(benesch): verify that every last range in the table has been

pkg/cmd/roachtest/cluster_test.go

@@ -74,7 +74,7 @@ func TestClusterMonitor(t *testing.T) {
 		m.Go(func(ctx context.Context) error {
 			<-ctx.Done()
 			fmt.Printf("worker done\n")
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		})
 
 		err := m.wait(`echo`, "1: 100\n1: dead")
@@ -92,7 +92,7 @@ func TestClusterMonitor(t *testing.T) {
 		})
 		m.Go(func(ctx context.Context) error {
 			<-ctx.Done()
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		})
 
 		err := m.wait(`echo`, `1`)

pkg/cmd/roachtest/hotspotsplits.go

@@ -85,7 +85,7 @@ func registerHotSpotSplits(r *registry) {
 
 				select {
 				case <-ctx.Done():
-					return ctx.Err()
+					return errors.WithStack(ctx.Err())
 				case <-time.After(5 * time.Second):
 				}
 			}

pkg/cmd/roachtest/rebalance_load.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -96,7 +97,7 @@ func registerRebalanceLoad(r *registry) {
 
 				select {
 				case <-ctx.Done():
-					return ctx.Err()
+					return errors.WithStack(ctx.Err())
 				case <-time.After(5 * time.Second):
 				}
 			}

pkg/cmd/roachtest/tpcc.go

@@ -334,7 +334,7 @@ func loadTPCCBench(
 	select {
 	case <-time.After(rebalanceWait):
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	}
 
 	_, err = db.ExecContext(ctx, `SET CLUSTER SETTING kv.snapshot_rebalance.max_rate='2MiB'`)

pkg/cmd/roachtest/upgrade.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	_ "github.com/lib/pq"
+	"github.com/pkg/errors"
 
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/binfetcher"
@@ -65,7 +66,7 @@ func registerUpgrade(r *registry) {
 			t.WorkerStatus("sleeping")
 			select {
 			case <-ctx.Done():
-				return ctx.Err()
+				return errors.WithStack(ctx.Err())
 			case <-time.After(ts):
 				return nil
 			}

pkg/cmd/roachtest/version.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	_ "github.com/lib/pq"
+	"github.com/pkg/errors"
 
 	"github.com/cockroachdb/cockroach/pkg/util/binfetcher"
 )
@@ -88,7 +89,7 @@ func registerVersion(r *registry) {
 				t.WorkerStatus("sleeping")
 				select {
 				case <-ctx.Done():
-					return ctx.Err()
+					return errors.WithStack(ctx.Err())
 				case <-time.After(stageDuration):
 				}
 				// Make sure everyone is still running.

pkg/gossip/server.go

@@ -121,7 +121,7 @@ func (s *server) Gossip(stream Gossip_GossipServer) error {
 	send := func(reply *Response) error {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return errors.WithStack(ctx.Err())
 		case syncChan <- struct{}{}:
 			defer func() { <-syncChan }()
 

pkg/internal/client/lease.go

@@ -142,7 +142,7 @@ func (m *LeaseManager) timeRemaining(val *LeaseVal) time.Duration {
 func (m *LeaseManager) ExtendLease(ctx context.Context, l *Lease) error {
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	case l.val.sem <- struct{}{}:
 	}
 	defer func() { <-l.val.sem }()
@@ -173,7 +173,7 @@ func (m *LeaseManager) ExtendLease(ctx context.Context, l *Lease) error {
 func (m *LeaseManager) ReleaseLease(ctx context.Context, l *Lease) error {
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.WithStack(ctx.Err())
 	case l.val.sem <- struct{}{}:
 	}
 	defer func() { <-l.val.sem }()

pkg/internal/client/range_lookup.go

@@ -251,7 +251,7 @@ func RangeLookup(
 			key, prefetchedRanges)
 	}
 
-	ctxErr := ctx.Err()
+	ctxErr := errors.WithStack(ctx.Err())
 	if ctxErr == nil {
 		log.Fatalf(ctx, "retry loop broke before context expired")
 	}

pkg/internal/client/txn.go

@@ -687,7 +687,7 @@ func (txn *Txn) exec(ctx context.Context, fn func(context.Context, *Txn) error)
 	// error condition this loop isn't capable of handling.
 	for {
 		if err := ctx.Err(); err != nil {
-			return err
+			return errors.WithStack(err)
 		}
 		err = fn(ctx, txn)
 

pkg/jobs/jobs.go

@@ -647,7 +647,7 @@ func RunAndWaitForTerminalState(
 	for r := retry.StartWithCtx(ctx, retry.Options{}); ; {
 		select {
 		case <-ctx.Done():
-			return 0, "", ctx.Err()
+			return 0, "", errors.WithStack(ctx.Err())
 		case execErr = <-execErrCh:
 			// The closure finished, try to fetch a job id one more time. Close
 			// and nil out execErrCh so it blocks from now on.
@@ -671,7 +671,7 @@ func RunAndWaitForTerminalState(
 	for r := retry.StartWithCtx(ctx, retry.Options{}); ; {
 		select {
 		case <-ctx.Done():
-			return jobID, "", ctx.Err()
+			return jobID, "", errors.WithStack(ctx.Err())
 		case execErr = <-execErrCh:
 			// The closure finished, this is a nice hint to wake up, but it only
 			// works once. Close and nil out execErrCh so it blocks from now on.