使用http做地址探测 (#73)

* 使用http做地址探测

Author: humingcheng

Diff for: .github/workflows/db-ci.yml

@@ -33,7 +33,7 @@ jobs:
         uses: actions/checkout@v1
       - name: UT for etcd
         run: |
-          time docker run -d -p 2379:2379 --name etcd quay.io/coreos/etcd etcd -name etcd --advertise-client-urls http://0.0.0.0:2379 --listen-client-urls http://0.0.0.0:2379
+          time docker run -d -p 2379:2379 --name etcd quay.io/coreos/etcd:v3.5.15 etcd -name etcd --advertise-client-urls http://0.0.0.0:2379 --listen-client-urls http://0.0.0.0:2379
           while ! nc -z 127.0.0.1 2379; do
             sleep 1
           done

Diff for: .gitignore

@@ -12,5 +12,5 @@
 *.out
 
 # Dependency directories (remove the comment below to include it)
-# vendor/
+vendor/
 .idea

Diff for: addresspool/pool.go

@@ -1,13 +1,17 @@
 package addresspool
 
 import (
+	"context"
 	"fmt"
+	"io"
 	"net"
+	"net/http"
 	"os"
 	"strconv"
 	"sync"
 	"time"
 
+	"github.com/go-chassis/foundation/httpclient"
 	"github.com/go-chassis/openlog"
 
 	"github.com/go-chassis/cari/discovery"
@@ -24,6 +28,16 @@ const (
 	healthProbeTimeout             = time.Second
 )
 
+type HttpProbeOptions struct {
+	Client   *httpclient.Requests
+	Protocol string
+	Path     string
+}
+
+type Options struct {
+	HttpProbeOptions *HttpProbeOptions // used in available check if set, tcp will be used if not set
+}
+
 // Pool cloud server address pool
 type Pool struct {
 	mutex          sync.RWMutex
@@ -33,10 +47,11 @@ type Pool struct {
 	sameAzAddress []string
 	diffAzAddress []string
 
-	status      map[string]string
-	onceMonitor sync.Once
-	quit        chan struct{}
-	onceQuit    sync.Once
+	status           map[string]string
+	onceMonitor      sync.Once
+	quit             chan struct{}
+	onceQuit         sync.Once
+	httpProbeOptions *HttpProbeOptions
 }
 
 func (p *Pool) Close() {
@@ -46,23 +61,21 @@ func (p *Pool) Close() {
 }
 
 // NewPool Get registry pool instance
-func NewPool(addresses []string) *Pool {
+func NewPool(addresses []string, opts ...Options) *Pool {
 	p := &Pool{
 		defaultAddress: removeDuplicates(addresses),
 		status:         make(map[string]string),
 	}
-	p.appendAddressToStatus(addresses)
-	p.monitor()
-	return p
-}
 
-func (p *Pool) appendAddressToStatus(addresses []string) {
-	for _, v := range addresses {
-		if _, ok := p.status[v]; ok {
-			continue
+	if len(opts) > 0 && opts[0].HttpProbeOptions != nil {
+		p.httpProbeOptions = opts[0].HttpProbeOptions
+		if p.httpProbeOptions.Client == nil {
+			openlog.Info(fmt.Sprintf("http client nil, make one with default options"))
+			p.httpProbeOptions.Client, _ = httpclient.New(nil)
 		}
-		p.status[v] = statusAvailable
 	}
+	p.monitor()
+	return p
 }
 
 func (p *Pool) ResetAddress(addresses []string) {
@@ -72,7 +85,6 @@ func (p *Pool) ResetAddress(addresses []string) {
 	p.diffAzAddress = []string{}
 	p.sameAzAddress = []string{}
 	p.status = make(map[string]string)
-	p.appendAddressToStatus(addresses)
 }
 
 func (p *Pool) SetAddressByInstances(instances []*discovery.MicroServiceInstance) error {
@@ -88,12 +100,10 @@ func (p *Pool) SetAddressByInstances(instances []*discovery.MicroServiceInstance
 		uniqueAddrList := removeDuplicates(addrList)
 		if p.isSameAzAddr(uniqueAddrList) {
 			p.sameAzAddress = uniqueAddrList
-			p.appendAddressToStatus(uniqueAddrList)
 			openlog.Info(fmt.Sprintf("sync same az endpoints: %s", uniqueAddrList))
 			continue
 		}
 		p.diffAzAddress = uniqueAddrList
-		p.appendAddressToStatus(uniqueAddrList)
 		openlog.Info(fmt.Sprintf("sync different az endpoints: %s", addrList))
 	}
 	return nil
@@ -167,13 +177,12 @@ func (p *Pool) checkConnectivity() {
 		if _, exist := status[v]; exist {
 			continue
 		}
-		conn, err := net.DialTimeout("tcp", v, healthProbeTimeout)
+		err := p.doCheckConnectivity(v)
 		if err != nil {
-			openlog.Error("connectivity unavailable: " + v)
+			openlog.Error(fmt.Sprintf("%s connectivity unavailable: %s", v, err))
 			status[v] = statusUnavailable
 		} else {
 			status[v] = statusAvailable
-			conn.Close()
 		}
 	}
 
@@ -182,6 +191,54 @@ func (p *Pool) checkConnectivity() {
 	p.mutex.Unlock()
 }
 
+func (p *Pool) doCheckConnectivity(endpoint string) error {
+	if p.httpProbeOptions != nil {
+		return p.doCheckConnectivityWithHttp(endpoint)
+	}
+
+	return p.doCheckConnectivityWithTcp(endpoint)
+}
+
+func (p *Pool) doCheckConnectivityWithTcp(endpoint string) error {
+	conn, err := net.DialTimeout("tcp", endpoint, healthProbeTimeout)
+	if err != nil {
+		return err
+	}
+
+	err = conn.Close()
+	if err != nil {
+		openlog.Error(fmt.Sprintf("close conn failed when check connectivity: %s", err))
+	}
+
+	return nil
+}
+
+func (p *Pool) doCheckConnectivityWithHttp(endpoint string) error {
+	u := p.httpProbeOptions.Protocol + "://" + endpoint + p.httpProbeOptions.Path
+	resp, err := p.httpProbeOptions.Client.Get(context.Background(), u, nil)
+	if err != nil {
+		return err
+	}
+
+	if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices {
+		return nil
+	}
+	// do tcp check if api not exist, ensure to compatible with old scenes
+	if resp.StatusCode == http.StatusNotFound {
+		return p.doCheckConnectivityWithTcp(endpoint)
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("http status: %s, read resp error: %s", resp.Status, err)
+	}
+	err = resp.Body.Close()
+	if err != nil {
+		openlog.Error(fmt.Sprintf("close http resp.Body failed when check connectivity: %s", err))
+	}
+	return fmt.Errorf("http status: %s, resp: %s", resp.Status, string(body))
+}
+
 func (p *Pool) monitor() {
 	p.onceMonitor.Do(func() {
 		var interval time.Duration

Diff for: addresspool/pool_test.go

@@ -13,18 +13,28 @@ import (
 )
 
 func TestNewPool(t *testing.T) {
+	mockHttpServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		writer.WriteHeader(http.StatusOK)
+		return
+	}))
+
 	os.Setenv("CHASSIS_SC_HEALTH_CHECK_INTERVAL", "1")
 
-	defaultAddr := "127.0.0.1:30000"
+	defaultAddr := mockHttpServer.Listener.Addr().String()
 	// can get an address
 	pool := NewPool([]string{defaultAddr})
 	addr := pool.GetAvailableAddress()
 	assert.Equal(t, defaultAddr, addr)
 
 	// check monitor started
+	assert.NotEqual(t, statusAvailable, pool.status[defaultAddr]) // unavailable by default
+
+	time.Sleep(2*time.Second + 100*time.Millisecond)
 	assert.Equal(t, statusAvailable, pool.status[defaultAddr])
+
+	mockHttpServer.Close()
 	time.Sleep(2*time.Second + 100*time.Millisecond)
-	assert.Equal(t, statusUnavailable, pool.status[defaultAddr]) // the status should be unavailable after check interval
+	assert.NotEqual(t, statusAvailable, pool.status[defaultAddr]) // the status should be unavailable again
 }
 
 func TestAddressPool_GetAvailableAddress_priority(t *testing.T) {
@@ -49,21 +59,27 @@ func TestAddressPool_GetAvailableAddress_priority(t *testing.T) {
 				p.defaultAddress = []string{defaultAddr}
 				p.sameAzAddress = []string{sameAzAddr}
 				p.diffAzAddress = []string{diffAzAddr}
-				p.appendAddressToStatus([]string{defaultAddr, sameAzAddr, diffAzAddr})
+				p.status[sameAzAddr] = statusAvailable
+				p.status[diffAzAddr] = statusAvailable
+				p.status[defaultAddr] = statusAvailable
 			},
 			want: sameAzAddr,
 		},
 		{
 			name: "diff az address available, return diff az address",
 			preDo: func() {
 				p.status[sameAzAddr] = statusUnavailable
+				p.status[diffAzAddr] = statusAvailable
+				p.status[defaultAddr] = statusAvailable
 			},
 			want: diffAzAddr,
 		},
 		{
 			name: "same az/diff az address unavailable, return default address",
 			preDo: func() {
+				p.status[sameAzAddr] = statusUnavailable
 				p.status[diffAzAddr] = statusUnavailable
+				p.status[defaultAddr] = statusAvailable
 			},
 			want: defaultAddr,
 		},
@@ -144,10 +160,6 @@ func TestAddressPool_SetAddressByInstances(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, []string{"192.168.2.1:30100", "192.168.2.2:30100"}, p.sameAzAddress)
 	assert.Equal(t, []string{"192.168.1.1:30100", "192.168.1.2:30100"}, p.diffAzAddress)
-	assert.Equal(t, statusAvailable, p.status["192.168.1.1:30100"])
-	assert.Equal(t, statusAvailable, p.status["192.168.1.2:30100"])
-	assert.Equal(t, statusAvailable, p.status["192.168.2.1:30100"])
-	assert.Equal(t, statusAvailable, p.status["192.168.2.2:30100"])
 }
 
 func TestAddressPool_checkConnectivity(t *testing.T) {
@@ -164,9 +176,9 @@ func TestAddressPool_checkConnectivity(t *testing.T) {
 	// init, all address is available
 	defaultAddr := "127.0.0.1:30000"
 	p := NewPool([]string{defaultAddr, server1Addr, server2Addr})
-	assert.Equal(t, statusAvailable, p.status[defaultAddr])
-	assert.Equal(t, statusAvailable, p.status[server1Addr])
-	assert.Equal(t, statusAvailable, p.status[server2Addr])
+	assert.NotEqual(t, statusAvailable, p.status[defaultAddr])
+	assert.NotEqual(t, statusAvailable, p.status[server1Addr])
+	assert.NotEqual(t, statusAvailable, p.status[server2Addr])
 
 	// check connectivity, default address status should be unavailable, as it is fake
 	p.checkConnectivity()