DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Improve HTTP server startup, add wrapping handler

Browse Source 
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.
This commit is contained in:
Denis Arh
2022-02-12 16:38:32 +01:00
parent f697511902
commit d74239c735
8 changed files with 335 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/app.go
+2 -1
View File
@@ -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 {
app/cli.go
+32 -1
View File
@@ -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) {
app/servers.go
-32
View File
@@ -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
pkg/api/server/demux.go
+53
View File
@@ -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)
}
pkg/api/server/handlers.go
+185
View File
@@ -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)
}
}
pkg/api/server/middleware.go
+18 -24
View File
@@ -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
}
}
pkg/api/server/server.go
+42 -115
View File
@@ -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")
}
}
pkg/errors/http.go
+3 -3
View File
@@ -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")