sql: add SHOW TENANT name WITH REPLICATION STATUS

Extending SHOW TENANT to also allow showing replication status which contains info such as the protected timestamp on the destination cluster and the source cluster name. Command output right after the destination cluster is created: ``` [email protected]:26257/defaultdb> show tenant dest5 with replication status; id | name | status | source_tenant_name | source_cluster_uri | replication_job_id | replicated_time | retained_time -----+-------+--------+--------------------+--------------------+--------------------+-----------------+---------------- 7 | dest5 | ADD | NULL | NULL | 819890711267737601 | NULL | NULL (1 row) ``` A bit later we have most stats (manually adjusting the source_cluster_uri): ``` [email protected]:26257/defaultdb> show tenant dest5 with replication status; id | name | status | source_tenant_name | source_cluster_uri | replication_job_id | replicated_time | retained_time -----+-------+--------+--------------------+-------------------------------------------------------+--------------------+-----------------+----------------------------- 7 | dest5 | ADD | src | postgresql://[email protected]:26257/defaultdb?ssl...crt | 819890711267737601 | NULL | 2022-12-05 23:00:04.516331 (1 row) ``` And a moment later the replication time is populated. Informs: #91261 Epic: CRDB-18749 Release note: None
cockroachdb · Dec 8, 2022 · dcd183c · dcd183c
1 parent 2629aee
commit dcd183c
Show file tree

Hide file tree

Showing 17 changed files with 260 additions and 40 deletions.
diff --git a/docs/generated/sql/bnf/show_tenant_stmt.bnf b/docs/generated/sql/bnf/show_tenant_stmt.bnf
@@ -1,2 +1,3 @@
 show_tenant_stmt ::=
-	'SHOW' 'TENANT' name
+	'SHOW' 'TENANT' d_expr
+	| 'SHOW' 'TENANT' d_expr 'WITH' 'REPLICATION' 'STATUS'
diff --git a/docs/generated/sql/bnf/stmt_block.bnf b/docs/generated/sql/bnf/stmt_block.bnf
@@ -922,7 +922,8 @@ show_tables_stmt ::=
 	| 'SHOW' 'TABLES' with_comment
 
 show_tenant_stmt ::=
-	'SHOW' 'TENANT' name
+	'SHOW' 'TENANT' d_expr
+	| 'SHOW' 'TENANT' d_expr 'WITH' 'REPLICATION' 'STATUS'
 
 show_trace_stmt ::=
 	'SHOW' opt_compact 'TRACE' 'FOR' 'SESSION'

diff --git a/pkg/ccl/streamingccl/streamclient/client.go b/pkg/ccl/streamingccl/streamclient/client.go
@@ -172,6 +172,9 @@ func NewStreamClient(
 // GetFirstActiveClient iterates through each provided stream address
 // and returns the first client it's able to successfully Dial.
 func GetFirstActiveClient(ctx context.Context, streamAddresses []string) (Client, error) {
+	if len(streamAddresses) == 0 {
+		return nil, errors.Newf("failed to connect, no partition addresses")
+	}
 	var combinedError error = nil
 	for _, address := range streamAddresses {
 		streamAddress := streamingccl.StreamAddress(address)

diff --git a/pkg/ccl/streamingccl/streamclient/client_test.go b/pkg/ccl/streamingccl/streamclient/client_test.go
@@ -122,6 +122,15 @@ func (t testStreamSubscription) Err() error {
 	return nil
 }
 
+func TestGetFirstActiveClientEmpty(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	var streamAddresses []string
+	activeClient, err := GetFirstActiveClient(context.Background(), streamAddresses)
+	require.ErrorContains(t, err, "failed to connect, no partition addresses")
+	require.Nil(t, activeClient)
+}
+
 func TestGetFirstActiveClient(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 

diff --git a/pkg/ccl/streamingccl/streamingest/stream_replication_e2e_test.go b/pkg/ccl/streamingccl/streamingest/stream_replication_e2e_test.go
@@ -44,6 +44,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
@@ -1244,3 +1245,46 @@ func TestTenantReplicationProtectedTimestampManagement(t *testing.T) {
 		})
 	})
 }
