store TiDB server info to PD and add http api handle

ddl/mock.go

@@ -39,6 +39,11 @@ func NewMockSchemaSyncer() SchemaSyncer {
 	return &mockSchemaSyncer{}
 }
 
+// GetSessionLeaseID implements SchemaSyncer.GetSessionLeaseID interface.
+func (s *mockSchemaSyncer) GetSessionLeaseID() clientv3.LeaseID {
+	return 0
+}
+
 // Init implements SchemaSyncer.Init interface.
 func (s *mockSchemaSyncer) Init(ctx context.Context) error {
 	s.globalVerCh = make(chan clientv3.WatchResponse, 1)

ddl/syncer.go

@@ -83,6 +83,8 @@ type SchemaSyncer interface {
 	// the latest schema version. If the result is false, wait for a while and check again util the processing time reach 2 * lease.
 	// It returns until all servers' versions are equal to the latest version or the ctx is done.
 	OwnerCheckAllVersions(ctx context.Context, latestVer int64) error
+	// GetSession return the session lease id of SchemaSyncer.
+	GetSessionLeaseID() clientv3.LeaseID
 }
 
 type schemaVersionSyncer struct {
@@ -127,6 +129,11 @@ func PutKVToEtcd(ctx context.Context, etcdCli *clientv3.Client, retryCnt int, ke
 	return errors.Trace(err)
 }
 
+// GetSessionLeaseID implements SchemaSyncer.GetSessionLeaseID interface.
+func (s *schemaVersionSyncer) GetSessionLeaseID() clientv3.LeaseID {
+	return s.session.Lease()
+}
+
 // Init implements SchemaSyncer.Init interface.
 func (s *schemaVersionSyncer) Init(ctx context.Context) error {
 	startTime := time.Now()

domain/domain.go

@@ -365,6 +365,7 @@ func (do *Domain) loadSchemaInLoop(lease time.Duration) {
 				break
 			}
 			do.SchemaValidator.Restart()
+			do.info.Restart(syncer.GetSessionLeaseID())
 		case <-do.exit:
 			return
 		}
@@ -509,7 +510,7 @@ func (do *Domain) Init(ddlLease time.Duration, sysFactory func(*Domain) (pools.R
 	if err != nil {
 		return errors.Trace(err)
 	}
-	do.info = NewInfoSyncer(do.ddl.GetID(), do.etcdClient)
+	do.info = NewInfoSyncer(do.ddl.GetID(), do.etcdClient, do.ddl.SchemaSyncer().GetSessionLeaseID())
 	err = do.info.StoreServerInfo(ctx)
 	if err != nil {
 		return errors.Trace(err)

domain/info.go

@@ -43,6 +43,8 @@ type InfoSyncer struct {
 	etcdCli        *clientv3.Client
 	info           *ServerInfo
 	serverInfoPath string
+	// InfoSyncer will reuse the leaseID of ddl.SchemaSyncer.
+	sessionLeaseID clientv3.LeaseID
 }
 
 // ServerInfo is server static information.
@@ -63,28 +65,15 @@ type ServerVersionInfo struct {
 }
 
 // NewInfoSyncer return new InfoSyncer. It is exported for testing.
-func NewInfoSyncer(id string, etcdCli *clientv3.Client) *InfoSyncer {
+func NewInfoSyncer(id string, etcdCli *clientv3.Client, leaseID clientv3.LeaseID) *InfoSyncer {
 	return &InfoSyncer{
 		etcdCli:        etcdCli,
 		info:           getServerInfo(id),
 		serverInfoPath: fmt.Sprintf("%s/%s", ServerInformationPath, id),
+		sessionLeaseID: leaseID,
 	}
 }
 
-func getServerInfo(id string) *ServerInfo {
-	cfg := config.GetGlobalConfig()
-	info := &ServerInfo{
-		ID:         id,
-		IP:         cfg.AdvertiseAddress,
-		Port:       cfg.Port,
-		StatusPort: cfg.Status.StatusPort,
-		Lease:      cfg.Lease,
-	}
-	info.Version = mysql.ServerVersion
-	info.GitHash = printer.TiDBGitHash
-	return info
-}
-
 //GetServerInfo gets self server static information.
 func (is *InfoSyncer) GetServerInfo() *ServerInfo {
 	return is.info
@@ -130,7 +119,7 @@ func (is *InfoSyncer) StoreServerInfo(ctx context.Context) error {
 	if err != nil {
 		return errors.Trace(err)
 	}
-	err = ddl.PutKVToEtcd(ctx, is.etcdCli, keyOpDefaultRetryCnt, is.serverInfoPath, hack.String(infoBuf))
+	err = ddl.PutKVToEtcd(ctx, is.etcdCli, keyOpDefaultRetryCnt, is.serverInfoPath, hack.String(infoBuf), clientv3.WithLease(is.sessionLeaseID))
 	return errors.Trace(err)
 }
 
@@ -145,6 +134,12 @@ func (is *InfoSyncer) RemoveServerInfo() {
 	}
 }
 
+// Restart restart the info syncer with new session leaseID and store server info to etcd again.
+func (is *InfoSyncer) Restart(leaseID clientv3.LeaseID) {
+	is.sessionLeaseID = leaseID
+	is.StoreServerInfo(context.Background())
+}
+
 // getInfo gets server information from Etcd according to the key and opts.
 func getInfo(ctx context.Context, etcdCli *clientv3.Client, key string, retryCnt int, timeout time.Duration, opts ...clientv3.OpOption) (map[string]*ServerInfo, error) {
 	var err error
@@ -178,3 +173,18 @@ func getInfo(ctx context.Context, etcdCli *clientv3.Client, key string, retryCnt
 	}
 	return nil, errors.Trace(err)
 }
+
+// getServerInfo gets self tidb server information.
+func getServerInfo(id string) *ServerInfo {
+	cfg := config.GetGlobalConfig()
+	info := &ServerInfo{
+		ID:         id,
+		IP:         cfg.AdvertiseAddress,
+		Port:       cfg.Port,
+		StatusPort: cfg.Status.StatusPort,
+		Lease:      cfg.Lease,
+	}
+	info.Version = mysql.ServerVersion
+	info.GitHash = printer.TiDBGitHash
+	return info
+}

server/http_handler.go

@@ -1269,6 +1269,7 @@ func (h serverInfoHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	do, err := session.GetDomain(h.store.(kv.Storage))
 	if err != nil {
 		writeError(w, errors.New("create session error"))
+		log.Error(err)
 		return
 	}
 	info := serverInfo{}
@@ -1291,6 +1292,7 @@ func (h allServerInfoHandler) ServeHTTP(w http.ResponseWriter, req *http.Request
 	do, err := session.GetDomain(h.store.(kv.Storage))
 	if err != nil {
 		writeError(w, errors.New("create session error"))
+		log.Error(err)
 		return
 	}
 	ctx := context.Background()