diff --git a/cmd/climc/shell/hosts.go b/cmd/climc/shell/hosts.go index 9e8f6f31da..c34a23ecfd 100644 --- a/cmd/climc/shell/hosts.go +++ b/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 + }) } diff --git a/pkg/apis/compute/host_const.go b/pkg/apis/compute/host_const.go index b4ca9513f7..5c74da65f7 100644 --- a/pkg/apis/compute/host_const.go +++ b/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" +) diff --git a/pkg/cloudcommon/cronman/cronman.go b/pkg/cloudcommon/cronman/cronman.go index 569cc48855..151c2f6f6e 100644 --- a/pkg/cloudcommon/cronman/cronman.go +++ b/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() { diff --git a/pkg/cloudcommon/db/opslog.go b/pkg/cloudcommon/db/opslog.go index 40156f1ec4..fcd9e6392a 100644 --- a/pkg/cloudcommon/db/opslog.go +++ b/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" diff --git a/pkg/cloudcommon/elect/elect.go b/pkg/cloudcommon/elect/elect.go index 24f11857f3..54c165e737 100644 --- a/pkg/cloudcommon/elect/elect.go +++ b/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" diff --git a/pkg/cloudcommon/etcd/etcd.go b/pkg/cloudcommon/etcd/etcd.go index 35ea6672d7..28a4ebdec7 100644 --- a/pkg/cloudcommon/etcd/etcd.go +++ b/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) diff --git a/pkg/cloudcommon/etcd/global.go b/pkg/cloudcommon/etcd/global.go index fa1f3b9868..f720a46570 100644 --- a/pkg/cloudcommon/etcd/global.go +++ b/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 } diff --git a/pkg/cloudcommon/etcd/options.go b/pkg/cloudcommon/etcd/options.go index 3309ba0b17..eb8cb86d44 100644 --- a/pkg/cloudcommon/etcd/options.go +++ b/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"` } diff --git a/pkg/cloudcommon/options/options.go b/pkg/cloudcommon/options/options.go index c8b11fa3e1..73486a4a6f 100644 --- a/pkg/cloudcommon/options/options.go +++ b/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 diff --git a/pkg/cloudir/service/service.go b/pkg/cloudir/service/service.go index d3f09a9c98..6ddd435e64 100644 --- a/pkg/cloudir/service/service.go +++ b/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 diff --git a/pkg/cloutpost/service/service.go b/pkg/cloutpost/service/service.go index d694b0181f..b861feccf6 100644 --- a/pkg/cloutpost/service/service.go +++ b/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) } diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index 772d84bfac..3f1796bde3 100644 --- a/pkg/compute/models/guest_actions.go +++ b/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 } diff --git a/pkg/compute/models/host_health.go b/pkg/compute/models/host_health.go new file mode 100644 index 0000000000..611a8837f8 --- /dev/null +++ b/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)) +} diff --git a/pkg/compute/models/hosts.go b/pkg/compute/models/hosts.go index fc82a7235d..4b6437a720 100644 --- a/pkg/compute/models/hosts.go +++ b/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 { diff --git a/pkg/compute/options/options.go b/pkg/compute/options/options.go index 22d3625ba7..83f3154d6c 100644 --- a/pkg/compute/options/options.go +++ b/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 diff --git a/pkg/compute/service/service.go b/pkg/compute/service/service.go index e12597b199..821037f7e0 100644 --- a/pkg/compute/service/service.go +++ b/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 +} diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index b9ddefab60..38a95a79da 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/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) diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index a2ec420110..8bd3a08c25 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/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 +} diff --git a/pkg/hostman/host_health/doc.go b/pkg/hostman/host_health/doc.go new file mode 100644 index 0000000000..2f5277bbe1 --- /dev/null +++ b/pkg/hostman/host_health/doc.go @@ -0,0 +1 @@ +package host_health // import "yunion.io/x/onecloud/pkg/hostman/host_health" diff --git a/pkg/hostman/host_health/etcd.go b/pkg/hostman/host_health/etcd.go new file mode 100644 index 0000000000..5f54e21ba4 --- /dev/null +++ b/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 +} diff --git a/pkg/hostman/host_health/health_manager.go b/pkg/hostman/host_health/health_manager.go new file mode 100644 index 0000000000..eae2749625 --- /dev/null +++ b/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 +} diff --git a/pkg/hostman/hostinfo/hostinfo.go b/pkg/hostman/hostinfo/hostinfo.go index 57aeec138a..c05a95d76d 100644 --- a/pkg/hostman/hostinfo/hostinfo.go +++ b/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 } diff --git a/pkg/hostman/options/options.go b/pkg/hostman/options/options.go index fb3420a3f3..d9f7735325 100644 --- a/pkg/hostman/options/options.go +++ b/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 ( diff --git a/pkg/util/seclib2/tls.go b/pkg/util/seclib2/tls.go index 6fc66d69b3..4a95dccd88 100644 --- a/pkg/util/seclib2/tls.go +++ b/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 {