From d74239c7353c6813834d1a34a02b72d6bb3afbe5 Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Sat, 12 Feb 2022 16:38:32 +0100 Subject: [PATCH] Improve HTTP server startup, add wrapping handler HTTP server now initializes much earlier and mounts "waiting" router with debugging, version and health check routes. When server is fully activated it switches to "active" router. Shutdown stage is also handled with catch-all route. --- app/app.go | 3 +- app/cli.go | 33 ++++++- app/servers.go | 32 ------ pkg/api/server/demux.go | 53 ++++++++++ pkg/api/server/handlers.go | 185 +++++++++++++++++++++++++++++++++++ pkg/api/server/middleware.go | 42 ++++---- pkg/api/server/server.go | 157 ++++++++--------------------- pkg/errors/http.go | 6 +- 8 files changed, 335 insertions(+), 176 deletions(-) create mode 100644 pkg/api/server/demux.go create mode 100644 pkg/api/server/handlers.go diff --git a/app/app.go b/app/app.go index ba74b8ccb..0491d61b0 100644 --- a/app/app.go +++ b/app/app.go @@ -19,8 +19,9 @@ import ( type ( httpApiServer interface { - MountRoutes(mm ...func(chi.Router)) Serve(ctx context.Context) + Activate(mm ...func(chi.Router)) + Shutdown() } grpcServer interface { diff --git a/app/cli.go b/app/cli.go index 24bf19121..61042e6a5 100644 --- a/app/cli.go +++ b/app/cli.go @@ -3,9 +3,12 @@ package app import ( "context" "fmt" + "sync" authCommands "github.com/cortezaproject/corteza-server/auth/commands" federationCommands "github.com/cortezaproject/corteza-server/federation/commands" + "github.com/cortezaproject/corteza-server/pkg/actionlog" + "github.com/cortezaproject/corteza-server/pkg/api/server" "github.com/cortezaproject/corteza-server/pkg/cli" "github.com/cortezaproject/corteza-server/pkg/options" "github.com/cortezaproject/corteza-server/pkg/plugin" @@ -65,11 +68,39 @@ func (app *CortezaApp) InitCLI() { "If no paths are provided, corteza loads .env file from the current directory (equivalent to --env-file .)") serveCmd := cli.ServeCommand(func() (err error) { + wg := &sync.WaitGroup{} + + { // @todo refactor wait-for out of HTTP API server. + app.HttpServer = server.New(app.Log, app.Opt.Environment, app.Opt.HTTPServer, app.Opt.WaitFor) + + wg.Add(1) + go func() { + app.HttpServer.Serve(actionlog.RequestOriginToContext(ctx, actionlog.RequestOrigin_API_REST)) + wg.Done() + }() + } + + { + // @todo add other server/listeners here... + //wg.Add(1) + //go func(ctx context.Context) { + // grpcApi.Serve(actionlog.RequestOriginToContext(ctx, actionlog.RequestOrigin_API_GRPC)) + // wg.Done() + //}(ctx) + } + if err = app.Activate(ctx); err != nil { return } - return app.Serve(ctx) + app.HttpServer.Activate(app.mountHttpRoutes) + + // Wait for all servers to be done + wg.Wait() + + app.HttpServer.Shutdown() + + return nil }) upgradeCmd := cli.UpgradeCommand(func() (err error) { diff --git a/app/servers.go b/app/servers.go index 58c8749e7..47a5c6775 100644 --- a/app/servers.go +++ b/app/servers.go @@ -1,19 +1,15 @@ package app import ( - "context" "net/http" "path" "regexp" "strings" - "sync" automationRest "github.com/cortezaproject/corteza-server/automation/rest" composeRest "github.com/cortezaproject/corteza-server/compose/rest" "github.com/cortezaproject/corteza-server/docs" federationRest "github.com/cortezaproject/corteza-server/federation/rest" - "github.com/cortezaproject/corteza-server/pkg/actionlog" - "github.com/cortezaproject/corteza-server/pkg/api/server" "github.com/cortezaproject/corteza-server/pkg/logger" "github.com/cortezaproject/corteza-server/pkg/options" "github.com/cortezaproject/corteza-server/pkg/webapp" @@ -23,34 +19,6 @@ import ( "go.uber.org/zap" ) -func (app *CortezaApp) Serve(ctx context.Context) (err error) { - wg := &sync.WaitGroup{} - - { // @todo refactor wait-for out of HTTP API server. - app.HttpServer = server.New(app.Log, app.Opt.Environment, app.Opt.HTTPServer, app.Opt.WaitFor) - app.HttpServer.MountRoutes(app.mountHttpRoutes) - - wg.Add(1) - go func() { - app.HttpServer.Serve(actionlog.RequestOriginToContext(ctx, actionlog.RequestOrigin_API_REST)) - wg.Done() - }() - } - - { - //wg.Add(1) - //go func(ctx context.Context) { - // grpcApi.Serve(actionlog.RequestOriginToContext(ctx, actionlog.RequestOrigin_API_GRPC)) - // wg.Done() - //}(ctx) - } - - // Wait for all servers to be done - wg.Wait() - - return nil -} - func (app *CortezaApp) mountHttpRoutes(r chi.Router) { var ( ho = app.Opt.HTTPServer diff --git a/pkg/api/server/demux.go b/pkg/api/server/demux.go new file mode 100644 index 000000000..30d3c32de --- /dev/null +++ b/pkg/api/server/demux.go @@ -0,0 +1,53 @@ +package server + +import ( + "fmt" + "net/http" + + "github.com/go-chi/chi/v5" + "go.uber.org/atomic" +) + +type ( + // demux (demultiplexer) routes request to one of the underlying routers + // according to current state + demux struct { + state *atomic.Uint32 + routers map[uint32]chi.Router + } +) + +var _ http.Handler = &demux{} + +func Demux(state uint32, r chi.Router) *demux { + return &demux{ + state: atomic.NewUint32(state), + routers: map[uint32]chi.Router{state: r}, + } +} + +func (d *demux) State(s uint32) { + d.state.Store(s) +} + +func (d *demux) Router(s uint32, r chi.Router) { + d.routers[s] = r +} + +func (d *demux) ServeHTTP(w http.ResponseWriter, r *http.Request) { + defer panicRecovery(r.Context(), w) + + var ( + state = d.state.Load() + router, exists = d.routers[state] + ) + + if !exists { + _, _ = fmt.Fprintf(w, "unconfigured request demultiplexor state (%d)", state) + w.WriteHeader(http.StatusInternalServerError) + return + } + + router.ServeHTTP(w, r) + +} diff --git a/pkg/api/server/handlers.go b/pkg/api/server/handlers.go new file mode 100644 index 000000000..b8de45d8c --- /dev/null +++ b/pkg/api/server/handlers.go @@ -0,0 +1,185 @@ +package server + +import ( + "fmt" + "net/http" + "path" + "strings" + + "github.com/cortezaproject/corteza-server/pkg/api" + "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/pkg/errors" + "github.com/cortezaproject/corteza-server/pkg/healthcheck" + "github.com/cortezaproject/corteza-server/pkg/options" + "github.com/cortezaproject/corteza-server/pkg/version" + "github.com/go-chi/chi/v5" + "github.com/go-chi/chi/v5/middleware" + "go.uber.org/zap" +) + +// routes used when server is in waiting mode +func waitingRoutes(log *zap.Logger, httpOpt options.HttpServerOpt) (r chi.Router) { + r = chi.NewRouter() + mountServiceHandlers(r, log, httpOpt) + + r.HandleFunc("/*", func(w http.ResponseWriter, r *http.Request) { + // For non GET requests, return 503 (service unavailable) + errors.ServeHTTPWithCode(w, r, + http.StatusServiceUnavailable, + fmt.Errorf("corteza server initializing"), + true, + ) + }) + + r.Get("/*", func(w http.ResponseWriter, r *http.Request) { + // Refresh the page in 15 seconds + w.Header().Set("Refresh", "15; url=/") + _, _ = fmt.Fprint(w, "Corteza server initializing\n\n") + if httpOpt.EnableHealthcheckRoute { + healthcheck.Defaults().Run(r.Context()).WriteTo(w) + } + }) + + return +} + +// routes used when server in shutdown mode +func shutdownRoutes() (r chi.Router) { + r = chi.NewRouter() + + r.HandleFunc("/*", func(w http.ResponseWriter, r *http.Request) { + // For non GET requests, return 503 (service unavailable) + errors.ServeHTTPWithCode(w, r, + http.StatusServiceUnavailable, + fmt.Errorf("corteza server shutting down"), + true, + ) + }) + + r.Get("/*", func(w http.ResponseWriter, r *http.Request) { + // Refresh the page in 15 seconds + w.Header().Set("Refresh", "15; url=/") + _, _ = fmt.Fprint(w, "corteza server shutting down") + }) + return +} + +// routes used when in active mode +func activeRoutes(log *zap.Logger, mountable []func(r chi.Router), envOpt options.EnvironmentOpt, httpOpt options.HttpServerOpt) (r chi.Router) { + r = chi.NewRouter() + + r.Route("/"+strings.TrimPrefix(httpOpt.BaseUrl, "/"), func(r chi.Router) { + // Reports error to Sentry if enabled + if httpOpt.EnablePanicReporting { + r.Use(sentryMiddleware()) + } + + if httpOpt.EnableMetrics { + // Metrics tracking middleware + r.Use(metricsMiddleware(httpOpt.MetricsServiceLabel)) + } + + // Handle panic (sets 500 server error headers) + //r.Use(handlePanic) + + // Base middleware, CORS, RealIP, RequestID, context-logger + r.Use(BaseMiddleware(envOpt.IsProduction(), log)...) + + // Logging request if enabled + if httpOpt.LogRequest { + r.Use(LogRequest) + } + + // Logging response if enabled + if httpOpt.LogResponse { + r.Use(LogResponse) + } + + // Verifies JWT in headers, cookies, ... + r.Use(auth.HttpTokenVerifier) + + for _, mount := range mountable { + mount(r) + } + + if httpOpt.EnableMetrics { + metricsMount(r, httpOpt.MetricsUsername, httpOpt.MetricsPassword) + } + + }) + + if httpOpt.BaseUrl != "/" { + r.Handle("/", http.RedirectHandler(httpOpt.BaseUrl, http.StatusTemporaryRedirect)) + + } + + mountServiceHandlers(r, log, httpOpt) + return +} + +func mountServiceHandlers(r chi.Router, log *zap.Logger, opt options.HttpServerOpt) { + if opt.EnableDebugRoute { + mountDebugHandler(r, log) + } + + if opt.EnableVersionRoute { + mountVersionHandler(r, log, opt.BaseUrl) + } + + if opt.EnableHealthcheckRoute { + mountHealthCheckHandler(r, log, opt.BaseUrl) + } +} + +func mountDebugHandler(r chi.Router, log *zap.Logger) { + log.Debug("route debugger enabled: /__routes") + r.Get("/__routes", debugRoutes(r)) + + log.Debug("profiler enabled: /__profiler") + r.Mount("/__profiler", middleware.Profiler()) + + log.Debug("eventbus handlers debug enabled: /__eventbus") + r.Get("/__eventbus", debugEventbus()) + + log.Debug("corredor service debug enabled: /__corredor") + r.Get("/__corredor", debugCorredor()) +} + +func mountVersionHandler(r chi.Router, log *zap.Logger, basePath string) { + var ( + dPath = "/version" + sPath = path.Join(basePath, dPath) + handler = func(w http.ResponseWriter, r *http.Request) { + api.Send(w, r, struct { + BuildTime string `json:"buildTime"` + Version string `json:"version"` + }{version.BuildTime, version.Version}) + } + ) + + r.Get(dPath, handler) + log.Debug("version route enabled: " + dPath) + + if dPath != sPath { + r.Get(sPath, handler) + log.Debug("version route enabled: " + sPath) + } + +} + +func mountHealthCheckHandler(r chi.Router, log *zap.Logger, basePath string) { + // default & sub path for health-check endpoint + var ( + dPath = "/healthcheck" + sPath = path.Join(basePath, dPath) + ) + + r.Get(dPath, healthcheck.HttpHandler()) + log.Debug("health check route enabled: " + sPath) + + if dPath != sPath { + r.Get(sPath, healthcheck.HttpHandler()) + log.Debug("health check route enabled: " + sPath) + } + +} diff --git a/pkg/api/server/middleware.go b/pkg/api/server/middleware.go index ae944bc42..7c70ba79c 100644 --- a/pkg/api/server/middleware.go +++ b/pkg/api/server/middleware.go @@ -1,6 +1,7 @@ package server import ( + "context" "net/http" "os" "runtime/debug" @@ -31,32 +32,25 @@ func sentryMiddleware() func(http.Handler) http.Handler { }).Handle } -// HandlePanic sends 500 error when panic occurs inside the request call -func handlePanic(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { - defer func() { - if err := recover(); err != nil { - log := logger.Default() - if err, ok := err.(error); ok { - log = log.With(zap.Error(err)) - } else { - log = log.With(zap.Any("recover-value", err)) - } +func panicRecovery(ctx context.Context, w http.ResponseWriter) { + if err := recover(); err != nil { + log := logger.ContextValue(ctx) + if err, ok := err.(error); ok { + log = log.With(zap.Error(err)) + } else { + log = log.With(zap.Any("recover-value", err)) + } - log.Debug("crashed on http request", zap.ByteString("stack", debug.Stack())) + log.Debug("crashed on http request", zap.ByteString("stack", debug.Stack())) - w.WriteHeader(500) + w.WriteHeader(500) - if _, has := os.LookupEnv("DEBUG_DUMP_STACK_IN_RESPONSE"); has { - // Provide nice call stack on endpoint when - // we crash - w.Write(debug.Stack()) - } + if _, has := os.LookupEnv("DEBUG_DUMP_STACK_IN_RESPONSE"); has { + // Provide nice call stack on endpoint when + // we crash + _, _ = w.Write(debug.Stack()) + } - return - } - }() - - next.ServeHTTP(w, req) - }) + return + } } diff --git a/pkg/api/server/server.go b/pkg/api/server/server.go index 9db06877a..63946d7a0 100644 --- a/pkg/api/server/server.go +++ b/pkg/api/server/server.go @@ -4,16 +4,9 @@ import ( "context" "net" "net/http" - "path" - "strings" - "github.com/cortezaproject/corteza-server/pkg/api" - "github.com/cortezaproject/corteza-server/pkg/auth" - "github.com/cortezaproject/corteza-server/pkg/healthcheck" "github.com/cortezaproject/corteza-server/pkg/options" - "github.com/cortezaproject/corteza-server/pkg/version" "github.com/go-chi/chi/v5" - "github.com/go-chi/chi/v5/middleware" "go.uber.org/zap" ) @@ -24,11 +17,29 @@ type ( waitForOpt options.WaitForOpt environmentOpt options.EnvironmentOpt endpoints []func(r chi.Router) + + demux *demux } ) +const ( + waiting uint32 = iota + active + shutdown +) + +// New initializes new HTTP server with special powers +// Server is started as early as possible and with a special request handler +// that demultiplexes request to one of the configured routers according to the server state. +// +// Waiting state +// This is initial state that with some simple route handlers: +// - /version +// - /healthcheck +// - /healthcheck + func New(log *zap.Logger, envOpt options.EnvironmentOpt, httpOpt options.HttpServerOpt, waitForOpt options.WaitForOpt) *server { - return &server{ + s := &server{ endpoints: make([]func(r chi.Router), 0), log: log.Named("http"), @@ -36,10 +47,25 @@ func New(log *zap.Logger, envOpt options.EnvironmentOpt, httpOpt options.HttpSer httpOpt: httpOpt, waitForOpt: waitForOpt, } + + s.demux = Demux(waiting, waitingRoutes(s.log, s.httpOpt)) + s.demux.Router(shutdown, shutdownRoutes()) + + return s } -func (s *server) MountRoutes(mm ...func(chi.Router)) { - s.endpoints = append(s.endpoints, mm...) +// Activate reconfigures server to use active routes +func (s *server) Activate(mm ...func(chi.Router)) { + s.demux.Router(active, activeRoutes(s.log, mm, s.environmentOpt, s.httpOpt)) + + s.log.Debug("entering active state") + s.demux.State(active) +} + +// Shutdown reconfigures server to use shutdown routes +func (s *server) Shutdown() { + s.log.Debug("entering shutdown state") + s.demux.State(shutdown) } func (s server) Serve(ctx context.Context) { @@ -56,61 +82,14 @@ func (s server) Serve(ctx context.Context) { return } - router := chi.NewRouter() - - router.Route("/"+strings.TrimPrefix(s.httpOpt.BaseUrl, "/"), func(r chi.Router) { - // Reports error to Sentry if enabled - if s.httpOpt.EnablePanicReporting { - r.Use(sentryMiddleware()) - } - - if s.httpOpt.EnableMetrics { - // Metrics tracking middleware - r.Use(metricsMiddleware(s.httpOpt.MetricsServiceLabel)) - } - - // Handle panic (sets 500 server error headers) - r.Use(handlePanic) - - // Base middleware, CORS, RealIP, RequestID, context-logger - r.Use(BaseMiddleware(s.environmentOpt.IsProduction(), s.log)...) - - // Logging request if enabled - if s.httpOpt.LogRequest { - r.Use(LogRequest) - } - - // Logging response if enabled - if s.httpOpt.LogResponse { - r.Use(LogResponse) - } - - // Verifies JWT in headers, cookies, ... - r.Use(auth.HttpTokenVerifier) - - for _, mountRoutes := range s.endpoints { - mountRoutes(r) - } - - if s.httpOpt.EnableMetrics { - metricsMount(r, s.httpOpt.MetricsUsername, s.httpOpt.MetricsPassword) - } - - }) - - if s.httpOpt.BaseUrl != "/" { - router.Handle("/", http.RedirectHandler(s.httpOpt.BaseUrl, http.StatusTemporaryRedirect)) - - } - - s.bindMiscRoutes(router) - go func() { srv := http.Server{ - Handler: router, - BaseContext: func(listener net.Listener) context.Context { - return ctx - }, + Handler: s.demux, + + // use root context as server's base context and as a basis for + // context for all requests + // this enables us to send cancellation down to every request + BaseContext: func(listener net.Listener) context.Context { return ctx }, } err = srv.Serve(listener) }() @@ -125,55 +104,3 @@ func (s server) Serve(ctx context.Context) { s.log.Info("HTTP server stopped", zap.Error(err)) } - -func (s server) bindMiscRoutes(r chi.Router) { - if s.httpOpt.EnableDebugRoute { - s.log.Debug("route debugger enabled: /__routes") - r.Get("/__routes", debugRoutes(r)) - - s.log.Debug("profiler enabled: /__profiler") - r.Mount("/__profiler", middleware.Profiler()) - - s.log.Debug("eventbus handlers debug enabled: /__eventbus") - r.Get("/__eventbus", debugEventbus()) - - s.log.Debug("corredor service debug enabled: /__corredor") - r.Get("/__corredor", debugCorredor()) - } - - if s.httpOpt.EnableVersionRoute { - var ( - dPath = "/version" - sPath = path.Join(s.httpOpt.BaseUrl, dPath) - v = func(w http.ResponseWriter, r *http.Request) { - api.Send(w, r, struct { - BuildTime string `json:"buildTime"` - Version string `json:"version"` - }{version.BuildTime, version.Version}) - } - ) - - r.Get(dPath, v) - if dPath != sPath { - r.Get(sPath, v) - } - } - - if s.httpOpt.EnableHealthcheckRoute { - // default & sub path for healthcheck endpoint - var ( - dPath = "/healthcheck" - sPath = path.Join(s.httpOpt.BaseUrl, dPath) - log = s.log.With(zap.String("url", dPath)) - ) - - r.Get(dPath, healthcheck.HttpHandler()) - - if dPath != sPath { - r.Get(sPath, healthcheck.HttpHandler()) - log = log.With(zap.String("url", sPath)) - } - - log.Info("healthcheck endpoint enabled") - } -} diff --git a/pkg/errors/http.go b/pkg/errors/http.go index 10d87eb75..eca5f5f50 100644 --- a/pkg/errors/http.go +++ b/pkg/errors/http.go @@ -27,7 +27,7 @@ type ( func ServeHTTP(w http.ResponseWriter, r *http.Request, err error, mask bool) { // due to backward compatibility, // custom HTTP statuses are disabled for now. - serveHTTP(w, r, http.StatusOK, err, mask) + ServeHTTPWithCode(w, r, http.StatusOK, err, mask) } // ProperlyServeHTTP Prepares and encodes given error for HTTP transport, same as ServeHTTP but with proper status codes @@ -40,11 +40,11 @@ func ProperlyServeHTTP(w http.ResponseWriter, r *http.Request, err error, mask b code = e.kind.httpStatus() } - serveHTTP(w, r, code, err, mask) + ServeHTTPWithCode(w, r, code, err, mask) } // Serves error via -func serveHTTP(w http.ResponseWriter, r *http.Request, code int, err error, mask bool) { +func ServeHTTPWithCode(w http.ResponseWriter, r *http.Request, code int, err error, mask bool) { var ( // Very naive approach on parsing accept headers acceptsJson = strings.Contains(r.Header.Get("accept"), "application/json")