healthcheck latency of the provider is also stored in the extra, without compromising rest api compatibility

This commit is contained in:
tommytag 2023-12-06 17:11:24 +08:00
parent 2d7538aca6
commit f63acc0202
7 changed files with 69 additions and 109 deletions

View File

@ -26,21 +26,20 @@ const (
defaultHistoriesNum = 10 defaultHistoriesNum = 10
) )
type extraProxyState struct { type internalProxyState struct {
history *queue.Queue[C.DelayHistory]
alive atomic.Bool alive atomic.Bool
history *queue.Queue[C.DelayHistory]
} }
type Proxy struct { type Proxy struct {
C.ProxyAdapter C.ProxyAdapter
history *queue.Queue[C.DelayHistory] history *queue.Queue[C.DelayHistory]
alive atomic.Bool alive atomic.Bool
url string extra *xsync.MapOf[string, *internalProxyState]
extra *xsync.MapOf[string, *extraProxyState]
} }
// AliveForTestUrl implements C.Proxy // Alive implements C.Proxy
func (p *Proxy) AliveForTestUrl(url string) bool { func (p *Proxy) Alive(url string) bool {
if state, ok := p.extra.Load(url); ok { if state, ok := p.extra.Load(url); ok {
return state.alive.Load() return state.alive.Load()
} }
@ -48,10 +47,6 @@ func (p *Proxy) AliveForTestUrl(url string) bool {
return p.alive.Load() return p.alive.Load()
} }
func (p *Proxy) OriginalHealthCheckUrl(url string) {
p.url = url
}
// Dial implements C.Proxy // Dial implements C.Proxy
func (p *Proxy) Dial(metadata *C.Metadata) (C.Conn, error) { func (p *Proxy) Dial(metadata *C.Metadata) (C.Conn, error) {
ctx, cancel := context.WithTimeout(context.Background(), C.DefaultTCPTimeout) ctx, cancel := context.WithTimeout(context.Background(), C.DefaultTCPTimeout)
@ -81,7 +76,7 @@ func (p *Proxy) ListenPacketContext(ctx context.Context, metadata *C.Metadata, o
// DelayHistory implements C.Proxy // DelayHistory implements C.Proxy
func (p *Proxy) DelayHistory() []C.DelayHistory { func (p *Proxy) DelayHistory() []C.DelayHistory {
queueM := p.history.Copy() queueM := p.history.Copy()
histories := []C.DelayHistory{} var histories []C.DelayHistory
for _, item := range queueM { for _, item := range queueM {
histories = append(histories, item) histories = append(histories, item)
} }
@ -97,71 +92,52 @@ func (p *Proxy) DelayHistoryForTestUrl(url string) []C.DelayHistory {
queueM = state.history.Copy() queueM = state.history.Copy()
} }
if queueM == nil { var histories []C.DelayHistory
queueM = p.history.Copy()
}
histories := []C.DelayHistory{}
for _, item := range queueM { for _, item := range queueM {
histories = append(histories, item) histories = append(histories, item)
} }
return histories return histories
} }
func (p *Proxy) ExtraDelayHistory() map[string][]C.DelayHistory { // ExtraDelayHistories return all delay histories for each test URL
extraHistory := map[string][]C.DelayHistory{} // implements C.Proxy
func (p *Proxy) ExtraDelayHistories() map[string]C.ProxyState {
p.extra.Range(func(k string, v *extraProxyState) bool { histories := map[string]C.ProxyState{}
p.extra.Range(func(k string, v *internalProxyState) bool {
testUrl := k testUrl := k
state := v state := v
histories := []C.DelayHistory{}
queueM := state.history.Copy() queueM := state.history.Copy()
var history []C.DelayHistory
for _, item := range queueM { for _, item := range queueM {
histories = append(histories, item) history = append(history, item)
} }
extraHistory[testUrl] = histories histories[testUrl] = C.ProxyState{
Alive: state.alive.Load(),
History: history,
}
return true return true
}) })
return extraHistory return histories
} }
// LastDelay return last history record. if proxy is not alive, return the max value of uint16. // LastDelayForTestUrl return last history record of the specified URL. if proxy is not alive, return the max value of uint16.
// implements C.Proxy // implements C.Proxy
func (p *Proxy) LastDelay() (delay uint16) {
var max uint16 = 0xffff
if !p.alive.Load() {
return max
}
history := p.history.Last()
if history.Delay == 0 {
return max
}
return history.Delay
}
// LastDelayForTestUrl implements C.Proxy
func (p *Proxy) LastDelayForTestUrl(url string) (delay uint16) { func (p *Proxy) LastDelayForTestUrl(url string) (delay uint16) {
var max uint16 = 0xffff var max uint16 = 0xffff
alive := p.alive.Load() alive := false
history := p.history.Last() var history C.DelayHistory
if state, ok := p.extra.Load(url); ok { if state, ok := p.extra.Load(url); ok {
alive = state.alive.Load() alive = state.alive.Load()
history = state.history.Last() history = state.history.Last()
} }
if !alive { if !alive || history.Delay == 0 {
return max
}
if history.Delay == 0 {
return max return max
} }
return history.Delay return history.Delay
@ -177,8 +153,8 @@ func (p *Proxy) MarshalJSON() ([]byte, error) {
mapping := map[string]any{} mapping := map[string]any{}
_ = json.Unmarshal(inner, &mapping) _ = json.Unmarshal(inner, &mapping)
mapping["history"] = p.DelayHistory() mapping["history"] = p.DelayHistory()
mapping["extra"] = p.ExtraDelayHistory() mapping["extra"] = p.ExtraDelayHistories()
mapping["alive"] = p.AliveForTestUrl(p.url) mapping["alive"] = p.alive.Load()
mapping["name"] = p.Name() mapping["name"] = p.Name()
mapping["udp"] = p.SupportUDP() mapping["udp"] = p.SupportUDP()
mapping["xudp"] = p.SupportXUDP() mapping["xudp"] = p.SupportXUDP()
@ -191,31 +167,20 @@ func (p *Proxy) MarshalJSON() ([]byte, error) {
func (p *Proxy) URLTest(ctx context.Context, url string, expectedStatus utils.IntRanges[uint16]) (t uint16, err error) { func (p *Proxy) URLTest(ctx context.Context, url string, expectedStatus utils.IntRanges[uint16]) (t uint16, err error) {
defer func() { defer func() {
alive := err == nil alive := err == nil
if len(p.url) == 0 || url == p.url {
p.alive.Store(alive)
record := C.DelayHistory{Time: time.Now()} record := C.DelayHistory{Time: time.Now()}
if alive { if alive {
record.Delay = t record.Delay = t
} }
p.alive.Store(alive)
p.history.Put(record) p.history.Put(record)
if p.history.Len() > defaultHistoriesNum { if p.history.Len() > defaultHistoriesNum {
p.history.Pop() p.history.Pop()
} }
// test URL configured by the proxy provider
if len(p.url) == 0 {
p.url = url
}
} else {
record := C.DelayHistory{Time: time.Now()}
if alive {
record.Delay = t
}
state, ok := p.extra.Load(url) state, ok := p.extra.Load(url)
if !ok { if !ok {
state = &extraProxyState{ state = &internalProxyState{
history: queue.New[C.DelayHistory](defaultHistoriesNum), history: queue.New[C.DelayHistory](defaultHistoriesNum),
alive: atomic.NewBool(true), alive: atomic.NewBool(true),
} }
@ -227,7 +192,7 @@ func (p *Proxy) URLTest(ctx context.Context, url string, expectedStatus utils.In
if state.history.Len() > defaultHistoriesNum { if state.history.Len() > defaultHistoriesNum {
state.history.Pop() state.history.Pop()
} }
}
}() }()
unifiedDelay := UnifiedDelay.Load() unifiedDelay := UnifiedDelay.Load()
@ -304,8 +269,7 @@ func NewProxy(adapter C.ProxyAdapter) *Proxy {
ProxyAdapter: adapter, ProxyAdapter: adapter,
history: queue.New[C.DelayHistory](defaultHistoriesNum), history: queue.New[C.DelayHistory](defaultHistoriesNum),
alive: atomic.NewBool(true), alive: atomic.NewBool(true),
url: "", extra: xsync.NewMapOf[string, *internalProxyState]()}
extra: xsync.NewMapOf[string, *extraProxyState]()}
} }
func urlToMetadata(rawURL string) (addr C.Metadata, err error) { func urlToMetadata(rawURL string) (addr C.Metadata, err error) {

View File

@ -102,12 +102,12 @@ func (f *Fallback) findAliveProxy(touch bool) C.Proxy {
proxies := f.GetProxies(touch) proxies := f.GetProxies(touch)
for _, proxy := range proxies { for _, proxy := range proxies {
if len(f.selected) == 0 { if len(f.selected) == 0 {
if proxy.AliveForTestUrl(f.testUrl) { if proxy.Alive(f.testUrl) {
return proxy return proxy
} }
} else { } else {
if proxy.Name() == f.selected { if proxy.Name() == f.selected {
if proxy.AliveForTestUrl(f.testUrl) { if proxy.Alive(f.testUrl) {
return proxy return proxy
} else { } else {
f.selected = "" f.selected = ""
@ -133,7 +133,7 @@ func (f *Fallback) Set(name string) error {
} }
f.selected = name f.selected = name
if !p.AliveForTestUrl(f.testUrl) { if !p.Alive(f.testUrl) {
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(5000)) ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(5000))
defer cancel() defer cancel()
expectedStatus, _ := utils.NewIntRanges[uint16](f.expectedStatus) expectedStatus, _ := utils.NewIntRanges[uint16](f.expectedStatus)

View File

@ -150,8 +150,7 @@ func strategyRoundRobin(url string) strategyFn {
for ; i < length; i++ { for ; i < length; i++ {
id := (idx + i) % length id := (idx + i) % length
proxy := proxies[id] proxy := proxies[id]
// if proxy.Alive() { if proxy.Alive(url) {
if proxy.AliveForTestUrl(url) {
i++ i++
return proxy return proxy
} }
@ -169,16 +168,14 @@ func strategyConsistentHashing(url string) strategyFn {
for i := 0; i < maxRetry; i, key = i+1, key+1 { for i := 0; i < maxRetry; i, key = i+1, key+1 {
idx := jumpHash(key, buckets) idx := jumpHash(key, buckets)
proxy := proxies[idx] proxy := proxies[idx]
// if proxy.Alive() { if proxy.Alive(url) {
if proxy.AliveForTestUrl(url) {
return proxy return proxy
} }
} }
// when availability is poor, traverse the entire list to get the available nodes // when availability is poor, traverse the entire list to get the available nodes
for _, proxy := range proxies { for _, proxy := range proxies {
// if proxy.Alive() { if proxy.Alive(url) {
if proxy.AliveForTestUrl(url) {
return proxy return proxy
} }
} }
@ -204,8 +201,7 @@ func strategyStickySessions(url string) strategyFn {
nowIdx := idx nowIdx := idx
for i := 1; i < maxRetry; i++ { for i := 1; i < maxRetry; i++ {
proxy := proxies[nowIdx] proxy := proxies[nowIdx]
// if proxy.Alive() { if proxy.Alive(url) {
if proxy.AliveForTestUrl(url) {
if nowIdx != idx { if nowIdx != idx {
lruCache.Delete(key) lruCache.Delete(key)
lruCache.Set(key, nowIdx) lruCache.Set(key, nowIdx)

View File

@ -101,7 +101,7 @@ func (u *URLTest) fast(touch bool) C.Proxy {
proxies := u.GetProxies(touch) proxies := u.GetProxies(touch)
if u.selected != "" { if u.selected != "" {
for _, proxy := range proxies { for _, proxy := range proxies {
if !proxy.AliveForTestUrl(u.testUrl) { if !proxy.Alive(u.testUrl) {
continue continue
} }
if proxy.Name() == u.selected { if proxy.Name() == u.selected {
@ -121,7 +121,7 @@ func (u *URLTest) fast(touch bool) C.Proxy {
fastNotExist = false fastNotExist = false
} }
if !proxy.AliveForTestUrl(u.testUrl) { if !proxy.Alive(u.testUrl) {
continue continue
} }
@ -133,7 +133,7 @@ func (u *URLTest) fast(touch bool) C.Proxy {
} }
// tolerance // tolerance
if u.fastNode == nil || fastNotExist || !u.fastNode.AliveForTestUrl(u.testUrl) || u.fastNode.LastDelayForTestUrl(u.testUrl) > fast.LastDelayForTestUrl(u.testUrl)+u.tolerance { if u.fastNode == nil || fastNotExist || !u.fastNode.Alive(u.testUrl) || u.fastNode.LastDelayForTestUrl(u.testUrl) > fast.LastDelayForTestUrl(u.testUrl)+u.tolerance {
u.fastNode = fast u.fastNode = fast
} }
return u.fastNode, nil return u.fastNode, nil

View File

@ -202,7 +202,7 @@ func (hc *HealthCheck) execute(b *batch.Batch[bool], url, uid string, option *ex
defer cancel() defer cancel()
log.Debugln("Health Checking, proxy: %s, url: %s, id: {%s}", p.Name(), url, uid) log.Debugln("Health Checking, proxy: %s, url: %s, id: {%s}", p.Name(), url, uid)
_, _ = p.URLTest(ctx, url, expectedStatus) _, _ = p.URLTest(ctx, url, expectedStatus)
log.Debugln("Health Checked, proxy: %s, url: %s, alive: %t, delay: %d ms uid: {%s}", p.Name(), url, p.AliveForTestUrl(url), p.LastDelayForTestUrl(url), uid) log.Debugln("Health Checked, proxy: %s, url: %s, alive: %t, delay: %d ms uid: {%s}", p.Name(), url, p.Alive(url), p.LastDelayForTestUrl(url), uid)
return false, nil return false, nil
}) })
} }

View File

@ -114,10 +114,6 @@ func (pp *proxySetProvider) RegisterHealthCheckTask(url string, expectedStatus u
func (pp *proxySetProvider) setProxies(proxies []C.Proxy) { func (pp *proxySetProvider) setProxies(proxies []C.Proxy) {
pp.proxies = proxies pp.proxies = proxies
for _, proxy := range pp.proxies {
proxy.OriginalHealthCheckUrl(pp.healthCheck.url)
}
pp.healthCheck.setProxy(proxies) pp.healthCheck.setProxy(proxies)
if pp.healthCheck.auto() { if pp.healthCheck.auto() {
go pp.healthCheck.check() go pp.healthCheck.check()

View File

@ -147,15 +147,19 @@ type DelayHistory struct {
Delay uint16 `json:"delay"` Delay uint16 `json:"delay"`
} }
type ProxyState struct {
Alive bool `json:"alive"`
History []DelayHistory `json:"history"`
}
type DelayHistoryStoreType int type DelayHistoryStoreType int
type Proxy interface { type Proxy interface {
ProxyAdapter ProxyAdapter
AliveForTestUrl(url string) bool Alive(url string) bool
DelayHistory() []DelayHistory DelayHistory() []DelayHistory
ExtraDelayHistory() map[string][]DelayHistory ExtraDelayHistories() map[string]ProxyState
LastDelayForTestUrl(url string) uint16 LastDelayForTestUrl(url string) uint16
OriginalHealthCheckUrl(url string)
URLTest(ctx context.Context, url string, expectedStatus utils.IntRanges[uint16]) (uint16, error) URLTest(ctx context.Context, url string, expectedStatus utils.IntRanges[uint16]) (uint16, error)
// Deprecated: use DialContext instead. // Deprecated: use DialContext instead.