grpctmclient: validate tablet record on dial

Signed-off-by: Tim Vaillancourt <[email protected]>

Author: timvaillancourt

go/test/endtoend/reparent/plannedreparent/reparent_test.go

@@ -29,6 +29,7 @@ import (
 	"google.golang.org/protobuf/encoding/protojson"
 
 	"vitess.io/vitess/go/mysql/replication"
+	"vitess.io/vitess/go/protoutil"
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/reparent/utils"
 	"vitess.io/vitess/go/vt/log"
@@ -112,16 +113,43 @@ func TestReparentReplicaOffline(t *testing.T) {
 	clusterInstance := utils.SetupReparentCluster(t, policy.DurabilitySemiSync)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+	killTablet := tablets[3]
 
-	// Kill one tablet so we seem offline
-	utils.StopTablet(t, tablets[3], true)
+	// TabletShutdownTime is v24+.
+	if clusterInstance.VtctlMajorVersion >= 24 {
+		tabletInfo, err := clusterInstance.VtctldClientProcess.GetTablet(killTablet.Alias)
+		require.NoError(t, err)
+		require.Nil(t, tabletInfo.TabletShutdownTime)
+	}
+
+	// Gracefully kill one tablet so we seem offline. Use a SIGKILL-fallback delay of 30s, like kube.
+	startKillTime := time.Now()
+	killTablet.VttabletProcess.TearDownWithTimeout(30 * time.Second)
+
+	// Confirm the tablet shutdown via the topo.
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		tabletInfo, err := clusterInstance.VtctldClientProcess.GetTablet(killTablet.Alias)
+		require.NoError(c, err)
+
+		// TabletShutdownTime is v24+. Test this is the vttablet and vtctld are v24+.
+		if clusterInstance.VtctlMajorVersion >= 24 && clusterInstance.VtTabletMajorVersion >= 24 {
+			require.NotNil(c, tabletInfo.TabletShutdownTime)
+			shutdownTime := protoutil.TimeFromProto(tabletInfo.TabletShutdownTime)
+			require.WithinRange(c, shutdownTime, startKillTime, time.Now())
+		} else {
+			// TODO: remove this test after v25.
+			require.Empty(c, tabletInfo.Hostname)
+			require.Empty(c, tabletInfo.MysqlHostname)
+			require.Empty(c, tabletInfo.PortMap)
+		}
+	}, time.Second, time.Second*31)
 
 	// Perform a graceful reparent operation.
 	out, err := utils.PrsWithTimeout(t, clusterInstance, tablets[1], false, "", "31s")
 	require.Error(t, err)
 
 	// Assert that PRS failed
-	assert.Contains(t, out, "rpc error: code = DeadlineExceeded desc")
+	assert.Contains(t, out, "rpc error: code = Unknown desc = tablet is shutdown")
 	utils.CheckPrimaryTablet(t, clusterInstance, tablets[0])
 }
 
@@ -151,7 +179,7 @@ func TestReparentAvoid(t *testing.T) {
 	utils.StopTablet(t, tablets[0], true)
 	out, err := utils.PrsAvoid(t, clusterInstance, tablets[1])
 	require.Error(t, err)
-	assert.Contains(t, out, "rpc error: code = DeadlineExceeded desc = latest balancer error")
+	assert.Contains(t, out, "rpc error: code = Unknown desc = tablet is shutdown")
 	utils.ValidateTopology(t, clusterInstance, false)
 	utils.CheckPrimaryTablet(t, clusterInstance, tablets[1])
 

go/vt/proto/topodata/topodata.pb.go

@@ -191,6 +191,11 @@ func ReadTopologyInstanceBufferable(tabletAlias string, latency *stopwatch.Named
 		goto Cleanup
 	}
 
+	// Don't poll the tablet if we know it is down.
+	if tablet.TabletShutdownTime != nil {
+		goto Cleanup
+	}
+
 	fs, err = fullStatus(tablet)
 	if err != nil {
 		goto Cleanup

go/vt/proto/topodata/topodata_vtproto.pb.go

@@ -45,6 +45,7 @@ import (
 	tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
 	tabletmanagerservicepb "vitess.io/vitess/go/vt/proto/tabletmanagerservice"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
 type DialPoolGroup int
@@ -160,9 +161,39 @@ func NewClient() *Client {
 	}
 }
 
+// validateTablet confirms the tablet record contains the necessary fields
+// for talking gRPC.
+func validateTablet(tablet *topodatapb.Tablet) error {
+	// v24+ adds a TabletShutdownTime field that is set to "now" on clean shutdown.
+	if tablet.TabletShutdownTime != nil {
+		return vterrors.New(vtrpcpb.Code_UNAVAILABLE, "tablet is shutdown")
+	}
+
+	// <= v23 compatibility to similuate missing TabletShutdownTime field. Remove in v25.
+	if tablet.Hostname == "" && tablet.MysqlHostname == "" && tablet.PortMap == nil && tablet.Type != topodatapb.TabletType_UNKNOWN {
+		return vterrors.New(vtrpcpb.Code_UNAVAILABLE, "tablet is shutdown")
+	}
+
+	if tablet.Hostname == "" {
+		return vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "invalid tablet hostname")
+	}
+	if tablet.PortMap == nil {
+		return vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "invalid tablet port map")
+	}
+	grpcPort := int32(tablet.PortMap["grpc"])
+	if grpcPort <= 0 {
+		return vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "invalid tablet grpc port")
+	}
+	return nil
+}
+
 // dial returns a client to use
 func (client *grpcClient) dial(ctx context.Context, tablet *topodatapb.Tablet) (tabletmanagerservicepb.TabletManagerClient, io.Closer, error) {
-	addr := netutil.JoinHostPort(tablet.Hostname, int32(tablet.PortMap["grpc"]))
+	if err := validateTablet(tablet); err != nil {
+		return nil, nil, err
+	}
+
+	addr := netutil.JoinHostPort(tablet.Hostname, tablet.PortMap["grpc"])
 	opt, err := grpcclient.SecureDialOption(cert, key, ca, crl, name)
 	if err != nil {
 		return nil, nil, err
@@ -187,6 +218,10 @@ func (client *grpcClient) createTmc(ctx context.Context, addr string, opt grpc.D
 }
 
 func (client *grpcClient) dialPool(ctx context.Context, tablet *topodatapb.Tablet) (tabletmanagerservicepb.TabletManagerClient, error) {
+	if err := validateTablet(tablet); err != nil {
+		return nil, err
+	}
+
 	addr := netutil.JoinHostPort(tablet.Hostname, int32(tablet.PortMap["grpc"]))
 	opt, err := grpcclient.SecureDialOption(cert, key, ca, crl, name)
 	if err != nil {
@@ -227,6 +262,10 @@ func (client *grpcClient) dialPool(ctx context.Context, tablet *topodatapb.Table
 }
 
 func (client *grpcClient) dialDedicatedPool(ctx context.Context, dialPoolGroup DialPoolGroup, tablet *topodatapb.Tablet) (tabletmanagerservicepb.TabletManagerClient, invalidatorFunc, error) {
+	if err := validateTablet(tablet); err != nil {
+		return nil, nil, err
+	}
+
 	addr := netutil.JoinHostPort(tablet.Hostname, int32(tablet.PortMap["grpc"]))
 	opt, err := grpcclient.SecureDialOption(cert, key, ca, crl, name)
 	if err != nil {