// checkRestart checks that there are no errors left to inject.
func checkRestarts(t *testing.T, magicVals *filterVals) {
	magicVals.Lock()
	defer magicVals.Unlock()
	for key, count := range magicVals.restartCounts {
		if count != 0 {
			file, line, _ := caller.Lookup(1)
			t.Errorf("%s:%d: INSERT for \"%s\" still has to be retried %d times",
				file, line, key, count)
		}
	}
	for key, count := range magicVals.abortCounts {
		if count != 0 {
			file, line, _ := caller.Lookup(1)
			t.Errorf("%s:%d: INSERT for \"%s\" still has to be aborted %d times",
				file, line, key, count)
		}
	}
	for key, count := range magicVals.endTxnRestartCounts {
		if count != 0 {
			file, line, _ := caller.Lookup(1)
			t.Errorf("%s:%d: txn writing \"%s\" still has to be aborted %d times",
				file, line, key, count)
		}
	}
	if len(magicVals.txnsToFail) > 0 {
		file, line, _ := caller.Lookup(1)
		t.Errorf("%s:%d: txns still to be failed: %v", file, line, magicVals.txnsToFail)
	}
	if t.Failed() {
		t.Fatalf("checking error injection failed")
	}
}

// newTxn begins a transaction. For testing purposes, this comes with a ID
// pre-initialized, but with the Writing flag set to false.
func newTxn(clock *hlc.Clock, baseKey proto.Key) *proto.Transaction {
	f, l, fun := caller.Lookup(1)
	name := fmt.Sprintf("%s:%d %s", f, l, fun)
	txn := proto.NewTransaction("test", baseKey, 1, proto.SERIALIZABLE, clock.Now(), clock.MaxOffset().Nanoseconds())
	txn.Name = name
	return txn
}

// StartCluster starts a cluster from the relevant flags. All test clusters
// should be created through this command since it sets up the logging in a
// unified way.
func StartCluster(t *testing.T) cluster.Cluster {
	if *numLocal > 0 {
		if *numRemote > 0 {
			t.Fatal("cannot both specify -num-local and -num-remote")
		}
		logDir := *logDir
		if logDir != "" {
			if _, _, fun := caller.Lookup(1); fun != "" {
				logDir = filepath.Join(logDir, fun)
			}
		}
		l := cluster.CreateLocal(*numLocal, *numStores, logDir, stopper)
		l.Start()
		checkRangeReplication(t, l, 20*time.Second)
		return l
	}
	if *numRemote == 0 {
		t.Fatal("need to either specify -num-local or -num-remote")
	}
	f := farmer(t)
	if err := f.Resize(*numRemote, 0); err != nil {
		t.Fatal(err)
	}
	if err := f.WaitReady(5 * time.Minute); err != nil {
		_ = f.Destroy()
		t.Fatalf("cluster not ready in time: %v", err)
	}
	checkRangeReplication(t, f, 20*time.Second)
	return f
}

func TestLogBacktraceAt(t *testing.T) {
	setFlags()
	defer logging.swap(logging.newBuffers())
	// The peculiar style of this code simplifies line counting and maintenance of the
	// tracing block below.
	var infoLine string
	setTraceLocation := func(file string, line int, delta int) {
		_, file = filepath.Split(file)
		infoLine = fmt.Sprintf("%s:%d", file, line+delta)
		err := logging.traceLocation.Set(infoLine)
		if err != nil {
			t.Fatal("error setting log_backtrace_at: ", err)
		}
	}
	{
		// Start of tracing block. These lines know about each other's relative position.
		file, line, _ := caller.Lookup(0)
		setTraceLocation(file, line, +2) // Two lines between Caller and Info calls.
		Info(context.Background(), "we want a stack trace here")
		if err := logging.traceLocation.Set(""); err != nil {
			t.Fatal(err)
		}
	}
	numAppearances := strings.Count(contents(InfoLog), infoLine)
	if numAppearances < 2 {
		// Need 2 appearances, one in the log header and one in the trace:
		//   log_test.go:281: I0511 16:36:06.952398 02238 log_test.go:280] we want a stack trace here
		//   ...
		//   github.com/clog/glog_test.go:280 (0x41ba91)
		//   ...
		// We could be more precise but that would require knowing the details
		// of the traceback format, which may not be dependable.
		t.Fatal("got no trace back; log is ", contents(InfoLog))
	}
}

