Merge pull request #854 from krishnasrinivas/servicehandlers2

Controller Service proxies rpc calls to the corresponding servers
master
Harshavardhana 9 years ago
commit d9328d25e9
  1. 4
      controller-router.go
  2. 119
      controller-rpc-server.go
  3. 4
      controller.go
  4. 28
      controller_rpc_test.go
  5. 7
      flags.go
  6. 1
      main.go
  7. 80
      rpc-definitions.go
  8. 1
      server-api-definitions.go
  9. 2
      server-main.go
  10. 11
      server-router.go
  11. 63
      server-rpc-server.go
  12. 50
      server.go

@ -25,13 +25,13 @@ import (
) )
// getRPCHandler rpc handler // getRPCHandler rpc handler
func getRPCHandler() http.Handler { func getRPCCtrlHandler() http.Handler {
s := jsonrpc.NewServer() s := jsonrpc.NewServer()
s.RegisterCodec(json.NewCodec(), "application/json") s.RegisterCodec(json.NewCodec(), "application/json")
s.RegisterService(new(VersionService), "Version") s.RegisterService(new(VersionService), "Version")
s.RegisterService(new(DonutService), "Donut") s.RegisterService(new(DonutService), "Donut")
s.RegisterService(new(AuthService), "Auth") s.RegisterService(new(AuthService), "Auth")
s.RegisterService(new(ServerService), "Server") s.RegisterService(new(controllerServerRPCService), "Server")
// Add new RPC services here // Add new RPC services here
return registerRPC(router.NewRouter(), s) return registerRPC(router.NewRouter(), s)
} }

@ -18,118 +18,51 @@ package main
import ( import (
"net/http" "net/http"
"os"
"runtime"
"syscall"
"github.com/gorilla/rpc/v2/json"
"github.com/minio/minio/pkg/probe" "github.com/minio/minio/pkg/probe"
) )
// MinioServer - container for minio server data type controllerServerRPCService struct {
type MinioServer struct { serverList []ServerArg
IP string `json:"ip"`
ID string `json:"id"`
Name string `json:"name"`
Status string `json:"status"`
} }
// ServerArgs - server arg func proxyRequest(method string, ip string, arg interface{}, res interface{}) error {
type ServerArgs struct { op := rpcOperation{
MinioServers []MinioServer `json:"servers"` Method: "Server." + method,
Request: arg,
} }
// ServerAddReply - server add reply request, _ := newRPCRequest("http://"+ip+":9002/rpc", op, nil)
type ServerAddReply struct { resp, err := request.Do()
ServersAdded []MinioServer `json:"serversAdded"` if err != nil {
return probe.WrapError(err)
} }
decodeerr := json.DecodeClientResponse(resp.Body, res)
// MemStatsReply memory statistics return decodeerr
type MemStatsReply struct {
runtime.MemStats `json:"memstats"`
} }
// DiskStatsReply disk statistics func (s *controllerServerRPCService) Add(r *http.Request, arg *ServerArg, res *DefaultRep) error {
type DiskStatsReply struct { err := proxyRequest("Add", arg.IP, arg, res)
DiskStats syscall.Statfs_t `json:"diskstats"` if err == nil {
s.serverList = append(s.serverList, *arg)
} }
return err
// SysInfoReply system info
type SysInfoReply struct {
Hostname string `json:"hostname"`
SysARCH string `json:"sysArch"`
SysOS string `json:"sysOS"`
SysCPUS int `json:"sysNCPUs"`
GORoutines int `json:"golangRoutines"`
GOVersion string `json:"golangVersion"`
} }
// ServerListReply list of minio servers func (s *controllerServerRPCService) MemStats(r *http.Request, arg *ServerArg, res *MemStatsRep) error {
type ServerListReply struct { return proxyRequest("MemStats", arg.IP, arg, res)
ServerList []MinioServer `json:"servers"`
} }
// ServerService server json rpc service func (s *controllerServerRPCService) DiskStats(r *http.Request, arg *ServerArg, res *DiskStatsRep) error {
type ServerService struct { return proxyRequest("DiskStats", arg.IP, arg, res)
serverList []MinioServer
} }
// Add - add new server func (s *controllerServerRPCService) SysInfo(r *http.Request, arg *ServerArg, res *SysInfoRep) error {
func (s *ServerService) Add(r *http.Request, arg *ServerArgs, reply *ServerAddReply) error { return proxyRequest("SysInfo", arg.IP, arg, res)
for _, server := range arg.MinioServers {
server.Status = "connected"
reply.ServersAdded = append(reply.ServersAdded, server)
}
return nil
} }
// MemStats - memory statistics on the server func (s *controllerServerRPCService) List(r *http.Request, arg *ServerArg, res *ListRep) error {
func (s *ServerService) MemStats(r *http.Request, arg *ServerArgs, reply *MemStatsReply) error { res.List = s.serverList
runtime.ReadMemStats(&reply.MemStats)
return nil
}
// DiskStats - disk statistics on the server
func (s *ServerService) DiskStats(r *http.Request, arg *ServerArgs, reply *DiskStatsReply) error {
syscall.Statfs("/", &reply.DiskStats)
return nil
}
// SysInfo - system info for the server
func (s *ServerService) SysInfo(r *http.Request, arg *ServerArgs, reply *SysInfoReply) error {
reply.SysOS = runtime.GOOS
reply.SysARCH = runtime.GOARCH
reply.SysCPUS = runtime.NumCPU()
reply.GOVersion = runtime.Version()
reply.GORoutines = runtime.NumGoroutine()
var err error
reply.Hostname, err = os.Hostname()
if err != nil {
return probe.WrapError(probe.NewError(err))
}
return nil
}
// List of servers in the cluster
func (s *ServerService) List(r *http.Request, arg *ServerArgs, reply *ServerListReply) error {
reply.ServerList = []MinioServer{
{
"server.one",
"192.168.1.1",
"192.168.1.1",
"connected",
},
{
"server.two",
"192.168.1.2",
"192.168.1.2",
"connected",
},
{
"server.three",
"192.168.1.3",
"192.168.1.3",
"connected",
},
}
return nil return nil
} }

