guerrilla.go 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574
  1. package guerrilla
  2. import (
  3. "errors"
  4. "fmt"
  5. "os"
  6. "sync"
  7. "sync/atomic"
  8. "github.com/flashmob/go-guerrilla/backends"
  9. "github.com/flashmob/go-guerrilla/log"
  10. )
  11. const (
  12. // all configured servers were just been created
  13. daemonStateNew = iota
  14. // ... been started and running
  15. daemonStateStarted
  16. // ... been stopped
  17. daemonStateStopped
  18. )
  19. type Errors []error
  20. // implement the Error interface
  21. func (e Errors) Error() string {
  22. if len(e) == 1 {
  23. return e[0].Error()
  24. }
  25. // multiple errors
  26. msg := ""
  27. for _, err := range e {
  28. msg += "\n" + err.Error()
  29. }
  30. return msg
  31. }
  32. type Guerrilla interface {
  33. Start() error
  34. Shutdown()
  35. Subscribe(topic Event, fn interface{}) error
  36. Publish(topic Event, args ...interface{})
  37. Unsubscribe(topic Event, handler interface{}) error
  38. SetLogger(log.Logger)
  39. }
  40. type guerrilla struct {
  41. Config AppConfig
  42. servers map[string]*server
  43. // guard controls access to g.servers
  44. guard sync.Mutex
  45. state int8
  46. EventHandler
  47. logStore
  48. backendStore
  49. }
  50. type logStore struct {
  51. atomic.Value
  52. }
  53. type backendStore struct {
  54. atomic.Value
  55. }
  56. type daemonEvent func(c *AppConfig)
  57. type serverEvent func(sc *ServerConfig)
  58. // Get loads the log.logger in an atomic operation. Returns a stderr logger if not able to load
  59. func (ls *logStore) mainlog() log.Logger {
  60. if v, ok := ls.Load().(log.Logger); ok {
  61. return v
  62. }
  63. l, _ := log.GetLogger(log.OutputStderr.String(), log.InfoLevel.String())
  64. return l
  65. }
  66. // setMainlog stores the log value in an atomic operation
  67. func (ls *logStore) setMainlog(log log.Logger) {
  68. ls.Store(log)
  69. }
  70. // Returns a new instance of Guerrilla with the given config, not yet running. Backend started.
  71. func New(ac *AppConfig, b backends.Backend, l log.Logger) (Guerrilla, error) {
  72. g := &guerrilla{
  73. Config: *ac, // take a local copy
  74. servers: make(map[string]*server, len(ac.Servers)),
  75. }
  76. g.backendStore.Store(b)
  77. g.setMainlog(l)
  78. if ac.LogLevel != "" {
  79. if h, ok := l.(*log.HookedLogger); ok {
  80. if h, err := log.GetLogger(h.GetLogDest(), ac.LogLevel); err == nil {
  81. g.setMainlog(h)
  82. }
  83. }
  84. }
  85. // Write the process id (pid) to a file
  86. // we should still be able to continue even if we can't write the pid, error will be logged by writePid()
  87. _ = g.writePid()
  88. g.state = daemonStateNew
  89. err := g.makeServers()
  90. if err != nil {
  91. return g, err
  92. }
  93. // start backend for processing email
  94. err = g.backend().Start()
  95. if err != nil {
  96. return g, err
  97. }
  98. // subscribe for any events that may come in while running
  99. g.subscribeEvents()
  100. return g, err
  101. }
  102. // Instantiate servers
  103. func (g *guerrilla) makeServers() error {
  104. g.mainlog().Debug("making servers")
  105. var errs Errors
  106. for _, sc := range g.Config.Servers {
  107. if _, ok := g.servers[sc.ListenInterface]; ok {
  108. // server already instantiated
  109. continue
  110. }
  111. if err := sc.Validate(); err != nil {
  112. g.mainlog().WithError(errs).Errorf("Failed to create server [%s]", sc.ListenInterface)
  113. errs = append(errs, err)
  114. continue
  115. } else {
  116. sc := sc // pin!
  117. server, err := newServer(&sc, g.backend(), g.mainlog())
  118. if err != nil {
  119. g.mainlog().WithError(err).Errorf("Failed to create server [%s]", sc.ListenInterface)
  120. errs = append(errs, err)
  121. }
  122. if server != nil {
  123. g.servers[sc.ListenInterface] = server
  124. server.setAllowedHosts(g.Config.AllowedHosts)
  125. }
  126. }
  127. }
  128. if len(g.servers) == 0 {
  129. errs = append(errs, errors.New("there are no servers that can start, please check your config"))
  130. }
  131. if len(errs) == 0 {
  132. return nil
  133. }
  134. return errs
  135. }
  136. // findServer finds a server by iface (interface), retuning the server or err
  137. func (g *guerrilla) findServer(iface string) (*server, error) {
  138. g.guard.Lock()
  139. defer g.guard.Unlock()
  140. if server, ok := g.servers[iface]; ok {
  141. return server, nil
  142. }
  143. return nil, errors.New("server not found in g.servers")
  144. }
  145. // removeServer removes a server from the list of servers
  146. func (g *guerrilla) removeServer(iface string) {
  147. g.guard.Lock()
  148. defer g.guard.Unlock()
  149. delete(g.servers, iface)
  150. }
  151. // setConfig sets the app config
  152. func (g *guerrilla) setConfig(c *AppConfig) {
  153. g.guard.Lock()
  154. defer g.guard.Unlock()
  155. g.Config = *c
  156. }
  157. // setServerConfig config updates the server's config, which will update for the next connected client
  158. func (g *guerrilla) setServerConfig(sc *ServerConfig) {
  159. g.guard.Lock()
  160. defer g.guard.Unlock()
  161. if _, ok := g.servers[sc.ListenInterface]; ok {
  162. g.servers[sc.ListenInterface].setConfig(sc)
  163. }
  164. }
  165. // mapServers calls a callback on each server in g.servers map
  166. // It locks the g.servers map before mapping
  167. func (g *guerrilla) mapServers(callback func(*server)) map[string]*server {
  168. defer g.guard.Unlock()
  169. g.guard.Lock()
  170. for _, server := range g.servers {
  171. callback(server)
  172. }
  173. return g.servers
  174. }
  175. // subscribeEvents subscribes event handlers for configuration change events
  176. func (g *guerrilla) subscribeEvents() {
  177. events := map[Event]interface{}{}
  178. // main config changed
  179. events[EventConfigNewConfig] = daemonEvent(func(c *AppConfig) {
  180. g.setConfig(c)
  181. })
  182. // allowed_hosts changed, set for all servers
  183. events[EventConfigAllowedHosts] = daemonEvent(func(c *AppConfig) {
  184. g.mapServers(func(server *server) {
  185. server.setAllowedHosts(c.AllowedHosts)
  186. })
  187. g.mainlog().Infof("allowed_hosts config changed, a new list was set")
  188. })
  189. // the main log file changed
  190. events[EventConfigLogFile] = daemonEvent(func(c *AppConfig) {
  191. var err error
  192. var l log.Logger
  193. if l, err = log.GetLogger(c.LogFile, c.LogLevel); err == nil {
  194. g.setMainlog(l)
  195. g.mapServers(func(server *server) {
  196. // it will change server's logger when the next client gets accepted
  197. server.mainlogStore.Store(l)
  198. })
  199. g.mainlog().Infof("main log for new clients changed to [%s]", c.LogFile)
  200. } else {
  201. g.mainlog().WithError(err).Errorf("main logging change failed [%s]", c.LogFile)
  202. }
  203. })
  204. // re-open the main log file (file not changed)
  205. events[EventConfigLogReopen] = daemonEvent(func(c *AppConfig) {
  206. err := g.mainlog().Reopen()
  207. if err != nil {
  208. g.mainlog().WithError(err).Errorf("main log file [%s] failed to re-open", c.LogFile)
  209. return
  210. }
  211. g.mainlog().Infof("re-opened main log file [%s]", c.LogFile)
  212. })
  213. // when log level changes, apply to mainlog and server logs
  214. events[EventConfigLogLevel] = daemonEvent(func(c *AppConfig) {
  215. l, err := log.GetLogger(g.mainlog().GetLogDest(), c.LogLevel)
  216. if err == nil {
  217. g.logStore.Store(l)
  218. g.mapServers(func(server *server) {
  219. server.logStore.Store(l)
  220. })
  221. g.mainlog().Infof("log level changed to [%s]", c.LogLevel)
  222. }
  223. })
  224. // write out our pid whenever the file name changes in the config
  225. events[EventConfigPidFile] = daemonEvent(func(ac *AppConfig) {
  226. _ = g.writePid()
  227. })
  228. // server config was updated
  229. events[EventConfigServerConfig] = serverEvent(func(sc *ServerConfig) {
  230. g.setServerConfig(sc)
  231. g.mainlog().Infof("server %s config change event, a new config has been saved", sc.ListenInterface)
  232. })
  233. // add a new server to the config & start
  234. events[EventConfigServerNew] = serverEvent(func(sc *ServerConfig) {
  235. g.mainlog().Debugf("event fired [%s] %s", EventConfigServerNew, sc.ListenInterface)
  236. if _, err := g.findServer(sc.ListenInterface); err != nil {
  237. // not found, lets add it
  238. //
  239. if err := g.makeServers(); err != nil {
  240. g.mainlog().WithError(err).Errorf("cannot add server [%s]", sc.ListenInterface)
  241. return
  242. }
  243. g.mainlog().Infof("New server added [%s]", sc.ListenInterface)
  244. if g.state == daemonStateStarted {
  245. err := g.Start()
  246. if err != nil {
  247. g.mainlog().WithError(err).Info("Event server_change:new_server returned errors when starting")
  248. }
  249. }
  250. } else {
  251. g.mainlog().Debugf("new event, but server already fund")
  252. }
  253. })
  254. // start a server that already exists in the config and has been enabled
  255. events[EventConfigServerStart] = serverEvent(func(sc *ServerConfig) {
  256. if server, err := g.findServer(sc.ListenInterface); err == nil {
  257. if server.state == ServerStateStopped || server.state == ServerStateNew {
  258. g.mainlog().Infof("Starting server [%s]", server.listenInterface)
  259. err := g.Start()
  260. if err != nil {
  261. g.mainlog().WithError(err).Info("Event server_change:start_server returned errors when starting")
  262. }
  263. }
  264. }
  265. })
  266. // stop running a server
  267. events[EventConfigServerStop] = serverEvent(func(sc *ServerConfig) {
  268. if server, err := g.findServer(sc.ListenInterface); err == nil {
  269. if server.state == ServerStateRunning {
  270. server.Shutdown()
  271. g.mainlog().Infof("Server [%s] stopped.", sc.ListenInterface)
  272. }
  273. }
  274. })
  275. // server was removed from config
  276. events[EventConfigServerRemove] = serverEvent(func(sc *ServerConfig) {
  277. if server, err := g.findServer(sc.ListenInterface); err == nil {
  278. server.Shutdown()
  279. g.removeServer(sc.ListenInterface)
  280. g.mainlog().Infof("Server [%s] removed from config, stopped it.", sc.ListenInterface)
  281. }
  282. })
  283. // TLS changes
  284. events[EventConfigServerTLSConfig] = serverEvent(func(sc *ServerConfig) {
  285. if server, err := g.findServer(sc.ListenInterface); err == nil {
  286. if err := server.configureTLS(); err == nil {
  287. g.mainlog().Infof("Server [%s] new TLS configuration loaded", sc.ListenInterface)
  288. } else {
  289. g.mainlog().WithError(err).Errorf("Server [%s] failed to load the new TLS configuration", sc.ListenInterface)
  290. }
  291. }
  292. })
  293. // when server's timeout change.
  294. events[EventConfigServerTimeout] = serverEvent(func(sc *ServerConfig) {
  295. g.mapServers(func(server *server) {
  296. server.setTimeout(sc.Timeout)
  297. })
  298. })
  299. // when server's max clients change.
  300. events[EventConfigServerMaxClients] = serverEvent(func(sc *ServerConfig) {
  301. g.mapServers(func(server *server) {
  302. // TODO resize the pool somehow
  303. })
  304. })
  305. // when a server's log file changes
  306. events[EventConfigServerLogFile] = serverEvent(func(sc *ServerConfig) {
  307. if server, err := g.findServer(sc.ListenInterface); err == nil {
  308. var err error
  309. var l log.Logger
  310. level := g.mainlog().GetLevel()
  311. if l, err = log.GetLogger(sc.LogFile, level); err == nil {
  312. g.setMainlog(l)
  313. backends.Svc.SetMainlog(l)
  314. // it will change to the new logger on the next accepted client
  315. server.logStore.Store(l)
  316. g.mainlog().Infof("Server [%s] changed, new clients will log to: [%s]",
  317. sc.ListenInterface,
  318. sc.LogFile,
  319. )
  320. } else {
  321. g.mainlog().WithError(err).Errorf(
  322. "Server [%s] log change failed to: [%s]",
  323. sc.ListenInterface,
  324. sc.LogFile,
  325. )
  326. }
  327. }
  328. })
  329. // when the daemon caught a sighup, event for individual server
  330. events[EventConfigServerLogReopen] = serverEvent(func(sc *ServerConfig) {
  331. if server, err := g.findServer(sc.ListenInterface); err == nil {
  332. if err = server.log().Reopen(); err != nil {
  333. g.mainlog().WithError(err).Errorf("server [%s] log file [%s] failed to re-open", sc.ListenInterface, sc.LogFile)
  334. return
  335. }
  336. g.mainlog().Infof("Server [%s] re-opened log file [%s]", sc.ListenInterface, sc.LogFile)
  337. }
  338. })
  339. // when the backend changes
  340. events[EventConfigBackendConfig] = daemonEvent(func(appConfig *AppConfig) {
  341. logger, _ := log.GetLogger(appConfig.LogFile, appConfig.LogLevel)
  342. // shutdown the backend first.
  343. var err error
  344. if err = g.backend().Shutdown(); err != nil {
  345. logger.WithError(err).Warn("Backend failed to shutdown")
  346. return
  347. }
  348. // init a new backend, Revert to old backend config if it fails
  349. if newBackend, newErr := backends.New(appConfig.BackendConfig, logger); newErr != nil {
  350. logger.WithError(newErr).Error("Error while loading the backend")
  351. err = g.backend().Reinitialize()
  352. if err != nil {
  353. logger.WithError(err).Fatal("failed to revert to old backend config")
  354. return
  355. }
  356. err = g.backend().Start()
  357. if err != nil {
  358. logger.WithError(err).Fatal("failed to start backend with old config")
  359. return
  360. }
  361. logger.Info("reverted to old backend config")
  362. } else {
  363. // swap to the bew backend (assuming old backend was shutdown so it can be safely swapped)
  364. if err := newBackend.Start(); err != nil {
  365. logger.WithError(err).Error("backend could not start")
  366. }
  367. logger.Info("new backend started")
  368. g.storeBackend(newBackend)
  369. }
  370. })
  371. var err error
  372. for topic, fn := range events {
  373. switch f := fn.(type) {
  374. case daemonEvent:
  375. err = g.Subscribe(topic, f)
  376. case serverEvent:
  377. err = g.Subscribe(topic, f)
  378. }
  379. if err != nil {
  380. g.mainlog().WithError(err).Errorf("failed to subscribe on topic [%s]", topic)
  381. break
  382. }
  383. }
  384. }
  385. func (g *guerrilla) storeBackend(b backends.Backend) {
  386. g.backendStore.Store(b)
  387. g.mapServers(func(server *server) {
  388. server.setBackend(b)
  389. })
  390. }
  391. func (g *guerrilla) backend() backends.Backend {
  392. if b, ok := g.backendStore.Load().(backends.Backend); ok {
  393. return b
  394. }
  395. return nil
  396. }
  397. // Entry point for the application. Starts all servers.
  398. func (g *guerrilla) Start() error {
  399. var startErrors Errors
  400. g.guard.Lock()
  401. defer func() {
  402. g.state = daemonStateStarted
  403. g.guard.Unlock()
  404. }()
  405. if len(g.servers) == 0 {
  406. return append(startErrors, errors.New("no servers to start, please check the config"))
  407. }
  408. if g.state == daemonStateStopped {
  409. // when a backend is shutdown, we need to re-initialize before it can be started again
  410. if err := g.backend().Reinitialize(); err != nil {
  411. startErrors = append(startErrors, err)
  412. }
  413. if err := g.backend().Start(); err != nil {
  414. startErrors = append(startErrors, err)
  415. }
  416. }
  417. // channel for reading errors
  418. errs := make(chan error, len(g.servers))
  419. var startWG sync.WaitGroup
  420. // start servers, send any errors back to errs channel
  421. for ListenInterface := range g.servers {
  422. if !g.servers[ListenInterface].isEnabled() {
  423. // not enabled
  424. continue
  425. }
  426. if g.servers[ListenInterface].state != ServerStateNew &&
  427. g.servers[ListenInterface].state != ServerStateStopped {
  428. continue
  429. }
  430. startWG.Add(1)
  431. go func(s *server) {
  432. g.mainlog().Infof("Starting: %s", s.listenInterface)
  433. if err := s.Start(&startWG); err != nil {
  434. errs <- err
  435. }
  436. }(g.servers[ListenInterface])
  437. }
  438. // wait for all servers to start (or fail)
  439. startWG.Wait()
  440. // close, then read any errors
  441. close(errs)
  442. for err := range errs {
  443. if err != nil {
  444. startErrors = append(startErrors, err)
  445. }
  446. }
  447. if len(startErrors) > 0 {
  448. return startErrors
  449. }
  450. return nil
  451. }
  452. func (g *guerrilla) Shutdown() {
  453. // shut down the servers first
  454. g.mapServers(func(s *server) {
  455. if s.state == ServerStateRunning {
  456. s.Shutdown()
  457. g.mainlog().Infof("shutdown completed for [%s]", s.listenInterface)
  458. }
  459. })
  460. g.guard.Lock()
  461. defer func() {
  462. g.state = daemonStateStopped
  463. defer g.guard.Unlock()
  464. }()
  465. if err := g.backend().Shutdown(); err != nil {
  466. g.mainlog().WithError(err).Warn("Backend failed to shutdown")
  467. } else {
  468. g.mainlog().Infof("Backend shutdown completed")
  469. }
  470. }
  471. // SetLogger sets the logger for the app and propagates it to sub-packages (eg.
  472. func (g *guerrilla) SetLogger(l log.Logger) {
  473. g.setMainlog(l)
  474. backends.Svc.SetMainlog(l)
  475. }
  476. // writePid writes the pid (process id) to the file specified in the config.
  477. // Won't write anything if no file specified
  478. func (g *guerrilla) writePid() (err error) {
  479. var f *os.File
  480. defer func() {
  481. if f != nil {
  482. if closeErr := f.Close(); closeErr != nil {
  483. err = closeErr
  484. }
  485. }
  486. if err != nil {
  487. g.mainlog().WithError(err).Errorf("error while writing pidFile (%s)", g.Config.PidFile)
  488. }
  489. }()
  490. if len(g.Config.PidFile) > 0 {
  491. if f, err = os.Create(g.Config.PidFile); err != nil {
  492. return err
  493. }
  494. pid := os.Getpid()
  495. if _, err := f.WriteString(fmt.Sprintf("%d", pid)); err != nil {
  496. return err
  497. }
  498. if err = f.Sync(); err != nil {
  499. return err
  500. }
  501. g.mainlog().Infof("pid_file (%s) written with pid:%v", g.Config.PidFile, pid)
  502. }
  503. return nil
  504. }
  505. // CheckFileLimit checks the number of files we can open (works on OS'es that support the ulimit command)
  506. func CheckFileLimit(c *AppConfig) (bool, int, uint64) {
  507. fileLimit, err := getFileLimit()
  508. maxClients := 0
  509. if err != nil {
  510. // since we can't get the limit, return true to indicate the check passed
  511. return true, maxClients, fileLimit
  512. }
  513. if c.Servers == nil {
  514. // no servers have been configured, assuming default
  515. maxClients = defaultMaxClients
  516. } else {
  517. for _, s := range c.Servers {
  518. maxClients += s.MaxClients
  519. }
  520. }
  521. if uint64(maxClients) > fileLimit {
  522. return false, maxClients, fileLimit
  523. }
  524. return true, maxClients, fileLimit
  525. }