Repository: cloudwego/hertz
Pull request #380 (2022-11-17 06:25:25)

feat: improve observability of hostclient
#### What type of PR is this?
feat

#### Check the PR title.
<!--
The description of the title will be attached in Release Notes, 
so please describe it from user-oriented, what this PR does / why we need it.
Please check your PR title with the below requirements:
-->
- [x] This PR title match the format: \<type\>(optional scope): \<description\>
- [x] The description of this PR title is user-oriented and clear enough for others to understand.


#### (Optional) Translate the PR title into Chinese.
提高 hostclient 的可观测性

#### (Optional) More detail description for this PR(en: English/zh: Chinese).
<!--
Provide more detailed info for review. If it is a perf type PR, perf data is suggested to give.
-->
en: Improve observability of hostclient. Add HostClientState interface and provide method to inject observation function. Launch a goroutine to call the observation function after creating HostClient
zh(optional): 提高 hostclient 的可观测性。增加 HostClientState 接口，并可注入观测函数。在创建 HostClient 后，启动协程调用观测函数

#### Which issue(s) this PR fixes:
<!--
Automatically closes linked issue when PR is merged.
Eg: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->


codecov[bot]: # [Codecov](https://codecov.io/gh/cloudwego/hertz/pull/380?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) Report
Base: **67.34**% // Head: **67.25**% // Decreases project coverage by **`-0.08%`** :warning:
> Coverage data is based on head [(`bf72ee8`)](https://codecov.io/gh/cloudwego/hertz/pull/380?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) compared to base [(`76aa67b`)](https://codecov.io/gh/cloudwego/hertz/commit/76aa67bc22b0de1bd6fab358a976b09c60dd1846?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).
> Patch coverage: 6.66% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           develop     #380      +/-   ##
===========================================
- Coverage    67.34%   67.25%   -0.09%     
===========================================
  Files           89       89              
  Lines         8607     8615       +8     
===========================================
- Hits          5796     5794       -2     
- Misses        2450     2459       +9     
- Partials       361      362       +1     
```


| [Impacted Files](https://codecov.io/gh/cloudwego/hertz/pull/380?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) | Coverage Δ | |
|---|---|---|
| [pkg/app/client/option.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2FwcC9jbGllbnQvb3B0aW9uLmdv) | `94.73% <0.00%> (-5.27%)` | :arrow_down: |
| [pkg/common/config/client\_option.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2NvbW1vbi9jb25maWcvY2xpZW50X29wdGlvbi5nbw==) | `100.00% <ø> (ø)` | |
| [pkg/common/errors/errors.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2NvbW1vbi9lcnJvcnMvZXJyb3JzLmdv) | `87.95% <ø> (ø)` | |
| [pkg/protocol/http1/server.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL3NlcnZlci5nbw==) | `0.00% <0.00%> (ø)` | |
| [pkg/protocol/http1/client.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL2NsaWVudC5nbw==) | `48.97% <16.66%> (-0.31%)` | :arrow_down: |
| [pkg/common/timer/timer.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2NvbW1vbi90aW1lci90aW1lci5nbw==) | `61.90% <0.00%> (-19.05%)` | :arrow_down: |
| [pkg/app/client/client.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2FwcC9jbGllbnQvY2xpZW50Lmdv) | `72.15% <0.00%> (-1.27%)` | :arrow_down: |
| [pkg/app/server/option.go](https://codecov.io/gh/cloudwego/hertz/pull/380/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL2FwcC9zZXJ2ZXIvb3B0aW9uLmdv) | `100.00% <0.00%> (+3.25%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/cloudwego/hertz/pull/380?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).

CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/cloudwego/hertz?pullRequest=380) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/cloudwego/hertz?pullRequest=380) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/cloudwego/hertz?pullRequest=380) it.</sub>

