Merge #148013 #148057 #148171

148013: randgen: add schema filters for ldr r=jeffswenson a=jeffswenson

The first two commits in the PR are intended to be purely mechanical. The final commit adds support for schema filters that limit which schemas can be randomly generated.

## randgen: rework TableOpt to follow option pattern
This commit reworks randgen so that TableOpt follows the standard go
option convention. This allows for spelling non-binary options like type
filters or transformations.

Release note: none

## randgen: extract table logic into stand alone file
Previously, the logic for generating schemas was mixed up with the logic
for generating tables. This moves table specific logic into its own
file.

Release note: none

## randgen: add filter options for ldr
Add options to filter the columns, primary key, and secondary indexes
generated by randgen. This will be used when testing LDR to constraint
schema generation to schemas that are compatible with LDR.

Release note: none

Informs: CRDB-44094


148057: *: consolidate `AdminClient` RPC client creation r=cthumuluru-crdb a=cthumuluru-crdb

This commit consolidates `AdminClient` RPC client creation logic in cli, kv, and server packages. It is a continuation of the work done in #147606.

Epic: CRDB-48923
Fixes: #148158
Release note: none

148171: github: change check-pebble-deps action to post Slack message r=RaduBerinde a=RaduBerinde

This action checks that Pebble dependency is up-to-date for recent
release branches. It currently files issues but they are annoying to
triage and close.

This commit removes the github issue posting for the script and
instead sends a message to the storage-notifications channel on Slack.

Epic: none
Release note: None

Co-authored-by: Jeff Swenson <[email protected]>
Co-authored-by: Chandra Thumuluru <[email protected]>
Co-authored-by: Radu Berinde <[email protected]>

Author: 4 people

.github/workflows/check-pebble-dep.yml

@@ -2,17 +2,42 @@ name: Check Pebble dep
 on:
   schedule:
     - cron: '0 8 * * 0,3' # Every Sun and Wed at 8:00 UTC
+  workflow_dispatch:
 
 jobs:
   check-pebble-dep:
     runs-on: ubuntu-latest
     env:
       GH_TOKEN: ${{ github.token }}
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
         with:
           fetch-depth: 0
 
       - name: Check Pebble deps
-        shell: bash
-        run: scripts/check-pebble-dep.sh
+        id: run_script
+        run: |
+          EXIT_CODE=0
+          OUTPUT=$(scripts/check-pebble-dep.sh 2>&1) || EXIT_CODE=$?
+          echo "$OUTPUT"
+          # Set output as a multi-line string.
+          echo "output<<EOF" >> $GITHUB_OUTPUT
+          echo "$OUTPUT"     >> $GITHUB_OUTPUT
+          echo "EOF"         >> $GITHUB_OUTPUT
+          # Set exit code.
+          echo "exitcode=$EXIT_CODE" >> $GITHUB_OUTPUT
+
+      - name: Notify Slack on failure
+        if: steps.run_script.outputs.exitcode != '0'
+        uses: slackapi/[email protected]
+        with:
+          method: chat.postMessage
+          token: ${{ secrets.PEBBLE_SLACK_BOT_TOKEN }}
+          # The channel ID is for #storage-notifications.
+          payload: |
+            channel: C08JE13CM9S
+            text: |
+            Some Pebble dependencies are not up to date. Details below:
+            ```
+            ${{ steps.run_script.outputs.output }}
+            ```

pkg/BUILD.bazel

