grpc_client_server.go 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288
  1. package pb
  2. import (
  3. "context"
  4. "fmt"
  5. "google.golang.org/grpc/metadata"
  6. "math/rand"
  7. "net/http"
  8. "strconv"
  9. "strings"
  10. "sync"
  11. "time"
  12. "github.com/seaweedfs/seaweedfs/weed/glog"
  13. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  14. "github.com/seaweedfs/seaweedfs/weed/util"
  15. "google.golang.org/grpc"
  16. "google.golang.org/grpc/keepalive"
  17. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  18. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  19. "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
  20. )
  21. const (
  22. Max_Message_Size = 1 << 30 // 1 GB
  23. )
  24. var (
  25. // cache grpc connections
  26. grpcClients = make(map[string]*versionedGrpcClient)
  27. grpcClientsLock sync.Mutex
  28. )
  29. type versionedGrpcClient struct {
  30. *grpc.ClientConn
  31. version int
  32. errCount int
  33. }
  34. func init() {
  35. http.DefaultTransport.(*http.Transport).MaxIdleConnsPerHost = 1024
  36. http.DefaultTransport.(*http.Transport).MaxIdleConns = 1024
  37. }
  38. func NewGrpcServer(opts ...grpc.ServerOption) *grpc.Server {
  39. var options []grpc.ServerOption
  40. options = append(options,
  41. grpc.KeepaliveParams(keepalive.ServerParameters{
  42. Time: 10 * time.Second, // wait time before ping if no activity
  43. Timeout: 20 * time.Second, // ping timeout
  44. // MaxConnectionAge: 10 * time.Hour,
  45. }),
  46. grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
  47. MinTime: 60 * time.Second, // min time a client should wait before sending a ping
  48. PermitWithoutStream: true,
  49. }),
  50. grpc.MaxRecvMsgSize(Max_Message_Size),
  51. grpc.MaxSendMsgSize(Max_Message_Size),
  52. )
  53. for _, opt := range opts {
  54. if opt != nil {
  55. options = append(options, opt)
  56. }
  57. }
  58. return grpc.NewServer(options...)
  59. }
  60. func GrpcDial(ctx context.Context, address string, waitForReady bool, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
  61. // opts = append(opts, grpc.WithBlock())
  62. // opts = append(opts, grpc.WithTimeout(time.Duration(5*time.Second)))
  63. var options []grpc.DialOption
  64. options = append(options,
  65. // grpc.WithTransportCredentials(insecure.NewCredentials()),
  66. grpc.WithDefaultCallOptions(
  67. grpc.MaxCallSendMsgSize(Max_Message_Size),
  68. grpc.MaxCallRecvMsgSize(Max_Message_Size),
  69. grpc.WaitForReady(waitForReady),
  70. ),
  71. grpc.WithKeepaliveParams(keepalive.ClientParameters{
  72. Time: 30 * time.Second, // client ping server if no activity for this long
  73. Timeout: 20 * time.Second,
  74. PermitWithoutStream: true,
  75. }))
  76. for _, opt := range opts {
  77. if opt != nil {
  78. options = append(options, opt)
  79. }
  80. }
  81. return grpc.DialContext(ctx, address, options...)
  82. }
  83. func getOrCreateConnection(address string, waitForReady bool, opts ...grpc.DialOption) (*versionedGrpcClient, error) {
  84. grpcClientsLock.Lock()
  85. defer grpcClientsLock.Unlock()
  86. existingConnection, found := grpcClients[address]
  87. if found {
  88. return existingConnection, nil
  89. }
  90. ctx := context.Background()
  91. grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...)
  92. if err != nil {
  93. return nil, fmt.Errorf("fail to dial %s: %v", address, err)
  94. }
  95. vgc := &versionedGrpcClient{
  96. grpcConnection,
  97. rand.Int(),
  98. 0,
  99. }
  100. grpcClients[address] = vgc
  101. return vgc, nil
  102. }
  103. // WithGrpcClient In streamingMode, always use a fresh connection. Otherwise, try to reuse an existing connection.
  104. func WithGrpcClient(streamingMode bool, signature int32, fn func(*grpc.ClientConn) error, address string, waitForReady bool, opts ...grpc.DialOption) error {
  105. if !streamingMode {
  106. vgc, err := getOrCreateConnection(address, waitForReady, opts...)
  107. if err != nil {
  108. return fmt.Errorf("getOrCreateConnection %s: %v", address, err)
  109. }
  110. executionErr := fn(vgc.ClientConn)
  111. if executionErr != nil {
  112. if strings.Contains(executionErr.Error(), "transport") ||
  113. strings.Contains(executionErr.Error(), "connection closed") {
  114. grpcClientsLock.Lock()
  115. if t, ok := grpcClients[address]; ok {
  116. if t.version == vgc.version {
  117. vgc.Close()
  118. delete(grpcClients, address)
  119. }
  120. }
  121. grpcClientsLock.Unlock()
  122. }
  123. }
  124. return executionErr
  125. } else {
  126. ctx := context.Background()
  127. if signature != 0 {
  128. md := metadata.New(map[string]string{"sw-client-id": fmt.Sprintf("%d", signature)})
  129. ctx = metadata.NewOutgoingContext(ctx, md)
  130. }
  131. grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...)
  132. if err != nil {
  133. return fmt.Errorf("fail to dial %s: %v", address, err)
  134. }
  135. defer grpcConnection.Close()
  136. executionErr := fn(grpcConnection)
  137. if executionErr != nil {
  138. return executionErr
  139. }
  140. return nil
  141. }
  142. }
  143. func ParseServerAddress(server string, deltaPort int) (newServerAddress string, err error) {
  144. host, port, parseErr := hostAndPort(server)
  145. if parseErr != nil {
  146. return "", fmt.Errorf("server port parse error: %v", parseErr)
  147. }
  148. newPort := int(port) + deltaPort
  149. return util.JoinHostPort(host, newPort), nil
  150. }
  151. func hostAndPort(address string) (host string, port uint64, err error) {
  152. colonIndex := strings.LastIndex(address, ":")
  153. if colonIndex < 0 {
  154. return "", 0, fmt.Errorf("server should have hostname:port format: %v", address)
  155. }
  156. port, err = strconv.ParseUint(address[colonIndex+1:], 10, 64)
  157. if err != nil {
  158. return "", 0, fmt.Errorf("server port parse error: %v", err)
  159. }
  160. return address[:colonIndex], port, err
  161. }
  162. func ServerToGrpcAddress(server string) (serverGrpcAddress string) {
  163. host, port, parseErr := hostAndPort(server)
  164. if parseErr != nil {
  165. glog.Fatalf("server address %s parse error: %v", server, parseErr)
  166. }
  167. grpcPort := int(port) + 10000
  168. return util.JoinHostPort(host, grpcPort)
  169. }
  170. func GrpcAddressToServerAddress(grpcAddress string) (serverAddress string) {
  171. host, grpcPort, parseErr := hostAndPort(grpcAddress)
  172. if parseErr != nil {
  173. glog.Fatalf("server grpc address %s parse error: %v", grpcAddress, parseErr)
  174. }
  175. port := int(grpcPort) - 10000
  176. return util.JoinHostPort(host, port)
  177. }
  178. func WithMasterClient(streamingMode bool, master ServerAddress, grpcDialOption grpc.DialOption, waitForReady bool, fn func(client master_pb.SeaweedClient) error) error {
  179. return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
  180. client := master_pb.NewSeaweedClient(grpcConnection)
  181. return fn(client)
  182. }, master.ToGrpcAddress(), waitForReady, grpcDialOption)
  183. }
  184. func WithVolumeServerClient(streamingMode bool, volumeServer ServerAddress, grpcDialOption grpc.DialOption, fn func(client volume_server_pb.VolumeServerClient) error) error {
  185. return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
  186. client := volume_server_pb.NewVolumeServerClient(grpcConnection)
  187. return fn(client)
  188. }, volumeServer.ToGrpcAddress(), false, grpcDialOption)
  189. }
  190. func WithBrokerClient(streamingMode bool, broker ServerAddress, grpcDialOption grpc.DialOption, fn func(client mq_pb.SeaweedMessagingClient) error) error {
  191. return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
  192. client := mq_pb.NewSeaweedMessagingClient(grpcConnection)
  193. return fn(client)
  194. }, broker.ToGrpcAddress(), false, grpcDialOption)
  195. }
  196. func WithOneOfGrpcMasterClients(streamingMode bool, masterGrpcAddresses map[string]ServerAddress, grpcDialOption grpc.DialOption, fn func(client master_pb.SeaweedClient) error) (err error) {
  197. for _, masterGrpcAddress := range masterGrpcAddresses {
  198. err = WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
  199. client := master_pb.NewSeaweedClient(grpcConnection)
  200. return fn(client)
  201. }, masterGrpcAddress.ToGrpcAddress(), false, grpcDialOption)
  202. if err == nil {
  203. return nil
  204. }
  205. }
  206. return err
  207. }
  208. func WithBrokerGrpcClient(streamingMode bool, brokerGrpcAddress string, grpcDialOption grpc.DialOption, fn func(client mq_pb.SeaweedMessagingClient) error) error {
  209. return WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
  210. client := mq_pb.NewSeaweedMessagingClient(grpcConnection)
  211. return fn(client)
  212. }, brokerGrpcAddress, false, grpcDialOption)
  213. }
  214. func WithFilerClient(streamingMode bool, signature int32, filer ServerAddress, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) error {
  215. return WithGrpcFilerClient(streamingMode, signature, filer, grpcDialOption, fn)
  216. }
  217. func WithGrpcFilerClient(streamingMode bool, signature int32, filerGrpcAddress ServerAddress, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) error {
  218. return WithGrpcClient(streamingMode, signature, func(grpcConnection *grpc.ClientConn) error {
  219. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  220. return fn(client)
  221. }, filerGrpcAddress.ToGrpcAddress(), false, grpcDialOption)
  222. }
  223. func WithOneOfGrpcFilerClients(streamingMode bool, filerAddresses []ServerAddress, grpcDialOption grpc.DialOption, fn func(client filer_pb.SeaweedFilerClient) error) (err error) {
  224. for _, filerAddress := range filerAddresses {
  225. err = WithGrpcClient(streamingMode, 0, func(grpcConnection *grpc.ClientConn) error {
  226. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  227. return fn(client)
  228. }, filerAddress.ToGrpcAddress(), false, grpcDialOption)
  229. if err == nil {
  230. return nil
  231. }
  232. }
  233. return err
  234. }