roachpb: add lock table metadata structures

This change adds the protobuf structures, as well as the method, needed
to capture the state of a replica's lock table.  This is part of the
work coming out of #75541, and is needed to be able to implement the
`QueryLocks` RPC.

Release justification: (Category 2) The structures created here are
entirely new, for the purposes of observability, and do not represent a
risk to any existing code or functionality.

Release note: None

docs/generated/redact_safe.md

@@ -12,6 +12,8 @@ pkg/kv/bulk/sst_batcher.go | `sz`
 pkg/kv/bulk/sst_batcher.go | `timing`
 pkg/kv/kvserver/closedts/ctpb/service.go | `LAI`
 pkg/kv/kvserver/closedts/ctpb/service.go | `SeqNum`
+pkg/kv/kvserver/concurrency/lock/locking.go | `Durability`
+pkg/kv/kvserver/concurrency/lock/locking.go | `Strength`
 pkg/kv/kvserver/concurrency/lock/locking.go | `WaitPolicy`
 pkg/kv/kvserver/kvserverpb/raft.go | `SnapshotRequest_Type`
 pkg/roachpb/data.go | `LeaseSequence`

pkg/BUILD.bazel

@@ -139,6 +139,7 @@ ALL_TESTS = [
     "//pkg/kv/kvserver/closedts/sidetransport:sidetransport_test",
     "//pkg/kv/kvserver/closedts/tracker:tracker_test",
     "//pkg/kv/kvserver/closedts:closedts_test",
+    "//pkg/kv/kvserver/concurrency/lock:lock_test",
     "//pkg/kv/kvserver/concurrency/poison:poison_test",
     "//pkg/kv/kvserver/concurrency:concurrency_test",
     "//pkg/kv/kvserver/gc:gc_test",

pkg/kv/kvserver/concurrency/lock/BUILD.bazel

@@ -1,21 +1,32 @@
 load("@rules_proto//proto:defs.bzl", "proto_library")
 load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "lock",
-    srcs = ["locking.go"],
+    srcs = [
+        "lock_waiter.go",
+        "locking.go",
+    ],
     embed = [":lock_go_proto"],
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock",
     visibility = ["//visibility:public"],
+    deps = ["@com_github_cockroachdb_redact//:redact"],
 )
 
 proto_library(
     name = "lock_proto",
-    srcs = ["locking.proto"],
+    srcs = [
+        "lock_waiter.proto",
+        "locking.proto",
+    ],
     strip_import_prefix = "/pkg",
     visibility = ["//visibility:public"],
-    deps = ["@com_github_gogo_protobuf//gogoproto:gogo_proto"],
+    deps = [
+        "//pkg/storage/enginepb:enginepb_proto",
+        "@com_github_gogo_protobuf//gogoproto:gogo_proto",
+        "@com_google_protobuf//:duration_proto",
+    ],
 )
 
 go_proto_library(
@@ -24,5 +35,22 @@ go_proto_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock",
     proto = ":lock_proto",
     visibility = ["//visibility:public"],
-    deps = ["@com_github_gogo_protobuf//gogoproto"],
+    deps = [
+        "//pkg/storage/enginepb",
+        "@com_github_gogo_protobuf//gogoproto",
+    ],
+)
+
+go_test(
+    name = "lock_test",
+    srcs = ["lock_waiter_test.go"],
+    deps = [
+        ":lock",
+        "//pkg/roachpb",
+        "//pkg/storage/enginepb",
+        "//pkg/util/hlc",
+        "//pkg/util/uuid",
+        "@com_github_cockroachdb_redact//:redact",
+        "@com_github_stretchr_testify//require",
+    ],
 )

pkg/kv/kvserver/concurrency/lock/lock_waiter.go

@@ -0,0 +1,30 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+// Package lock provides type definitions for locking-related concepts used by
+// concurrency control in the key-value layer.
+package lock
+
+import "github.com/cockroachdb/redact"
+
+// SafeFormat implements redact.SafeFormatter.
+func (lw Waiter) SafeFormat(w redact.SafePrinter, _ rune) {
+	expand := w.Flag('+')
+
+	txnIDRedactableString := redact.Sprint(nil)
+	if lw.WaitingTxn != nil {
+		if expand {
+			txnIDRedactableString = redact.Sprint(lw.WaitingTxn.ID)
+		} else {
+			txnIDRedactableString = redact.Sprint(lw.WaitingTxn.Short())
+		}
+	}
+	w.Printf("waiting_txn:%s active_waiter:%t strength:%s wait_duration:%s", txnIDRedactableString, lw.ActiveWaiter, lw.Strength, lw.WaitDuration)
+}

