host health check

cmd/climc/shell/hosts.go

@@ -793,4 +793,26 @@ func init() {
 		printObject(result)
 		return nil
 	})
+
+	type HostAutoMigrateOnHostDownOptions struct {
+		ID      string `help:"ID or name of host"`
+		Enable  bool   `help:"enable auto migrate"`
+		Disable bool   `help:"disable auto migrate"`
+	}
+	R(&HostAutoMigrateOnHostDownOptions{}, "host-auto-migrate-on-host-down", "Get change owner candidate domain list", func(s *mcclient.ClientSession, args *HostAutoMigrateOnHostDownOptions) error {
+		params := jsonutils.NewDict()
+		if args.Disable {
+			params.Set("auto_migrate_on_host_down", jsonutils.NewString("enable"))
+		} else if args.Enable {
+			params.Set("auto_migrate_on_host_down", jsonutils.NewString("disable"))
+		} else {
+			return fmt.Errorf("missing input enable or disable")
+		}
+		result, err := modules.Hosts.GetSpecific(s, args.ID, "auto-migrate-on-host-down", params)
+		if err != nil {
+			return err
+		}
+		printObject(result)
+		return nil
+	})
 }

pkg/apis/compute/host_const.go

@@ -123,3 +123,9 @@ const (
 	BOOT_MODE_PXE = "pxe"
 	BOOT_MODE_ISO = "iso"
 )
+
+const (
+	HOST_HEALTH_PREFIX         = "/onecloud/kvm/host/health"
+	HOST_HEALTH_STATUS_RUNNING = "running"
+	HOST_HEALTH_LOCK_PREFIX    = "host-health"
+)

pkg/cloudcommon/cronman/cronman.go

@@ -304,25 +304,7 @@ func (self *SCronJobManager) Start2(ctx context.Context, electObj *elect.Elect)
 		self.start(ctx)
 		return
 	}
