refactor gateway proxy request api

model/gateway/instance.go

@@ -5,6 +5,7 @@
 package gateway
 
 import (
+	"infini.sh/framework/core/agent"
 	"infini.sh/framework/core/orm"
 )
 
@@ -16,13 +17,8 @@ type Instance struct {
 	Name        string `json:"name,omitempty" elastic_mapping:"name:{type:keyword,fields:{text: {type: text}}}"`
 	Endpoint string `json:"endpoint,omitempty" elastic_mapping:"endpoint: { type: keyword }"`
 	Version map[string]interface{} `json:"version,omitempty" elastic_mapping:"version: { type: object }"`
-	BasicAuth BasicAuth `config:"basic_auth" json:"basic_auth,omitempty" elastic_mapping:"basic_auth:{type:object}"`
+	BasicAuth agent.BasicAuth `config:"basic_auth" json:"basic_auth,omitempty" elastic_mapping:"basic_auth:{type:object}"`
 	Owner string `json:"owner,omitempty" config:"owner" elastic_mapping:"owner:{type:keyword}"`
 	Tags [] string `json:"tags,omitempty"`
 	Description string `json:"description,omitempty" config:"description" elastic_mapping:"description:{type:keyword}"`
 }
-
-type BasicAuth struct {
-	Username string `json:"username,omitempty" config:"username" elastic_mapping:"username:{type:keyword}"`
-	Password string `json:"password,omitempty" config:"password" elastic_mapping:"password:{type:keyword}"`
-}

plugin/api/gateway/helper.go

@@ -1,162 +0,0 @@
-/* Copyright © INFINI Ltd. All rights reserved.
- * web: https://infinilabs.com
- * mail: hello#infini.ltd */
-
-package gateway
-
-import (
-	"bytes"
-	"crypto/tls"
-	"github.com/segmentio/encoding/json"
-	"infini.sh/console/model/gateway"
-	"infini.sh/framework/lib/fasthttp"
-	"io"
-	"time"
-)
-
-//import (
-//	"fmt"
-//	"infini.sh/console/model/gateway"
-//	"infini.sh/framework/core/orm"
-//	"infini.sh/framework/core/util"
-//)
-//
-//func fetchInstanceGroup(instanceID string) (string, error){
-//	// fetch gateway instance group
-//	q := orm.Query{}
-//	q.RawQuery = []byte(fmt.Sprintf(`{"size": 1, "query":{"term":{"instance_id":{"value":"%s"}}}}`, instanceID))
-//	err, res := orm.Search(&gateway.InstanceGroup{}, &q)
-//	if err != nil {
-//		return "", err
-//	}
-//	if len(res.Result) > 0 {
-//		if rowMap, ok := res.Result[0].(map[string]interface{}); ok {
-//			return rowMap["group_id"].(string), nil
-//		}
-//	}
-//	return "", nil
-//}
-//
-//func fetchInstanceGroupByID(instanceIDs []interface{})([]interface{}, error){
-//	if len(instanceIDs) == 0 {
-//		return nil, nil
-//	}
-//	// fetch gateway instance groups
-//	esQuery := util.MapStr{
-//		"query": util.MapStr{
-//			"terms": util.MapStr{
-//				"instance_id": instanceIDs,
-//			},
-//		},
-//	}
-//	q := orm.Query{}
-//	q.RawQuery = util.MustToJSONBytes(esQuery)
-//	err, res := orm.Search(&gateway.InstanceGroup{}, &q)
-//	return res.Result, err
-//}
-//func fetchGroupByID(groupIDs []interface{})([]interface{}, error){
-//	if len(groupIDs) == 0 {
-//		return nil, nil
-//	}
-//	// fetch gateway groups
-//	esQuery := util.MapStr{
-//		"query": util.MapStr{
-//			"terms": util.MapStr{
-//				"_id": groupIDs,
-//			},
-//		},
-//	}
-//	q := orm.Query{}
-//	q.RawQuery = util.MustToJSONBytes(esQuery)
-//	err, res := orm.Search(&gateway.Group{}, &q)
-//	return res.Result, err
-//}
-//
-//func pickElasticsearchColumnValues(result []interface{}, columnName string) []interface{}{
-//	if len(result) == 0 {
-//		return nil
-//	}
-//	columnValues := make([]interface{}, 0, len(result))
-//	for _, row := range result {
-//		if rowMap, ok := row.(map[string]interface{}); ok {
-//			columnValues = append(columnValues, rowMap[columnName])
-//		}
-//	}
-//	return columnValues
-//}
-//
-//func getRelationshipMap(result []interface{}, key string, value string) map[string]interface{}{
-//	if len(result) == 0 {
-//		return nil
-//	}
-//	resultMap := map[string]interface{}{}
-//	for _, row := range result {
-//		if rowMap, ok := row.(map[string]interface{}); ok {
-//			resultMap[rowMap[key].(string)] = rowMap[value]
-//		}
-//	}
-//	return resultMap
-//}
-
-type ProxyRequest struct {
-	Endpoint string
-	Path string
-	Method string
-	BasicAuth gateway.BasicAuth
-	Body interface{}
-}
-
-type ProxyResponse struct {
-	Body []byte
-	StatusCode int
-}
-
-func doGatewayRequest(req *ProxyRequest) (*ProxyResponse, error){
-	var (
-		freq = fasthttp.AcquireRequest()
-		fres = fasthttp.AcquireResponse()
-	)
-	defer func() {
-		fasthttp.ReleaseRequest(freq)
-		fasthttp.ReleaseResponse(fres)
-	}()
-	freq.SetRequestURI(req.Endpoint+ req.Path)
-	freq.Header.SetMethod(req.Method)
-	if req.BasicAuth.Username != ""{
-		freq.SetBasicAuth(req.BasicAuth.Username, req.BasicAuth.Password)
-	}
-	if req.Body != nil {
-		switch req.Body.(type) {
-		case []byte:
-			freq.SetBody(req.Body.([]byte))
-		case string:
-			 freq.SetBody([]byte(req.Body.(string)))
-		case io.Reader:
-			 freq.SetBodyStream(req.Body.(io.Reader), -1)
-		default:
-			rw := &bytes.Buffer{}
-			enc := json.NewEncoder(rw)
-			err := enc.Encode(req.Body)
-			if err != nil {
-				return nil, err
-			}
-			freq.SetBody(rw.Bytes())
-		}
-	}
-
-	client := &fasthttp.Client{
-		MaxConnsPerHost: 1000,
-		TLSConfig:       &tls.Config{InsecureSkipVerify: true},
-		ReadTimeout: time.Second * 5,
-	}
-	err := client.Do(freq, fres)
-	if err != nil {
-		return nil, err
-	}
-
-	return &ProxyResponse{
-		Body: fres.Body(),
-		StatusCode: fres.StatusCode(),
-	}, nil
-
-}

