From 3cb578488ca9d1a5eafd084c54390402342f85cf Mon Sep 17 00:00:00 2001 From: John Guo Date: Thu, 27 Jan 2022 16:50:31 +0800 Subject: [PATCH] add service registry feature for package ghttp/gclient --- contrib/balancer/balancer_node.go | 4 ++ contrib/registry/etcd/etcd.go | 1 - contrib/registry/etcd/etcd_registry.go | 8 +-- contrib/registry/etcd/etcd_watcher.go | 4 ++ example/registry/http/client/main.go | 21 +++++++ example/registry/http/server/main.go | 18 ++++++ .../rawgrpc/greeter_client/main.go | 2 +- .../rawgrpc/greeter_server/main.go | 2 +- .../rawgrpc/helloworld/helloworld.pb.go | 0 .../rawgrpc/helloworld/helloworld.proto | 0 .../rawgrpc/helloworld/helloworld_grpc.pb.go | 0 frame/gins/gins_server.go | 4 +- net/gclient/gclient.go | 2 +- net/gclient/gclient_discovery.go | 39 +++++++++++++ net/gclient/gclient_tracing.go | 4 +- net/ghttp/ghttp.go | 4 +- net/ghttp/ghttp_server.go | 32 +++++------ net/ghttp/ghttp_server_config.go | 24 ++++++-- net/ghttp/ghttp_server_registry.go | 56 +++++++++++++++++++ net/gsel/gsel.go | 1 + net/gsvc/gsvc.go | 6 ++ net/gsvc/gsvc_service.go | 9 +++ 22 files changed, 205 insertions(+), 36 deletions(-) create mode 100644 example/registry/http/client/main.go create mode 100644 example/registry/http/server/main.go rename example/{ => registry}/rawgrpc/greeter_client/main.go (94%) rename example/{ => registry}/rawgrpc/greeter_server/main.go (95%) rename example/{ => registry}/rawgrpc/helloworld/helloworld.pb.go (100%) rename example/{ => registry}/rawgrpc/helloworld/helloworld.proto (100%) rename example/{ => registry}/rawgrpc/helloworld/helloworld_grpc.pb.go (100%) create mode 100644 net/gclient/gclient_discovery.go create mode 100644 net/ghttp/ghttp_server_registry.go diff --git a/contrib/balancer/balancer_node.go b/contrib/balancer/balancer_node.go index f5ccf36c4..007332ee4 100644 --- a/contrib/balancer/balancer_node.go +++ b/contrib/balancer/balancer_node.go @@ -19,3 +19,7 @@ type Node struct { func (n *Node) Service() *gsvc.Service { return n.service } + +func (n *Node) Address() string { + return n.service.Address() +} diff --git a/contrib/registry/etcd/etcd.go b/contrib/registry/etcd/etcd.go index 7f5997a85..a6fb9453d 100644 --- a/contrib/registry/etcd/etcd.go +++ b/contrib/registry/etcd/etcd.go @@ -21,7 +21,6 @@ import ( var ( _ gsvc.Registry = &Registry{} - _ gsvc.Watcher = &watcher{} ) type Registry struct { diff --git a/contrib/registry/etcd/etcd_registry.go b/contrib/registry/etcd/etcd_registry.go index db629c6d7..f9beb1056 100644 --- a/contrib/registry/etcd/etcd_registry.go +++ b/contrib/registry/etcd/etcd_registry.go @@ -46,12 +46,10 @@ func (r *Registry) Register(ctx context.Context, service *gsvc.Service) error { } func (r *Registry) Deregister(ctx context.Context, service *gsvc.Service) error { - defer func() { - if r.lease != nil { - _ = r.lease.Close() - } - }() _, err := r.client.Delete(ctx, service.Key()) + if r.lease != nil { + _ = r.lease.Close() + } return err } diff --git a/contrib/registry/etcd/etcd_watcher.go b/contrib/registry/etcd/etcd_watcher.go index 9928696a7..f5197956b 100644 --- a/contrib/registry/etcd/etcd_watcher.go +++ b/contrib/registry/etcd/etcd_watcher.go @@ -13,6 +13,10 @@ import ( etcd3 "go.etcd.io/etcd/client/v3" ) +var ( + _ gsvc.Watcher = &watcher{} +) + type watcher struct { key string ctx context.Context diff --git a/example/registry/http/client/main.go b/example/registry/http/client/main.go new file mode 100644 index 000000000..8282fb11b --- /dev/null +++ b/example/registry/http/client/main.go @@ -0,0 +1,21 @@ +package main + +import ( + "fmt" + + "github.com/gogf/gf/contrib/registry/etcd/v2" + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/net/gsvc" + "github.com/gogf/gf/v2/os/gctx" +) + +func main() { + gsvc.SetRegistry(etcd.New(`127.0.0.1:2379`)) + + res, err := g.Client().Get(gctx.New(), `http://hello.svc:12345/`) + if err != nil { + panic(err) + } + defer res.Close() + fmt.Println(res.ReadAllString()) +} diff --git a/example/registry/http/server/main.go b/example/registry/http/server/main.go new file mode 100644 index 000000000..619b487ae --- /dev/null +++ b/example/registry/http/server/main.go @@ -0,0 +1,18 @@ +package main + +import ( + "github.com/gogf/gf/contrib/registry/etcd/v2" + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/net/ghttp" + "github.com/gogf/gf/v2/net/gsvc" +) + +func main() { + gsvc.SetRegistry(etcd.New(`127.0.0.1:2379`)) + + s := g.Server(`hello.svc`) + s.BindHandler("/", func(r *ghttp.Request) { + r.Response.Write(`Hello world`) + }) + s.Run() +} diff --git a/example/rawgrpc/greeter_client/main.go b/example/registry/rawgrpc/greeter_client/main.go similarity index 94% rename from example/rawgrpc/greeter_client/main.go rename to example/registry/rawgrpc/greeter_client/main.go index 01c0ecb7a..de2accf30 100644 --- a/example/rawgrpc/greeter_client/main.go +++ b/example/registry/rawgrpc/greeter_client/main.go @@ -6,7 +6,7 @@ import ( "github.com/gogf/gf/contrib/balancer/v2" "github.com/gogf/gf/contrib/registry/etcd/v2" "github.com/gogf/gf/contrib/resolver/v2" - pb "github.com/gogf/gf/example/rawgrpc/helloworld" + pb "github.com/gogf/gf/example/registry/rawgrpc/helloworld" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/gsvc" "github.com/gogf/gf/v2/os/gctx" diff --git a/example/rawgrpc/greeter_server/main.go b/example/registry/rawgrpc/greeter_server/main.go similarity index 95% rename from example/rawgrpc/greeter_server/main.go rename to example/registry/rawgrpc/greeter_server/main.go index 6768675c4..2b70fb4e5 100644 --- a/example/rawgrpc/greeter_server/main.go +++ b/example/registry/rawgrpc/greeter_server/main.go @@ -7,7 +7,7 @@ import ( "github.com/gogf/gf/contrib/registry/etcd/v2" "github.com/gogf/gf/contrib/resolver/v2" - pb "github.com/gogf/gf/example/rawgrpc/helloworld" + pb "github.com/gogf/gf/example/registry/rawgrpc/helloworld" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/gipv4" "github.com/gogf/gf/v2/net/gsvc" diff --git a/example/rawgrpc/helloworld/helloworld.pb.go b/example/registry/rawgrpc/helloworld/helloworld.pb.go similarity index 100% rename from example/rawgrpc/helloworld/helloworld.pb.go rename to example/registry/rawgrpc/helloworld/helloworld.pb.go diff --git a/example/rawgrpc/helloworld/helloworld.proto b/example/registry/rawgrpc/helloworld/helloworld.proto similarity index 100% rename from example/rawgrpc/helloworld/helloworld.proto rename to example/registry/rawgrpc/helloworld/helloworld.proto diff --git a/example/rawgrpc/helloworld/helloworld_grpc.pb.go b/example/registry/rawgrpc/helloworld/helloworld_grpc.pb.go similarity index 100% rename from example/rawgrpc/helloworld/helloworld_grpc.pb.go rename to example/registry/rawgrpc/helloworld/helloworld_grpc.pb.go diff --git a/frame/gins/gins_server.go b/frame/gins/gins_server.go index 4c176d373..febea4e2c 100644 --- a/frame/gins/gins_server.go +++ b/frame/gins/gins_server.go @@ -63,7 +63,9 @@ func Server(name ...interface{}) *ghttp.Server { // The configuration is not necessary, so it just prints internal logs. intlog.Printf(ctx, `missing configuration from configuration component for HTTP server "%s"`, instanceName) } - + if s.GetName() == "" || s.GetName() == ghttp.DefaultServerName { + s.SetName(instanceName) + } // Server logger configuration checks. serverLoggerNodeName := fmt.Sprintf(`%s.%s.%s`, configNodeName, s.GetName(), configNodeNameLogger) if v, _ := Config().Get(ctx, serverLoggerNodeName); !v.IsEmpty() { diff --git a/net/gclient/gclient.go b/net/gclient/gclient.go index d80c1e411..6881f82d2 100644 --- a/net/gclient/gclient.go +++ b/net/gclient/gclient.go @@ -78,7 +78,7 @@ func New() *Client { } c.header[httpHeaderUserAgent] = defaultClientAgent // It enables OpenTelemetry for client in default. - c.Use(internalMiddlewareTracing) + c.Use(internalMiddlewareTracing, internalMiddlewareDiscovery) return c } diff --git a/net/gclient/gclient_discovery.go b/net/gclient/gclient_discovery.go new file mode 100644 index 000000000..c6fe5cbc1 --- /dev/null +++ b/net/gclient/gclient_discovery.go @@ -0,0 +1,39 @@ +// Copyright GoFrame Author(https://goframe.org). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. + +package gclient + +import ( + "net/http" + + "github.com/gogf/gf/v2/net/gsvc" + "github.com/gogf/gf/v2/os/gctx" +) + +const ( + discoveryMiddlewareHandled gctx.StrKey = `MiddlewareClientDiscoveryHandled` +) + +// internalMiddlewareDiscovery is a client middleware that enables service discovery feature for client. +func internalMiddlewareDiscovery(c *Client, r *http.Request) (response *Response, err error) { + var ctx = r.Context() + // Mark this request is handled by server tracing middleware, + // to avoid repeated handling by the same middleware. + if ctx.Value(discoveryMiddlewareHandled) != nil { + return c.Next(r) + } + if gsvc.GetRegistry() != nil { + service, err := gsvc.Get(ctx, r.URL.Host) + if err != nil { + return nil, err + } + if service != nil { + r.URL.Host = service.Address() + r.Host = service.Address() + } + } + return c.Next(r) +} diff --git a/net/gclient/gclient_tracing.go b/net/gclient/gclient_tracing.go index 56eafd79f..86a2828c7 100644 --- a/net/gclient/gclient_tracing.go +++ b/net/gclient/gclient_tracing.go @@ -48,9 +48,7 @@ const ( // internalMiddlewareTracing is a client middleware that enables tracing feature using standards of OpenTelemetry. func internalMiddlewareTracing(c *Client, r *http.Request) (response *Response, err error) { - var ( - ctx = r.Context() - ) + var ctx = r.Context() // Mark this request is handled by server tracing middleware, // to avoid repeated handling by the same middleware. if ctx.Value(tracingMiddlewareHandled) != nil { diff --git a/net/ghttp/ghttp.go b/net/ghttp/ghttp.go index 474463137..c4e21cca2 100644 --- a/net/ghttp/ghttp.go +++ b/net/ghttp/ghttp.go @@ -12,6 +12,7 @@ import ( "reflect" "time" + "github.com/gogf/gf/v2/net/gsvc" "github.com/gorilla/websocket" "github.com/gogf/gf/v2/container/gmap" @@ -24,7 +25,7 @@ import ( type ( // Server wraps the http.Server and provides more rich features. Server struct { - name string // Unique name for instance management. + instance string // Instance name. config ServerConfig // Configuration. plugins []Plugin // Plugin array to extend server functionality. servers []*gracefulServer // Underlying http.Server array. @@ -36,6 +37,7 @@ type ( statusHandlerMap map[string][]HandlerFunc // Custom status handler map. sessionManager *gsession.Manager // Session manager. openapi *goai.OpenApiV3 // The OpenApi specification management object. + service *gsvc.Service // The service for Registry. } // Router object. diff --git a/net/ghttp/ghttp_server.go b/net/ghttp/ghttp_server.go index 57714ac32..b04d91e58 100644 --- a/net/ghttp/ghttp_server.go +++ b/net/ghttp/ghttp_server.go @@ -47,9 +47,7 @@ func init() { // serverProcessInit initializes some process configurations, which can only be done once. func serverProcessInit() { - var ( - ctx = context.TODO() - ) + var ctx = context.TODO() if !serverProcessInitialized.Cas(false, true) { return } @@ -98,7 +96,7 @@ func GetServer(name ...interface{}) *Server { return s.(*Server) } s := &Server{ - name: serverName, + instance: serverName, plugins: make([]Plugin, 0), servers: make([]*gracefulServer, 0), closeChan: make(chan struct{}, 10000), @@ -123,9 +121,7 @@ func GetServer(name ...interface{}) *Server { // Start starts listening on configured port. // This function does not block the process, you can use function Wait blocking the process. func (s *Server) Start() error { - var ( - ctx = context.TODO() - ) + var ctx = context.TODO() // Swagger UI. if s.config.SwaggerPath != "" { @@ -183,7 +179,7 @@ func (s *Server) Start() error { if s.config.SessionStorage == nil { path := "" if s.config.SessionPath != "" { - path = gfile.Join(s.config.SessionPath, s.name) + path = gfile.Join(s.config.SessionPath, s.config.Name) if !gfile.Exists(path) { if err := gfile.Mkdir(path); err != nil { return gerror.Wrapf(err, `mkdir failed for "%s"`, path) @@ -231,7 +227,7 @@ func (s *Server) Start() error { fdMapStr := genv.Get(adminActionReloadEnvKey).String() if len(fdMapStr) > 0 { sfm := bufferToServerFdMap([]byte(fdMapStr)) - if v, ok := sfm[s.name]; ok { + if v, ok := sfm[s.config.Name]; ok { s.startServer(v) reloaded = true } @@ -249,6 +245,7 @@ func (s *Server) Start() error { }) } s.initOpenApi() + s.doServiceRegister() s.dumpRouterMap() return nil } @@ -352,7 +349,7 @@ func (s *Server) GetRoutes() []RouterItem { array, _ := gregex.MatchString(`(.*?)%([A-Z]+):(.+)@(.+)`, k) for index, registeredItem := range registeredItems { item := RouterItem{ - Server: s.name, + Server: s.config.Name, Address: address, Domain: array[4], Type: registeredItem.Handler.Type, @@ -417,9 +414,8 @@ func (s *Server) GetRoutes() []RouterItem { // Run starts server listening in blocking way. // It's commonly used for single server situation. func (s *Server) Run() { - var ( - ctx = context.TODO() - ) + var ctx = context.TODO() + if err := s.Start(); err != nil { s.Logger().Fatalf(ctx, `%+v`, err) } @@ -434,15 +430,15 @@ func (s *Server) Run() { } } } + s.doServiceDeregister() s.Logger().Printf(ctx, "%d: all servers shutdown", gproc.Pid()) } // Wait blocks to wait for all servers done. -// It's commonly used in multiple servers situation. +// It's commonly used in multiple server situation. func Wait() { - var ( - ctx = context.TODO() - ) + var ctx = context.TODO() + <-allDoneChan // Remove plugins. serverMapping.Iterator(func(k string, v interface{}) bool { @@ -560,7 +556,7 @@ func (s *Server) startServer(fdMap listenerFdMap) { if s.serverCount.Add(-1) < 1 { s.closeChan <- struct{}{} if serverRunning.Add(-1) < 1 { - serverMapping.Remove(s.name) + serverMapping.Remove(s.instance) allDoneChan <- struct{}{} } } diff --git a/net/ghttp/ghttp_server_config.go b/net/ghttp/ghttp_server_config.go index 7d3944718..f70161d16 100644 --- a/net/ghttp/ghttp_server_config.go +++ b/net/ghttp/ghttp_server_config.go @@ -14,11 +14,13 @@ import ( "time" "github.com/gogf/gf/v2/internal/intlog" + "github.com/gogf/gf/v2/net/gtcp" "github.com/gogf/gf/v2/os/gfile" "github.com/gogf/gf/v2/os/glog" "github.com/gogf/gf/v2/os/gres" "github.com/gogf/gf/v2/os/gsession" "github.com/gogf/gf/v2/os/gview" + "github.com/gogf/gf/v2/text/gstr" "github.com/gogf/gf/v2/util/gconv" "github.com/gogf/gf/v2/util/gutil" ) @@ -38,6 +40,9 @@ type ServerConfig struct { // Basic. // ====================================================================================================== + // Service name, which is for service registry and discovery. + Name string `json:"name"` + // Address specifies the server listening address like "port" or ":port", // multiple addresses joined using ','. Address string `json:"address"` @@ -234,6 +239,7 @@ type ServerConfig struct { // some pointer attributes that may be shared in different servers. func NewConfig() ServerConfig { return ServerConfig{ + Name: DefaultServerName, Address: "", HTTPSAddr: "", Handler: nil, @@ -311,6 +317,13 @@ func (s *Server) SetConfigWithMap(m map[string]interface{}) error { // SetConfig sets the configuration for the server. func (s *Server) SetConfig(c ServerConfig) error { s.config = c + // Address, check and use a random free port. + array := gstr.Split(s.config.Address, ":") + if s.config.Address == "" || len(array) < 2 || array[1] == "0" { + s.config.Address = gstr.Join([]string{ + array[0], gconv.String(gtcp.MustGetFreePort()), + }, ":") + } // Static. if c.ServerRoot != "" { s.SetServerRoot(c.ServerRoot) @@ -382,9 +395,7 @@ func (s *Server) SetHTTPSPort(port ...int) { // EnableHTTPS enables HTTPS with given certification and key files for the server. // The optional parameter `tlsConfig` specifies custom TLS configuration. func (s *Server) EnableHTTPS(certFile, keyFile string, tlsConfig ...*tls.Config) { - var ( - ctx = context.TODO() - ) + var ctx = context.TODO() certFileRealPath := gfile.RealPath(certFile) if certFileRealPath == "" { certFileRealPath = gfile.RealPath(gfile.Pwd() + gfile.Separator + certFile) @@ -462,7 +473,12 @@ func (s *Server) SetView(view *gview.View) { // GetName returns the name of the server. func (s *Server) GetName() string { - return s.name + return s.config.Name +} + +// SetName sets the name for the server. +func (s *Server) SetName(name string) { + s.config.Name = name } // Handler returns the request handler of the server. diff --git a/net/ghttp/ghttp_server_registry.go b/net/ghttp/ghttp_server_registry.go new file mode 100644 index 000000000..ad1f8f59a --- /dev/null +++ b/net/ghttp/ghttp_server_registry.go @@ -0,0 +1,56 @@ +// Copyright GoFrame Author(https://goframe.org). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. + +package ghttp + +import ( + "context" + "fmt" + + "github.com/gogf/gf/v2/net/gipv4" + "github.com/gogf/gf/v2/net/gsvc" + "github.com/gogf/gf/v2/text/gstr" +) + +// doServiceRegister registers current service to Registry. +func (s *Server) doServiceRegister() { + var ( + ctx = context.Background() + protocol = `http` + insecure = true + address = s.config.Address + ) + if address == "" { + address = s.config.HTTPSAddr + } + var ( + array = gstr.Split(address, ":") + ip = array[0] + port = array[1] + ) + if ip == "" { + ip = gipv4.MustGetIntranetIp() + } + if s.config.TLSConfig != nil { + protocol = `https` + insecure = false + } + metadata := gsvc.Metadata{ + gsvc.MDProtocol: protocol, + gsvc.MDInsecure: insecure, + } + s.service = &gsvc.Service{ + Name: s.GetName(), + Endpoints: []string{fmt.Sprintf(`%s:%s`, ip, port)}, + Metadata: metadata, + } + _ = gsvc.Register(ctx, s.service) +} + +// doServiceDeregister de-registers current service from Registry. +func (s *Server) doServiceDeregister() { + _ = gsvc.Deregister(context.Background(), s.service) +} diff --git a/net/gsel/gsel.go b/net/gsel/gsel.go index 8996b6f8d..096e03baa 100644 --- a/net/gsel/gsel.go +++ b/net/gsel/gsel.go @@ -25,6 +25,7 @@ type Selector interface { // Node is node interface. type Node interface { Service() *gsvc.Service + Address() string } // DoneFunc is callback function when RPC invoke done. diff --git a/net/gsvc/gsvc.go b/net/gsvc/gsvc.go index 3f3e8e2a9..e626c4d49 100644 --- a/net/gsvc/gsvc.go +++ b/net/gsvc/gsvc.go @@ -105,3 +105,9 @@ func SetRegistry(registry Registry) { } defaultRegistry = registry } + +// GetRegistry returns the default Registry that is previously set. +// It returns nil if no Registry is set. +func GetRegistry() Registry { + return defaultRegistry +} diff --git a/net/gsvc/gsvc_service.go b/net/gsvc/gsvc_service.go index 1a8eef6cc..ed8d2599f 100644 --- a/net/gsvc/gsvc_service.go +++ b/net/gsvc/gsvc_service.go @@ -88,6 +88,15 @@ func (s *Service) Value() string { return string(b) } +// Address returns the first endpoint of Service. +// Eg: 192.168.1.12:9000. +func (s *Service) Address() string { + if len(s.Endpoints) == 0 { + return "" + } + return s.Endpoints[0] +} + func (s *Service) autoFillDefaultAttributes() { if s.Prefix == "" { s.Prefix = gcmd.GetOptWithEnv(EnvPrefix, DefaultPrefix).String()