// RunLimitedAsyncTask runs function f in a goroutine, using the given channel
// as a semaphore to limit the number of tasks that are run concurrently to
// the channel's capacity. Blocks until the semaphore is available in order to
// push back on callers that may be trying to create many tasks. Returns an
// error if the Stopper is draining, in which case the function is not
// executed.
func (s *Stopper) RunLimitedAsyncTask(sem chan struct{}, f func()) error {
	file, line, _ := caller.Lookup(1)
	key := taskKey{file, line}

	// Wait for permission to run from the semaphore.
	select {
	case sem <- struct{}{}:
	case <-s.ShouldDrain():
		return errUnavailable
	default:
		log.Printf("stopper throttling task from %s:%d due to semaphore", file, line)
		// Retry the select without the default.
		select {
		case sem <- struct{}{}:
		case <-s.ShouldDrain():
			return errUnavailable
		}
	}

	if !s.runPrelude(key) {
		<-sem
		return errUnavailable
	}
	go func() {
		defer s.runPostlude(key)
		defer func() { <-sem }()
		f()
	}()
	return nil
}

// SetDebugName sets the debug name associated with the transaction which will
// appear in log files and the web UI. Each transaction starts out with an
// automatically assigned debug name composed of the file and line number where
// the transaction was created.
func (txn *Txn) SetDebugName(name string, depth int) {
	file, line, fun := caller.Lookup(depth + 1)
	if name == "" {
		name = fun
	}
	txn.Proto.Name = fmt.Sprintf("%s:%d %s", file, line, name)
}

// SetDebugName sets the debug name associated with the transaction which will
// appear in log files and the web UI. Each transaction starts out with an
// automatically assigned debug name composed of the file and line number where
// the transaction was created.
func (txn *Txn) SetDebugName(name string, depth int) {
	file, line, fun := caller.Lookup(depth + 1)
	if name == "" {
		name = fun
	}
	txn.Proto.Name = file + ":" + strconv.Itoa(line) + " " + name
}

// TestErrorf verifies the pass through to fmt.Errorf as well as
// file/line prefix.
func TestErrorf(t *testing.T) {
	err := Errorf("foo: %d %f", 1, 3.14)
	file, line, _ := caller.Lookup(0)
	expected := fmt.Sprintf("%sfoo: 1 3.140000", fmt.Sprintf(errorPrefixFormat, file, line-1))
	if expected != err.Error() {
		t.Errorf("expected %s, got %s", expected, err.Error())
	}
}

// EnsureContext checks whether the given context.Context contains a Span. If
// not, it creates one using the provided Tracer and wraps it in the returned
// Span. The returned closure must be called after the request has been fully
// processed.
func EnsureContext(ctx context.Context, tracer opentracing.Tracer) (context.Context, func()) {
	_, _, funcName := caller.Lookup(1)
	if opentracing.SpanFromContext(ctx) == nil {
		sp := tracer.StartSpan(funcName)
		return opentracing.ContextWithSpan(ctx, sp), sp.Finish
	}
	return ctx, func() {}
}

// addStructured creates a structured log entry to be written to the
// specified facility of the logger.
func addStructured(ctx context.Context, s Severity, depth int, format string, args []interface{}) {
	if ctx == nil {
		panic("nil context")
	}
	file, line, _ := caller.Lookup(depth + 1)
	msg := makeMessage(ctx, format, args)
	Trace(ctx, msg)
	logging.outputLogEntry(s, file, line, msg)
}

// RunTask adds one to the count of tasks left to drain in the system. Any
// worker which is a "first mover" when starting tasks must call this method
// before starting work on a new task. First movers include
// goroutines launched to do periodic work and the kv/db.go gateway which
// accepts external client requests.
//
// Returns false to indicate that the system is currently draining and
// function f was not called.
func (s *Stopper) RunTask(f func()) bool {
	file, line, _ := caller.Lookup(1)
	key := taskKey{file, line}
	if !s.runPrelude(key) {
		return false
	}
	// Call f.
	defer s.runPostlude(key)
	f()
	return true
}

// TestErrorSkipFrames verifies ErrorSkipFrames with an additional
// stack frame.
func TestErrorSkipFrames(t *testing.T) {
	var err error
	func() {
		err = ErrorSkipFrames(1, "foo ", 1, 3.14)
	}()
	file, line, _ := caller.Lookup(0)
	expected := fmt.Sprintf("%sfoo 1 3.14", fmt.Sprintf(errorPrefixFormat, file, line-1))
	if expected != err.Error() {
		t.Errorf("expected %s, got %s", expected, err.Error())
	}
}

