Feature: make every provider support health check

This commit is contained in:
gVisor bot 2019-12-26 18:41:06 +08:00
parent 5314d8bb19
commit ddd62418ff
4 changed files with 70 additions and 85 deletions

View file

@ -55,7 +55,8 @@ func ParseProxyGroup(config map[string]interface{}, proxyMap map[string]C.Proxy,
// if Use not empty, drop health check options // if Use not empty, drop health check options
if len(groupOption.Use) != 0 { if len(groupOption.Use) != 0 {
pd, err := provider.NewCompatibleProvier(groupName, ps, nil) hc := provider.NewHealthCheck(ps, "", 0)
pd, err := provider.NewCompatibleProvier(groupName, ps, hc)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -64,7 +65,8 @@ func ParseProxyGroup(config map[string]interface{}, proxyMap map[string]C.Proxy,
} else { } else {
// select don't need health check // select don't need health check
if groupOption.Type == "select" { if groupOption.Type == "select" {
pd, err := provider.NewCompatibleProvier(groupName, ps, nil) hc := provider.NewHealthCheck(ps, "", 0)
pd, err := provider.NewCompatibleProvier(groupName, ps, hc)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -76,11 +78,8 @@ func ParseProxyGroup(config map[string]interface{}, proxyMap map[string]C.Proxy,
return nil, errMissHealthCheck return nil, errMissHealthCheck
} }
healthOption := &provider.HealthCheckOption{ hc := provider.NewHealthCheck(ps, groupOption.URL, uint(groupOption.Interval))
URL: groupOption.URL, pd, err := provider.NewCompatibleProvier(groupName, ps, hc)
Interval: uint(groupOption.Interval),
}
pd, err := provider.NewCompatibleProvier(groupName, ps, healthOption)
if err != nil { if err != nil {
return nil, err return nil, err
} }

View file

@ -16,20 +16,37 @@ type HealthCheckOption struct {
Interval uint Interval uint
} }
type healthCheck struct { type HealthCheck struct {
url string url string
proxies []C.Proxy proxies []C.Proxy
ticker *time.Ticker interval uint
done chan struct{}
} }
func (hc *healthCheck) process() { func (hc *HealthCheck) process() {
ticker := time.NewTicker(time.Duration(hc.interval) * time.Second)
go hc.check() go hc.check()
for range hc.ticker.C { for {
select {
case <-ticker.C:
hc.check() hc.check()
case <-hc.done:
ticker.Stop()
return
}
} }
} }
func (hc *healthCheck) check() { func (hc *HealthCheck) setProxy(proxies []C.Proxy) {
hc.proxies = proxies
}
func (hc *HealthCheck) auto() bool {
return hc.interval != 0
}
func (hc *HealthCheck) check() {
ctx, cancel := context.WithTimeout(context.Background(), defaultURLTestTimeout) ctx, cancel := context.WithTimeout(context.Background(), defaultURLTestTimeout)
for _, proxy := range hc.proxies { for _, proxy := range hc.proxies {
go proxy.URLTest(ctx, hc.url) go proxy.URLTest(ctx, hc.url)
@ -39,15 +56,15 @@ func (hc *healthCheck) check() {
cancel() cancel()
} }
func (hc *healthCheck) close() { func (hc *HealthCheck) close() {
hc.ticker.Stop() hc.done <- struct{}{}
} }
func newHealthCheck(proxies []C.Proxy, url string, interval uint) *healthCheck { func NewHealthCheck(proxies []C.Proxy, url string, interval uint) *HealthCheck {
ticker := time.NewTicker(time.Duration(interval) * time.Second) return &HealthCheck{
return &healthCheck{
proxies: proxies, proxies: proxies,
url: url, url: url,
ticker: ticker, interval: interval,
done: make(chan struct{}, 1),
} }
} }

View file

@ -35,13 +35,11 @@ func ParseProxyProvider(name string, mapping map[string]interface{}) (ProxyProvi
return nil, err return nil, err
} }
var healthCheckOption *HealthCheckOption var hcInterval uint = 0
if schema.HealthCheck.Enable { if schema.HealthCheck.Enable {
healthCheckOption = &HealthCheckOption{ hcInterval = uint(schema.HealthCheck.Interval)
URL: schema.HealthCheck.URL,
Interval: uint(schema.HealthCheck.Interval),
}
} }
hc := NewHealthCheck([]C.Proxy{}, schema.HealthCheck.URL, hcInterval)
path := C.Path.Reslove(schema.Path) path := C.Path.Reslove(schema.Path)
@ -56,5 +54,5 @@ func ParseProxyProvider(name string, mapping map[string]interface{}) (ProxyProvi
} }
interval := time.Duration(uint(schema.Interval)) * time.Second interval := time.Duration(uint(schema.Interval)) * time.Second
return NewProxySetProvider(name, interval, vehicle, healthCheckOption), nil return NewProxySetProvider(name, interval, vehicle, hc), nil
} }

View file

@ -7,9 +7,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"io/ioutil" "io/ioutil"
"net/url"
"os" "os"
"sync"
"time" "time"
"github.com/Dreamacro/clash/adapters/outbound" "github.com/Dreamacro/clash/adapters/outbound"
@ -72,13 +70,9 @@ type ProxySetProvider struct {
vehicle Vehicle vehicle Vehicle
hash [16]byte hash [16]byte
proxies []C.Proxy proxies []C.Proxy
healthCheck *healthCheck healthCheck *HealthCheck
healthCheckOption *HealthCheckOption
ticker *time.Ticker ticker *time.Ticker
updatedAt *time.Time updatedAt *time.Time
// mux for avoiding creating new goroutines when pulling
mux sync.Mutex
} }
func (pp *ProxySetProvider) MarshalJSON() ([]byte, error) { func (pp *ProxySetProvider) MarshalJSON() ([]byte, error) {
@ -100,24 +94,15 @@ func (pp *ProxySetProvider) Reload() error {
} }
func (pp *ProxySetProvider) HealthCheck() { func (pp *ProxySetProvider) HealthCheck() {
pp.mux.Lock()
defer pp.mux.Unlock()
if pp.healthCheck != nil {
pp.healthCheck.check() pp.healthCheck.check()
} }
}
func (pp *ProxySetProvider) Update() error { func (pp *ProxySetProvider) Update() error {
return pp.pull() return pp.pull()
} }
func (pp *ProxySetProvider) Destroy() error { func (pp *ProxySetProvider) Destroy() error {
pp.mux.Lock()
defer pp.mux.Unlock()
if pp.healthCheck != nil {
pp.healthCheck.close() pp.healthCheck.close()
pp.healthCheck = nil
}
if pp.ticker != nil { if pp.ticker != nil {
pp.ticker.Stop() pp.ticker.Stop()
@ -241,35 +226,32 @@ func (pp *ProxySetProvider) parse(buf []byte) ([]C.Proxy, error) {
func (pp *ProxySetProvider) setProxies(proxies []C.Proxy) { func (pp *ProxySetProvider) setProxies(proxies []C.Proxy) {
pp.proxies = proxies pp.proxies = proxies
if pp.healthCheckOption != nil { pp.healthCheck.setProxy(proxies)
pp.mux.Lock() go pp.healthCheck.check()
if pp.healthCheck != nil {
pp.healthCheck.close()
}
pp.healthCheck = newHealthCheck(proxies, pp.healthCheckOption.URL, pp.healthCheckOption.Interval)
go pp.healthCheck.process()
pp.mux.Unlock()
}
} }
func NewProxySetProvider(name string, interval time.Duration, vehicle Vehicle, option *HealthCheckOption) *ProxySetProvider { func NewProxySetProvider(name string, interval time.Duration, vehicle Vehicle, hc *HealthCheck) *ProxySetProvider {
var ticker *time.Ticker var ticker *time.Ticker
if interval != 0 { if interval != 0 {
ticker = time.NewTicker(interval) ticker = time.NewTicker(interval)
} }
if hc.auto() {
go hc.process()
}
return &ProxySetProvider{ return &ProxySetProvider{
name: name, name: name,
vehicle: vehicle, vehicle: vehicle,
proxies: []C.Proxy{}, proxies: []C.Proxy{},
healthCheckOption: option, healthCheck: hc,
ticker: ticker, ticker: ticker,
} }
} }
type CompatibleProvier struct { type CompatibleProvier struct {
name string name string
healthCheck *healthCheck healthCheck *HealthCheck
proxies []C.Proxy proxies []C.Proxy
} }
@ -291,26 +273,19 @@ func (cp *CompatibleProvier) Reload() error {
} }
func (cp *CompatibleProvier) Destroy() error { func (cp *CompatibleProvier) Destroy() error {
if cp.healthCheck != nil {
cp.healthCheck.close() cp.healthCheck.close()
}
return nil return nil
} }
func (cp *CompatibleProvier) HealthCheck() { func (cp *CompatibleProvier) HealthCheck() {
if cp.healthCheck != nil {
cp.healthCheck.check() cp.healthCheck.check()
} }
}
func (cp *CompatibleProvier) Update() error { func (cp *CompatibleProvier) Update() error {
return nil return nil
} }
func (cp *CompatibleProvier) Initial() error { func (cp *CompatibleProvier) Initial() error {
if cp.healthCheck != nil {
go cp.healthCheck.process()
}
return nil return nil
} }
@ -326,17 +301,13 @@ func (cp *CompatibleProvier) Proxies() []C.Proxy {
return cp.proxies return cp.proxies
} }
func NewCompatibleProvier(name string, proxies []C.Proxy, option *HealthCheckOption) (*CompatibleProvier, error) { func NewCompatibleProvier(name string, proxies []C.Proxy, hc *HealthCheck) (*CompatibleProvier, error) {
if len(proxies) == 0 { if len(proxies) == 0 {
return nil, errors.New("Provider need one proxy at least") return nil, errors.New("Provider need one proxy at least")
} }
var hc *healthCheck if hc.auto() {
if option != nil { go hc.process()
if _, err := url.Parse(option.URL); err != nil {
return nil, fmt.Errorf("URL format error: %w", err)
}
hc = newHealthCheck(proxies, option.URL, option.Interval)
} }
return &CompatibleProvier{ return &CompatibleProvier{