opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-10-29 17:58:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(region): host health use sync map (#15067)

Browse Source 
* fix(region): host health use sync map

Signed-off-by: wanyaoqi <d3lx.yq@gmail.com>

* fix auto migrate on host shutdown

Signed-off-by: wanyaoqi <d3lx.yq@gmail.com>
This commit is contained in:
wanyaoqi
2022-09-28 14:12:46 +08:00
committed by GitHub
parent 5e6935bc3a
commit 59fd4f61be
2 changed files with 17 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/compute/models/host_health.go
+16 -9
View File
@@ -17,6 +17,7 @@ package models
import (
"context"
"fmt"
"sync"
"time"
"yunion.io/x/log"
@@ -37,7 +38,7 @@ type SHostHealthChecker struct {
// time of wait host reconnect
timeout time.Duration
// hosts chan
hc map[string]chan struct{}
hc *sync.Map
}
func hostKey(hostname string) string {
@@ -51,7 +52,7 @@ func InitHostHealthChecker(cli *etcd.SEtcdClient, timeout int) *SHostHealthCheck
hostHealthChecker = &SHostHealthChecker{
cli: cli,
timeout: time.Duration(timeout) * time.Second,
hc: make(map[string]chan struct{}),
hc: new(sync.Map),
}
return hostHealthChecker
}
@@ -61,6 +62,11 @@ func (h *SHostHealthChecker) StartHostsHealthCheck(ctx context.Context) error {
return h.startHealthCheck(ctx)
}
func (h *SHostHealthChecker) load(hostname string) chan struct{} {
v, _ := h.hc.Load(hostname)
return v.(chan struct{})
}
func (h *SHostHealthChecker) startHealthCheck(ctx context.Context) error {
q := HostManager.Query().IsTrue("enabled").IsTrue("enable_health_check").Equals("host_type", api.HOST_TYPE_HYPERVISOR)
rows, err := q.Rows()
@@ -88,8 +94,8 @@ func (h *SHostHealthChecker) startWatcher(ctx context.Context, hostname string)
log.Infof("Start watch host %s", hostname)
var key = hostKey(hostname)
if _, ok := h.hc[hostname]; !ok {
h.hc[hostname] = make(chan struct{})
if _, ok := h.hc.Load(hostname); !ok {
h.hc.Store(hostname, make(chan struct{}))
}
if err := h.cli.Watch(
ctx, key,
@@ -108,7 +114,7 @@ func (h *SHostHealthChecker) startWatcher(ctx context.Context, hostname string)
select {
case <-time.NewTimer(h.timeout).C:
h.onHostUnhealthy(ctx, hostname)
case <-h.hc[hostname]:
case <-h.load(hostname):
if _err := h.startWatcher(ctx, hostname); _err != nil {
log.Errorf("failed start watcher %s", _err)
}
@@ -136,8 +142,9 @@ func (h *SHostHealthChecker) onHostUnhealthy(ctx context.Context, hostname strin
func (h *SHostHealthChecker) onHostOnline(ctx context.Context, hostname string) etcd.TEtcdCreateEventFunc {
return func(ctx context.Context, key, value []byte) {
log.Infof("Got host online %s", hostname)
if h.hc[hostname] != nil {
h.hc[hostname] <- struct{}{}
if v, ok := h.hc.Load(hostname); ok {
c := v.(chan struct{})
c <- struct{}{}
}
}
}
@@ -148,7 +155,7 @@ func (h *SHostHealthChecker) processHostOffline(ctx context.Context, hostname st
select {
case <-time.NewTimer(h.timeout).C:
h.onHostUnhealthy(ctx, hostname)
case <-h.hc[hostname]:
case <-h.load(hostname):
if err := h.startWatcher(ctx, hostname); err != nil {
log.Errorf("failed start watcher %s", err)
}
@@ -178,5 +185,5 @@ func (h *SHostHealthChecker) WatchHost(ctx context.Context, hostname string) err
func (h *SHostHealthChecker) UnwatchHost(ctx context.Context, hostname string) {
log.Infof("Unwatch host %s", hostname)
h.cli.Unwatch(hostKey(hostname))
delete(h.hc, hostname)
h.hc.Delete(hostname)
}
pkg/compute/models/hosts.go
+1 -1
View File
@@ -5630,7 +5630,7 @@ func (host *SHost) OnHostDown(ctx context.Context, userCred mcclient.TokenCreden
hostHealthChecker.UnwatchHost(ctx, host.GetHostnameByName())
if host.HostStatus == api.HOST_OFFLINE && !host.EnableHealthCheck &&
host.autoMigrateOnHostShutdown(ctx) {
!host.autoMigrateOnHostShutdown(ctx) {
// hostagent requested offline, and not enable auto migrate on host shutdown
log.Infof("host not need auto migrate on host shutdown")
return