Patch:
diff --git a/pkg/app/client/client.go b/pkg/app/client/client.go
index 3fd2571..8d3a03b 100644
--- a/pkg/app/client/client.go
+++ b/pkg/app/client/client.go
@@ -629,5 +629,6 @@ func newHttp1OptionFromClient(c *Client) *http1.ClientOptions {
 		ResponseBodyStream:            c.options.ResponseBodyStream,
 		RetryConfig:                   c.options.RetryConfig,
 		RetryIfFunc:                   c.RetryIfFunc,
+		StateObserve:                  c.options.HostClientStateObserve,
 	}
 }
diff --git a/pkg/app/client/option.go b/pkg/app/client/option.go
index a8c7510..37e0864 100644
--- a/pkg/app/client/option.go
+++ b/pkg/app/client/option.go
@@ -149,6 +149,13 @@ func WithWriteTimeout(t time.Duration) config.ClientOption {
 	}}
 }
 
+// WithConnStateObserve sets the connection state observation function.
+func WithConnStateObserve(hs config.HostClientStateFunc) config.ClientOption {
+	return config.ClientOption{F: func(o *config.ClientOptions) {
+		o.HostClientStateObserve = hs
+	}}
+}
+
 // WithDialFunc is used to set dialer function.
 // Note: WithDialFunc will overwrite custom dialer.
 func WithDialFunc(f network.DialFunc, dialers ...network.Dialer) config.ClientOption {
diff --git a/pkg/common/config/client_option.go b/pkg/common/config/client_option.go
index 8b44d24..99a7c46 100644
--- a/pkg/common/config/client_option.go
+++ b/pkg/common/config/client_option.go
@@ -25,6 +25,18 @@ import (
 	"github.com/cloudwego/hertz/pkg/protocol/consts"
 )
 
+type ConnPoolState struct {
+	ConnNum     int
+	WaitConnNum int
+	Addr        string
+}
+
+type HostClientState interface {
+	ConnPoolState() ConnPoolState
+}
+
+type HostClientStateFunc func(HostClientState)
+
 // ClientOption is the only struct that can be used to set ClientOptions.
 type ClientOption struct {
 	F func(o *ClientOptions)
@@ -106,6 +118,8 @@ type ClientOptions struct {
 
 	// all configurations related to retry
 	RetryConfig *retry.Config
+
+	HostClientStateObserve HostClientStateFunc
 }
 
 func NewClientOptions(opts []ClientOption) *ClientOptions {
diff --git a/pkg/common/errors/errors.go b/pkg/common/errors/errors.go
index e9cccdb..4bfc79c 100644
--- a/pkg/common/errors/errors.go
+++ b/pkg/common/errors/errors.go
@@ -60,6 +60,7 @@ var (
 	ErrDialTimeout      = errors.New("dial timeout")
 	ErrNothingRead      = errors.New("nothing read")
 	ErrShortConnection  = errors.New("short connection")
+	ErrNoFreeConns      = errors.New("no free connections available to host")
 	ErrConnectionClosed = errors.New("connection closed")
 )
 
diff --git a/pkg/protocol/http1/client.go b/pkg/protocol/http1/client.go
index 6d42636..e9c1504 100644
--- a/pkg/protocol/http1/client.go
+++ b/pkg/protocol/http1/client.go
@@ -71,11 +71,8 @@ import (
 	respI "github.com/cloudwego/hertz/pkg/protocol/http1/resp"
 )
 
-var (
-	errNoFreeConns      = errs.NewPublic("no free connections available to host")
-	errConnectionClosed = errs.NewPublic("the server closed connection before returning the first response byte. " +
-		"Make sure the server returns 'Connection: close' response header before closing the connection")
-)
+var errConnectionClosed = errs.NewPublic("the server closed connection before returning the first response byte. " +
+	"Make sure the server returns 'Connection: close' response header before closing the connection")
 
 // HostClient balances http requests among hosts listed in Addr.
 //
@@ -129,6 +126,11 @@ func (c *HostClient) SetDynamicConfig(dc *client.DynamicConfig) {
 	c.Addr = dc.Addr
 	c.ProxyURI = dc.ProxyURI
 	c.IsTLS = dc.IsTLS
+
+	// start observation after setting addr
+	if c.StateObserve != nil {
+		go c.StateObserve(c)
+	}
 }
 
 type clientConn struct {
@@ -167,6 +169,20 @@ func (c *HostClient) WantConnectionCount() (count int) {
 	return c.connsWait.len()
 }
 
+func (c *HostClient) ConnPoolState() config.ConnPoolState {
+	c.connsLock.Lock()
+	defer c.connsLock.Unlock()
+	cps := config.ConnPoolState{
+		ConnNum: len(c.conns),
+		Addr:    c.Addr,
+	}
+
+	if c.connsWait != nil {
+		cps.WaitConnNum = c.connsWait.len()
+	}
+	return cps
+}
+
 // GetTimeout returns the status code and body of url.
 //
 // The contents of dst will be replaced by the body and returned, if the dst
@@ -252,7 +268,7 @@ type wantConn struct {
 // errTimeout is returned if the response wasn't returned during
 // the given timeout.
 //
-// errNoFreeConns is returned if all HostClient.MaxConns connections
+// ErrNoFreeConns is returned if all HostClient.MaxConns connections
 // to the host are busy.
 //
 // It is recommended obtaining req and resp via AcquireRequest
@@ -279,7 +295,7 @@ func (c *HostClient) DoTimeout(ctx context.Context, req *protocol.Request, resp
 // errTimeout is returned if the response wasn't returned until
 // the given deadline.
 //
-// errNoFreeConns is returned if all HostClient.MaxConns connections
+// ErrNoFreeConns is returned if all HostClient.MaxConns connections
 // to the host are busy.
 //
 // It is recommended obtaining req and resp via AcquireRequest
@@ -302,7 +318,7 @@ func (c *HostClient) DoDeadline(ctx context.Context, req *protocol.Request, resp
 //
 // Response is ignored if resp is nil.
 //
-// errNoFreeConns is returned if all DefaultMaxConnsPerHost connections
+// ErrNoFreeConns is returned if all DefaultMaxConnsPerHost connections
 // to the requested host are busy.
 //
 // It is recommended obtaining req and resp via AcquireRequest
@@ -321,7 +337,7 @@ func (c *HostClient) DoRedirects(ctx context.Context, req *protocol.Request, res
 //
 // Response is ignored if resp is nil.
 //
-// errNoFreeConns is returned if all HostClient.MaxConns connections
+// ErrNoFreeConns is returned if all HostClient.MaxConns connections
 // to the host are busy.
 //
 // It is recommended obtaining req and resp via AcquireRequest
@@ -627,7 +643,7 @@ func (c *HostClient) acquireConn(dialTimeout time.Duration) (cc *clientConn, err
 	}
 	if !createConn {
 		if c.MaxConnWaitTimeout <= 0 {
-			return nil, errNoFreeConns
+			return nil, errs.ErrNoFreeConns
 		}
 
 		timeout := c.MaxConnWaitTimeout
@@ -656,7 +672,7 @@ func (c *HostClient) acquireConn(dialTimeout time.Duration) (cc *clientConn, err
 		case <-w.ready:
 			return w.conn, w.err
 		case <-tc.C:
-			return nil, errNoFreeConns
+			return nil, errs.ErrNoFreeConns
 		}
 	}
 
@@ -1116,9 +1132,11 @@ func (q *wantConnQueue) clearFront() (cleaned bool) {
 }
 
 func NewHostClient(c *ClientOptions) client.HostClient {
-	return &HostClient{
+	hc := &HostClient{
 		ClientOptions: c,
 	}
+
+	return hc
 }
 
 type ClientOptions struct {
@@ -1218,7 +1236,7 @@ type ClientOptions struct {
 
 	// Maximum duration for waiting for a free connection.
 	//
-	// By default will not wait, return errNoFreeConns immediately
+	// By default will not wait, return ErrNoFreeConns immediately
 	MaxConnWaitTimeout time.Duration
 
 	// ResponseBodyStream enables response body streaming
@@ -1228,4 +1246,6 @@ type ClientOptions struct {
 	RetryConfig *retry.Config
 
 	RetryIfFunc client.RetryIfFunc
+
+	StateObserve config.HostClientStateFunc
 }


Test patch:
diff --git a/pkg/app/client/client_test.go b/pkg/app/client/client_test.go
index ac0307e..1891de8 100644
--- a/pkg/app/client/client_test.go
+++ b/pkg/app/client/client_test.go
@@ -81,10 +81,7 @@ import (
 	"github.com/cloudwego/hertz/pkg/route"
 )
 
-var (
-	errTooManyRedirects = errors.New("too many redirects detected when doing the request")
-	errNoFreeConns      = errors.New("no free connections available to host")
-)
+var errTooManyRedirects = errors.New("too many redirects detected when doing the request")
 
 func TestCloseIdleConnections(t *testing.T) {
 	opt := config.NewOptions([]config.Option{})
@@ -834,7 +831,7 @@ func TestHostClientMaxConnsWithDeadline(t *testing.T) {
 
 			for {
 				if err := c.DoDeadline(context.Background(), req, resp, time.Now().Add(timeout)); err != nil {
-					if err.Error() == errNoFreeConns.Error() {
+					if err.Error() == errs.ErrNoFreeConns.Error() {
 						time.Sleep(time.Millisecond * 500)
 						continue
 					}
@@ -1179,8 +1176,8 @@ func TestHostClientMaxConnWaitTimeoutError(t *testing.T) {
 			resp := protocol.AcquireResponse()
 
 			if err := c.Do(context.Background(), req, resp); err != nil {
-				if err.Error() != errNoFreeConns.Error() {
-					t.Errorf("unexpected error: %s. Expecting %s", err.Error(), errNoFreeConns.Error())
+				if err.Error() != errs.ErrNoFreeConns.Error() {
+					t.Errorf("unexpected error: %s. Expecting %s", err.Error(), errs.ErrNoFreeConns.Error())
 				}
 				atomic.AddUint32(&errNoFreeConnsCount, 1)
 			} else {
@@ -2152,3 +2149,21 @@ func TestClientDoWithDialFunc(t *testing.T) {
 		t.Fatalf("timeout")
 	}
 }
+
+func TestClientState(t *testing.T) {
+	opt := config.NewOptions([]config.Option{})
+	opt.Addr = ":11000"
+	engine := route.NewEngine(opt)
+	go engine.Run()
+
+	client, _ := NewClient(WithConnStateObserve(func(hcs config.HostClientState) {
+		assert.DeepEqual(t, 0, hcs.ConnPoolState().ConnNum)
+		time.Sleep(time.Second * 2)
+		assert.DeepEqual(t, 1, hcs.ConnPoolState().ConnNum)
+		assert.DeepEqual(t, "127.0.0.1:11000", hcs.ConnPoolState().Addr)
+	}))
+
+	time.Sleep(time.Second)
+	client.Get(context.Background(), nil, "http://127.0.0.1:11000")
+	time.Sleep(time.Second * 3)
+}
diff --git a/pkg/protocol/http1/client_test.go b/pkg/protocol/http1/client_test.go
index ad63f41..f61e491 100644
--- a/pkg/protocol/http1/client_test.go
+++ b/pkg/protocol/http1/client_test.go
@@ -124,8 +124,8 @@ func TestHostClientMaxConnWaitTimeoutWithEarlierDeadline(t *testing.T) {
 			break
 		}
 		w.mu.Lock()
-		if w.err != nil && !errors.Is(w.err, errNoFreeConns) {
-			t.Errorf("unexpected error: %s. Expecting %s", w.err, errNoFreeConns)
+		if w.err != nil && !errors.Is(w.err, errs.ErrNoFreeConns) {
+			t.Errorf("unexpected error: %s. Expecting %s", w.err, errs.ErrNoFreeConns)
 		}
 		w.mu.Unlock()
 	}