gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(sdk): add common headers in sdk http client

Browse Source 
. #22115371

Signed-off-by: slasher <mcq.sejust@gmail.com>
This commit is contained in:
slasher 2024-02-06 11:40:54 +08:00 committed by zhumingze1108
parent ecb0b49016
commit bf26b4bc3d
5 changed files with 72 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cli/cmd/flashnode.go
View File

@ -127,7 +127,7 @@ func newCmdFlashNodeHTTPStat(_ *master.MasterClient) *cobra.Command {
Short: "show flashnode stat",
Args: cobra.MinimumNArgs(1),
RunE: func(_ *cobra.Command, args []string) (err error) {
stat, err := httpclient.New().WithAddr(addr2Prof(args[0])).FlashNode().Stat()
stat, err := httpclient.New().Addr(addr2Prof(args[0])).FlashNode().Stat()
if err != nil {
return
}
@ -145,13 +145,13 @@ func newCmdFlashNodeHTTPEvict(_ *master.MasterClient) *cobra.Command {
RunE: func(_ *cobra.Command, args []string) (err error) {
addr := args[0]
if len(args) == 1 {
if err = httpclient.New().WithAddr(addr2Prof(addr)).FlashNode().EvictAll(); err == nil {
if err = httpclient.New().Addr(addr2Prof(addr)).FlashNode().EvictAll(); err == nil {
stdoutlnf("%s evicts all [OK]", addr)
}
return
}
volume := args[1]
if err = httpclient.New().WithAddr(addr2Prof(addr)).FlashNode().EvictVol(volume); err == nil {
if err = httpclient.New().Addr(addr2Prof(addr)).FlashNode().EvictVol(volume); err == nil {
stdoutlnf("%s evicts volume(%s) [OK]", addr, volume)
}
return
@ -173,7 +173,7 @@ func showFlashNodesView(flashNodeViewInfos []*proto.FlashNodeViewInfo, showStat
hitRate, evicts, limit := "N/A", "N/A", "N/A"
if fn.IsActive && fn.IsEnable {
if stat, e := client.WithAddr(addr2Prof(fn.Addr)).FlashNode().Stat(); e == nil {
if stat, e := client.Addr(addr2Prof(fn.Addr)).FlashNode().Stat(); e == nil {
hitRate = fmt.Sprintf("%.2f%%", stat.CacheStatus.HitRate*100)
evicts = strconv.Itoa(stat.CacheStatus.Evicts)
limit = strconv.FormatUint(stat.NodeLimit, 10)

12
flashnode/serve_http_test.go
View File

@ -31,27 +31,27 @@ func testHTTP(t *testing.T) {
}
func testHTTPStat(t *testing.T) {
st, err := httpCli.WithAddr(httpServer.Addr).FlashNode().Stat()
st, err := httpCli.Addr(httpServer.Addr).FlashNode().Stat()
require.NoError(t, err)
t.Logf("node status %+v", st)
t.Logf("cache status %+v", *st.CacheStatus)
}
func testHTTPEvictVol(t *testing.T) {
require.Error(t, httpCli.WithAddr(httpServer.Addr).FlashNode().EvictVol(""))
st, err := httpCli.WithAddr(httpServer.Addr).FlashNode().Stat()
require.Error(t, httpCli.Addr(httpServer.Addr).FlashNode().EvictVol(""))
st, err := httpCli.Addr(httpServer.Addr).FlashNode().Stat()
require.NoError(t, err)
require.Equal(t, 1, len(st.CacheStatus.Keys))
require.Equal(t, cachengine.GenCacheBlockKey(_volume, _inode, _offset, _version), st.CacheStatus.Keys[0])
t.Logf("cache status before evicted, %+v", *st.CacheStatus)
require.NoError(t, httpCli.WithAddr(httpServer.Addr).FlashNode().EvictVol(_volume))
st, err = httpCli.WithAddr(httpServer.Addr).FlashNode().Stat()
require.NoError(t, httpCli.Addr(httpServer.Addr).FlashNode().EvictVol(_volume))
st, err = httpCli.Addr(httpServer.Addr).FlashNode().Stat()
require.NoError(t, err)
require.Equal(t, 0, len(st.CacheStatus.Keys))
t.Logf("cache status after evicted, %+v", *st.CacheStatus)
}
func testHTTPEvictAll(t *testing.T) {
require.NoError(t, httpCli.WithAddr(httpServer.Addr).FlashNode().EvictAll())
require.NoError(t, httpCli.Addr(httpServer.Addr).FlashNode().EvictAll())
}

83
sdk/httpclient/client.go
View File

@ -24,11 +24,13 @@ import (
"time"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/compressor"
"github.com/cubefs/cubefs/util/log"
)
const (
requestTimeout = 30 * time.Second
get = http.MethodGet
post = http.MethodPost
)
var ErrNoAddress = errors.New("no address")
@ -37,31 +39,58 @@ type Client struct {
http.Client
schema string
addr []string
header map[string]string
}
func New() *Client {
return &Client{schema: "http"}
}
func (c *Client) WithAddr(addresses ...string) *Client {
func (c *Client) Addr(addresses ...string) *Client {
c.addr = addresses[:]
return c
}
func (c *Client) WithTimeout(timeout time.Duration) *Client {
func (c *Client) Timeout(timeout time.Duration) *Client {
c.Client.Timeout = timeout
return c
}
func (c *Client) WithSSL(ssl bool) *Client {
func (c *Client) SSL(ssl bool) *Client {
c.schema = "http"
if ssl {
c.schema = "https"
} else {
c.schema = "http"
}
return c
}
// WithHeader return a new Client.
func (c *Client) WithHeader(headers map[string]string, added ...string) *Client {
if len(added)%2 == 1 {
added = added[:len(added)-1]
}
nc := &Client{
Client: c.Client,
schema: c.schema,
addr: c.addr[:],
header: make(map[string]string, len(c.header)),
}
for k, v := range c.header {
nc.header[k] = v
}
for k, v := range headers {
nc.header[k] = v
}
for idx := 0; idx < len(added); idx += 2 {
nc.header[added[idx]] = added[idx+1]
}
return nc
}
func (c *Client) EncodingGzip() *Client {
return c.WithHeader(nil, "x-cfs-Accept-Encoding", compressor.EncodingGzip)
}
func (c *Client) serve(r *request) (data []byte, err error) {
err = ErrNoAddress
for _, addr := range c.addr {
@ -74,6 +103,9 @@ func (c *Client) serve(r *request) (data []byte, err error) {
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Connection", "close")
for k, v := range c.header {
req.Header.Set(k, v)
}
for k, v := range r.header {
req.Header.Set(k, v)
}
@ -91,36 +123,33 @@ func (c *Client) serve(r *request) (data []byte, err error) {
continue
}
switch resp.StatusCode {
case http.StatusOK:
body := &struct {
Code int32 `json:"code"`
Msg string `json:"msg"`
Data json.RawMessage `json:"data"`
}{}
if err = json.Unmarshal(respData, body); err != nil {
log.LogErrorf("unmarshal response body %v", err)
return nil, fmt.Errorf("unmarshal response body %v", err)
}
if body.Code != proto.ErrCodeSuccess {
log.LogWarnf("serve: body code[%d] msg[%s] data[%v] ", body.Code, body.Msg, body.Data)
if body.Code == proto.ErrCodeInternalError && len(body.Msg) > 0 {
return nil, errors.New(body.Msg)
}
return nil, proto.ParseErrorCode(body.Code)
}
return []byte(body.Data), nil
default:
if resp.StatusCode != http.StatusOK {
err = fmt.Errorf("serve: url(%s) status(%d) body(%s)",
resp.Request.URL.String(), resp.StatusCode, string(respData))
log.LogError(err)
continue
}
respData, err = compressor.New(resp.Header.Get("x-cfs-Content-Encoding")).Decompress(respData)
if err != nil {
log.LogErrorf("decompress response body %v", err)
continue
}
reply := new(proto.HTTPReplyRaw)
if err = reply.Unmarshal(respData); err != nil {
log.LogError(err)
continue
}
if err = reply.Success(); err != nil {
log.LogError(err)
continue
}
return reply.Bytes(), nil
}
return
}
func (c *Client) serveWith(r *request, rst interface{}) error {
func (c *Client) serveWith(rst interface{}, r *request) error {
buf, err := c.serve(r)
if err != nil {
return err

10
sdk/httpclient/flashnode.go
View File

@ -15,8 +15,6 @@
package httpclient
import (
"net/http"
"github.com/cubefs/cubefs/proto"
)
@ -35,16 +33,16 @@ func (c *Client) FlashNode() FlashNode {
}
func (f *flashNode) EvictVol(volume string) error {
r := newRequest(http.MethodPost, "/evictVol")
r := newRequest(post, "/evictVol")
r.params.Add("volume", volume)
return f.client.serveWith(r, nil)
return f.client.serveWith(nil, r)
}
func (f *flashNode) EvictAll() error {
return f.client.serveWith(newRequest(http.MethodPost, "/evictAll"), nil)
return f.client.serveWith(nil, newRequest(post, "/evictAll"))
}
func (f *flashNode) Stat() (st proto.FlashNodeStat, err error) {
err = f.client.serveWith(newRequest(http.MethodGet, "/stat"), &st)
err = f.client.serveWith(&st, newRequest(get, "/stat"))
return
}

2
sdk/httpclient/request.go
View File

@ -21,6 +21,8 @@ import (
"github.com/cubefs/cubefs/proto"
)
// TODO: rename sdk/master/request.go to here.
type request struct {
method string
path string