filer_server_handlers.go 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234
  1. package weed_server
  2. import (
  3. "context"
  4. "errors"
  5. "net/http"
  6. "os"
  7. "strconv"
  8. "strings"
  9. "sync/atomic"
  10. "time"
  11. "github.com/seaweedfs/seaweedfs/weed/filer"
  12. "github.com/seaweedfs/seaweedfs/weed/glog"
  13. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  14. "github.com/seaweedfs/seaweedfs/weed/security"
  15. "github.com/seaweedfs/seaweedfs/weed/util"
  16. "github.com/seaweedfs/seaweedfs/weed/stats"
  17. )
  18. func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
  19. start := time.Now()
  20. statusRecorder := stats.NewStatusResponseWriter(w)
  21. w = statusRecorder
  22. origin := r.Header.Get("Origin")
  23. if origin != "" {
  24. if fs.option.AllowedOrigins == nil || len(fs.option.AllowedOrigins) == 0 || fs.option.AllowedOrigins[0] == "*" {
  25. origin = "*"
  26. } else {
  27. originFound := false
  28. for _, allowedOrigin := range fs.option.AllowedOrigins {
  29. if origin == allowedOrigin {
  30. originFound = true
  31. }
  32. }
  33. if !originFound {
  34. writeJsonError(w, r, http.StatusForbidden, errors.New("origin not allowed"))
  35. return
  36. }
  37. }
  38. w.Header().Set("Access-Control-Allow-Origin", origin)
  39. w.Header().Set("Access-Control-Expose-Headers", "*")
  40. w.Header().Set("Access-Control-Allow-Headers", "*")
  41. w.Header().Set("Access-Control-Allow-Credentials", "true")
  42. w.Header().Set("Access-Control-Allow-Methods", "PUT, POST, GET, DELETE, OPTIONS")
  43. }
  44. if r.Method == http.MethodOptions {
  45. OptionsHandler(w, r, false)
  46. return
  47. }
  48. // proxy to volume servers
  49. var fileId string
  50. if strings.HasPrefix(r.RequestURI, "/?proxyChunkId=") {
  51. fileId = r.RequestURI[len("/?proxyChunkId="):]
  52. }
  53. if fileId != "" {
  54. fs.proxyToVolumeServer(w, r, fileId)
  55. stats.FilerHandlerCounter.WithLabelValues(stats.ChunkProxy).Inc()
  56. stats.FilerRequestHistogram.WithLabelValues(stats.ChunkProxy).Observe(time.Since(start).Seconds())
  57. return
  58. }
  59. requestMethod := r.Method
  60. defer func(method *string) {
  61. stats.FilerRequestCounter.WithLabelValues(*method, strconv.Itoa(statusRecorder.Status)).Inc()
  62. stats.FilerRequestHistogram.WithLabelValues(*method).Observe(time.Since(start).Seconds())
  63. }(&requestMethod)
  64. isReadHttpCall := r.Method == http.MethodGet || r.Method == http.MethodHead
  65. if !fs.maybeCheckJwtAuthorization(r, !isReadHttpCall) {
  66. writeJsonError(w, r, http.StatusUnauthorized, errors.New("wrong jwt"))
  67. return
  68. }
  69. w.Header().Set("Server", "SeaweedFS "+util.VERSION)
  70. switch r.Method {
  71. case http.MethodGet, http.MethodHead:
  72. fs.GetOrHeadHandler(w, r)
  73. case http.MethodDelete:
  74. if _, ok := r.URL.Query()["tagging"]; ok {
  75. fs.DeleteTaggingHandler(w, r)
  76. } else {
  77. fs.DeleteHandler(w, r)
  78. }
  79. case http.MethodPost, http.MethodPut:
  80. // wait until in flight data is less than the limit
  81. contentLength := getContentLength(r)
  82. fs.inFlightDataLimitCond.L.Lock()
  83. inFlightDataSize := atomic.LoadInt64(&fs.inFlightDataSize)
  84. for fs.option.ConcurrentUploadLimit != 0 && inFlightDataSize > fs.option.ConcurrentUploadLimit {
  85. glog.V(4).Infof("wait because inflight data %d > %d", inFlightDataSize, fs.option.ConcurrentUploadLimit)
  86. fs.inFlightDataLimitCond.Wait()
  87. inFlightDataSize = atomic.LoadInt64(&fs.inFlightDataSize)
  88. }
  89. fs.inFlightDataLimitCond.L.Unlock()
  90. atomic.AddInt64(&fs.inFlightDataSize, contentLength)
  91. defer func() {
  92. atomic.AddInt64(&fs.inFlightDataSize, -contentLength)
  93. fs.inFlightDataLimitCond.Signal()
  94. }()
  95. if r.Method == http.MethodPut {
  96. if _, ok := r.URL.Query()["tagging"]; ok {
  97. fs.PutTaggingHandler(w, r)
  98. } else {
  99. fs.PostHandler(w, r, contentLength)
  100. }
  101. } else { // method == "POST"
  102. fs.PostHandler(w, r, contentLength)
  103. }
  104. default:
  105. requestMethod = "INVALID"
  106. w.WriteHeader(http.StatusMethodNotAllowed)
  107. }
  108. }
  109. func (fs *FilerServer) readonlyFilerHandler(w http.ResponseWriter, r *http.Request) {
  110. start := time.Now()
  111. statusRecorder := stats.NewStatusResponseWriter(w)
  112. w = statusRecorder
  113. os.Stdout.WriteString("Request: " + r.Method + " " + r.URL.String() + "\n")
  114. origin := r.Header.Get("Origin")
  115. if origin != "" {
  116. if fs.option.AllowedOrigins == nil || len(fs.option.AllowedOrigins) == 0 || fs.option.AllowedOrigins[0] == "*" {
  117. origin = "*"
  118. } else {
  119. originFound := false
  120. for _, allowedOrigin := range fs.option.AllowedOrigins {
  121. if origin == allowedOrigin {
  122. originFound = true
  123. }
  124. }
  125. if !originFound {
  126. writeJsonError(w, r, http.StatusForbidden, errors.New("origin not allowed"))
  127. return
  128. }
  129. }
  130. w.Header().Set("Access-Control-Allow-Origin", origin)
  131. w.Header().Set("Access-Control-Allow-Headers", "OPTIONS, GET, HEAD")
  132. w.Header().Set("Access-Control-Allow-Credentials", "true")
  133. }
  134. requestMethod := r.Method
  135. defer func(method *string) {
  136. stats.FilerRequestCounter.WithLabelValues(*method, strconv.Itoa(statusRecorder.Status)).Inc()
  137. stats.FilerRequestHistogram.WithLabelValues(*method).Observe(time.Since(start).Seconds())
  138. }(&requestMethod)
  139. // We handle OPTIONS first because it never should be authenticated
  140. if r.Method == http.MethodOptions {
  141. OptionsHandler(w, r, true)
  142. return
  143. }
  144. if !fs.maybeCheckJwtAuthorization(r, false) {
  145. writeJsonError(w, r, http.StatusUnauthorized, errors.New("wrong jwt"))
  146. return
  147. }
  148. w.Header().Set("Server", "SeaweedFS "+util.VERSION)
  149. switch r.Method {
  150. case http.MethodGet, http.MethodHead:
  151. fs.GetOrHeadHandler(w, r)
  152. default:
  153. requestMethod = "INVALID"
  154. w.WriteHeader(http.StatusMethodNotAllowed)
  155. }
  156. }
  157. func OptionsHandler(w http.ResponseWriter, r *http.Request, isReadOnly bool) {
  158. if isReadOnly {
  159. w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
  160. } else {
  161. w.Header().Set("Access-Control-Allow-Methods", "PUT, POST, GET, DELETE, OPTIONS")
  162. w.Header().Set("Access-Control-Expose-Headers", "*")
  163. }
  164. w.Header().Set("Access-Control-Allow-Headers", "*")
  165. w.Header().Set("Access-Control-Allow-Credentials", "true")
  166. }
  167. // maybeCheckJwtAuthorization returns true if access should be granted, false if it should be denied
  168. func (fs *FilerServer) maybeCheckJwtAuthorization(r *http.Request, isWrite bool) bool {
  169. var signingKey security.SigningKey
  170. if isWrite {
  171. if len(fs.filerGuard.SigningKey) == 0 {
  172. return true
  173. } else {
  174. signingKey = fs.filerGuard.SigningKey
  175. }
  176. } else {
  177. if len(fs.filerGuard.ReadSigningKey) == 0 {
  178. return true
  179. } else {
  180. signingKey = fs.filerGuard.ReadSigningKey
  181. }
  182. }
  183. tokenStr := security.GetJwt(r)
  184. if tokenStr == "" {
  185. glog.V(1).Infof("missing jwt from %s", r.RemoteAddr)
  186. return false
  187. }
  188. token, err := security.DecodeJwt(signingKey, tokenStr, &security.SeaweedFilerClaims{})
  189. if err != nil {
  190. glog.V(1).Infof("jwt verification error from %s: %v", r.RemoteAddr, err)
  191. return false
  192. }
  193. if !token.Valid {
  194. glog.V(1).Infof("jwt invalid from %s: %v", r.RemoteAddr, tokenStr)
  195. return false
  196. } else {
  197. return true
  198. }
  199. }
  200. func (fs *FilerServer) filerHealthzHandler(w http.ResponseWriter, r *http.Request) {
  201. w.Header().Set("Server", "SeaweedFS "+util.VERSION)
  202. if _, err := fs.filer.Store.FindEntry(context.Background(), filer.TopicsDir); err != nil && err != filer_pb.ErrNotFound {
  203. glog.Warningf("filerHealthzHandler FindEntry: %+v", err)
  204. w.WriteHeader(http.StatusServiceUnavailable)
  205. } else {
  206. w.WriteHeader(http.StatusOK)
  207. }
  208. }