diff --git a/app/main.go b/app/main.go index 8b89800d..64e4eb79 100644 --- a/app/main.go +++ b/app/main.go @@ -60,19 +60,21 @@ type Opts struct { WebRoot string `long:"web-root" env:"REMARK_WEB_ROOT" default:"./web" description:"web root directory"` } -var opts Opts var revision = "unknown" // Application holds all active objects type Application struct { Opts - srv api.Rest - importer api.Import - exporter migrator.Exporter + srv *api.Rest + importer *api.Import + exporter migrator.Exporter + terminated chan struct{} } func main() { fmt.Printf("remark %s\n", revision) + + var opts Opts p := flags.NewParser(&opts, flags.Default) if _, e := p.ParseArgs(os.Args[1:]); e != nil { os.Exit(1) @@ -80,9 +82,7 @@ func main() { log.Print("[INFO] started remark") ctx, cancel := context.WithCancel(context.Background()) - - // catch signal and invoke graceful termination - go func() { + go func() { // catch signal and invoke graceful termination stop := make(chan os.Signal, 1) signal.Notify(stop, os.Interrupt, syscall.SIGTERM) <-stop @@ -94,7 +94,8 @@ func main() { if err != nil { log.Fatalf("[ERROR] failed to setup application, %+v", err) } - log.Printf("[INFO] remark terminated %s", app.Run(ctx)) + err = app.Run(ctx) + log.Printf("[INFO] remark terminated %s", err) } // New prepares application and return it with all active parts @@ -114,7 +115,7 @@ func New(opts Opts) (*Application, error) { } cache := rest.NewLoadingCache(rest.MaxValSize(opts.MaxCachedValue), rest.MaxKeys(opts.MaxCachedItems), - rest.PostFlushFn(postFlushFn)) + rest.PostFlushFn(postFlushFn(opts.Sites, opts.Port))) jwtService := auth.NewJWT(opts.SecretKey, strings.HasPrefix(opts.RemarkURL, "https://"), 7*24*time.Hour) @@ -126,7 +127,7 @@ func New(opts Opts) (*Application, error) { exporter := &migrator.Remark{DataStore: &dataService} - importer := api.Import{ + importer := &api.Import{ Version: revision, Cache: cache, NativeImporter: &migrator.Remark{DataStore: &dataService}, @@ -134,7 +135,7 @@ func New(opts Opts) (*Application, error) { SecretKey: opts.SecretKey, } - srv := api.Rest{ + srv := &api.Rest{ Version: revision, DataService: dataService, Exporter: exporter, @@ -143,14 +144,15 @@ func New(opts Opts) (*Application, error) { Authenticator: auth.Authenticator{ JWTService: jwtService, Admins: opts.Admins, - Providers: makeAuthProviders(jwtService, avatarProxy), + Providers: makeAuthProviders(jwtService, avatarProxy, opts), AvatarProxy: avatarProxy, DevPasswd: opts.DevPasswd, }, Cache: cache, } srv.ScoreThresholds.Low, srv.ScoreThresholds.Critical = opts.LowScore, opts.CriticalScore - return &Application{srv: srv, importer: importer, exporter: exporter, Opts: opts}, nil + tch := make(chan struct{}) + return &Application{srv: srv, importer: importer, exporter: exporter, Opts: opts, terminated: tch}, nil } // Run all application objects @@ -159,15 +161,22 @@ func (a *Application) Run(ctx context.Context) error { log.Printf("[WARN] running in dev mode") } + go func() { + // shutdown on context cancellation + <-ctx.Done() + a.srv.Shutdown() + a.importer.Shutdown() + }() a.activateBackup(ctx) // runs in goroutine for each site go a.importer.Run(a.Port + 1) - go a.srv.Run(opts.Port) + a.srv.Run(a.Port) + close(a.terminated) + return nil +} - // shutdown on context cancellation - <-ctx.Done() - a.srv.Shutdown() - a.importer.Shutdown() - return ctx.Err() +// Wait for application completion (termination) +func (a *Application) Wait() { + <-a.terminated } // activateBackup runs background backups for each site @@ -226,7 +235,7 @@ func makeDirs(dirs ...string) error { return nil } -func makeAuthProviders(jwtService *auth.JWT, avatarProxy *proxy.Avatar) (providers []auth.Provider) { +func makeAuthProviders(jwtService *auth.JWT, avatarProxy *proxy.Avatar, opts Opts) (providers []auth.Provider) { makeParams := func(cid, secret string) auth.Params { return auth.Params{ @@ -257,23 +266,25 @@ func makeAuthProviders(jwtService *auth.JWT, avatarProxy *proxy.Avatar) (provide } // post-flush callback invoked by cache after each flush in async way -func postFlushFn() { +func postFlushFn(sites []string, port int) func() { - // list of heavy urls for pre-heating on cache change - urls := []string{ - "http://localhost:%d/api/v1/list?site=%s", - "http://localhost:%d/api/v1/last/50?site=%s", - } + return func() { + // list of heavy urls for pre-heating on cache change + urls := []string{ + "http://localhost:%d/api/v1/list?site=%s", + "http://localhost:%d/api/v1/last/50?site=%s", + } - for _, site := range opts.Sites { - for _, u := range urls { - resp, err := http.Get(fmt.Sprintf(u, opts.Port, site)) - if err != nil { - log.Printf("[WARN] failed to refresh cached list for %s, %s", site, err) - return - } - if err = resp.Body.Close(); err != nil { - log.Printf("[WARN] failed to close response body, %s", err) + for _, site := range sites { + for _, u := range urls { + resp, err := http.Get(fmt.Sprintf(u, port, site)) + if err != nil { + log.Printf("[WARN] failed to refresh cached list for %s, %s", site, err) + return + } + if err = resp.Body.Close(); err != nil { + log.Printf("[WARN] failed to close response body, %s", err) + } } } } diff --git a/app/main_test.go b/app/main_test.go new file mode 100644 index 00000000..85615b0e --- /dev/null +++ b/app/main_test.go @@ -0,0 +1,62 @@ +package main + +import ( + "context" + "fmt" + "io/ioutil" + "log" + "net/http" + "testing" + "time" + + flags "github.com/jessevdk/go-flags" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestApplication(t *testing.T) { + app, ctx := prepApp(t, 18080, 500*time.Millisecond) + go app.Run(ctx) + time.Sleep(100 * time.Millisecond) // let server start + + resp, err := http.Get("http://localhost:18080/api/v1/ping") + require.Nil(t, err) + defer resp.Body.Close() + + assert.Equal(t, 200, resp.StatusCode) + body, err := ioutil.ReadAll(resp.Body) + assert.Nil(t, err) + assert.Equal(t, "pong", string(body)) + app.Wait() +} + +func TestApplicationShutdown(t *testing.T) { + app, ctx := prepApp(t, 18090, 500*time.Millisecond) + st := time.Now() + app.Run(ctx) + assert.True(t, time.Since(st).Seconds() < 1, "should take about 500msec") + app.Wait() +} + +func prepApp(t *testing.T, port int, duration time.Duration) (*Application, context.Context) { + // prepare options + opts := Opts{} + p := flags.NewParser(&opts, flags.Default) + p.ParseArgs([]string{"--secret=123456"}) + opts.AvatarStore, opts.BackupLocation = "/tmp", "/tmp" + opts.BoltPath = fmt.Sprintf("/tmp/%d", port) + opts.GithubCSEC, opts.GithubCID = "csec", "cid" + opts.Port = port + + // create app + app, err := New(opts) + require.Nil(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(duration) + log.Print("[TEST] terminate app") + cancel() + }() + return app, ctx +} diff --git a/app/rest/api/import.go b/app/rest/api/import.go index c7318d3e..e04429bc 100644 --- a/app/rest/api/import.go +++ b/app/rest/api/import.go @@ -6,6 +6,7 @@ import ( "log" "net/http" "strings" + "sync" "time" "github.com/didip/tollbooth" @@ -27,6 +28,7 @@ type Import struct { SecretKey string httpServer *http.Server + lock sync.Mutex } // Run the listener and request's router, activate rest server @@ -34,7 +36,11 @@ type Import struct { func (s *Import) Run(port int) { log.Printf("[INFO] activate import server on port %d", port) router := s.routes() + + s.lock.Lock() s.httpServer = &http.Server{Addr: fmt.Sprintf("127.0.0.1:%d", port), Handler: router} + s.lock.Unlock() + err := s.httpServer.ListenAndServe() log.Printf("[WARN] http server terminated, %s", err) } @@ -44,9 +50,13 @@ func (s *Import) Shutdown() { log.Print("[WARN] shutdown import server") ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() + + s.lock.Lock() if err := s.httpServer.Shutdown(ctx); err != nil { log.Printf("[DEBUG] importer shutdown error, %s", err) } + s.lock.Unlock() + log.Print("[DEBUG] shutdown import server completed") } diff --git a/app/rest/api/rest.go b/app/rest/api/rest.go index 23356638..21242e2f 100644 --- a/app/rest/api/rest.go +++ b/app/rest/api/rest.go @@ -11,6 +11,7 @@ import ( "net/http" "strconv" "strings" + "sync" "time" "github.com/didip/tollbooth" @@ -44,7 +45,9 @@ type Rest struct { Critical int } - httpServer *http.Server + httpServer *http.Server + lock sync.Mutex + adminService admin } @@ -64,6 +67,7 @@ func (s *Rest) Run(port int) { router := s.routes() + s.lock.Lock() s.httpServer = &http.Server{ Addr: fmt.Sprintf(":%d", port), Handler: router, @@ -71,6 +75,8 @@ func (s *Rest) Run(port int) { WriteTimeout: 5 * time.Second, IdleTimeout: 30 * time.Second, } + s.lock.Unlock() + err := s.httpServer.ListenAndServe() log.Printf("[WARN] http server terminated, %s", err) } @@ -80,10 +86,12 @@ func (s *Rest) Shutdown() { log.Print("[WARN] shutdown rest server") ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() + s.lock.Lock() if err := s.httpServer.Shutdown(ctx); err != nil { log.Printf("[DEBUG] rest shutdown error, %s", err) } log.Print("[DEBUG] shutdown rest server completed") + s.lock.Unlock() } func (s *Rest) routes() chi.Router {