pkg/kv/kvserver/concurrency/lock/lock_waiter.proto

@@ -0,0 +1,35 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+syntax = "proto3";
+package cockroach.kv.kvserver.concurrency.lock;
+option go_package = "lock";
+
+import "kv/kvserver/concurrency/lock/locking.proto";
+import "storage/enginepb/mvcc3.proto";
+import "gogoproto/gogo.proto";
+import "google/protobuf/duration.proto";
+
+// Waiter represents a transaction (or non-transactional operation) that is
+// waiting in the wait queue of readers or writers on an individual lock.
+message Waiter {
+  // The transaction associated with this waiter, or nil in the case of a
+  // non-transactional waiter.
+  storage.enginepb.TxnMeta waiting_txn = 1;
+  // Represents if this operation is actively waiting on the lock.  While all
+  // readers are active waiters, there are some cases in which writers may not
+  // be actively waiting, for instance in the case of a broken reservation.
+  bool active_waiter = 2;
+  // The strength at which this waiter is attempting to acquire the lock.
+  Strength strength = 3;
+  // The wall clock duration since this operation began waiting on the lock.
+  google.protobuf.Duration wait_duration = 4 [(gogoproto.nullable) = false,
+    (gogoproto.stdduration) = true];
+}

pkg/kv/kvserver/concurrency/lock/lock_waiter_test.go

@@ -0,0 +1,79 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+// Package lock provides type definitions for locking-related concepts used by
+// concurrency control in the key-value layer.
+package lock_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/cockroachdb/redact"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWaiterSafeFormat(t *testing.T) {
+	ts := hlc.Timestamp{Logical: 1}
+	txnMeta := &enginepb.TxnMeta{
+		Key:               roachpb.Key("foo"),
+		ID:                uuid.NamespaceDNS,
+		Epoch:             2,
+		WriteTimestamp:    ts,
+		MinTimestamp:      ts,
+		Priority:          957356782,
+		Sequence:          123,
+		CoordinatorNodeID: 3,
+	}
+	waiter := &lock.Waiter{
+		WaitingTxn:   txnMeta,
+		ActiveWaiter: true,
+		Strength:     lock.Exclusive,
+		WaitDuration: 135 * time.Second,
+	}
+
+	require.EqualValues(t,
+		"waiting_txn:6ba7b810 active_waiter:true strength:Exclusive wait_duration:2m15s",
+		redact.Sprint(waiter).StripMarkers())
+	require.EqualValues(t,
+		"waiting_txn:6ba7b810-9dad-11d1-80b4-00c04fd430c8 active_waiter:true strength:Exclusive wait_duration:2m15s",
+		redact.Sprintf("%+v", waiter).StripMarkers())
+	require.EqualValues(t,
+		"waiting_txn:6ba7b810 active_waiter:true strength:Exclusive wait_duration:2m15s",
+		redact.Sprint(waiter).Redact())
+	require.EqualValues(t,
+		"waiting_txn:‹×› active_waiter:true strength:Exclusive wait_duration:2m15s",
+		redact.Sprintf("%+v", waiter).Redact())
+
+	nonTxnWaiter := &lock.Waiter{
+		WaitingTxn:   nil,
+		ActiveWaiter: false,
+		Strength:     lock.None,
+		WaitDuration: 17 * time.Millisecond,
+	}
+
+	require.EqualValues(t,
+		"waiting_txn:<nil> active_waiter:false strength:None wait_duration:17ms",
+		redact.Sprint(nonTxnWaiter).StripMarkers())
+	require.EqualValues(t,
+		"waiting_txn:<nil> active_waiter:false strength:None wait_duration:17ms",
+		redact.Sprintf("%+v", nonTxnWaiter).StripMarkers())
+	require.EqualValues(t,
+		"waiting_txn:<nil> active_waiter:false strength:None wait_duration:17ms",
+		redact.Sprint(nonTxnWaiter).Redact())
+	require.EqualValues(t,
+		"waiting_txn:<nil> active_waiter:false strength:None wait_duration:17ms",
+		redact.Sprintf("%+v", nonTxnWaiter).Redact())
+}

