Merge #96811 #98707 #98713 #98732

96811: loqrecovery: support mixed version recovery r=erikgrinaker a=aliher1911

This commit adds mixed version support for half-online loss of quorum recovery service and cli tools.
This change would allow user to use loq recovery in partially upgraded clusters by tracking version that generated data and produce recovery plans which will have identical version so that versions could be verified on all steps of recovery.
General rule is you can use data from the cluster that is not newer than a binary version to avoid new information being dropped. This rule applies to planning process where planner should understand replica info and also to cockroach node that applies the plan, which should be created by equal or lower version. Additional restriction is on planner to preserve version in the plan and don't use any new features if processed info is older than the binary version. This is no different on what version gates do in cockroach.

Release note: None

Fixes #95344

98707: keyvisualizer: pre-aggregate ranges r=zachlite a=zachlite

Previously, there was no bound on the number of ranges that could be propagated to the collectors. After collection, data was downsampled using a simple heurstic to decide if a bucket was worth keeping or if it should be aggregated with its neighbor.

In this commit, I've introduced a function, `maybeAggregateBoundaries`, to prevent more than `keyvisualizer.max_buckets` from being propagated to collectors. This pre-aggregation takes the place of the post-collection downsampling. For the first stable release of the key visualizer, I am intentionally sacrificing dynamic resolution and prioritizing boundary stability instead. This trade-off means that the key visualizer will demand less network, memory, and storage resources from the cluster while operating.

Additionally, this PR drops the sample retention time from 14 days to 7 days, and ensures that `keyvisualizer.max_buckets` is bounded between [1, 1024].

Resolves: #96740
Epic: None
Release note: None

98713: sql,kv: bubble up retry errors when creating leaf transactions r=arulajmani a=arulajmani

Previously, if we detected that the transaction was aborted when trying to construct leaf transaction state, we would handle the retry error instead of bubbling it up to the caller. When a transaction is aborted, the `TransactionRetryWithProtoRefreshError` carries with it a new transaction that should be used for subsequent attempts. Handling the retry error entailed swapping out the old `TxnCoordSender` with a new one -- one that is associated with this new transaction.

This is bug prone when trying to create multiple leaf transactions in parallel if the root has been aborted. We would expect the first leaf transaction to handle the error and all subsequent leaf transactions to point to the new transaction, as the `TxnCoordSender` has been swapped out. This wasn't an issue before as we never really created multiple leaf transactions in parallel. This recently change in 0f4b431, which started parallelizing FK and uniqueness checks. With this change, we could see FK or uniqueness violations when in fact the transaction needed to be retried.

This patch fixes the issue described above by not handling the retry error when creating leaf transactions. Instead, we expect the ConnExecutor to retry the entire transaction and prepare it for another iteration.

Fixes #97141

Epic: none

Release note: None

98732: cloud/gcp_test: add weird code 0/ok error to regex r=dt a=dt

Still unsure why we sometimes see this instead of the other more infromative errors but in the meanime, make the test pass.

Release note: none.
Epic: none.

Co-authored-by: Oleg Afanasyev <[email protected]>
Co-authored-by: zachlite <[email protected]>
Co-authored-by: Arul Ajmani <[email protected]>
Co-authored-by: David Taylor <[email protected]>

docs/generated/http/full.md

