kvflowcontrol,raftlog: interfaces for replication control

.github/CODEOWNERS

@@ -222,7 +222,8 @@
 /pkg/kv/kvserver/gc/                    @cockroachdb/kv-prs
 /pkg/kv/kvserver/idalloc/               @cockroachdb/kv-prs
 /pkg/kv/kvserver/intentresolver/        @cockroachdb/kv-prs
-/pkg/kv/kvserver/kvadmission/           @cockroachdb/kv-prs
+/pkg/kv/kvserver/kvadmission/           @cockroachdb/admission-control
+/pkg/kv/kvserver/kvflowcontrol/         @cockroachdb/admission-control
 /pkg/kv/kvserver/kvserverbase/          @cockroachdb/kv-prs
 /pkg/kv/kvserver/kvserverpb/            @cockroachdb/kv-prs
 /pkg/kv/kvserver/kvstorage/             @cockroachdb/repl-prs

pkg/BUILD.bazel

@@ -1214,6 +1214,8 @@ GO_TARGETS = [
     "//pkg/kv/kvserver/intentresolver:intentresolver",
     "//pkg/kv/kvserver/intentresolver:intentresolver_test",
     "//pkg/kv/kvserver/kvadmission:kvadmission",
+    "//pkg/kv/kvserver/kvflowcontrol/kvflowcontrolpb:kvflowcontrolpb",
+    "//pkg/kv/kvserver/kvflowcontrol:kvflowcontrol",
     "//pkg/kv/kvserver/kvserverbase:kvserverbase",
     "//pkg/kv/kvserver/kvserverpb:kvserverpb",
     "//pkg/kv/kvserver/kvstorage:kvstorage",
@@ -2607,6 +2609,8 @@ GET_X_DATA_TARGETS = [
     "//pkg/kv/kvserver/idalloc:get_x_data",
     "//pkg/kv/kvserver/intentresolver:get_x_data",
     "//pkg/kv/kvserver/kvadmission:get_x_data",
+    "//pkg/kv/kvserver/kvflowcontrol:get_x_data",
+    "//pkg/kv/kvserver/kvflowcontrol/kvflowcontrolpb:get_x_data",
     "//pkg/kv/kvserver/kvserverbase:get_x_data",
     "//pkg/kv/kvserver/kvserverpb:get_x_data",
     "//pkg/kv/kvserver/kvstorage:get_x_data",

pkg/gen/protobuf.bzl

@@ -30,6 +30,7 @@ PROTOBUF_SRCS = [
     "//pkg/kv/kvserver/closedts/ctpb:ctpb_go_proto",
     "//pkg/kv/kvserver/concurrency/lock:lock_go_proto",
     "//pkg/kv/kvserver/concurrency/poison:poison_go_proto",
+    "//pkg/kv/kvserver/kvflowcontrol/kvflowcontrolpb:kvflowcontrolpb_go_proto",
     "//pkg/kv/kvserver/kvserverpb:kvserverpb_go_proto",
     "//pkg/kv/kvserver/liveness/livenesspb:livenesspb_go_proto",
     "//pkg/kv/kvserver/loqrecovery/loqrecoverypb:loqrecoverypb_go_proto",

pkg/kv/kvserver/kvadmission/BUILD.bazel

@@ -18,6 +18,7 @@ go_library(
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_pebble//:pebble",
+        "@io_etcd_go_raft_v3//raftpb",
     ],
 )
 

pkg/kv/kvserver/kvadmission/kvadmission.go

@@ -29,6 +29,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"go.etcd.io/raft/v3/raftpb"
 )
 
 // elasticCPUDurationPerExportRequest controls how many CPU tokens are allotted
@@ -81,6 +82,14 @@ var rangefeedCatchupScanElasticControlEnabled = settings.RegisterBoolSetting(
 	true,
 )
 
+// ProvisionedBandwidth set a value of the provisioned
+// bandwidth for each store in the cluster.
+var ProvisionedBandwidth = settings.RegisterByteSizeSetting(
+	settings.SystemOnly, "kvadmission.store.provisioned_bandwidth",
+	"if set to a non-zero value, this is used as the provisioned bandwidth (in bytes/s), "+
+		"for each store. It can be over-ridden on a per-store basis using the --store flag",
+	0).WithPublic()
+
 // Controller provides admission control for the KV layer.
 type Controller interface {
 	// AdmitKVWork must be called before performing KV work.
@@ -108,6 +117,9 @@ type Controller interface {
 	// replicated to a raft follower, that have not been subject to admission
 	// control.
 	FollowerStoreWriteBytes(roachpb.StoreID, FollowerStoreWriteBytes)
+	// AdmitRaftEntry informs admission control of a raft log entry being
+	// written to storage.
+	AdmitRaftEntry(roachpb.TenantID, roachpb.StoreID, roachpb.RangeID, raftpb.Entry)
 }
 
 // TenantWeightProvider can be periodically asked to provide the tenant
@@ -394,13 +406,12 @@ func (n *controllerImpl) FollowerStoreWriteBytes(
 		followerWriteBytes.NumEntries, followerWriteBytes.StoreWorkDoneInfo)
 }
 
-// ProvisionedBandwidth set a value of the provisioned
-// bandwidth for each store in the cluster.
-var ProvisionedBandwidth = settings.RegisterByteSizeSetting(
-	settings.SystemOnly, "kvadmission.store.provisioned_bandwidth",
-	"if set to a non-zero value, this is used as the provisioned bandwidth (in bytes/s), "+
-		"for each store. It can be over-ridden on a per-store basis using the --store flag",
-	0).WithPublic()
+// AdmitRaftEntry implements the Controller interface.
+func (n *controllerImpl) AdmitRaftEntry(
+	roachpb.TenantID, roachpb.StoreID, roachpb.RangeID, raftpb.Entry,
+) {
+	panic("unimplemented")
+}
 
 // FollowerStoreWriteBytes captures stats about writes done to a store by a
 // replica that is not the leaseholder. These are used for admission control.

pkg/kv/kvserver/kvflowcontrol/BUILD.bazel

@@ -0,0 +1,19 @@
+load("//build/bazelutil/unused_checker:unused.bzl", "get_x_data")
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "kvflowcontrol",
+    srcs = [
+        "doc.go",
+        "kvflowcontrol.go",
+    ],
+    importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvflowcontrol",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/kv/kvserver/kvflowcontrol/kvflowcontrolpb",
+        "//pkg/roachpb",
+        "//pkg/util/admission/admissionpb",
+    ],
+)
+
+get_x_data(name = "get_x_data")