pkg/kv/kvserver/concurrency/lock/locking.go

@@ -25,5 +25,11 @@ func init() {
 	}
 }
 
+// SafeValue implements redact.SafeValue.
+func (Strength) SafeValue() {}
+
+// SafeValue implements redact.SafeValue.
+func (Durability) SafeValue() {}
+
 // SafeValue implements redact.SafeValue.
 func (WaitPolicy) SafeValue() {}

pkg/roachpb/data.go

@@ -2053,6 +2053,38 @@ func (u *LockUpdate) SetTxn(txn *Transaction) {
 	u.IgnoredSeqNums = txn.IgnoredSeqNums
 }
 
+// SafeFormat implements redact.SafeFormatter.
+func (ls LockStateInfo) SafeFormat(w redact.SafePrinter, r rune) {
+	expand := w.Flag('+')
+	w.Printf("range_id=%d key=%s ", ls.RangeID, ls.Key)
+	redactableLockHolder := redact.Sprint(nil)
+	if ls.LockHolder != nil {
+		if expand {
+			redactableLockHolder = redact.Sprint(ls.LockHolder.ID)
+		} else {
+			redactableLockHolder = redact.Sprint(ls.LockHolder.Short())
+		}
+	}
+	w.Printf("holder=%s ", redactableLockHolder)
+	w.Printf("durability=%s ", ls.Durability)
+	w.Printf("duration=%s", ls.HoldDuration)
+	if len(ls.Waiters) > 0 {
+		w.Printf("\n waiters:")
+
+		for _, lw := range ls.Waiters {
+			if expand {
+				w.Printf("\n  %+v", lw)
+			} else {
+				w.Printf("\n  %s", lw)
+			}
+		}
+	}
+}
+
+func (ls LockStateInfo) String() string {
+	return redact.StringWithoutMarkers(ls)
+}
+
 // EqualValue is Equal.
 //
 // TODO(tbg): remove this passthrough.

pkg/roachpb/data.proto

@@ -12,13 +12,15 @@ syntax = "proto3";
 package cockroach.roachpb;
 option go_package = "roachpb";
 
+import "kv/kvserver/concurrency/lock/lock_waiter.proto";
 import "kv/kvserver/concurrency/lock/locking.proto";
 import "kv/kvserver/readsummary/rspb/summary.proto";
 import "roachpb/metadata.proto";
 import "storage/enginepb/mvcc.proto";
 import "storage/enginepb/mvcc3.proto";
 import "util/hlc/timestamp.proto";
 import "gogoproto/gogo.proto";
+import "google/protobuf/duration.proto";
 
 // Span is a key range with an inclusive start Key and an exclusive end Key.
 message Span {
@@ -542,6 +544,28 @@ message LockUpdate {
   repeated storage.enginepb.IgnoredSeqNumRange ignored_seqnums = 4 [(gogoproto.nullable) = false, (gogoproto.customname) = "IgnoredSeqNums"];
 }
 
+// A LockStateInfo represents the metadata of a lock tracked in a replica's
+// lock table.  This includes any current lock holder, as well as waiting
+// readers and writers.
+message LockStateInfo {
+  option (gogoproto.goproto_stringer) = false;
+
+  // The range that owns the Lock Table containing this lock.
+  int64 range_id = 1 [(gogoproto.customname) = "RangeID", (gogoproto.casttype) = "RangeID"];
+  // The key that this lock controls access to.
+  bytes key = 2 [(gogoproto.casttype) = "Key"];
+  // The current lock holder, or nil if the lock is not held.
+  storage.enginepb.TxnMeta lock_holder = 3;
+  // The durability that the lock is held at, or Unreplicated if not held.
+  kv.kvserver.concurrency.lock.Durability durability = 4;
+  // The wall clock duration since this lock was acquired (or zero, if not held).
+  google.protobuf.Duration hold_duration = 5 [(gogoproto.nullable) = false,
+    (gogoproto.stdduration) = true];
+  // The readers and writers currently waiting on the lock.  Stable ordering
+  // is not guaranteed.
+  repeated kv.kvserver.concurrency.lock.Waiter waiters = 6 [(gogoproto.nullable) = false];
+}
+
 // A SequencedWrite is a point write to a key with a certain sequence number.
 message SequencedWrite {
   option (gogoproto.populate) = true;