Implement HTTP POST based RPC (#5840)

Added support for new RPC support using HTTP POST.  RPC's 
arguments and reply are Gob encoded and sent as HTTP 
request/response body.

This patch also removes Go RPC based implementation.
This commit is contained in:
Bala FA
2018-06-06 14:21:56 +05:30
committed by Nitish Tiwari
parent 9d41051e91
commit 6a53dd1701
59 changed files with 5272 additions and 4169 deletions

128
cmd/rpc/client.go Normal file
View File

@@ -0,0 +1,128 @@
/*
* Minio Cloud Storage, (C) 2018 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rpc
import (
"bytes"
"context"
"crypto/tls"
"encoding/gob"
"errors"
"fmt"
"net"
"net/http"
"reflect"
"time"
xhttp "github.com/minio/minio/cmd/http"
xnet "github.com/minio/minio/pkg/net"
)
// DefaultRPCTimeout - default RPC timeout is one minute.
const DefaultRPCTimeout = 1 * time.Minute
// Client - http based RPC client.
type Client struct {
httpClient *http.Client
serviceURL *xnet.URL
}
// Call - calls service method on RPC server.
func (client *Client) Call(serviceMethod string, args, reply interface{}) error {
replyKind := reflect.TypeOf(reply).Kind()
if replyKind != reflect.Ptr {
return fmt.Errorf("rpc reply must be a pointer type, but found %v", replyKind)
}
data, err := gobEncode(args)
if err != nil {
return err
}
callRequest := CallRequest{
Method: serviceMethod,
ArgBytes: data,
}
var buf bytes.Buffer
if err = gob.NewEncoder(&buf).Encode(callRequest); err != nil {
return err
}
response, err := client.httpClient.Post(client.serviceURL.String(), "", &buf)
if err != nil {
return err
}
defer response.Body.Close()
if response.StatusCode != http.StatusOK {
return fmt.Errorf("%v rpc call failed with error code %v", serviceMethod, response.StatusCode)
}
var callResponse CallResponse
if err := gob.NewDecoder(response.Body).Decode(&callResponse); err != nil {
return err
}
if callResponse.Error != "" {
return errors.New(callResponse.Error)
}
return gobDecode(callResponse.ReplyBytes, reply)
}
// Close - does nothing and presents for interface compatibility.
func (client *Client) Close() error {
return nil
}
func newCustomDialContext(timeout time.Duration) func(ctx context.Context, network, addr string) (net.Conn, error) {
return func(ctx context.Context, network, addr string) (net.Conn, error) {
dialer := &net.Dialer{
Timeout: timeout,
KeepAlive: timeout,
DualStack: true,
}
conn, err := dialer.DialContext(ctx, network, addr)
if err != nil {
return nil, err
}
return xhttp.NewTimeoutConn(conn, timeout, timeout), nil
}
}
// NewClient - returns new RPC client.
func NewClient(serviceURL *xnet.URL, tlsConfig *tls.Config, timeout time.Duration) *Client {
return &Client{
httpClient: &http.Client{
// Transport is exactly same as Go default in https://golang.org/pkg/net/http/#RoundTripper
// except custom DialContext and TLSClientConfig.
Transport: &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: newCustomDialContext(timeout),
MaxIdleConns: 100,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
TLSClientConfig: tlsConfig,
},
},
serviceURL: serviceURL,
}
}

72
cmd/rpc/client_test.go Normal file
View File

@@ -0,0 +1,72 @@
/*
* Minio Cloud Storage, (C) 2018 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rpc
import (
"net/http"
"net/http/httptest"
"testing"
xnet "github.com/minio/minio/pkg/net"
)
func TestClientCall(t *testing.T) {
rpcServer := NewServer()
if err := rpcServer.RegisterName("Arith", &Arith{}); err != nil {
t.Fatalf("unexpected error %v", err)
}
httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
rpcServer.ServeHTTP(w, r)
}))
defer httpServer.Close()
url, err := xnet.ParseURL(httpServer.URL)
if err != nil {
t.Fatalf("unexpected error %v", err)
}
rpcClient := NewClient(url, nil, DefaultRPCTimeout)
var reply int
var boolReply bool
var intArg int
testCases := []struct {
serviceMethod string
args interface{}
reply interface{}
expectErr bool
}{
{"Arith.Multiply", Args{7, 8}, &reply, false},
{"Arith.Multiply", &Args{7, 8}, &reply, false},
// rpc reply must be a pointer type but found int error.
{"Arith.Multiply", &Args{7, 8}, reply, true},
// gob: type mismatch in decoder: want struct type rpc.Args; got non-struct error.
{"Arith.Multiply", intArg, &reply, true},
// gob: decoding into local type *bool, received remote type int error.
{"Arith.Multiply", &Args{7, 8}, &boolReply, true},
}
for i, testCase := range testCases {
err := rpcClient.Call(testCase.serviceMethod, testCase.args, testCase.reply)
expectErr := (err != nil)
if expectErr != testCase.expectErr {
t.Fatalf("case %v: expected: %v, got: %v", i+1, testCase.expectErr, expectErr)
}
}
}

264
cmd/rpc/server.go Normal file
View File

@@ -0,0 +1,264 @@
/*
* Minio Cloud Storage, (C) 2018 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rpc
import (
"bytes"
"encoding/gob"
"errors"
"fmt"
"net/http"
"reflect"
"strings"
"unicode"
"unicode/utf8"
)
// Authenticator - validator of first argument of any RPC call.
type Authenticator interface {
// Method to validate first argument of any RPC call.
Authenticate() error
}
// reflect.Type of error interface.
var errorType = reflect.TypeOf((*error)(nil)).Elem()
// reflect.Type of Authenticator interface.
var authenticatorType = reflect.TypeOf((*Authenticator)(nil)).Elem()
func gobEncode(e interface{}) ([]byte, error) {
var buf bytes.Buffer
if err := gob.NewEncoder(&buf).Encode(e); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
func gobDecode(data []byte, e interface{}) error {
return gob.NewDecoder(bytes.NewReader(data)).Decode(e)
}
// Returns whether given type is exported or builin type or not.
func isExportedOrBuiltinType(t reflect.Type) bool {
for t.Kind() == reflect.Ptr {
t = t.Elem()
}
rune, _ := utf8.DecodeRuneInString(t.Name())
return unicode.IsUpper(rune) || t.PkgPath() == ""
}
// Makes method name map from given type.
func getMethodMap(receiverType reflect.Type) map[string]reflect.Method {
methodMap := make(map[string]reflect.Method)
for i := 0; i < receiverType.NumMethod(); i++ {
// Method.PkgPath is empty for this package.
method := receiverType.Method(i)
// Methods must have three arguments (receiver, args, reply)
if method.Type.NumIn() != 3 {
continue
}
// First argument must be exported.
if !isExportedOrBuiltinType(method.Type.In(1)) {
continue
}
// First argument must be Authenticator.
if !method.Type.In(1).Implements(authenticatorType) {
continue
}
// Second argument must be exported or builtin type.
if !isExportedOrBuiltinType(method.Type.In(2)) {
continue
}
// Second argument must be a pointer.
if method.Type.In(2).Kind() != reflect.Ptr {
continue
}
// Method must return one value.
if method.Type.NumOut() != 1 {
continue
}
// The return type of the method must be error.
if method.Type.Out(0) != errorType {
continue
}
methodMap[method.Name] = method
}
return methodMap
}
// Server - HTTP based RPC server.
type Server struct {
serviceName string
receiverValue reflect.Value
methodMap map[string]reflect.Method
}
// RegisterName - registers receiver with given name to handle RPC requests.
func (server *Server) RegisterName(name string, receiver interface{}) error {
server.serviceName = name
server.receiverValue = reflect.ValueOf(receiver)
if !reflect.Indirect(server.receiverValue).IsValid() {
return fmt.Errorf("nil receiver")
}
receiverName := reflect.Indirect(server.receiverValue).Type().Name()
receiverType := reflect.TypeOf(receiver)
server.methodMap = getMethodMap(receiverType)
if len(server.methodMap) == 0 {
str := "rpc.Register: type " + receiverName + " has no exported methods of suitable type"
// To help the user, see if a pointer receiver would work.
if len(getMethodMap(reflect.PtrTo(receiverType))) != 0 {
str += " (hint: pass a pointer to value of that type)"
}
return errors.New(str)
}
return nil
}
// call - call service method in receiver.
func (server *Server) call(serviceMethod string, argBytes []byte) (replyBytes []byte, err error) {
tokens := strings.SplitN(serviceMethod, ".", 2)
if len(tokens) != 2 {
return nil, fmt.Errorf("invalid service/method request ill-formed %v", serviceMethod)
}
serviceName := tokens[0]
if serviceName != server.serviceName {
return nil, fmt.Errorf("can't find service %v", serviceName)
}
methodName := tokens[1]
method, found := server.methodMap[methodName]
if !found {
return nil, fmt.Errorf("can't find method %v", methodName)
}
var argv reflect.Value
// Decode the argument value.
argIsValue := false // if true, need to indirect before calling.
if method.Type.In(1).Kind() == reflect.Ptr {
argv = reflect.New(method.Type.In(1).Elem())
} else {
argv = reflect.New(method.Type.In(1))
argIsValue = true
}
if err = gobDecode(argBytes, argv.Interface()); err != nil {
return nil, err
}
if argIsValue {
argv = argv.Elem()
}
// call Authenticate() method.
authMethod, ok := method.Type.In(1).MethodByName("Authenticate")
if !ok {
panic("Authenticate() method not found. This should not happen.")
}
returnValues := authMethod.Func.Call([]reflect.Value{argv})
errInter := returnValues[0].Interface()
if errInter != nil {
err = errInter.(error)
}
if err != nil {
return nil, err
}
replyv := reflect.New(method.Type.In(2).Elem())
switch method.Type.In(2).Elem().Kind() {
case reflect.Map:
replyv.Elem().Set(reflect.MakeMap(method.Type.In(2).Elem()))
case reflect.Slice:
replyv.Elem().Set(reflect.MakeSlice(method.Type.In(2).Elem(), 0, 0))
}
returnValues = method.Func.Call([]reflect.Value{server.receiverValue, argv, replyv})
errInter = returnValues[0].Interface()
if errInter != nil {
err = errInter.(error)
}
if err != nil {
return nil, err
}
return gobEncode(replyv.Interface())
}
// CallRequest - RPC call request parameters.
type CallRequest struct {
Method string
ArgBytes []byte
}
// CallResponse - RPC call response parameters.
type CallResponse struct {
Error string
ReplyBytes []byte
}
// ServeHTTP - handles RPC on HTTP request.
func (server *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if req.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
var callRequest CallRequest
if err := gob.NewDecoder(req.Body).Decode(&callRequest); err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
var callResponse CallResponse
var err error
callResponse.ReplyBytes, err = server.call(callRequest.Method, callRequest.ArgBytes)
if err != nil {
callResponse.Error = err.Error()
}
data, err := gobEncode(callResponse)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
return
}
w.Write(data)
}
// NewServer - returns new RPC server.
func NewServer() *Server {
return &Server{}
}

345
cmd/rpc/server_test.go Normal file
View File

@@ -0,0 +1,345 @@
/*
* Minio Cloud Storage, (C) 2018 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rpc
import (
"bytes"
"errors"
"net/http"
"net/http/httptest"
"reflect"
"testing"
)
type Args struct {
A, B int
}
func (a *Args) Authenticate() (err error) {
if a.A == 0 && a.B == 0 {
err = errors.New("authenticated failed")
}
return
}
type Quotient struct {
Quo, Rem int
}
type Arith struct{}
func (t *Arith) Multiply(args *Args, reply *int) error {
*reply = args.A * args.B
return nil
}
func (t *Arith) Divide(args *Args, quo *Quotient) error {
if args.B == 0 {
return errors.New("divide by zero")
}
quo.Quo = args.A / args.B
quo.Rem = args.A % args.B
return nil
}
type mytype int
type Auth struct{}
func (a Auth) Authenticate() error {
return nil
}
// exported method.
func (t mytype) Foo(a *Auth, b *int) error {
return nil
}
// incompatible method because of unexported method.
func (t mytype) foo(a *Auth, b *int) error {
return nil
}
// incompatible method because of first argument is not Authenticator.
func (t *mytype) Bar(a, b *int) error {
return nil
}
// incompatible method because of error is not returned.
func (t mytype) IncompatFoo(a, b *int) {
}
// incompatible method because of second argument is not a pointer.
func (t *mytype) IncompatBar(a *int, b int) error {
return nil
}
func TestIsExportedOrBuiltinType(t *testing.T) {
var i int
case1Type := reflect.TypeOf(i)
var iptr *int
case2Type := reflect.TypeOf(iptr)
var a Arith
case3Type := reflect.TypeOf(a)
var aptr *Arith
case4Type := reflect.TypeOf(aptr)
var m mytype
case5Type := reflect.TypeOf(m)
var mptr *mytype
case6Type := reflect.TypeOf(mptr)
testCases := []struct {
t reflect.Type
expectedResult bool
}{
{case1Type, true},
{case2Type, true},
{case3Type, true},
{case4Type, true},
// Type.Name() starts with lower case and Type.PkgPath() is not empty.
{case5Type, false},
// Type.Name() starts with lower case and Type.PkgPath() is not empty.
{case6Type, false},
}
for i, testCase := range testCases {
result := isExportedOrBuiltinType(testCase.t)
if result != testCase.expectedResult {
t.Fatalf("case %v: expected: %v, got: %v\n", i+1, testCase.expectedResult, result)
}
}
}
func TestGetMethodMap(t *testing.T) {
var a Arith
case1Type := reflect.TypeOf(a)
var aptr *Arith
case2Type := reflect.TypeOf(aptr)
var m mytype
case3Type := reflect.TypeOf(m)
var mptr *mytype
case4Type := reflect.TypeOf(mptr)
testCases := []struct {
t reflect.Type
expectedResult int
}{
// No methods exported.
{case1Type, 0},
// Multiply and Divide methods are exported.
{case2Type, 2},
// Foo method is exported.
{case3Type, 1},
// Foo method is exported.
{case4Type, 1},
}
for i, testCase := range testCases {
m := getMethodMap(testCase.t)
result := len(m)
if result != testCase.expectedResult {
t.Fatalf("case %v: expected: %v, got: %v\n", i+1, testCase.expectedResult, result)
}
}
}
func TestServerRegisterName(t *testing.T) {
case1Receiver := &Arith{}
var case2Receiver mytype
var case3Receiver *Arith
i := 0
var case4Receiver = &i
var case5Receiver Arith
testCases := []struct {
name string
receiver interface{}
expectErr bool
}{
{"Arith", case1Receiver, false},
{"arith", case1Receiver, false},
{"Arith", case2Receiver, false},
// nil receiver error.
{"Arith", nil, true},
// nil receiver error.
{"Arith", case3Receiver, true},
// rpc.Register: type Arith has no exported methods of suitable type error.
{"Arith", case4Receiver, true},
// rpc.Register: type Arith has no exported methods of suitable type (hint: pass a pointer to value of that type) error.
{"Arith", case5Receiver, true},
}
for i, testCase := range testCases {
err := NewServer().RegisterName(testCase.name, testCase.receiver)
expectErr := (err != nil)
if expectErr != testCase.expectErr {
t.Fatalf("case %v: expected: %v, got: %v\n", i+1, testCase.expectErr, expectErr)
}
}
}
func TestServerCall(t *testing.T) {
server1 := NewServer()
if err := server1.RegisterName("Arith", &Arith{}); err != nil {
t.Fatalf("unexpected error %v", err)
}
server2 := NewServer()
if err := server2.RegisterName("arith", &Arith{}); err != nil {
t.Fatalf("unexpected error %v", err)
}
case1ArgBytes, err := gobEncode(&Args{7, 8})
if err != nil {
t.Fatalf("unexpected error %v", err)
}
reply := 7 * 8
case1ExpectedResult, err := gobEncode(&reply)
if err != nil {
t.Fatalf("unexpected error %v", err)
}
case2ArgBytes, err := gobEncode(&Args{})
if err != nil {
t.Fatalf("unexpected error %v", err)
}
testCases := []struct {
server *Server
serviceMethod string
argBytes []byte
expectedResult []byte
expectErr bool
}{
{server1, "Arith.Multiply", case1ArgBytes, case1ExpectedResult, false},
{server2, "arith.Multiply", case1ArgBytes, case1ExpectedResult, false},
// invalid service/method request ill-formed error.
{server1, "Multiply", nil, nil, true},
// can't find service error.
{server1, "arith.Multiply", nil, nil, true},
// can't find method error.
{server1, "Arith.Add", nil, nil, true},
// gob decode error.
{server1, "Arith.Multiply", []byte{10}, nil, true},
// authentication error.
{server1, "Arith.Multiply", case2ArgBytes, nil, true},
}
for i, testCase := range testCases {
result, err := testCase.server.call(testCase.serviceMethod, testCase.argBytes)
expectErr := (err != nil)
if expectErr != testCase.expectErr {
t.Fatalf("case %v: error: expected: %v, got: %v\n", i+1, testCase.expectErr, expectErr)
}
if !testCase.expectErr {
if !reflect.DeepEqual(result, testCase.expectedResult) {
t.Fatalf("case %v: result: expected: %v, got: %v\n", i+1, testCase.expectedResult, result)
}
}
}
}
func TestServerServeHTTP(t *testing.T) {
server1 := NewServer()
if err := server1.RegisterName("Arith", &Arith{}); err != nil {
t.Fatalf("unexpected error %v", err)
}
argBytes, err := gobEncode(&Args{7, 8})
if err != nil {
t.Fatalf("unexpected error %v", err)
}
requestBodyData, err := gobEncode(CallRequest{Method: "Arith.Multiply", ArgBytes: argBytes})
if err != nil {
t.Fatalf("unexpected error %v", err)
}
case1Request, err := http.NewRequest("POST", "http://localhost:12345/", bytes.NewReader(requestBodyData))
if err != nil {
t.Fatalf("unexpected error %v", err)
}
reply := 7 * 8
replyBytes, err := gobEncode(&reply)
if err != nil {
t.Fatalf("unexpected error %v", err)
}
case1Result, err := gobEncode(CallResponse{ReplyBytes: replyBytes})
if err != nil {
t.Fatalf("unexpected error %v", err)
}
case2Request, err := http.NewRequest("GET", "http://localhost:12345/", bytes.NewReader([]byte{}))
if err != nil {
t.Fatalf("unexpected error %v", err)
}
case3Request, err := http.NewRequest("POST", "http://localhost:12345/", bytes.NewReader([]byte{10, 20}))
if err != nil {
t.Fatalf("unexpected error %v", err)
}
requestBodyData, err = gobEncode(CallRequest{Method: "Arith.Add", ArgBytes: argBytes})
if err != nil {
t.Fatalf("unexpected error %v", err)
}
case4Request, err := http.NewRequest("POST", "http://localhost:12345/", bytes.NewReader(requestBodyData))
if err != nil {
t.Fatalf("unexpected error %v", err)
}
case4Result, err := gobEncode(CallResponse{Error: "can't find method Add"})
if err != nil {
t.Fatalf("unexpected error %v", err)
}
testCases := []struct {
server *Server
httpRequest *http.Request
expectedCode int
expectedResult []byte
}{
{server1, case1Request, http.StatusOK, case1Result},
{server1, case2Request, http.StatusMethodNotAllowed, nil},
{server1, case3Request, http.StatusBadRequest, nil},
{server1, case4Request, http.StatusOK, case4Result},
}
for i, testCase := range testCases {
writer := httptest.NewRecorder()
testCase.server.ServeHTTP(writer, testCase.httpRequest)
if writer.Code != testCase.expectedCode {
t.Fatalf("case %v: code: expected: %v, got: %v\n", i+1, testCase.expectedCode, writer.Code)
}
if testCase.expectedCode == http.StatusOK {
result := writer.Body.Bytes()
if !reflect.DeepEqual(result, testCase.expectedResult) {
t.Fatalf("case %v: result: expected: %v, got: %v\n", i+1, testCase.expectedResult, result)
}
}
}
}