func newTxn(db DB, depth int) *Txn {
	txn := &Txn{
		db:      db,
		wrapped: db.Sender,
	}
	txn.db.Sender = (*txnSender)(txn)

	file, line, fun := caller.Lookup(depth + 1)
	txn.txn.Name = fmt.Sprintf("%s:%d %s", file, line, fun)
	return txn
}

// RunTask adds one to the count of tasks left to drain in the system. Any
// worker which is a "first mover" when starting tasks must call this method
// before starting work on a new task. First movers include
// goroutines launched to do periodic work and the kv/db.go gateway which
// accepts external client requests.
//
// Returns false to indicate that the system is currently draining and
// function f was not called.
func (s *Stopper) RunTask(f func()) bool {
	file, line, _ := caller.Lookup(1)
	taskKey := fmt.Sprintf("%s:%d", file, line)
	if !s.runPrelude(taskKey) {
		return false
	}
	// Call f.
	defer s.runPostlude(taskKey)
	f()
	return true
}

// addStructured creates a structured log entry to be written to the
// specified facility of the logger.
func addStructured(ctx context.Context, s Severity, depth int, format string, args []interface{}) {
	if ctx == nil {
		panic("nil context")
	}
	file, line, _ := caller.Lookup(depth + 1)
	msg := makeMessage(ctx, format, args)
	// makeMessage already added the tags when forming msg, we don't want
	// eventInternal to prepend them again.
	eventInternal(ctx, (s >= ErrorLog), false /*withTags*/, "%s", msg)
	logging.outputLogEntry(s, file, line, msg)
}

// RunTask adds one to the count of tasks left to drain in the system. Any
// worker which is a "first mover" when starting tasks must call this method
// before starting work on a new task. First movers include
// goroutines launched to do periodic work and the kv/db.go gateway which
// accepts external client requests.
//
// Returns an error to indicate that the system is currently draining and
// function f was not called.
func (s *Stopper) RunTask(f func()) error {
	file, line, _ := caller.Lookup(1)
	key := taskKey{file, line}
	if !s.runPrelude(key) {
		return errUnavailable
	}
	// Call f.
	defer s.runPostlude(key)
	f()
	return nil
}

func TestDebugName(t *testing.T) {
	defer leaktest.AfterTest(t)
	s, db := setup()
	defer s.Stop()

	file, _, _ := caller.Lookup(0)
	_ = db.Txn(func(txn *client.Txn) error {
		if !strings.HasPrefix(txn.DebugName(), file+":") {
			t.Fatalf("expected \"%s\" to have the prefix \"%s:\"", txn.DebugName(), file)
		}
		return nil
	})
}

// RunAsyncTask runs function f in a goroutine. It returns an error when the
// Stopper is quiescing, in which case the function is not executed.
func (s *Stopper) RunAsyncTask(f func()) error {
	file, line, _ := caller.Lookup(1)
	key := taskKey{file, line}
	if !s.runPrelude(key) {
		return errUnavailable
	}
	// Call f.
	go func() {
		defer s.maybeHandlePanic()
		defer s.runPostlude(key)
		f()
	}()
	return nil
}

func (t *Trace) add(name string) int {
	// Must be called with two callers to the client.
	// (Client->Event|Epoch->epoch->add)
	file, line, fun := caller.Lookup(3)
	t.Content = append(t.Content, TraceItem{
		depth:     t.depth,
		Origin:    t.tracer.origin,
		File:      file,
		Line:      line,
		Func:      fun,
		Timestamp: t.tracer.now(),
		Name:      name,
	})
	return len(t.Content) - 1
}

func TestDebugName(t *testing.T) {
	defer leaktest.AfterTest(t)
	s, db := setup()
	defer s.Stop()

	file, _, _ := caller.Lookup(0)
	if pErr := db.Txn(func(txn *client.Txn) *roachpb.Error {
		if !strings.HasPrefix(txn.DebugName(), file+":") {
			t.Fatalf("expected \"%s\" to have the prefix \"%s:\"", txn.DebugName(), file)
		}
		return nil
	}); pErr != nil {
		t.Errorf("txn failed: %s", pErr)
	}
}