@ -28,7 +28,7 @@ import (
) )
// getRPCServer instance // getRPCServer instance
func getRPCServer(rpcHandler http.Handler) (*http.Server, *probe.Error) { func getControllerRPCServer(rpcHandler http.Handler) (*http.Server, *probe.Error) {
// Minio server config // Minio server config
httpServer := &http.Server{ httpServer := &http.Server{
Addr: ":9001", // TODO make this configurable Addr: ":9001", // TODO make this configurable
@ -56,7 +56,7 @@ func getRPCServer(rpcHandler http.Handler) (*http.Server, *probe.Error) {
// StartController starts a minio controller // StartController starts a minio controller
func StartController() *probe.Error { func StartController() *probe.Error {
rpcServer, err := getRPCServer(getRPCHandler()) rpcServer, err := getControllerRPCServer(getRPCCtrlHandler())
if err != nil { if err != nil {
return err.Trace() return err.Trace()
} }

@ -1,3 +1,5 @@
// +build ignore
/* /*
* Minio Cloud Storage, (C) 2014 Minio, Inc. * Minio Cloud Storage, (C) 2014 Minio, Inc.
* *
@ -38,7 +40,7 @@ func (s *ControllerRPCSuite) SetUpSuite(c *C) {
c.Assert(err, IsNil) c.Assert(err, IsNil)
auth.SetAuthConfigPath(root) auth.SetAuthConfigPath(root)
testRPCServer = httptest.NewServer(getRPCHandler()) testRPCServer = httptest.NewServer(getRPCCtrlHandler())
} }
func (s *ControllerRPCSuite) TearDownSuite(c *C) { func (s *ControllerRPCSuite) TearDownSuite(c *C) {
@ -48,7 +50,7 @@ func (s *ControllerRPCSuite) TearDownSuite(c *C) {
func (s *ControllerRPCSuite) TestMemStats(c *C) { func (s *ControllerRPCSuite) TestMemStats(c *C) {
op := rpcOperation{ op := rpcOperation{
Method: "Server.MemStats", Method: "Server.MemStats",
Request: ServerArgs{}, Request: ServerArg{},
} }
req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport) req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport)
c.Assert(err, IsNil) c.Assert(err, IsNil)
@ -57,16 +59,16 @@ func (s *ControllerRPCSuite) TestMemStats(c *C) {
c.Assert(err, IsNil) c.Assert(err, IsNil)
c.Assert(resp.StatusCode, Equals, http.StatusOK) c.Assert(resp.StatusCode, Equals, http.StatusOK)
var reply MemStatsReply var reply MemStatsRep
c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil) c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil)
resp.Body.Close() resp.Body.Close()
c.Assert(reply, Not(DeepEquals), MemStatsReply{}) c.Assert(reply, Not(DeepEquals), MemStatsRep{})
} }
func (s *ControllerRPCSuite) TestSysInfo(c *C) { func (s *ControllerRPCSuite) TestSysInfo(c *C) {
op := rpcOperation{ op := rpcOperation{
Method: "Server.SysInfo", Method: "Server.SysInfo",
Request: ServerArgs{}, Request: ServerArg{},
} }
req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport) req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport)
c.Assert(err, IsNil) c.Assert(err, IsNil)
@ -75,16 +77,16 @@ func (s *ControllerRPCSuite) TestSysInfo(c *C) {
c.Assert(err, IsNil) c.Assert(err, IsNil)
c.Assert(resp.StatusCode, Equals, http.StatusOK) c.Assert(resp.StatusCode, Equals, http.StatusOK)
var reply SysInfoReply var reply SysInfoRep
c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil) c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil)
resp.Body.Close() resp.Body.Close()
c.Assert(reply, Not(DeepEquals), SysInfoReply{}) c.Assert(reply, Not(DeepEquals), SysInfoRep{})
} }
func (s *ControllerRPCSuite) TestServerList(c *C) { func (s *ControllerRPCSuite) TestServerList(c *C) {
op := rpcOperation{ op := rpcOperation{
Method: "Server.List", Method: "Server.List",
Request: ServerArgs{}, Request: ServerArg{},
} }
req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport) req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport)
c.Assert(err, IsNil) c.Assert(err, IsNil)
@ -93,16 +95,16 @@ func (s *ControllerRPCSuite) TestServerList(c *C) {
c.Assert(err, IsNil) c.Assert(err, IsNil)
c.Assert(resp.StatusCode, Equals, http.StatusOK) c.Assert(resp.StatusCode, Equals, http.StatusOK)
var reply ServerListReply var reply ServerListRep
c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil) c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil)
resp.Body.Close() resp.Body.Close()
c.Assert(reply, Not(DeepEquals), ServerListReply{}) c.Assert(reply, Not(DeepEquals), ServerListRep{})
} }
func (s *ControllerRPCSuite) TestServerAdd(c *C) { func (s *ControllerRPCSuite) TestServerAdd(c *C) {
op := rpcOperation{ op := rpcOperation{
Method: "Server.Add", Method: "Server.Add",
Request: ServerArgs{MinioServers: []MinioServer{}}, Request: ServerArg{},
} }
req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport) req, err := newRPCRequest(testRPCServer.URL+"/rpc", op, http.DefaultTransport)
c.Assert(err, IsNil) c.Assert(err, IsNil)
@ -111,10 +113,10 @@ func (s *ControllerRPCSuite) TestServerAdd(c *C) {
c.Assert(err, IsNil) c.Assert(err, IsNil)
c.Assert(resp.StatusCode, Equals, http.StatusOK) c.Assert(resp.StatusCode, Equals, http.StatusOK)
var reply ServerAddReply var reply DefaultRep
c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil) c.Assert(json.DecodeClientResponse(resp.Body, &reply), IsNil)
resp.Body.Close() resp.Body.Close()
c.Assert(reply, Not(DeepEquals), ServerAddReply{ServersAdded: []MinioServer{}}) c.Assert(reply, Not(DeepEquals), DefaultRep{0, "Added"})
} }
func (s *ControllerRPCSuite) TestAuth(c *C) { func (s *ControllerRPCSuite) TestAuth(c *C) {

@ -35,6 +35,13 @@ var (
Usage: "ADDRESS:PORT for management console access", Usage: "ADDRESS:PORT for management console access",
} }
addressRPCServerFlag = cli.StringFlag{
Name: "address-rpcserver",
Hide: true,
Value: ":9002",
Usage: "ADDRESS:PORT for management console access",
}
ratelimitFlag = cli.IntFlag{ ratelimitFlag = cli.IntFlag{
Name: "ratelimit", Name: "ratelimit",
Value: 16, Value: 16,

@ -91,6 +91,7 @@ func registerApp() *cli.App {
registerFlag(certFlag) registerFlag(certFlag)
registerFlag(keyFlag) registerFlag(keyFlag)
registerFlag(debugFlag) registerFlag(debugFlag)
registerFlag(addressRPCServerFlag)
// set up app // set up app
app := cli.NewApp() app := cli.NewApp()

@ -0,0 +1,80 @@
/*
* Minio Cloud Storage, (C) 2014 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 main
// Ethernet interface
type Ifc struct {
IP string `json:"ip"`
Mask string `json:"mask"`
Eth string `json:"ifc"`
}
// Identify a server
type ServerArg struct {
Name string `json:"name"`
IP string `json:"ip"`
ID string `json:"id"`
}
// Needed for Reply for Server.List
type ServerRep struct {
Name string `json:"name"`
IP string `json:"ip"`
ID string `json:"id"`
}
// Default reply
type DefaultRep struct {
Error int64 `json:"error"`
Message string `json:"message"`
}
// Needed for Reply List
type ServerListRep struct {
List []ServerRep
}
// Reply DiskStats
type DiskStatsRep struct {
Disks []string
}
// Reply MemStats
type MemStatsRep struct {
Total uint64 `json:"total"`
Free uint64 `json:"free"`
}
// Reply NetStats
type NetStatsRep struct {
Interfaces []Ifc
}
// Reply SysInfo
type SysInfoRep struct {
Hostname string `json:"hostname"`
SysARCH string `json:"sys.arch"`
SysOS string `json:"sys.os"`
SysCPUS int `json:"sys.ncpus"`
Routines int `json:"goroutines"`
GOVersion string `json:"goversion"`
}
// Reply List
type ListRep struct {
List []ServerArg `json:"list"`
}

@ -21,6 +21,7 @@ import "encoding/xml"
// APIConfig - http server config // APIConfig - http server config
type APIConfig struct { type APIConfig struct {
Address string Address string
AddressRPC string
TLS bool TLS bool
CertFile string CertFile string
KeyFile string KeyFile string

@ -42,8 +42,10 @@ func getServerConfig(c *cli.Context) APIConfig {
Fatalln("Both certificate and key are required to enable https.") Fatalln("Both certificate and key are required to enable https.")
} }
tls := (certFile != "" && keyFile != "") tls := (certFile != "" && keyFile != "")
return APIConfig{ return APIConfig{
Address: c.GlobalString("address"), Address: c.GlobalString("address"),
AddressRPC: c.GlobalString("address-rpcserver"),
TLS: tls, TLS: tls,
CertFile: certFile, CertFile: certFile,
KeyFile: keyFile, KeyFile: keyFile,

@ -20,6 +20,8 @@ import (
"net/http" "net/http"
router "github.com/gorilla/mux" router "github.com/gorilla/mux"
jsonrpc "github.com/gorilla/rpc/v2"
"github.com/gorilla/rpc/v2/json"
) )
// registerAPI - register all the object API handlers to their respective paths // registerAPI - register all the object API handlers to their respective paths
@ -70,3 +72,12 @@ func getAPIHandler(conf APIConfig) (http.Handler, MinioAPI) {
apiHandler := registerCustomMiddleware(mux, mwHandlers...) apiHandler := registerCustomMiddleware(mux, mwHandlers...)
return apiHandler, minioAPI return apiHandler, minioAPI
} }
func getRPCServerHandler() http.Handler {
s := jsonrpc.NewServer()
s.RegisterCodec(json.NewCodec(), "application/json")
s.RegisterService(new(serverServerService), "Server")
mux := router.NewRouter()
mux.Handle("/rpc", s)
return mux
}

@ -0,0 +1,63 @@
/*
* Minio Cloud Storage, (C) 2015 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 main
import (
"net/http"
"os"
"runtime"
"github.com/minio/minio/pkg/probe"
)
type serverServerService struct{}
func (s *serverServerService) Add(r *http.Request, arg *ServerArg, rep *DefaultRep) error {
rep.Error = 0
rep.Message = "Added successfully"
return nil
}
func (s *serverServerService) MemStats(r *http.Request, arg *ServerArg, rep *MemStatsRep) error {
rep.Total = 64 * 1024 * 1024 * 1024
rep.Free = 9 * 1024 * 1024 * 1024
return nil
}
func (s *serverServerService) DiskStats(r *http.Request, arg *ServerArg, rep *DiskStatsRep) error {
rep.Disks = []string{"/mnt/disk1", "/mnt/disk2", "/mnt/disk3", "/mnt/disk4", "/mnt/disk5", "/mnt/disk6"}
return nil
}
func (s *serverServerService) SysInfo(r *http.Request, arg *ServerArg, rep *SysInfoRep) error {
rep.SysARCH = runtime.GOARCH
rep.SysOS = runtime.GOOS
rep.SysCPUS = runtime.NumCPU()
rep.Routines = runtime.NumGoroutine()
rep.GOVersion = runtime.Version()
var err error
rep.Hostname, err = os.Hostname()
if err != nil {
return probe.WrapError(probe.NewError(err))
}
return nil
}
func (s *serverServerService) NetStats(r *http.Request, arg *ServerArg, rep *NetStatsRep) error {
rep.Interfaces = []Ifc{{"192.168.1.1", "255.255.255.0", "eth0"}}
return nil
}

@ -28,28 +28,20 @@ import (
"github.com/minio/minio/pkg/probe" "github.com/minio/minio/pkg/probe"
) )
// getAPI server instance func configureServer(conf APIConfig, httpServer *http.Server) *probe.Error {
func getAPIServer(conf APIConfig, apiHandler http.Handler) (*http.Server, *probe.Error) {
// Minio server config
httpServer := &http.Server{
Addr: conf.Address,
Handler: apiHandler,
MaxHeaderBytes: 1 << 20,
}
if conf.TLS { if conf.TLS {
var err error var err error
httpServer.TLSConfig = &tls.Config{} httpServer.TLSConfig = &tls.Config{}
httpServer.TLSConfig.Certificates = make([]tls.Certificate, 1) httpServer.TLSConfig.Certificates = make([]tls.Certificate, 1)
httpServer.TLSConfig.Certificates[0], err = tls.LoadX509KeyPair(conf.CertFile, conf.KeyFile) httpServer.TLSConfig.Certificates[0], err = tls.LoadX509KeyPair(conf.CertFile, conf.KeyFile)
if err != nil { if err != nil {
return nil, probe.NewError(err) return probe.NewError(err)
} }
} }
host, port, err := net.SplitHostPort(conf.Address) host, port, err := net.SplitHostPort(conf.Address)
if err != nil { if err != nil {
return nil, probe.NewError(err) return probe.NewError(err)
} }
var hosts []string var hosts []string
@ -59,7 +51,7 @@ func getAPIServer(conf APIConfig, apiHandler http.Handler) (*http.Server, *probe
default: default:
addrs, err := net.InterfaceAddrs() addrs, err := net.InterfaceAddrs()
if err != nil { if err != nil {
return nil, probe.NewError(err) return probe.NewError(err)
} }
for _, addr := range addrs { for _, addr := range addrs {
if addr.Network() == "ip+net" { if addr.Network() == "ip+net" {
@ -77,7 +69,20 @@ func getAPIServer(conf APIConfig, apiHandler http.Handler) (*http.Server, *probe
} else { } else {
fmt.Printf("Starting minio server on: http://%s:%s, PID: %d\n", host, port, os.Getpid()) fmt.Printf("Starting minio server on: http://%s:%s, PID: %d\n", host, port, os.Getpid())
} }
}
return nil
}
// getAPI server instance
func getAPIServer(conf APIConfig, apiHandler http.Handler) (*http.Server, *probe.Error) {
// Minio server config
httpServer := &http.Server{
Addr: conf.Address,
Handler: apiHandler,
MaxHeaderBytes: 1 << 20,
}
if err := configureServer(conf, httpServer); err != nil {
return nil, err
} }
return httpServer, nil return httpServer, nil
} }
@ -91,7 +96,19 @@ func startTM(a MinioAPI) {
} }
} }
// StartServer starts an s3 compatible cloud storage server func getServerRPCServer(conf APIConfig, handler http.Handler) (*http.Server, *probe.Error) {
httpServer := &http.Server{
Addr: conf.AddressRPC,
Handler: handler,
MaxHeaderBytes: 1 << 20,
}
if err := configureServer(conf, httpServer); err != nil {
return nil, err
}
return httpServer, nil
}
// Start starts a s3 compatible cloud storage server
func StartServer(conf APIConfig) *probe.Error { func StartServer(conf APIConfig) *probe.Error {
apiHandler, minioAPI := getAPIHandler(conf) apiHandler, minioAPI := getAPIHandler(conf)
apiServer, err := getAPIServer(conf, apiHandler) apiServer, err := getAPIServer(conf, apiHandler)
@ -100,7 +117,12 @@ func StartServer(conf APIConfig) *probe.Error {
} }
// start ticket master // start ticket master
go startTM(minioAPI) go startTM(minioAPI)
if err := minhttp.ListenAndServeLimited(conf.RateLimit, apiServer); err != nil { rpcHandler := getRPCServerHandler()
rpcServer, err := getServerRPCServer(conf, rpcHandler)
if err != nil {
return err.Trace()
}
if err := minhttp.ListenAndServeLimited(conf.RateLimit, apiServer, rpcServer); err != nil {
return err.Trace() return err.Trace()
} }
return nil return nil

Loading…
Cancel
Save