plugin/api/gateway/instance.go

@@ -9,8 +9,10 @@ import (
 	log "github.com/cihub/seelog"
 	"github.com/segmentio/encoding/json"
 	"infini.sh/console/model/gateway"
+	"infini.sh/framework/core/agent"
 	httprouter "infini.sh/framework/core/api/router"
 	"infini.sh/framework/core/orm"
+	"infini.sh/framework/core/proxy"
 	"infini.sh/framework/core/util"
 	"net/http"
 	"strconv"
@@ -214,11 +216,11 @@ func (h *GatewayAPI) getInstanceStatus(w http.ResponseWriter, req *http.Request,
 			password = ""
 		}
 		gid, _ := instance.GetValue("id")
-		res, err := doGatewayRequest(&ProxyRequest{
+		res, err :=  proxy.DoProxyRequest(&proxy.Request{
 			Endpoint: endpoint.(string),
 			Method: http.MethodGet,
 			Path: "/stats",
-			BasicAuth: gateway.BasicAuth{
+			BasicAuth: agent.BasicAuth{
 				Username: username.(string),
 				Password: password.(string),
 			},
@@ -262,12 +264,13 @@ func (h *GatewayAPI) proxy(w http.ResponseWriter, req *http.Request, ps httprout
 		}, http.StatusNotFound)
 		return
 	}
-	res, err := doGatewayRequest(&ProxyRequest{
+	res, err :=  proxy.DoProxyRequest(&proxy.Request{
 		Method: method,
 		Endpoint: obj.Endpoint,
 		Path: path,
 		Body: req.Body,
 		BasicAuth: obj.BasicAuth,
+		ContentLength: int(req.ContentLength),
 	})
 	if err != nil {
 		h.WriteError(w, err.Error(), http.StatusInternalServerError)
@@ -290,8 +293,8 @@ type GatewayConnectResponse struct {
 	} `json:"version"`
 
 }
-func (h *GatewayAPI) doConnect(endpoint string, basicAuth gateway.BasicAuth) (*GatewayConnectResponse, error) {
+func (h *GatewayAPI) doConnect(endpoint string, basicAuth agent.BasicAuth) (*GatewayConnectResponse, error) {
-	res, err := doGatewayRequest(&ProxyRequest{
+	res, err := proxy.DoProxyRequest(&proxy.Request{
 		Method: http.MethodGet,
 		Endpoint: endpoint,
 		Path: "/_framework/api/_info",
@@ -313,7 +316,7 @@ func (h *GatewayAPI) doConnect(endpoint string, basicAuth gateway.BasicAuth) (*G
 func (h *GatewayAPI) tryConnect(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	var reqBody = struct {
 		Endpoint string `json:"endpoint"`
-		BasicAuth gateway.BasicAuth
+		BasicAuth agent.BasicAuth
 	}{}
 	err := h.DecodeJSON(req, &reqBody)
 	if err != nil {