server: validate store addresses on PutStore

server/grpc_service.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
+	"net/url"
 	"path"
 	"runtime"
 	"runtime/trace"
@@ -804,6 +806,47 @@ func checkStore(rc *cluster.RaftCluster, storeID uint64) *pdpb.Error {
 	return nil
 }
 
+func dialAddress(ctx context.Context, address string) error {
+	if strings.HasPrefix(address, "mock://") {
+		return nil
+	}
+
+	hostPort := address
+	if u, err := url.Parse(address); err == nil && u.Host != "" {
+		hostPort = u.Host
+	}
+	if _, _, err := net.SplitHostPort(hostPort); err != nil {
+		return errors.WithStack(err)
+	}
+
+	dialCtx, cancel := context.WithTimeout(ctx, defaultGRPCDialTimeout)
+	defer cancel()
+	conn, err := (&net.Dialer{}).DialContext(dialCtx, "tcp", hostPort)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	return conn.Close()
+}
+
+func validateStoreAddress(ctx context.Context, store *metapb.Store) error {
+	addresses := make(map[string]struct{}, 2)
+	if address := store.GetAddress(); address != "" {
+		addresses[address] = struct{}{}
+	}
+	if address := store.GetStatusAddress(); address != "" {
+		addresses[address] = struct{}{}
+	}
+	if address := store.GetPeerAddress(); address != "" {
+		addresses[address] = struct{}{}
+	}
+	for address := range addresses {
+		if err := dialAddress(ctx, address); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // PutStore implements gRPC PDServer.
 func (s *GrpcServer) PutStore(ctx context.Context, request *pdpb.PutStoreRequest) (*pdpb.PutStoreResponse, error) {
 	done, err := s.rateLimitCheck()
@@ -849,6 +892,15 @@ func (s *GrpcServer) PutStore(ctx context.Context, request *pdpb.PutStoreRequest
 		}, nil
 	}
 
+	// TiKV puts the store before listening on its store address, so PD can only
+	// validate the configured address by checking whether it is dialable.
+	if err := validateStoreAddress(ctx, store); err != nil {
+		return &pdpb.PutStoreResponse{
+			Header: grpcutil.WrapErrorToHeader(pdpb.ErrorType_INVALID_VALUE,
+				fmt.Sprintf("invalid store address %s: %s", store.GetAddress(), err)),
+		}, nil
+	}
+
 	if err := rc.PutMetaStore(store); err != nil {
 		return &pdpb.PutStoreResponse{
 			Header: grpcutil.WrapErrorToHeader(pdpb.ErrorType_UNKNOWN, err.Error()),

tests/server/cluster/cluster_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"net"
 	"os"
 	"strconv"
 	"strings"
@@ -2219,3 +2220,55 @@ func TestPutStoreInvalidEngineLabel(t *testing.T) {
 		})
 	}
 }
+
+func TestPutStoreValidateStoreAddress(t *testing.T) {
+	re := require.New(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	tc, err := tests.NewTestCluster(ctx, 1)
+	re.NoError(err)
+	defer tc.Destroy()
+
+	err = tc.RunInitialServers()
+	re.NoError(err)
+	tc.WaitLeader()
+	leaderServer := tc.GetLeaderServer()
+	grpcPDClient, conn := testutil.MustNewGrpcClient(re, leaderServer.GetAddr())
+	defer conn.Close()
+	clusterID := leaderServer.GetClusterID()
+
+	bootstrapCluster(re, clusterID, grpcPDClient)
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	re.NoError(err)
+	defer re.NoError(listener.Close())
+
+	resp, err := putStore(grpcPDClient, clusterID, &metapb.Store{
+		Id:      101,
+		Address: listener.Addr().String(),
+		Version: "2.0.1",
+	})
+	re.NoError(err)
+	re.Nil(resp.GetHeader().GetError())
+
+	unreachableListener, err := net.Listen("tcp", "127.0.0.1:0")
+	re.NoError(err)
+	unreachableAddr := unreachableListener.Addr().String()
+	re.NoError(unreachableListener.Close())
+
+	resp, err = putStore(grpcPDClient, clusterID, &metapb.Store{
+		Id:      102,
+		Address: unreachableAddr,
+		Version: "2.0.1",
+	})
+	re.NoError(err)
+	re.Equal(pdpb.ErrorType_INVALID_VALUE, resp.GetHeader().GetError().GetType())
+
+	resp, err = putStore(grpcPDClient, clusterID, &metapb.Store{
+		Id:      103,
+		Address: "mock://tikv-103:103",
+		Version: "2.0.1",
+	})
+	re.NoError(err)
+	re.Nil(resp.GetHeader().GetError())
+}