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")