http urls monitor.

server.go 7.4KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293
  1. /*
  2. Package manners provides a wrapper for a standard net/http server that
  3. ensures all active HTTP client have completed their current request
  4. before the server shuts down.
  5. It can be used a drop-in replacement for the standard http package,
  6. or can wrap a pre-configured Server.
  7. eg.
  8. http.Handle("/hello", func(w http.ResponseWriter, r *http.Request) {
  9. w.Write([]byte("Hello\n"))
  10. })
  11. log.Fatal(manners.ListenAndServe(":8080", nil))
  12. or for a customized server:
  13. s := manners.NewWithServer(&http.Server{
  14. Addr: ":8080",
  15. Handler: myHandler,
  16. ReadTimeout: 10 * time.Second,
  17. WriteTimeout: 10 * time.Second,
  18. MaxHeaderBytes: 1 << 20,
  19. })
  20. log.Fatal(s.ListenAndServe())
  21. The server will shut down cleanly when the Close() method is called:
  22. go func() {
  23. sigchan := make(chan os.Signal, 1)
  24. signal.Notify(sigchan, os.Interrupt, os.Kill)
  25. <-sigchan
  26. log.Info("Shutting down...")
  27. manners.Close()
  28. }()
  29. http.Handle("/hello", myHandler)
  30. log.Fatal(manners.ListenAndServe(":8080", nil))
  31. */
  32. package manners
  33. import (
  34. "crypto/tls"
  35. "net"
  36. "net/http"
  37. "sync"
  38. "sync/atomic"
  39. )
  40. // A GracefulServer maintains a WaitGroup that counts how many in-flight
  41. // requests the server is handling. When it receives a shutdown signal,
  42. // it stops accepting new requests but does not actually shut down until
  43. // all in-flight requests terminate.
  44. //
  45. // GracefulServer embeds the underlying net/http.Server making its non-override
  46. // methods and properties avaiable.
  47. //
  48. // It must be initialized by calling NewWithServer.
  49. type GracefulServer struct {
  50. *http.Server
  51. shutdown chan bool
  52. shutdownFinished chan bool
  53. wg waitGroup
  54. routinesCount int
  55. lcsmu sync.RWMutex
  56. connections map[net.Conn]bool
  57. up chan net.Listener // Only used by test code.
  58. }
  59. // NewServer creates a new GracefulServer.
  60. func NewServer() *GracefulServer {
  61. return NewWithServer(new(http.Server))
  62. }
  63. // NewWithServer wraps an existing http.Server object and returns a
  64. // GracefulServer that supports all of the original Server operations.
  65. func NewWithServer(s *http.Server) *GracefulServer {
  66. return &GracefulServer{
  67. Server: s,
  68. shutdown: make(chan bool),
  69. shutdownFinished: make(chan bool, 1),
  70. wg: new(sync.WaitGroup),
  71. routinesCount: 0,
  72. connections: make(map[net.Conn]bool),
  73. }
  74. }
  75. // Close stops the server from accepting new requets and begins shutting down.
  76. // It returns true if it's the first time Close is called.
  77. func (s *GracefulServer) Close() bool {
  78. return <-s.shutdown
  79. }
  80. // BlockingClose is similar to Close, except that it blocks until the last
  81. // connection has been closed.
  82. func (s *GracefulServer) BlockingClose() bool {
  83. result := s.Close()
  84. <-s.shutdownFinished
  85. return result
  86. }
  87. // ListenAndServe provides a graceful equivalent of net/http.Serve.ListenAndServe.
  88. func (s *GracefulServer) ListenAndServe() error {
  89. addr := s.Addr
  90. if addr == "" {
  91. addr = ":http"
  92. }
  93. listener, err := net.Listen("tcp", addr)
  94. if err != nil {
  95. return err
  96. }
  97. return s.Serve(listener)
  98. }
  99. // ListenAndServeTLS provides a graceful equivalent of net/http.Serve.ListenAndServeTLS.
  100. func (s *GracefulServer) ListenAndServeTLS(certFile, keyFile string) error {
  101. // direct lift from net/http/server.go
  102. addr := s.Addr
  103. if addr == "" {
  104. addr = ":https"
  105. }
  106. config := &tls.Config{}
  107. if s.TLSConfig != nil {
  108. *config = *s.TLSConfig
  109. }
  110. if config.NextProtos == nil {
  111. config.NextProtos = []string{"http/1.1"}
  112. }
  113. var err error
  114. config.Certificates = make([]tls.Certificate, 1)
  115. config.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile)
  116. if err != nil {
  117. return err
  118. }
  119. ln, err := net.Listen("tcp", addr)
  120. if err != nil {
  121. return err
  122. }
  123. return s.Serve(tls.NewListener(ln, config))
  124. }
  125. // Serve provides a graceful equivalent net/http.Server.Serve.
  126. func (s *GracefulServer) Serve(listener net.Listener) error {
  127. // Wrap the server HTTP handler into graceful one, that will close kept
  128. // alive connections if a new request is received after shutdown.
  129. gracefulHandler := newGracefulHandler(s.Server.Handler)
  130. s.Server.Handler = gracefulHandler
  131. // Start a goroutine that waits for a shutdown signal and will stop the
  132. // listener when it receives the signal. That in turn will result in
  133. // unblocking of the http.Serve call.
  134. go func() {
  135. s.shutdown <- true
  136. close(s.shutdown)
  137. gracefulHandler.Close()
  138. s.Server.SetKeepAlivesEnabled(false)
  139. listener.Close()
  140. }()
  141. originalConnState := s.Server.ConnState
  142. // s.ConnState is invoked by the net/http.Server every time a connection
  143. // changes state. It keeps track of each connection's state over time,
  144. // enabling manners to handle persisted connections correctly.
  145. s.ConnState = func(conn net.Conn, newState http.ConnState) {
  146. s.lcsmu.RLock()
  147. protected := s.connections[conn]
  148. s.lcsmu.RUnlock()
  149. switch newState {
  150. case http.StateNew:
  151. // New connection -> StateNew
  152. protected = true
  153. s.StartRoutine()
  154. case http.StateActive:
  155. // (StateNew, StateIdle) -> StateActive
  156. if gracefulHandler.IsClosed() {
  157. conn.Close()
  158. break
  159. }
  160. if !protected {
  161. protected = true
  162. s.StartRoutine()
  163. }
  164. default:
  165. // (StateNew, StateActive) -> (StateIdle, StateClosed, StateHiJacked)
  166. if protected {
  167. s.FinishRoutine()
  168. protected = false
  169. }
  170. }
  171. s.lcsmu.Lock()
  172. if newState == http.StateClosed || newState == http.StateHijacked {
  173. delete(s.connections, conn)
  174. } else {
  175. s.connections[conn] = protected
  176. }
  177. s.lcsmu.Unlock()
  178. if originalConnState != nil {
  179. originalConnState(conn, newState)
  180. }
  181. }
  182. // A hook to allow the server to notify others when it is ready to receive
  183. // requests; only used by tests.
  184. if s.up != nil {
  185. s.up <- listener
  186. }
  187. err := s.Server.Serve(listener)
  188. // An error returned on shutdown is not worth reporting.
  189. if err != nil && gracefulHandler.IsClosed() {
  190. err = nil
  191. }
  192. // Wait for pending requests to complete regardless the Serve result.
  193. s.wg.Wait()
  194. s.shutdownFinished <- true
  195. return err
  196. }
  197. // StartRoutine increments the server's WaitGroup. Use this if a web request
  198. // starts more goroutines and these goroutines are not guaranteed to finish
  199. // before the request.
  200. func (s *GracefulServer) StartRoutine() {
  201. s.lcsmu.Lock()
  202. defer s.lcsmu.Unlock()
  203. s.wg.Add(1)
  204. s.routinesCount++
  205. }
  206. // FinishRoutine decrements the server's WaitGroup. Use this to complement
  207. // StartRoutine().
  208. func (s *GracefulServer) FinishRoutine() {
  209. s.lcsmu.Lock()
  210. defer s.lcsmu.Unlock()
  211. s.wg.Done()
  212. s.routinesCount--
  213. }
  214. // RoutinesCount returns the number of currently running routines
  215. func (s *GracefulServer) RoutinesCount() int {
  216. s.lcsmu.RLock()
  217. defer s.lcsmu.RUnlock()
  218. return s.routinesCount
  219. }
  220. // gracefulHandler is used by GracefulServer to prevent calling ServeHTTP on
  221. // to be closed kept-alive connections during the server shutdown.
  222. type gracefulHandler struct {
  223. closed int32 // accessed atomically.
  224. wrapped http.Handler
  225. }
  226. func newGracefulHandler(wrapped http.Handler) *gracefulHandler {
  227. return &gracefulHandler{
  228. wrapped: wrapped,
  229. }
  230. }
  231. func (gh *gracefulHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  232. if atomic.LoadInt32(&gh.closed) == 0 {
  233. gh.wrapped.ServeHTTP(w, r)
  234. return
  235. }
  236. r.Body.Close()
  237. // Server is shutting down at this moment, and the connection that this
  238. // handler is being called on is about to be closed. So we do not need to
  239. // actually execute the handler logic.
  240. }
  241. func (gh *gracefulHandler) Close() {
  242. atomic.StoreInt32(&gh.closed, 1)
  243. }
  244. func (gh *gracefulHandler) IsClosed() bool {
  245. return atomic.LoadInt32(&gh.closed) == 1
  246. }