-
-	go func() {
-		ch := make(chan elect.ElectEvent)
-		electObj.Subscribe(ch)
-		for {
-			select {
-			case ev := <-ch:
-				log.Infof("cronman: elect event %s: cronman", ev)
-				switch ev {
-				case elect.ElectEventWin:
-					self.start(ctx)
-				case elect.ElectEventLost:
-					self.Stop()
-				}
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
+	electObj.SubscribeWithAction(ctx, func() { self.start(ctx) }, self.Stop)
 }
 
 func (self *SCronJobManager) Start() {

pkg/cloudcommon/db/opslog.go

@@ -261,6 +261,7 @@ const (
 	ACT_GUEST_CREATE_FROM_IMPORT_FAIL    = "guest_create_from_import_fail"
 	ACT_GUEST_PANICKED                   = "guest_panicked"
 	ACT_HOST_MAINTENANCE                 = "host_maintenance"
+	ACT_HOST_DOWN                        = "host_down"
 
 	ACT_UPLOAD_OBJECT = "upload_obj"
 	ACT_DELETE_OBJECT = "delete_obj"

pkg/cloudcommon/elect/elect.go

@@ -72,7 +72,7 @@ type Elect struct {
 	config *EtcdConfig
 
 	mutex       *sync.Mutex
-	subscribers []chan ElectEvent
+	subscribers []chan electEvent
 }
 
 type ticket struct {
@@ -119,18 +119,18 @@ func (elect *Elect) Stop() {
 func (elect *Elect) Start(ctx context.Context) {
 	ctx, elect.stopFunc = context.WithCancel(ctx)
 
-	prev := ElectEventLost
+	prev := electEventLost
 	for {
 		select {
 		case <-ctx.Done():
 			log.Infof("elect bye")
 			return
 		default:
-			now := ElectEventWin
+			now := electEventWin
 			ticket, err := elect.do(ctx)
 			if err != nil {
 				ticket.tearup(ctx)
-				now = ElectEventLost
+				now = electEventLost
 				log.Errorf("elect error: %v", err)
 			}
 			if now != prev {
@@ -173,13 +173,13 @@ func (elect *Elect) do(ctx context.Context) (*ticket, error) {
 	return r, err
 }
 
-func (elect *Elect) Subscribe(ch chan ElectEvent) {
+func (elect *Elect) subscribe(ch chan electEvent) {
 	elect.mutex.Lock()
 	defer elect.mutex.Unlock()
 	elect.subscribers = append(elect.subscribers, ch)
 }
 
-func (elect *Elect) notify(ctx context.Context, ev ElectEvent) {
+func (elect *Elect) notify(ctx context.Context, ev electEvent) {
 	elect.mutex.Lock()
 	defer elect.mutex.Unlock()
 
@@ -193,18 +193,49 @@ func (elect *Elect) notify(ctx context.Context, ev ElectEvent) {
 	}
 }
 
-type ElectEvent int
+func (elect *Elect) SubscribeWithAction(ctx context.Context, onWin, onLost func()) {
+	go func() {
+		ch := make(chan electEvent, 3)
+		var ev electEvent
+		elect.subscribe(ch)
+		for {
+			select {
+			case ev = <-ch:
+			case <-ctx.Done():
+				return
+			}
+		drain:
+			for {
+				select {
+				case ev = <-ch:
+					continue
+				default:
+					break drain
+				}
+			}
+			log.Infof("elect event %s", ev)
+			switch ev {
+			case electEventWin:
+				onWin()
+			case electEventLost:
+				onLost()
+			}
+		}
+	}()
+}
+
+type electEvent int
 
 const (
-	ElectEventWin ElectEvent = iota
-	ElectEventLost
+	electEventWin electEvent = iota
+	electEventLost
 )
 
-func (ev ElectEvent) String() string {
+func (ev electEvent) String() string {
 	switch ev {
-	case ElectEventWin:
+	case electEventWin:
 		return "win"
-	case ElectEventLost:
+	case electEventLost:
 		return "lost"
 	default:
 		return "unexpected"

pkg/cloudcommon/etcd/etcd.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"go.etcd.io/etcd/clientv3"
+	"google.golang.org/grpc"
 
 	"yunion.io/x/log"
 
@@ -39,24 +40,43 @@ type SEtcdClient struct {
 
 	namespace string
 
-	leaseId clientv3.LeaseID
+	leaseId            clientv3.LeaseID
+	onKeepaliveFailure func()
+	leaseLiving        bool
 
 	watchers map[string]*SEtcdWatcher
 }
 
-func NewEtcdClient(opt *SEtcdOptions) (*SEtcdClient, error) {
+func defaultOnKeepAliveFailed() {
+	log.Fatalf("etcd keepalive failed")
+}
+
+func NewEtcdClient(opt *SEtcdOptions, onKeepaliveFailure func()) (*SEtcdClient, error) {
 	var err error
 	var tlsConfig *tls.Config
 
 	if opt.EtcdEnabldSsl {
-		tlsConfig, err = seclib2.InitTLSConfig(opt.EtcdSslCertfile, opt.EtcdSslKeyfile)
-		if err != nil {
-			log.Errorf("init tls config fail %s", err)
-			return nil, err
+		if opt.TLSConfig == nil {
+			if len(opt.EtcdSslCaCertfile) > 0 {
+				tlsConfig, err = seclib2.InitTLSConfigWithCA(
+					opt.EtcdSslCertfile, opt.EtcdSslKeyfile, opt.EtcdSslCaCertfile)
+			} else {
+				tlsConfig, err = seclib2.InitTLSConfig(opt.EtcdSslCertfile, opt.EtcdSslKeyfile)
+			}
+			if err != nil {
+				log.Errorf("init tls config fail %s", err)
+				return nil, err
+			}
+		} else {
+			tlsConfig = opt.TLSConfig
 		}
 	}
 
 	etcdClient := &SEtcdClient{}
+	if onKeepaliveFailure == nil {
+		onKeepaliveFailure = defaultOnKeepAliveFailed
+	}
+	etcdClient.onKeepaliveFailure = onKeepaliveFailure
 
 	timeoutSeconds := opt.EtcdTimeoutSeconds
 	if timeoutSeconds == 0 {
@@ -69,6 +89,10 @@ func NewEtcdClient(opt *SEtcdOptions) (*SEtcdClient, error) {
 		Username:    opt.EtcdUsername,
 		Password:    opt.EtcdPassword,
 		TLS:         tlsConfig,
+
+		DialOptions: []grpc.DialOption{
+			grpc.WithBlock(),
+		},
 	})
 	if err != nil {
 		return nil, err
@@ -95,7 +119,9 @@ func NewEtcdClient(opt *SEtcdOptions) (*SEtcdClient, error) {
 
 	err = etcdClient.startSession()
 	if err != nil {
-		etcdClient.Close()
+		if e := etcdClient.Close(); e != nil {
+			log.Errorf("etcd client close failed %s", e)
+		}
 		return nil, err
 	}
 	return etcdClient, nil
@@ -129,12 +155,18 @@ func (cli *SEtcdClient) startSession() error {
 	if err != nil {
 		return err
 	}
+	cli.leaseLiving = true
 
 	go func() {
 		for {
 			ka := <-ch
 			if ka == nil {
-				log.Fatalf("fail to keepalive")
+				cli.leaseLiving = false
+				log.Errorf("fail to keepalive sessoin")
+				if cli.onKeepaliveFailure != nil {
+					cli.onKeepaliveFailure()
+				}
+				break
 			} else {
 				log.Debugf("etcd session %d keepalive ttl: %d", ka.ID, ka.TTL)
 			}
@@ -144,6 +176,13 @@ func (cli *SEtcdClient) startSession() error {
 	return nil
 }
 
+func (cli *SEtcdClient) RestartSession() error {
+	if cli.leaseLiving {
+		return errors.New("session is living, can't restart")
+	}
+	return cli.startSession()
+}
+
 func (cli *SEtcdClient) getKey(key string) string {
 	if len(cli.namespace) > 0 {
 		return fmt.Sprintf("%s%s", cli.namespace, key)

pkg/cloudcommon/etcd/global.go

@@ -18,13 +18,13 @@ var (
 	defaultClient *SEtcdClient
 )
 
-func InitDefaultEtcdClient(opt *SEtcdOptions) error {
+func InitDefaultEtcdClient(opt *SEtcdOptions, onKeepaliveFailure func()) error {
 	if defaultClient != nil {
 		return nil
 	}
 
 	var err error
-	defaultClient, err = NewEtcdClient(opt)
+	defaultClient, err = NewEtcdClient(opt, onKeepaliveFailure)
 	return err
 }
 

pkg/cloudcommon/etcd/options.go

@@ -14,6 +14,8 @@
 
 package etcd
 
+import "crypto/tls"
+
 type SEtcdOptions struct {
 	EtcdEndpoint              []string `help:"etcd endpoints in format of addr:port"`
 	EtcdTimeoutSeconds        int      `default:"5" help:"etcd dial timeout in seconds"`
@@ -25,7 +27,9 @@ type SEtcdOptions struct {
 	EtcdUsername string `help:"etcd username"`
 	EtcdPassword string `help:"etcd password"`
 
-	EtcdEnabldSsl   bool   `help:"enable SSL/TLS"`
-	EtcdSslCertfile string `help:"ssl certification file"`
-	EtcdSslKeyfile  string `help:"ssl certification private key file"`
+	EtcdEnabldSsl     bool        `help:"enable SSL/TLS"`
+	EtcdSslCertfile   string      `help:"ssl certification file"`
+	EtcdSslKeyfile    string      `help:"ssl certification private key file"`
+	EtcdSslCaCertfile string      `help:"ssl ca certification file"`
+	TLSConfig         *tls.Config `help:"tls config"`
 }

pkg/cloudcommon/options/options.go

@@ -125,22 +125,25 @@ type DBOptions struct {
 
 	QueryOffsetOptimization bool `help:"apply query offset optimization"`
 
-	LockmanMethod  string   `help:"method for lock synchronization" choices:"inmemory|etcd" default:"inmemory"`
-	EtcdLockPrefix string   `help:"prefix of etcd lock records"`
-	EtcdLockTTL    int      `help:"ttl of etcd lock records"`
-	EtcdEndpoints  []string `help:"endpoints of etcd cluster"`
+	LockmanMethod string `help:"method for lock synchronization" choices:"inmemory|etcd" default:"inmemory"`
 
-	EtcdUsername string `help:"username of etcd cluster"`
-	EtcdPassword string `help:"password of etcd cluster"`
-
-	EtcdUseTLS        bool   `help:"use tls transport to connect etcd cluster" default:"false"`
-	EtcdSkipTLSVerify bool   `help:"skip tls verification" default:"false"`
-	EtcdCacert        string `help:"path to cacert for connecting to etcd cluster"`
-	EtcdCert          string `help:"path to cert file for connecting to etcd cluster"`
-	EtcdKey           string `help:"path to key file for connecting to etcd cluster"`
+	EtcdOptions
+	EtcdLockPrefix string `help:"prefix of etcd lock records"`
+	EtcdLockTTL    int    `help:"ttl of etcd lock records"`
 }
 
-func (this *DBOptions) GetEtcdTLSConfig() (*tls.Config, error) {
+type EtcdOptions struct {
+	EtcdEndpoints     []string `help:"endpoints of etcd cluster"`
+	EtcdUsername      string   `help:"username of etcd cluster"`
+	EtcdPassword      string   `help:"password of etcd cluster"`
+	EtcdUseTLS        bool     `help:"use tls transport to connect etcd cluster" default:"false"`
+	EtcdSkipTLSVerify bool     `help:"skip tls verification" default:"false"`
+	EtcdCacert        string   `help:"path to cacert for connecting to etcd cluster"`
+	EtcdCert          string   `help:"path to cert file for connecting to etcd cluster"`
+	EtcdKey           string   `help:"path to key file for connecting to etcd cluster"`
+}
+
+func (this *EtcdOptions) GetEtcdTLSConfig() (*tls.Config, error) {
 	var (
 		cert       tls.Certificate
 		certLoaded bool

pkg/cloudir/service/service.go

@@ -36,7 +36,7 @@ func StartService() {
 		log.Infof("Auth complete!!")
 	})
 
-	err := etcd.InitDefaultEtcdClient(&opts.SEtcdOptions)
+	err := etcd.InitDefaultEtcdClient(&opts.SEtcdOptions, nil)
 	if err != nil {
 		log.Fatalf("init etcd fail: %s", err)
 		return

pkg/cloutpost/service/service.go

@@ -41,7 +41,7 @@ func StartService() {
 		log.Infof("Auth complete!!")
 	})
 
-	err := etcd.InitDefaultEtcdClient(&opts.SEtcdOptions)
+	err := etcd.InitDefaultEtcdClient(&opts.SEtcdOptions, nil)
 	if err != nil {
 		log.Fatalf("init etcd fail: %s", err)
 	}

pkg/compute/models/guest_actions.go

@@ -3329,6 +3329,15 @@ func (self *SGuest) guestDisksStorageTypeIsLocal() bool {
 	return true
 }
 
+func (self *SGuest) guestDisksStorageTypeIsShared() bool {
+	for _, gd := range self.GetDisks() {
+		if gd.GetDisk().GetStorage().StorageType == api.STORAGE_LOCAL {
+			return false
+		}
+	}
+	return true
+}
+
 func (self *SGuest) PerformCreateBackup(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
 	if len(self.BackupHostId) > 0 {
 		return nil, httperrors.NewBadRequestError("Already have backup server")
@@ -4231,7 +4240,7 @@ func (manager *SGuestManager) AllowPerformBatchMigrate(ctx context.Context,
 	return db.IsAdminAllowPerform(userCred, manager, "batch-guest-migrate")
 }
 
-func (self *SGuest) validateForBatchMigrate(ctx context.Context) (*SGuest, error) {
+func (self *SGuest) validateForBatchMigrate(ctx context.Context, rescueMode bool) (*SGuest, error) {
 	guest := GuestManager.FetchGuestById(self.Id)
 	if guest.Hypervisor != api.HYPERVISOR_KVM {
 		return guest, httperrors.NewBadRequestError("guest %s hypervisor %s can't migrate",
@@ -4240,6 +4249,12 @@ func (self *SGuest) validateForBatchMigrate(ctx context.Context) (*SGuest, error
 	if len(guest.BackupHostId) > 0 {
 		return guest, httperrors.NewBadRequestError("guest %s has backup, can't migrate", guest.Name)
 	}
+	if rescueMode {
+		if !guest.guestDisksStorageTypeIsShared() {
+			return guest, httperrors.NewBadRequestError("can't rescue geust %s with local storage", guest.Name)
+		}
+		return guest, nil
+	}
 	if !utils.IsInStringArray(guest.Status, []string{api.VM_RUNNING, api.VM_READY, api.VM_UNKNOWN}) {
 		return guest, httperrors.NewBadRequestError("guest %s status %s can't migrate", guest.Name, guest.Status)
 	}
@@ -4300,7 +4315,7 @@ func (manager *SGuestManager) PerformBatchMigrate(ctx context.Context, userCred
 	for i := 0; i < len(guests); i++ {
 		lockman.LockObject(ctx, &guests[i])
 		defer lockman.ReleaseObject(ctx, &guests[i])
-		guest, err := guests[i].validateForBatchMigrate(ctx)
+		guest, err := guests[i].validateForBatchMigrate(ctx, false)
 		if err != nil {
 			return nil, err
 		}

pkg/compute/models/host_health.go

@@ -0,0 +1,118 @@
+package models
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"yunion.io/x/log"
+
+	api "yunion.io/x/onecloud/pkg/apis/compute"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
+	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
+)
+
+var hostHealthChecker *SHostHealthChecker
+
+type SHostHealthChecker struct {
+	// etcd client
+	cli *etcd.SEtcdClient
+	// time of wait host reconnect
+	timeout time.Duration
+}
+
+func hostKey(hostId string) string {
+	return fmt.Sprintf("%s/%s", api.HOST_HEALTH_PREFIX, hostId)
+}
+
+func InitHostHealthChecker(cli *etcd.SEtcdClient, timeout int) *SHostHealthChecker {
+	if hostHealthChecker != nil {
+		return hostHealthChecker
+	}
+	hostHealthChecker = &SHostHealthChecker{cli, time.Duration(timeout) * time.Second}
+	return hostHealthChecker
+}
+
+func (h *SHostHealthChecker) StartHostsHealthCheck(ctx context.Context) {
+	log.Infof("Start host health check......")
+	h.startHealthCheck(ctx)
+}
+
+func (h *SHostHealthChecker) startHealthCheck(ctx context.Context) {
+	q := HostManager.Query().IsTrue("enabled").IsTrue("enable_health_check").Equals("host_type", api.HOST_TYPE_HYPERVISOR)
+	rows, err := q.Rows()
+	if err != nil {
+		log.Errorf("HostHealth check Query hosts %s", err)
+		return
+	}
+	defer rows.Close()
+	for rows.Next() {
+		host := new(SHost)
+		q.Row2Struct(rows, host)
+		host.SetModelManager(HostManager, host)
+		h.startWatcher(ctx, host.Id)
+	}
+}
+
+func (h *SHostHealthChecker) startWatcher(ctx context.Context, hostId string) {
+	log.Debugf("Start watch host %s", hostId)
+	var (
+		ch  chan struct{}
+		key = hostKey(hostId)
+	)
+	_, err := h.cli.Get(ctx, key)
+	if err == etcd.ErrNoSuchKey {
+		log.Errorf("No such key %s", hostId)
+		ch = make(chan struct{})
+		go func() {
+			select {
+			case <-time.NewTimer(h.timeout).C:
+				h.onHostUnhealthy(ctx, hostId)
+			case <-ch:
+				h.startWatcher(ctx, hostId)
+			case <-ctx.Done():
+				log.Infof("exit watch host %s", hostId)
+			}
+		}()
+	}
+	h.cli.Watch(ctx, key, h.onHostOnline(hostId, ch), h.onHostOffline(hostId))
+}
+
+func (h *SHostHealthChecker) onHostUnhealthy(ctx context.Context, hostId string) {
+	lockman.LockRawObject(ctx, api.HOST_HEALTH_LOCK_PREFIX, hostId)
+	defer lockman.ReleaseRawObject(ctx, api.HOST_HEALTH_LOCK_PREFIX, hostId)
+	host := HostManager.FetchHostById(hostId)
+	if host.EnableHealthCheck == true {
+		host.OnHostDown(ctx, auth.AdminCredential())
+	}
+}
+
+func (h *SHostHealthChecker) onHostOnline(hostId string, ch chan struct{}) etcd.TEtcdCreateEventFunc {
+	return func(key, value []byte) {
+		log.Debugf("Got host online %s", hostId)
+		if ch != nil {
+			close(ch)
+		}
+	}
+}
+
+func (h *SHostHealthChecker) onHostOffline(hostId string) etcd.TEtcdModifyEventFunc {
+	return func(key, oldvalue, value []byte) {
+		log.Warningf("host %s disconnect with etcd", hostId)
+		host := HostManager.FetchHostById(hostId)
+		if host.EnableHealthCheck == true {
+			h.startWatcher(context.Background(), hostId)
+		}
+	}
+}
+
+func (h *SHostHealthChecker) WatchHost(ctx context.Context, hostId string) {
+	h.cli.Unwatch(hostKey(hostId))
+	h.startWatcher(ctx, hostId)
+}
+
+func (h *SHostHealthChecker) UnwatchHost(ctx context.Context, hostId string) {
+	log.Debugf("Unwatch host %s", hostId)
+	h.cli.Unwatch(hostKey(hostId))
+}

pkg/compute/models/hosts.go

@@ -165,7 +165,8 @@ type SHost struct {
 	// 是否处于维护状态
 	IsMaintenance bool `nullable:"true" default:"false" list:"domain"`
 
-	LastPingAt time.Time ``
+	LastPingAt        time.Time ``
+	EnableHealthCheck bool      `nullable:"true" default:"false"`
 
 	ResourceType string `width:"36" charset:"ascii" nullable:"false" list:"domain" update:"domain" create:"domain_optional" default:"shared"`
 
@@ -2521,7 +2522,6 @@ func (self *SHost) getGuestsResource(status string) *SHostGuestResourceUsage {
 }
 
 func (self *SHost) getMoreDetails(ctx context.Context, out api.HostDetails, showReason bool) api.HostDetails {
-
 	server := self.GetBaremetalServer()
 	if server != nil {
 		out.ServerId = server.Id
@@ -3413,11 +3413,17 @@ func (self *SHost) PerformOffline(ctx context.Context, userCred mcclient.TokenCr
 	if self.HostStatus != api.HOST_OFFLINE {
 		_, err := self.SaveUpdates(func() error {
 			self.HostStatus = api.HOST_OFFLINE
+			if jsonutils.QueryBoolean(data, "update_health_status", false) {
+				self.EnableHealthCheck = false
+			}
 			return nil
 		})
 		if err != nil {
 			return nil, err
 		}
+		if hostHealthChecker != nil {
+			hostHealthChecker.UnwatchHost(context.Background(), self.Id)
+		}
 		db.OpsLog.LogEvent(self, db.ACT_OFFLINE, "", userCred)
 		logclient.AddActionLogWithContext(ctx, self, logclient.ACT_OFFLINE, nil, userCred, true)
 		self.SyncAttachedStorageStatus()
@@ -3437,6 +3443,7 @@ func (self *SHost) PerformOnline(ctx context.Context, userCred mcclient.TokenCre
 		_, err := self.SaveUpdates(func() error {
 			self.LastPingAt = time.Now()
 			self.HostStatus = api.HOST_ONLINE
+			self.EnableHealthCheck = true
 			if !self.IsMaintaining() {
 				self.Status = api.BAREMETAL_RUNNING
 			}
@@ -3445,6 +3452,9 @@ func (self *SHost) PerformOnline(ctx context.Context, userCred mcclient.TokenCre
 		if err != nil {
 			return nil, err
 		}
+		if hostHealthChecker != nil {
+			hostHealthChecker.WatchHost(context.Background(), self.Id)
+		}
 		db.OpsLog.LogEvent(self, db.ACT_ONLINE, "", userCred)
 		logclient.AddActionLogWithContext(ctx, self, logclient.ACT_ONLINE, nil, userCred, true)
 		self.SyncAttachedStorageStatus()
@@ -3453,6 +3463,20 @@ func (self *SHost) PerformOnline(ctx context.Context, userCred mcclient.TokenCre
 	return nil, nil
 }
 
+func (self *SHost) AllowPerformAutoMigrateOnHostDown(ctx context.Context,
+	userCred mcclient.TokenCredential,
+	query jsonutils.JSONObject,
+	data jsonutils.JSONObject) bool {
+	return db.IsAdminAllowPerform(userCred, self, "auto-migrate-on-host-down")
+}
+
+func (self *SHost) PerformAutoMigrateOnHostDown(
+	ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject,
+) (jsonutils.JSONObject, error) {
+	val, _ := data.GetString("auto_migrate_on_host_down")
+	return nil, self.SetMetadata(ctx, "__auto_migrate_on_host_down", val, userCred)
+}
+
 func (self *SHost) StartSyncAllGuestsStatusTask(ctx context.Context, userCred mcclient.TokenCredential) error {
 	if task, err := taskman.TaskManager.NewTask(ctx, "BaremetalSyncAllGuestsStatusTask", self, userCred, nil, "", "", nil); err != nil {
 		log.Errorln(err)
@@ -4629,12 +4653,14 @@ func (manager *SHostManager) PingDetectionTask(ctx context.Context, userCred mcc
 	}
 	defer rows.Close()
 
+	data := jsonutils.NewDict()
+	data.Set("update_health_status", jsonutils.JSONFalse)
 	for rows.Next() {
 		var host = new(SHost)
 		q.Row2Struct(rows, host)
 		host.SetModelManager(manager, host)
 		lockman.LockObject(ctx, host)
-		host.PerformOffline(ctx, userCred, nil, nil)
+		host.PerformOffline(ctx, userCred, nil, data)
 		host.MarkGuestUnknown(userCred)
 		lockman.ReleaseObject(ctx, host)
 	}
@@ -4720,7 +4746,7 @@ func (host *SHost) PerformHostMaintenance(ctx context.Context, userCred mcclient
 	for i := 0; i < len(guests); i++ {
 		lockman.LockObject(ctx, &guests[i])
 		defer lockman.ReleaseObject(ctx, &guests[i])
-		guest, err := guests[i].validateForBatchMigrate(ctx)
+		guest, err := guests[i].validateForBatchMigrate(ctx, false)
 		if err != nil {
 			return nil, err
 		}
@@ -4749,6 +4775,50 @@ func (host *SHost) PerformHostMaintenance(ctx context.Context, userCred mcclient
 	return nil, host.StartMaintainTask(ctx, userCred, kwargs)
 }
 
+func (host *SHost) OnHostDown(ctx context.Context, userCred mcclient.TokenCredential) {
+	log.Errorf("watched host down %s", host.Id)
+	db.OpsLog.LogEvent(host, db.ACT_HOST_DOWN, "", userCred)
+	if host.GetMetadata("__auto_migrate_on_host_down", nil) == "enable" {
+		if err := host.MigrateSharedStorageServers(ctx, userCred); err != nil {
+			db.OpsLog.LogEvent(host, db.ACT_HOST_DOWN, fmt.Sprintf("migrate servers failed %s", err), userCred)
+		}
+	}
+	if _, err := host.SaveUpdates(func() error {
+		host.EnableHealthCheck = false
+		return nil
+	}); err != nil {
+		log.Errorf("update host %s failed %s", host.Id, err)
+	}
+}
+
+func (host *SHost) MigrateSharedStorageServers(ctx context.Context, userCred mcclient.TokenCredential) error {
+	var (
+		guests     = host.GetGuests()
+		hostGuests = []*api.GuestBatchMigrateParams{}
+	)
+
+	for i := 0; i < len(guests); i++ {
+		lockman.LockObject(ctx, &guests[i])
+		defer lockman.ReleaseObject(ctx, &guests[i])
+		_, err := guests[i].validateForBatchMigrate(ctx, true)
+		if err != nil {
+			continue
+		} else {
+			bmp := &api.GuestBatchMigrateParams{
+				Id:          guests[i].Id,
+				LiveMigrate: false,
+				RescueMode:  true,
+				OldStatus:   guests[i].Status,
+			}
+			guests[i].SetStatus(userCred, api.VM_START_MIGRATE, "host down")
+			hostGuests = append(hostGuests, bmp)
+		}
+	}
+	kwargs := jsonutils.NewDict()
+	kwargs.Set("guests", jsonutils.Marshal(hostGuests))
+	return GuestManager.StartHostGuestsMigrateTask(ctx, userCred, host, kwargs, "")
+}
+
 func (host *SHost) SetStatus(userCred mcclient.TokenCredential, status string, reason string) error {
 	err := host.SEnabledStatusInfrasResourceBase.SetStatus(userCred, status, reason)
 	if err != nil {

pkg/compute/options/options.go

@@ -129,6 +129,9 @@ type ComputeOptions struct {
 
 	BaremetalServerReuseHostIp bool `help:"baremetal server reuse host IP address, default true" default:"true"`
 
+	EnableHostHealthCheck bool `help:"enable host health check"`
+	HostHealthTimeout     int  `help:"second of wait host reconnect" default:"60"`
+
 	SCapabilityOptions
 	SASControllerOptions
 	common_options.CommonOptions

pkg/compute/service/service.go

@@ -22,6 +22,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 
 	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
 
 	api "yunion.io/x/onecloud/pkg/apis/compute"
 	"yunion.io/x/onecloud/pkg/cloudcommon"
@@ -29,6 +30,7 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/cronman"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
 	"yunion.io/x/onecloud/pkg/cloudcommon/elect"
+	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
 	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
 	_ "yunion.io/x/onecloud/pkg/compute/guestdrivers"
 	_ "yunion.io/x/onecloud/pkg/compute/hostdrivers"
@@ -43,7 +45,6 @@ import (
 )
 
 func StartService() {
-
 	opts := &options.Options
 	commonOpts := &options.Options.CommonOptions
 	baseOpts := &options.Options.BaseOptions
@@ -83,17 +84,25 @@ func StartService() {
 	defer cancelFunc()
 
 	if opts.LockmanMethod == common_options.LockMethodEtcd {
-		cfg, err := elect.NewEtcdConfigFromDBOptions(dbOpts)
+		etcdCfg, err := elect.NewEtcdConfigFromDBOptions(dbOpts)
 		if err != nil {
 			log.Fatalf("etcd config for elect: %v", err)
 		}
-		electObj, err = elect.NewElect(cfg, "@master-role")
+		electObj, err = elect.NewElect(etcdCfg, "@master-role")
 		if err != nil {
 			log.Fatalf("new elect instance: %v", err)
 		}
 		go electObj.Start(ctx)
 	}
 
+	if opts.EnableHostHealthCheck {
+		if err := initDefaultEtcdClient(dbOpts); err != nil {
+			log.Fatalf("init etcd client failed %s", err)
+		}
+		models.InitHostHealthChecker(etcd.Default(), opts.HostHealthTimeout).
+			StartHostsHealthCheck(context.Background())
+	}
+
 	if !opts.IsSlaveNode {
 		cron := cronman.InitCronJobManager(true, options.Options.CronJobWorkerCount)
 		cron.AddJobAtIntervals("CleanPendingDeleteServers", time.Duration(opts.PendingDeleteCheckSeconds)*time.Second, models.GuestManager.CleanPendingDeleteServers)
@@ -124,7 +133,6 @@ func StartService() {
 		cron.AddJobEveryFewDays("SyncDBInstanceSkus", opts.SyncSkusDay, opts.SyncSkusHour, 0, 0, models.SyncDBInstanceSkus, true)
 		cron.AddJobEveryFewDays("SyncElasticCacheSkus", opts.SyncSkusDay, opts.SyncSkusHour, 0, 0, models.SyncElasticCacheSkus, true)
 		cron.AddJobEveryFewDays("StorageSnapshotsRecycle", 1, 2, 0, 0, models.StorageManager.StorageSnapshotsRecycle, false)
-
 		go cron.Start2(ctx, electObj)
 
 		// init auto scaling controller
@@ -133,3 +141,24 @@ func StartService() {
 
 	app_common.ServeForever(app, baseOpts)
 }
+
+func initDefaultEtcdClient(opts *common_options.DBOptions) error {
+	if etcd.Default() != nil {
+		return nil
+	}
+	tlsConfig, err := opts.GetEtcdTLSConfig()
+	if err != nil {
+		return err
+	}
+	err = etcd.InitDefaultEtcdClient(&etcd.SEtcdOptions{
+		EtcdEndpoint:  opts.EtcdEndpoints,
+		EtcdUsername:  opts.EtcdUsername,
+		EtcdPassword:  opts.EtcdPassword,
+		EtcdEnabldSsl: opts.EtcdUseTLS,
+		TLSConfig:     tlsConfig,
+	}, nil)
+	if err != nil {
+		return errors.Wrap(err, "init default etcd client")
+	}
+	return nil
+}

pkg/hostman/guestman/guestman.go

@@ -276,7 +276,19 @@ func (m *SGuestManager) LoadServer(sid string) {
 	m.CandidateServers[sid] = guest
 }
 
-//isDeleted先不加，目测只是在ofp中用到了
+func (m *SGuestManager) ShutdownSharedStorageServers() {
+	m.Servers.Range(func(k, v interface{}) bool {
+		guest := v.(*SKVMGuestInstance)
+		if guest.IsSharedStorage() {
+			log.Infof("Start shutdown server %s", guest.GetName())
+			if !guest.scriptStop() {
+				log.Errorf("shutdown server %s failed", guest.GetName())
+			}
+		}
+		return true
+	})
+}
+
 func (m *SGuestManager) GetGuestNicDesc(mac, ip, port, bridge string, isCandidate bool) (jsonutils.JSONObject, jsonutils.JSONObject) {
 	if isCandidate {
 		return m.getGuestNicDescInCandidate(mac, ip, port, bridge)

pkg/hostman/guestman/qemu-kvm.go

@@ -1516,3 +1516,19 @@ func (s *SKVMGuestInstance) BlockIoThrottle(ctx context.Context, bps, iops int64
 	task := SGuestBlockIoThrottleTask{s, ctx, bps, iops}
 	return task.Start()
 }
+
+func (s *SKVMGuestInstance) IsSharedStorage() bool {
+	disks, _ := s.Desc.GetArray("disks")
+	for i := 0; i < len(disks); i++ {
+		diskPath, _ := disks[i].GetString("path")
+		disk := storageman.GetManager().GetDiskByPath(diskPath)
+		if disk == nil {
+			log.Errorf("failed find disk by path %s", diskPath)
+			return false
+		}
+		if !utils.IsInStringArray(disk.GetType(), compute.SHARED_STORAGE) {
+			return false
+		}
+	}
+	return true
+}

pkg/hostman/host_health/doc.go

@@ -0,0 +1 @@
+package host_health // import "yunion.io/x/onecloud/pkg/hostman/host_health"

pkg/hostman/host_health/etcd.go

@@ -0,0 +1,99 @@
+package host_health
+
+import (
+	"context"
+	"fmt"
+
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	api "yunion.io/x/onecloud/pkg/apis/compute"
+	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
+	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
+	"yunion.io/x/onecloud/pkg/hostman/options"
+)
+
+func NewEtcdOptions(
+	opt *common_options.EtcdOptions, leaseTimeout, dialTimeout, requestTimeout int,
+) (*etcd.SEtcdOptions, error) {
+	cfg, err := opt.GetEtcdTLSConfig()
+	if err != nil {
+		return nil, err
+	}
+	return &etcd.SEtcdOptions{
+		EtcdEndpoint:              opt.EtcdEndpoints,
+		EtcdLeaseExpireSeconds:    leaseTimeout,
+		EtcdTimeoutSeconds:        dialTimeout,
+		EtcdRequestTimeoutSeconds: requestTimeout,
+		EtcdEnabldSsl:             opt.EtcdUseTLS,
+		TLSConfig:                 cfg,
+	}, nil
+}
+
+type SEtcdClient struct {
+	cli *etcd.SEtcdClient
+
+	hostId        string
+	onUnhealthy   func()
+	timeout       int
+	requestExpend int
+}
+
+func NewEtcdClient(opt *common_options.EtcdOptions, hostId string) (*SEtcdClient, error) {
+	var dialTimeout, requestTimeout = 3, 2
+	cfg, err := NewEtcdOptions(opt, options.HostOptions.HostLeaseTimeout, dialTimeout, requestTimeout)
+	if err != nil {
+		return nil, err
+	}
+	cli := new(SEtcdClient)
+	err = etcd.InitDefaultEtcdClient(cfg, cli.OnKeepaliveFailure)
+	if err != nil {
+		return nil, errors.Wrap(err, "init default etcd client")
+	}
+	cli.cli = etcd.Default()
+	cli.hostId = hostId
+	cli.timeout = options.HostOptions.HostHealthTimeout - options.HostOptions.HostLeaseTimeout
+	cli.requestExpend = requestTimeout
+	return cli, nil
+}
+
+func (c *SEtcdClient) StartHostHealthCheck(ctx context.Context) error {
+	return c.cli.PutSession(ctx,
+		fmt.Sprintf("%s/%s", api.HOST_HEALTH_PREFIX, c.hostId),
+		api.HOST_HEALTH_STATUS_RUNNING,
+	)
+}
+
+func (c *SEtcdClient) SetOnUnhealthy(onUnhealthy func()) {
+	c.onUnhealthy = onUnhealthy
+}
+
+func (c *SEtcdClient) OnKeepaliveFailure() {
+	var timeout = c.timeout
+	for timeout > 0 {
+		timeout -= c.requestExpend
+		if err := c.cli.RestartSession(); err != nil {
+			log.Errorf("restart session failed %s", err)
+		} else {
+			break
+		}
+	}
+	if timeout > 0 {
+		if err := c.cli.PutSession(context.Background(),
+			fmt.Sprintf("%s/%s", api.HOST_HEALTH_PREFIX, c.hostId),
+			api.HOST_HEALTH_STATUS_RUNNING,
+		); err != nil {
+			log.Errorf("put host key failed %s", err)
+		} else {
+			return
+		}
+	}
+	log.Errorln("keep etcd lease failed")
+	if c.onUnhealthy != nil {
+		c.onUnhealthy()
+	}
+}
+
+func (c *SEtcdClient) Stop() error {
+	return nil
+}

pkg/hostman/host_health/health_manager.go

@@ -0,0 +1,76 @@
+package host_health
+
+import (
+	"context"
+	"fmt"
+
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/hostman/guestman"
+	"yunion.io/x/onecloud/pkg/hostman/options"
+)
+
+type Status int
+
+const (
+	UNKNOWN Status = iota
+	HEALTHY
+	UNHEALTHY
+)
+
+type SHostHealthManager struct {
+	cli          Client
+	status       Status
+	guestManager *guestman.SGuestManager
+}
+
+var manager *SHostHealthManager
+
+func InitHostHealthManager(hostId string) (*SHostHealthManager, error) {
+	if manager != nil {
+		return manager, nil
+	}
+	switch options.HostOptions.HealthDriver {
+	case "etcd":
+		cli, err := NewEtcdClient(&options.HostOptions.EtcdOptions, hostId)
+		if err != nil {
+			return nil, errors.Wrap(err, "new etcd client")
+		}
+		manager = new(SHostHealthManager)
+		manager.cli = cli
+	default:
+		return nil, fmt.Errorf("not support health driver %s", options.HostOptions.HealthDriver)
+	}
+	manager.guestManager = guestman.GetGuestManager()
+	manager.cli.SetOnUnhealthy(manager.OnUnhealth)
+	go manager.StartHealthCheck()
+	return manager, nil
+}
+
+func (m *SHostHealthManager) StartHealthCheck() error {
+	return m.cli.StartHostHealthCheck(context.Background())
+}
+
+func (m *SHostHealthManager) OnUnhealth() {
+	log.Debugf("Host unhealthy, going to shotdown servers")
+	m.status = UNHEALTHY
+	if options.HostOptions.HealthShutdownServers {
+		m.shutdownServers()
+	}
+}
+
+// shutdown servers used shared storage
+func (m *SHostHealthManager) shutdownServers() {
+	m.guestManager.ShutdownSharedStorageServers()
+}
+
+func (m *SHostHealthManager) Stop() error {
+	return m.cli.Stop()
+}
+
+type Client interface {
+	StartHostHealthCheck(context.Context) error
+	SetOnUnhealthy(func())
+	Stop() error
+}

pkg/hostman/hostinfo/hostinfo.go

@@ -37,6 +37,7 @@ import (
 
 	api "yunion.io/x/onecloud/pkg/apis/compute"
 	"yunion.io/x/onecloud/pkg/hostman/guestfs/fsdriver"
+	"yunion.io/x/onecloud/pkg/hostman/host_health"
 	deployapi "yunion.io/x/onecloud/pkg/hostman/hostdeployer/apis"
 	"yunion.io/x/onecloud/pkg/hostman/hostinfo/hostbridge"
 	"yunion.io/x/onecloud/pkg/hostman/hostutils"
@@ -993,8 +994,12 @@ func (h *SHostInfo) getReservedMem() int {
 }
 
 func (h *SHostInfo) PutHostOffline() {
+	data := jsonutils.NewDict()
+	if options.HostOptions.EnableHealthChecker {
+		data.Set("update_health_status", jsonutils.JSONTrue)
+	}
 	_, err := modules.Hosts.PerformAction(
-		h.GetSession(), h.HostId, "offline", nil)
+		h.GetSession(), h.HostId, "offline", data)
 	if err != nil {
 		h.onFail(err)
 	} else {
@@ -1003,9 +1008,17 @@ func (h *SHostInfo) PutHostOffline() {
 }
 
 func (h *SHostInfo) PutHostOnline() error {
+	data := jsonutils.NewDict()
+	if options.HostOptions.EnableHealthChecker {
+		_, err := host_health.InitHostHealthManager(h.HostId)
+		if err != nil {
+			log.Fatalf("Init host health manager failed %s", err)
+		}
+		data.Set("update_health_status", jsonutils.JSONTrue)
+	}
+
 	_, err := modules.Hosts.PerformAction(
-		h.GetSession(),
-		h.HostId, "online", nil)
+		h.GetSession(), h.HostId, "online", data)
 	return err
 }
 

pkg/hostman/options/options.go

@@ -22,6 +22,7 @@ import (
 
 type SHostOptions struct {
 	common_options.CommonOptions
+	common_options.EtcdOptions
 
 	HostType        string   `help:"Host server type, either hypervisor or kubelet" default:"hypervisor"`
 	ListenInterface string   `help:"Master address of host server"`
@@ -122,6 +123,12 @@ type SHostOptions struct {
 	OvnEncapIp           string `help:"encap ip for ovn datapath.  Default to output src address of default route" default:"$HOST_OVN_ENCAP_IP"`
 	OvnIntegrationBridge string `help:"name of integration bridge for logical ports" default:"brvpc" default:"$HOST_OVN_INTEGRATION_BRIDGE|brvpc"`
 	OvnMappedBridge      string `help:"name of bridge for mapped traffic management" default:"mapped" default:"$HOST_OVN_MAPPED_BRIDGE|brmapped"`
+
+	EnableHealthChecker   bool   `help:"enable host health checker"`
+	HealthDriver          string `help:"Component save host health state" default:"etcd"`
+	HealthShutdownServers bool   `help:"Host healthor disconenct with controller shutdown shared storage servers" default:"false"`
+	HostHealthTimeout     int    `help:"host health timeout" default:"30"`
+	HostLeaseTimeout      int    `help:"lease timeout" default:"10"`
 }
 
 var (

pkg/util/seclib2/tls.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/pem"
 	"io/ioutil"
 
 	"yunion.io/x/log"
@@ -56,6 +57,71 @@ func splitCert(certBytes []byte) [][]byte {
 	return ret
 }
 
+func InitTLSConfigWithCA(certFile, keyFile, caCertFile string) (*tls.Config, error) {
+	cert, err := NewCert(certFile, keyFile, nil)
+	if err != nil {
+		return nil, err
+	}
+	cfg := &tls.Config{}
+
+	cfg.RootCAs, err = NewCertPool([]string{caCertFile})
+	if err != nil {
+		return nil, err
+	}
+	cfg.Certificates = []tls.Certificate{*cert}
+	return cfg, nil
+}
+
+// NewCertPool creates x509 certPool with provided CA files.
+func NewCertPool(CAFiles []string) (*x509.CertPool, error) {
+	certPool := x509.NewCertPool()
+
+	for _, CAFile := range CAFiles {
+		pemByte, err := ioutil.ReadFile(CAFile)
+		if err != nil {
+			return nil, err
+		}
+
+		for {
+			var block *pem.Block
+			block, pemByte = pem.Decode(pemByte)
+			if block == nil {
+				break
+			}
+			cert, err := x509.ParseCertificate(block.Bytes)
+			if err != nil {
+				return nil, err
+			}
+			certPool.AddCert(cert)
+		}
+	}
+
+	return certPool, nil
+}
+
+// NewCert generates TLS cert by using the given cert,key and parse function.
+func NewCert(certfile, keyfile string, parseFunc func([]byte, []byte) (tls.Certificate, error)) (*tls.Certificate, error) {
+	cert, err := ioutil.ReadFile(certfile)
+	if err != nil {
+		return nil, err
+	}
+
+	key, err := ioutil.ReadFile(keyfile)
+	if err != nil {
+		return nil, err
+	}
+
+	if parseFunc == nil {
+		parseFunc = tls.X509KeyPair
+	}
+
+	tlsCert, err := parseFunc(cert, key)
+	if err != nil {
+		return nil, err
+	}
+	return &tlsCert, nil
+}
+
 func InitTLSConfig(certFile, keyFile string) (*tls.Config, error) {
 	allCertPEM, err := ioutil.ReadFile(certFile)
 	if err != nil {