guerrilla.go 16 KB

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