@@ -7465,6 +7465,7 @@ Support status: [reserved](#support-status)
 | range_descriptor | [cockroach.roachpb.RangeDescriptor](#cockroach.server.serverpb.RecoveryCollectReplicaInfoResponse-cockroach.roachpb.RangeDescriptor) |  |  | [reserved](#support-status) |
 | replica_info | [cockroach.kv.kvserver.loqrecovery.loqrecoverypb.ReplicaInfo](#cockroach.server.serverpb.RecoveryCollectReplicaInfoResponse-cockroach.kv.kvserver.loqrecovery.loqrecoverypb.ReplicaInfo) |  |  | [reserved](#support-status) |
 | node_stream_restarted | [RecoveryCollectReplicaRestartNodeStream](#cockroach.server.serverpb.RecoveryCollectReplicaInfoResponse-cockroach.server.serverpb.RecoveryCollectReplicaRestartNodeStream) |  |  | [reserved](#support-status) |
+| metadata | [cockroach.kv.kvserver.loqrecovery.loqrecoverypb.ClusterMetadata](#cockroach.server.serverpb.RecoveryCollectReplicaInfoResponse-cockroach.kv.kvserver.loqrecovery.loqrecoverypb.ClusterMetadata) |  |  | [reserved](#support-status) |
 
 
 
@@ -7553,7 +7554,8 @@ Support status: [reserved](#support-status)
 | ----- | ---- | ----- | ----------- | -------------- |
 | plan | [cockroach.kv.kvserver.loqrecovery.loqrecoverypb.ReplicaUpdatePlan](#cockroach.server.serverpb.RecoveryStagePlanRequest-cockroach.kv.kvserver.loqrecovery.loqrecoverypb.ReplicaUpdatePlan) |  | Plan is replica update plan to stage for application on next restart. Plan could be empty in that case existing plan is removed if present. | [reserved](#support-status) |
 | all_nodes | [bool](#cockroach.server.serverpb.RecoveryStagePlanRequest-bool) |  | If all nodes is true, then receiver should act as a coordinator and perform a fan-out to stage plan on all nodes of the cluster. | [reserved](#support-status) |
-| force_plan | [bool](#cockroach.server.serverpb.RecoveryStagePlanRequest-bool) |  | Force plan tells receiver to ignore any plan already staged on the node if it is present and replace it with new plan (including empty one). | [reserved](#support-status) |
+| force_plan | [bool](#cockroach.server.serverpb.RecoveryStagePlanRequest-bool) |  | ForcePlan tells receiver to ignore any plan already staged on the node if it is present and replace it with new plan (including empty one). | [reserved](#support-status) |
+| force_local_internal_version | [bool](#cockroach.server.serverpb.RecoveryStagePlanRequest-bool) |  | ForceLocalInternalVersion tells server to update internal component of plan version to the one of active cluster version. This option needs to be set if target cluster is stuck in recovery where only part of nodes were successfully migrated. | [reserved](#support-status) |
 
 
 

pkg/BUILD.bazel

@@ -178,6 +178,7 @@ ALL_TESTS = [
     "//pkg/jobs/jobsprotectedts:jobsprotectedts_test",
     "//pkg/jobs:jobs_test",
     "//pkg/keys:keys_test",
+    "//pkg/keyvisualizer/spanstatsconsumer:spanstatsconsumer_test",
     "//pkg/kv/bulk:bulk_test",
     "//pkg/kv/kvclient/kvcoord:kvcoord_disallowed_imports_test",
     "//pkg/kv/kvclient/kvcoord:kvcoord_test",
@@ -1172,6 +1173,7 @@ GO_TARGETS = [
     "//pkg/keyvisualizer/keyvissubscriber:keyvissubscriber",
     "//pkg/keyvisualizer/spanstatscollector:spanstatscollector",
     "//pkg/keyvisualizer/spanstatsconsumer:spanstatsconsumer",
+    "//pkg/keyvisualizer/spanstatsconsumer:spanstatsconsumer_test",
     "//pkg/keyvisualizer/spanstatskvaccessor:spanstatskvaccessor",
     "//pkg/keyvisualizer:keyvisualizer",
     "//pkg/kv/bulk/bulkpb:bulkpb",

pkg/ccl/cloudccl/gcp/gcp_connection_test.go

@@ -314,8 +314,8 @@ func TestGCPAssumeRoleExternalConnection(t *testing.T) {
 	sqlDB.Exec(t, `INSERT INTO foo VALUES (1), (2), (3)`)
 
 	disallowedCreateExternalConnection := func(t *testing.T, externalConnectionName, uri string) {
-		t.Log(uri)
-		sqlDB.ExpectErr(t, "(PermissionDenied|AccessDenied|PERMISSION_DENIED|does not have storage.objects.create access)",
+		// TODO(dt): remove `code 0/OK`. See https://github.com/cockroachdb/cockroach/issues/98733.
+		sqlDB.ExpectErr(t, "(PermissionDenied|AccessDenied|PERMISSION_DENIED|does not have storage.objects.create access)|code 0/OK",
 			fmt.Sprintf(`CREATE EXTERNAL CONNECTION '%s' AS '%s'`, externalConnectionName, uri))
 	}
 	createExternalConnection := func(t *testing.T, externalConnectionName, uri string) {

pkg/cli/BUILD.bazel

@@ -367,6 +367,7 @@ go_test(
         "//pkg/kv/kvserver",
         "//pkg/kv/kvserver/allocator/storepool",
         "//pkg/kv/kvserver/liveness/livenesspb",
+        "//pkg/kv/kvserver/loqrecovery",
         "//pkg/kv/kvserver/loqrecovery/loqrecoverypb",
         "//pkg/kv/kvserver/stateloader",
         "//pkg/roachpb",

pkg/cli/cliflags/flags.go

@@ -1913,6 +1913,16 @@ p - prompt interactively for a confirmation
 `,
 	}
 
+	RecoverIgnoreInternalVersion = FlagInfo{
+		Name: "ignore-internal-version",
+		Description: `
+When set, staging and local store plan application commands will ignore internal
+cluster version. This option must only be used to bypass version check if
+cluster is stuck in the middle of upgrade and locally stored versions differ
+from node to node and previous application or staging attempt failed.
+`,
+	}
+
 	PrintKeyLength = FlagInfo{
 		Name: "print-key-max-length",
 		Description: `

pkg/cli/debug.go

@@ -1451,6 +1451,8 @@ func init() {
 		cliflags.ConfirmActions.Usage())
 	f.UintVar(&formatHelper.maxPrintedKeyLength, cliflags.PrintKeyLength.Name,
 		formatHelper.maxPrintedKeyLength, cliflags.PrintKeyLength.Usage())
+	f.BoolVar(&debugRecoverExecuteOpts.ignoreInternalVersion, cliflags.RecoverIgnoreInternalVersion.Name,
+		debugRecoverExecuteOpts.ignoreInternalVersion, cliflags.RecoverIgnoreInternalVersion.Usage())
 
 	f = debugMergeLogsCmd.Flags()
 	f.Var(flagutil.Time(&debugMergeLogsOpts.from), "from",

pkg/cli/debug_recover_loss_of_quorum.go

@@ -21,14 +21,14 @@ import (
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/loqrecovery"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/loqrecovery/loqrecoverypb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/strutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
@@ -353,10 +353,9 @@ func runDebugDeadReplicaCollect(cmd *cobra.Command, args []string) error {
 		defer outFile.Close()
 		writer = outFile
 	}
-	jsonpb := protoutil.JSONPb{Indent: "  "}
-	out, err := jsonpb.Marshal(&replicaInfo)
+	out, err := loqrecovery.MarshalReplicaInfo(replicaInfo)
 	if err != nil {
-		return errors.Wrap(err, "failed to marshal collected replica info")
+		return err
 	}
 	if _, err := writer.Write(out); err != nil {
 		return errors.Wrap(err, "failed to write collected replica info")
@@ -564,33 +563,41 @@ Discarded live replicas: %d
 		planFile = path.Base(debugRecoverPlanOpts.outputFileName)
 	}
 
-	jsonpb := protoutil.JSONPb{Indent: "  "}
 	var out []byte
-	if out, err = jsonpb.Marshal(&plan); err != nil {
-		return errors.Wrap(err, "failed to marshal recovery plan")
+	if out, err = loqrecovery.MarshalPlan(plan); err != nil {
+		return err
 	}
 	if _, err = writer.Write(out); err != nil {
 		return errors.Wrap(err, "failed to write recovery plan")
 	}
 
-	// No args means we collected connection info from cluster and need to
-	// preserve flags for subsequent invocation.
-	remoteArgs := getCLIClusterFlags(len(args) == 0, cmd, func(flag string) bool {
-		_, filter := planSpecificFlags[flag]
-		return filter
-	})
+	v := clusterversion.ClusterVersion{
+		Version: plan.Version,
+	}
+	if v.IsActive(clusterversion.V23_1) {
+		// No args means we collected connection info from cluster and need to
+		// preserve flags for subsequent invocation.
+		remoteArgs := getCLIClusterFlags(len(args) == 0, cmd, func(flag string) bool {
+			_, filter := planSpecificFlags[flag]
+			return filter
+		})
 
-	_, _ = fmt.Fprintf(stderr, `Plan created.
+		_, _ = fmt.Fprintf(stderr, `Plan created.
 To stage recovery application in half-online mode invoke:
 
 cockroach debug recover apply-plan %s %s
 
 Alternatively distribute plan to below nodes and invoke 'debug recover apply-plan --store=<store-dir> %s' on:
 `, remoteArgs, planFile, planFile)
+	} else {
+		_, _ = fmt.Fprintf(stderr, `Plan created.
+To complete recovery, distribute plan to below nodes and invoke 'debug recover apply-plan --store=<store-dir> %s' on:
+`, planFile)
+	}
 	for _, node := range report.UpdatedNodes {
-		_, _ = fmt.Fprintf(stderr, "- node n%d, store(s) %s\n", node.NodeID, strutil.JoinIDs("s", node.StoreIDs))
+		_, _ = fmt.Fprintf(stderr, "- node n%d, store(s) %s\n", node.NodeID,
+			strutil.JoinIDs("s", node.StoreIDs))
 	}
-
 	return nil
 }
 
@@ -602,9 +609,8 @@ func readReplicaInfoData(fileNames []string) (loqrecoverypb.ClusterReplicaInfo,
 			return loqrecoverypb.ClusterReplicaInfo{}, errors.Wrapf(err, "failed to read replica info file %q", filename)
 		}
 
-		var nodeReplicas loqrecoverypb.ClusterReplicaInfo
-		jsonpb := protoutil.JSONPb{}
-		if err = jsonpb.Unmarshal(data, &nodeReplicas); err != nil {
+		nodeReplicas, err := loqrecovery.UnmarshalReplicaInfo(data)
+		if err != nil {
 			return loqrecoverypb.ClusterReplicaInfo{}, errors.WithHint(errors.Wrapf(err,
 				"failed to unmarshal replica info from file %q", filename),
 				"Ensure that replica info file is generated with the same binary version and file is not corrupted.")
@@ -633,8 +639,9 @@ See debug recover command help for more details on how to use this command.
 }
 
 var debugRecoverExecuteOpts struct {
-	Stores        base.StoreSpecList
-	confirmAction confirmActionFlag
+	Stores                base.StoreSpecList
+	confirmAction         confirmActionFlag
+	ignoreInternalVersion bool
 }
 
 // runDebugExecuteRecoverPlan is using the following pattern when performing command
@@ -655,20 +662,24 @@ func runDebugExecuteRecoverPlan(cmd *cobra.Command, args []string) error {
 		return errors.Wrapf(err, "failed to read plan file %q", planFile)
 	}
 
-	var nodeUpdates loqrecoverypb.ReplicaUpdatePlan
-	jsonpb := protoutil.JSONPb{Indent: "  "}
-	if err = jsonpb.Unmarshal(data, &nodeUpdates); err != nil {
+	nodeUpdates, err := loqrecovery.UnmarshalPlan(data)
+	if err != nil {
 		return errors.Wrapf(err, "failed to unmarshal plan from file %q", planFile)
 	}
 
 	if len(debugRecoverExecuteOpts.Stores.Specs) == 0 {
-		return stageRecoveryOntoCluster(ctx, cmd, planFile, nodeUpdates)
+		return stageRecoveryOntoCluster(ctx, cmd, planFile, nodeUpdates,
+			debugRecoverExecuteOpts.ignoreInternalVersion)
 	}
-	return applyRecoveryToLocalStore(ctx, nodeUpdates)
+	return applyRecoveryToLocalStore(ctx, nodeUpdates, debugRecoverExecuteOpts.ignoreInternalVersion)
 }
 
 func stageRecoveryOntoCluster(
-	ctx context.Context, cmd *cobra.Command, planFile string, plan loqrecoverypb.ReplicaUpdatePlan,
+	ctx context.Context,
+	cmd *cobra.Command,
+	planFile string,
+	plan loqrecoverypb.ReplicaUpdatePlan,
+	ignoreInternalVersion bool,
 ) error {
 	c, finish, err := getAdminClient(ctx, serverCfg)
 	if err != nil {
@@ -747,7 +758,11 @@ func stageRecoveryOntoCluster(
 			return err
 		}
 	}
-	sr, err := c.RecoveryStagePlan(ctx, &serverpb.RecoveryStagePlanRequest{Plan: &plan, AllNodes: true})
+	sr, err := c.RecoveryStagePlan(ctx, &serverpb.RecoveryStagePlanRequest{
+		Plan:                      &plan,
+		AllNodes:                  true,
+		ForceLocalInternalVersion: ignoreInternalVersion,
+	})
 	if err := maybeWrapStagingError("failed to stage loss of quorum recovery plan on cluster",
 		sr, err); err != nil {
 		return err
@@ -787,19 +802,21 @@ func sortedKeys[T ~int | ~int32 | ~int64](set map[T]any) []T {
 }
 
 func applyRecoveryToLocalStore(
-	ctx context.Context, nodeUpdates loqrecoverypb.ReplicaUpdatePlan,
+	ctx context.Context, nodeUpdates loqrecoverypb.ReplicaUpdatePlan, ignoreInternalVersion bool,
 ) error {
 	stopper := stop.NewStopper()
 	defer stopper.Stop(ctx)
 
 	var localNodeID roachpb.NodeID
 	batches := make(map[roachpb.StoreID]storage.Batch)
-	for _, storeSpec := range debugRecoverExecuteOpts.Stores.Specs {
+	stores := make([]storage.Engine, len(debugRecoverExecuteOpts.Stores.Specs))
+	for i, storeSpec := range debugRecoverExecuteOpts.Stores.Specs {
 		store, err := OpenEngine(storeSpec.Path, stopper, storage.MustExist)
 		if err != nil {
 			return errors.Wrapf(err, "failed to open store at path %q. ensure that store path is "+
 				"correct and that it is not used by another process", storeSpec.Path)
 		}
+		stores[i] = store
 		batch := store.NewBatch()
 		defer store.Close()
 		defer batch.Close()
@@ -818,6 +835,10 @@ func applyRecoveryToLocalStore(
 		batches[storeIdent.StoreID] = batch
 	}
 
+	if err := loqrecovery.CheckEnginesVersion(ctx, stores, nodeUpdates, ignoreInternalVersion); err != nil {
+		return err
+	}
+
 	updateTime := timeutil.Now()
 	prepReport, err := loqrecovery.PrepareUpdateReplicas(
 		ctx, nodeUpdates, uuid.DefaultGenerator, updateTime, localNodeID, batches)
@@ -911,8 +932,8 @@ func runDebugVerify(cmd *cobra.Command, args []string) error {
 		if err != nil {
 			return errors.Wrapf(err, "failed to read plan file %q", planFile)
 		}
-		jsonpb := protoutil.JSONPb{Indent: "  "}
-		if err = jsonpb.Unmarshal(data, &updatePlan); err != nil {
+		updatePlan, err = loqrecovery.UnmarshalPlan(data)
+		if err != nil {
 			return errors.Wrapf(err, "failed to unmarshal plan from file %q", planFile)
 		}
 	}