@@ -180,6 +180,7 @@ ALL_TESTS = [
     "//pkg/crosscluster/physical:physical_test",
     "//pkg/crosscluster/producer:producer_disallowed_imports_test",
     "//pkg/crosscluster/producer:producer_test",
+    "//pkg/crosscluster/replicationtestutils:replicationtestutils_test",
     "//pkg/crosscluster/replicationutils:replicationutils_test",
     "//pkg/crosscluster/streamclient/randclient:randclient_test",
     "//pkg/crosscluster/streamclient:streamclient_test",
@@ -1322,6 +1323,7 @@ GO_TARGETS = [
     "//pkg/crosscluster/producer:producer",
     "//pkg/crosscluster/producer:producer_test",
     "//pkg/crosscluster/replicationtestutils:replicationtestutils",
+    "//pkg/crosscluster/replicationtestutils:replicationtestutils_test",
     "//pkg/crosscluster/replicationutils:replicationutils",
     "//pkg/crosscluster/replicationutils:replicationutils_test",
     "//pkg/crosscluster/streamclient/randclient:randclient",

pkg/cli/debug_recover_loss_of_quorum.go

@@ -308,11 +308,12 @@ func runDebugDeadReplicaCollect(cmd *cobra.Command, args []string) error {
 	var stats loqrecovery.CollectionStats
 
 	if len(debugRecoverCollectInfoOpts.Stores.Specs) == 0 {
-		c, finish, err := getAdminClient(ctx, serverCfg)
+		c, finish, err := dialAdminClient(ctx, serverCfg)
 		if err != nil {
 			return errors.Wrapf(err, "failed to get admin connection to cluster")
 		}
 		defer finish()
+
 		replicaInfo, stats, err = loqrecovery.CollectRemoteReplicaInfo(ctx, c,
 			debugRecoverCollectInfoOpts.maxConcurrency, stderr /* logOutput */)
 		if err != nil {
@@ -426,11 +427,12 @@ func runDebugPlanReplicaRemoval(cmd *cobra.Command, args []string) error {
 	if len(args) == 0 {
 		// If no replica info is provided, try to connect to a cluster default or
 		// explicitly provided to retrieve replica info.
-		c, finish, err := getAdminClient(ctx, serverCfg)
+		c, finish, err := dialAdminClient(ctx, serverCfg)
 		if err != nil {
 			return errors.Wrapf(err, "failed to get admin connection to cluster")
 		}
 		defer finish()
+
 		replicas, stats, err = loqrecovery.CollectRemoteReplicaInfo(ctx, c,
 			debugRecoverPlanOpts.maxConcurrency, stderr /* logOutput */)
 		if err != nil {
@@ -675,7 +677,7 @@ func stageRecoveryOntoCluster(
 	ignoreInternalVersion bool,
 	maxConcurrency int,
 ) error {
-	c, finish, err := getAdminClient(ctx, serverCfg)
+	c, finish, err := dialAdminClient(ctx, serverCfg)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get admin connection to cluster")
 	}
@@ -686,6 +688,7 @@ func stageRecoveryOntoCluster(
 		nodeID roachpb.NodeID
 		planID string
 	}
+
 	vr, err := c.RecoveryVerify(ctx, &serverpb.RecoveryVerifyRequest{
 		MaxConcurrency: int32(maxConcurrency),
 	})
@@ -948,11 +951,12 @@ func runDebugVerify(cmd *cobra.Command, args []string) error {
 		_, _ = fmt.Printf("Checking application of recovery plan %s\n", updatePlan.PlanID)
 	}
 
-	c, finish, err := getAdminClient(ctx, serverCfg)
+	c, finish, err := dialAdminClient(ctx, serverCfg)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get admin connection to cluster")
 	}
 	defer finish()
+
 	req := serverpb.RecoveryVerifyRequest{
 		DecommissionedNodeIDs: updatePlan.DecommissionedNodeIDs,
 		MaxReportedRanges:     20,

pkg/cli/debug_reset_quorum.go

@@ -45,15 +45,15 @@ func runDebugResetQuorum(cmd *cobra.Command, args []string) error {
 	}
 
 	// Set up GRPC Connection for running ResetQuorum.
-	cc, finish, err := getClientGRPCConn(ctx, serverCfg)
+	conn, finish, err := newClientConn(ctx, serverCfg)
 	if err != nil {
 		log.Errorf(ctx, "connection to server failed: %v", err)
 		return err
 	}
 	defer finish()
 
 	// Call ResetQuorum to reset quorum for given range on target node.
-	_, err = kvpb.NewInternalClient(cc).ResetQuorum(ctx, &kvpb.ResetQuorumRequest{
+	_, err = conn.NewInternalClient().ResetQuorum(ctx, &kvpb.ResetQuorumRequest{
 		RangeID: int32(rangeID),
 	})
 	if err != nil {

pkg/cli/debug_send_kv_batch.go

@@ -166,12 +166,11 @@ func runSendKVBatch(cmd *cobra.Command, args []string) error {
 	// Send BatchRequest.
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	conn, finish, err := getClientGRPCConn(ctx, serverCfg)
+	admin, finish, err := dialAdminClient(ctx, serverCfg)
 	if err != nil {
 		return errors.Wrap(err, "failed to connect to the node")
 	}
 	defer finish()
-	admin := serverpb.NewAdminClient(conn)
 
 	br, rec, err := sendKVBatchRequestWithTracingOption(ctx, enableTracing, admin, &ba)
 	if err != nil {

pkg/cli/democluster/demo_cluster.go

@@ -795,7 +795,7 @@ func (c *transientCluster) waitForNodeIDReadiness(
 			if err != nil {
 				return err
 			}
-			c.servers[idx].adminClient = serverpb.NewAdminClient(conn)
+			c.servers[idx].adminClient = conn.NewAdminClient()
 
 		}
 		break
@@ -1219,7 +1219,7 @@ func (c *transientCluster) startServerInternal(
 
 	c.servers[serverIdx] = serverEntry{
 		TestServerInterface: s,
-		adminClient:         serverpb.NewAdminClient(conn),
+		adminClient:         conn.NewAdminClient(),
 		nodeID:              nodeID,
 	}
 

pkg/cli/gen.go

@@ -407,7 +407,7 @@ func generateMetricList(ctx context.Context, skipFiltering bool) (map[string]*La
 		if err != nil {
 			return err
 		}
-		client := serverpb.NewAdminClient(conn)
+		client := conn.NewAdminClient()
 
 		resp, err := client.ChartCatalog(ctx, &serverpb.ChartCatalogRequest{})
 		if err != nil {

pkg/cli/init.go

@@ -89,7 +89,7 @@ func dialAndCheckHealth(ctx context.Context) error {
 		// (Attempt to) establish the gRPC connection. If that fails,
 		// it may be that the server hasn't started to listen yet, in
 		// which case we'll retry.
-		conn, finish, err := getClientGRPCConn(ctx, serverCfg)
+		conn, finish, err := newClientConn(ctx, serverCfg)
 		if err != nil {
 			return err
 		}
@@ -98,7 +98,7 @@ func dialAndCheckHealth(ctx context.Context) error {
 		// Access the /health endpoint. Until/unless this succeeds, the
 		// node is not yet fully initialized and ready to accept
 		// Bootstrap requests.
-		ac := serverpb.NewAdminClient(conn)
+		ac := conn.NewAdminClient()
 		_, err = ac.Health(ctx, &serverpb.HealthRequest{})
 		return err
 	}

pkg/cli/rpc_clients.go

@@ -73,20 +73,18 @@ func makeRPCClientConfig(cfg server.Config) rpc.ClientConnConfig {
 func newClientConn(ctx context.Context, cfg server.Config) (rpcConn, func(), error) {
 	ccfg := makeRPCClientConfig(cfg)
 	cc, finish, err := rpc.NewClientConn(ctx, ccfg)
-	return &grpcConn{conn: cc}, finish, errors.Wrap(err, "failed to connect to the node")
-}
-
-func getClientGRPCConn(ctx context.Context, cfg server.Config) (*grpc.ClientConn, func(), error) {
-	ccfg := makeRPCClientConfig(cfg)
-	return rpc.NewClientConn(ctx, ccfg)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "failed to connect to the node")
+	}
+	return &grpcConn{conn: cc}, finish, nil
 }
 
-// getAdminClient returns an AdminClient and a closure that must be invoked
-// to free associated resources.
-func getAdminClient(ctx context.Context, cfg server.Config) (serverpb.AdminClient, func(), error) {
-	conn, finish, err := getClientGRPCConn(ctx, cfg)
+// dialAdminClient dials a client connection and returns an AdminClient and a
+// closure that must be invoked to free associated resources.
+func dialAdminClient(ctx context.Context, cfg server.Config) (serverpb.AdminClient, func(), error) {
+	cc, finish, err := newClientConn(ctx, cfg)
 	if err != nil {
-		return nil, nil, errors.Wrap(err, "failed to connect to the node")
+		return nil, nil, err
 	}
-	return serverpb.NewAdminClient(conn), finish, nil
+	return cc.NewAdminClient(), finish, nil
 }

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -913,7 +913,7 @@ func TestRandomTables(t *testing.T) {
 			rng,
 			tableName,
 			1,
-			randgen.TableOptSkipColumnFamilyMutations)
+			[]randgen.TableOption{randgen.WithSkipColumnFamilyMutations()})
 		stmt := tree.SerializeForDisplay(createStmt)
 		t.Log(stmt)
 		runnerA.Exec(t, stmt)