kvserver: implement closed ts side-transport publisher

[andreimatei]

The side-transport is a component running on each node and periodically
publishing closed timestamps on ranges with the lease on the respective
node. This complements the closing of timestamps through Raft commands
such that inactive ranges still have their timestamp advanced.

This commit introduces only the publishing side (the consumer is coming
separately) - the guy opening streaming connections to all other nodes
with follower replicas for some ranges with local leases and
periodically publishing closed timestamp updates on a bunch of ranges at
once.

Care has been taken to make the communication protocol efficient. Each
stream is stateful and the information in every message is nicely
compressed.

See the RFC for details.

Release justification: Needed for global tables.
Release note: None

[cockroach-teamcity]

This change is 

[nvanbenschoten]

This is looking good. I just pushed three new commits. The first two serve as partial code reviews, making a few bug fixes, cleaning up some code (apologies for a few of the renames), getting it in a better place to wrap unit tests around, and adding a few TODOs that we'll want to address in this PR or in a follow-on PR. The third commit fixes two of the more severe issues I found when reading over this change, relating to when it's safe for a leaseholder to publish on the side-transport.

I wasn't able to get to unit tests yet, but that's still on my radar. Let's talk tomorrow about how we want to split those up. Once we're happy with everything, we can squash it all down and go through a polishing code review pass.

[nvanbenschoten]

once the msg.Snapshot = msg.SeqNum == 1 code I broke gets fixed and CI is happy.

This was a surprisingly enjoyable way to work through a complex code review. The reviewable comments will be lost to the sands of time, but the memories of force pushes and merge skew will last forever.

Reviewed 2 of 2 files at r1, 10 of 10 files at r2, 4 of 4 files at r3, 15 of 15 files at r4, 1 of 1 files at r5, 17 of 17 files at r6, 4 of 4 files at r7, 2 of 2 files at r8, 25 of 28 files at r9, 1 of 1 files at r10, 4 of 4 files at r11, 2 of 2 files at r12.
Reviewable status: complete! 1 of 0 LGTMs obtained

[andreimatei]

bors r+

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @nvanbenschoten)

[nvanbenschoten]

Ugh

show_test.go:936: kv.closed_timestamp.side_transport_interval: description "the interval at which the closed-timestamp side-transport attempts to advance each range's closed timestamp; set to 0 to disable the side-transport." must end with period if and only if it contains a secondary sentence

bors r-

[craig]

Canceled.

[nvanbenschoten]

bors r+

[craig]

Build succeeded:

• GitHub CI (Cockroach)

[RaduBerinde]

@andreimatei I have been investigating some troubles with race tests in various packages. The one I was looking at was race: limit on 8128 simultaneously alive goroutines is exceeded, dying during importccl.

I have added some instrumentation to the test to panic if we observe more than 4000 goroutines (diff below). Since this PR, the panic is hit reliably within 30s (with make testrace PKG=./pkg/ccl/importccl/). Before this PR, it is never hit. I don't think any of the importccl tests do anything crazy, and 4000 goroutines is a lot. Most of them are goroutines set up by grpc.newClientStream. Can you investigate?

diff --git a/pkg/ccl/importccl/main_test.go b/pkg/ccl/importccl/main_test.go
index 9387b5e4da..a95a3edc4d 100644
--- a/pkg/ccl/importccl/main_test.go
+++ b/pkg/ccl/importccl/main_test.go
@@ -10,7 +10,9 @@ package importccl
 
 import (
 	"os"
+	"runtime"
 	"testing"
+	"time"
 
 	_ "github.com/cockroachdb/cockroach/pkg/ccl/storageccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
@@ -28,6 +30,22 @@ func TestMain(m *testing.M) {
 	randutil.SeedForTests()
 	serverutils.InitTestServerFactory(server.TestServerFactory)
 	serverutils.InitTestClusterFactory(testcluster.TestClusterFactory)
+
+	go func() {
+		ticker := time.NewTicker(1 * time.Millisecond)
+		for {
+			<-ticker.C
+			n := runtime.NumGoroutine()
+			if n > 4000 {
+				//buf := make([]byte, 1024*1024*40)
+				//n := runtime.Stack(buf, true)
+				//fmt.Println(string(buf[:n]))
+				panic("blargle")
+			}
+		}
+	}()
+
 	os.Exit(m.Run())
 }