+
+// TODO(lidor): consider rewriting this test as a data driven test when #92609 is merged.
+func TestTenantStreamingShowTenant(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	args := defaultTenantStreamingClustersArgs
+
+	c, cleanup := createTenantStreamingClusters(ctx, t, args)
+	defer cleanup()
+	testStartTime := timeutil.Now()
+	producerJobID, ingestionJobID := c.startStreamReplication()
+
+	jobutils.WaitForJobToRun(c.t, c.srcSysSQL, jobspb.JobID(producerJobID))
+	jobutils.WaitForJobToRun(c.t, c.destSysSQL, jobspb.JobID(ingestionJobID))
+	highWatermark := c.srcCluster.Server(0).Clock().Now()
+	c.waitUntilHighWatermark(highWatermark, jobspb.JobID(ingestionJobID))
+
+	var (
+		id            int
+		dest          string
+		status        string
+		source        string
+		sourceUri     string
+		jobId         int
+		maxReplTime   time.Time
+		protectedTime time.Time
+	)
+	row := c.destSysSQL.QueryRow(t, fmt.Sprintf("SHOW TENANT %s WITH REPLICATION STATUS", args.destTenantName))
+	row.Scan(&id, &dest, &status, &source, &sourceUri, &jobId, &maxReplTime, &protectedTime)
+	require.Equal(t, 2, id)
+	require.Equal(t, "destination", dest)
+	require.Equal(t, "ADD", status)
+	require.Equal(t, "source", source)
+	require.Equal(t, c.srcURL.String(), sourceUri)
+	require.Equal(t, ingestionJobID, jobId)
+	require.Less(t, maxReplTime, timeutil.Now())
+	require.Less(t, protectedTime, timeutil.Now())
+	require.GreaterOrEqual(t, maxReplTime, highWatermark.GoTime())
+	// TODO(lidor): replace this start time with the actual replication start time when we have it.
+	require.GreaterOrEqual(t, protectedTime, testStartTime)
+}
diff --git a/pkg/sql/BUILD.bazel b/pkg/sql/BUILD.bazel
@@ -307,6 +307,7 @@ go_library(
         "//pkg/obsservice/obspb/opentelemetry-proto/common/v1:common",
         "//pkg/obsservice/obspb/opentelemetry-proto/logs/v1:logs",
         "//pkg/repstream",
+        "//pkg/repstream/streampb",
         "//pkg/roachpb",
         "//pkg/rpc",
         "//pkg/rpc/nodedialer",

diff --git a/pkg/sql/catalog/colinfo/result_columns.go b/pkg/sql/catalog/colinfo/result_columns.go
@@ -278,3 +278,17 @@ var TenantColumns = ResultColumns{
 	{Name: "name", Typ: types.String},
 	{Name: "status", Typ: types.String},
 }
+
+var TenantColumnsWithReplication = ResultColumns{
+	{Name: "id", Typ: types.Int},
+	{Name: "name", Typ: types.String},
+	{Name: "status", Typ: types.String},
+	{Name: "source_tenant_name", Typ: types.String},
+	{Name: "source_cluster_uri", Typ: types.String},
+	{Name: "replication_job_id", Typ: types.Int},
+	// The latest fully replicated time.
+	{Name: "replicated_time", Typ: types.Timestamp},
+	// The protected timestamp on the destination cluster, meaning we cannot
+	// cutover to before this time.
+	{Name: "retained_time", Typ: types.Timestamp},
+}
diff --git a/pkg/sql/faketreeeval/BUILD.bazel b/pkg/sql/faketreeeval/BUILD.bazel
@@ -8,6 +8,8 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/clusterversion",
+        "//pkg/jobs/jobspb",
+        "//pkg/repstream/streampb",
         "//pkg/roachpb",
         "//pkg/security/username",
         "//pkg/sql/catalog/catpb",

diff --git a/pkg/sql/faketreeeval/evalctx.go b/pkg/sql/faketreeeval/evalctx.go
@@ -16,6 +16,8 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
@@ -594,6 +596,12 @@ func (c *DummyTenantOperator) GetTenantInfo(
 	return nil, errors.WithStack(errEvalTenant)
 }
 
+func (p *DummyTenantOperator) GetTenantReplicationInfo(
+	ctx context.Context, replicationJobId jobspb.JobID,
+) (*streampb.StreamIngestionStats, error) {
+	return nil, errors.WithStack(errEvalTenant)
+}
+
 // DestroyTenant is part of the tree.TenantOperator interface.
 func (c *DummyTenantOperator) DestroyTenant(
 	ctx context.Context, tenantName roachpb.TenantName, synchronous bool,

diff --git a/pkg/sql/logictest/testdata/logic_test/tenant b/pkg/sql/logictest/testdata/logic_test/tenant
@@ -60,6 +60,12 @@ id  name   status
 statement error tenant "seven" does not exist
 SHOW TENANT seven
 
+statement error tenant "tenant-one" does not have an active replication job
+SHOW TENANT "tenant-one" WITH REPLICATION STATUS
+
+statement error tenant "two" does not have an active replication job
+SHOW TENANT two WITH REPLICATION STATUS
+
 # Test creating a tenant with the same name as an existing tenant, but a unique
 # ID.
 statement error tenant with name "three" already exists

diff --git a/pkg/sql/parser/help_test.go b/pkg/sql/parser/help_test.go
@@ -411,6 +411,7 @@ func TestContextualHelp(t *testing.T) {
 		{`SHOW TABLES FROM blah ??`, `SHOW TABLES`},
 
 		{`SHOW TENANT ??`, `SHOW TENANT`},
+		{`SHOW TENANT ?? WITH REPLICATION STATUS`, `SHOW TENANT`},
 
 		{`SHOW TRANSACTION PRIORITY ??`, `SHOW TRANSACTION`},
 		{`SHOW TRANSACTION STATUS ??`, `SHOW TRANSACTION`},

diff --git a/pkg/sql/parser/sql.y b/pkg/sql/parser/sql.y
@@ -5420,11 +5420,21 @@ backup_kms:
 
 // %Help: SHOW TENANT - display tenant information
 // %Category: Misc
-// %Text: SHOW TENANT
+// %Text: SHOW TENANT <tenant_name> [WITH REPLICATION STATUS]
 show_tenant_stmt:
-  SHOW TENANT name
+  SHOW TENANT d_expr
   {
-   $$.val = &tree.ShowTenant{Name: tree.Name($3)}
+   $$.val = &tree.ShowTenant{
+     Name: $3.expr(),
+     WithReplication: false,
+   }
+  }
+| SHOW TENANT d_expr WITH REPLICATION STATUS
+  {
+   $$.val = &tree.ShowTenant{
+     Name: $3.expr(),
+     WithReplication: true,
+   }
   }
 | SHOW TENANT error // SHOW HELP: SHOW TENANT
 

diff --git a/pkg/sql/parser/testdata/show b/pkg/sql/parser/testdata/show
@@ -1799,6 +1799,14 @@ parse
 SHOW TENANT foo
 ----
 SHOW TENANT foo
-SHOW TENANT foo -- fully parenthesized
+SHOW TENANT (foo) -- fully parenthesized
 SHOW TENANT foo -- literals removed
 SHOW TENANT _ -- identifiers removed
+
+parse
+SHOW TENANT foo WITH REPLICATION STATUS
+----
+SHOW TENANT foo WITH REPLICATION STATUS
+SHOW TENANT (foo) WITH REPLICATION STATUS -- fully parenthesized
+SHOW TENANT foo WITH REPLICATION STATUS -- literals removed
+SHOW TENANT _ WITH REPLICATION STATUS -- identifiers removed
diff --git a/pkg/sql/sem/eval/deps.go b/pkg/sql/sem/eval/deps.go
@@ -573,9 +573,6 @@ type TenantOperator interface {
 	// the gc job will not wait for a GC ttl.
 	DestroyTenant(ctx context.Context, tenantName roachpb.TenantName, synchronous bool) error
 
-	// GetTenantInfo returns information about the specified tenant.
-	GetTenantInfo(ctx context.Context, tenantName roachpb.TenantName) (*descpb.TenantInfo, error)
-
 	// GCTenant attempts to garbage collect a DROP tenant from the system. Upon
 	// success it also removes the tenant record.
 	// It returns an error if the tenant does not exist.

diff --git a/pkg/sql/sem/tree/show.go b/pkg/sql/sem/tree/show.go
@@ -777,13 +777,18 @@ func (node *ShowTableStats) Format(ctx *FmtCtx) {
 
 // ShowTenant represents a SHOW TENANT statement.
 type ShowTenant struct {
-	Name Name
+	Name            Expr
+	WithReplication bool
 }
 
 // Format implements the NodeFormatter interface.
 func (node *ShowTenant) Format(ctx *FmtCtx) {
 	ctx.WriteString("SHOW TENANT ")
-	ctx.FormatNode(&node.Name)
+	ctx.FormatNode(node.Name)
+
+	if node.WithReplication {
+		ctx.WriteString(" WITH REPLICATION STATUS")
+	}
 }
 
 // ShowHistogram represents a SHOW HISTOGRAM statement.