Refactor & improve application initialization
This commit is contained in:
Denis Arh
+3
-9
@@ -7,16 +7,10 @@ import (
|
||||
"runtime"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/go-chi/chi/middleware"
|
||||
)
|
||||
|
||||
func Debug(r chi.Router) {
|
||||
r.Mount("/debug", middleware.Profiler())
|
||||
DebugRoutes(r)
|
||||
}
|
||||
|
||||
func DebugRoutes(r chi.Router) {
|
||||
r.Get("/debug/routes", func(w http.ResponseWriter, req *http.Request) {
|
||||
func debugRoutes(r chi.Routes) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, req *http.Request) {
|
||||
var printRoutes func(chi.Routes, string)
|
||||
|
||||
printRoutes = func(r chi.Routes, pfix string) {
|
||||
@@ -33,5 +27,5 @@ func DebugRoutes(r chi.Router) {
|
||||
}
|
||||
|
||||
printRoutes(r, "")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+3
-3
@@ -9,12 +9,12 @@ import (
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
)
|
||||
|
||||
// Middleware is the request logger that provides metrics to prometheus
|
||||
func Middleware(name string) func(http.Handler) http.Handler {
|
||||
// MetricsMiddleware is the request logger that provides metrics to prometheus
|
||||
func metricsMiddleware(name string) func(http.Handler) http.Handler {
|
||||
return chiprometheus.NewMiddleware(name)
|
||||
}
|
||||
|
||||
func Mount(r chi.Router, username, password string) {
|
||||
func metricsMount(r chi.Router, username, password string) {
|
||||
r.Group(func(r chi.Router) {
|
||||
r.Use(basicauth.New("Metrics", map[string][]string{
|
||||
username: {password},
|
||||
|
||||
@@ -8,12 +8,12 @@ import (
|
||||
"github.com/go-chi/chi/middleware"
|
||||
"go.uber.org/zap"
|
||||
|
||||
sentryhttp "github.com/getsentry/sentry-go/http"
|
||||
"github.com/getsentry/sentry-go/http"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/logger"
|
||||
)
|
||||
|
||||
func Base(log *zap.Logger) []func(http.Handler) http.Handler {
|
||||
func BaseMiddleware(log *zap.Logger) []func(http.Handler) http.Handler {
|
||||
return []func(http.Handler) http.Handler{
|
||||
handleCORS,
|
||||
middleware.RealIP,
|
||||
@@ -22,14 +22,14 @@ func Base(log *zap.Logger) []func(http.Handler) http.Handler {
|
||||
}
|
||||
}
|
||||
|
||||
func Sentry() func(http.Handler) http.Handler {
|
||||
func sentryMiddleware() func(http.Handler) http.Handler {
|
||||
return sentryhttp.New(sentryhttp.Options{
|
||||
Repanic: true,
|
||||
}).Handle
|
||||
}
|
||||
|
||||
// HandlePanic sends 500 error when panic occurs inside the request call
|
||||
func HandlePanic(next http.Handler) http.Handler {
|
||||
func handlePanic(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
|
||||
@@ -1,71 +0,0 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"expvar"
|
||||
"runtime"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/logger"
|
||||
)
|
||||
|
||||
type Monitor struct {
|
||||
Alloc,
|
||||
TotalAlloc,
|
||||
Sys,
|
||||
Mallocs,
|
||||
Frees,
|
||||
LiveObjects,
|
||||
PauseTotalNs uint64
|
||||
|
||||
NumGC uint32
|
||||
NumGoroutine int
|
||||
}
|
||||
|
||||
func NewMonitor(duration int) {
|
||||
var (
|
||||
m = Monitor{}
|
||||
rtm runtime.MemStats
|
||||
goroutines = expvar.NewInt("num_goroutine")
|
||||
)
|
||||
var interval = time.Duration(duration) * time.Second
|
||||
for {
|
||||
<-time.After(interval)
|
||||
|
||||
// Read full mem stats
|
||||
runtime.ReadMemStats(&rtm)
|
||||
|
||||
// Number of goroutines
|
||||
m.NumGoroutine = runtime.NumGoroutine()
|
||||
goroutines.Set(int64(m.NumGoroutine))
|
||||
|
||||
// Misc memory stats
|
||||
m.Alloc = rtm.Alloc
|
||||
m.TotalAlloc = rtm.TotalAlloc
|
||||
m.Sys = rtm.Sys
|
||||
m.Mallocs = rtm.Mallocs
|
||||
m.Frees = rtm.Frees
|
||||
|
||||
// Live objects = Mallocs - Frees
|
||||
m.LiveObjects = m.Mallocs - m.Frees
|
||||
|
||||
// GC Stats
|
||||
m.PauseTotalNs = rtm.PauseTotalNs
|
||||
m.NumGC = rtm.NumGC
|
||||
|
||||
logger.Default().
|
||||
With(
|
||||
zap.Uint64("alloc", m.Alloc),
|
||||
zap.Uint64("totalAlloc", m.TotalAlloc),
|
||||
zap.Uint64("sys", m.Sys),
|
||||
zap.Uint64("mallocs", m.Mallocs),
|
||||
zap.Uint64("frees", m.Frees),
|
||||
zap.Uint64("liveObjects", m.LiveObjects),
|
||||
zap.Uint64("pauseTotalNs", m.PauseTotalNs),
|
||||
zap.Uint32("numGC", m.NumGC),
|
||||
zap.Int("numGoRoutines", m.NumGoroutine),
|
||||
).
|
||||
Debug("monitor")
|
||||
}
|
||||
}
|
||||
+27
-230
@@ -4,81 +4,40 @@ import (
|
||||
"context"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/spf13/cobra"
|
||||
"github.com/go-chi/chi/middleware"
|
||||
"github.com/titpetric/factory/resputil"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/app/options"
|
||||
"github.com/cortezaproject/corteza-server/pkg/auth"
|
||||
"github.com/cortezaproject/corteza-server/pkg/cli/options"
|
||||
"github.com/cortezaproject/corteza-server/pkg/version"
|
||||
)
|
||||
|
||||
type (
|
||||
Server struct {
|
||||
name string
|
||||
|
||||
log *zap.Logger
|
||||
|
||||
httpOpt *options.HTTPOpt
|
||||
monitorOpt *options.MonitorOpt
|
||||
|
||||
endpoints []func(r chi.Router)
|
||||
server struct {
|
||||
log *zap.Logger
|
||||
httpOpt options.HTTPServerOpt
|
||||
waitForOpt options.WaitForOpt
|
||||
endpoints []func(r chi.Router)
|
||||
}
|
||||
)
|
||||
|
||||
var (
|
||||
Monolith = false
|
||||
BaseURL = "/"
|
||||
)
|
||||
|
||||
func NewServer(log *zap.Logger) *Server {
|
||||
return &Server{
|
||||
endpoints: make([]func(r chi.Router), 0),
|
||||
log: log.Named("http"),
|
||||
func New(log *zap.Logger, httpOpt options.HTTPServerOpt, waitForOpt options.WaitForOpt) *server {
|
||||
return &server{
|
||||
endpoints: make([]func(r chi.Router), 0),
|
||||
log: log.Named("http"),
|
||||
httpOpt: httpOpt,
|
||||
waitForOpt: waitForOpt,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) Command(ctx context.Context, cmdName, prefix string, preRun func(context.Context) error) (cmd *cobra.Command) {
|
||||
s.httpOpt = options.HTTP(prefix)
|
||||
s.monitorOpt = options.Monitor(prefix)
|
||||
|
||||
cmd = &cobra.Command{
|
||||
Use: cmdName,
|
||||
Short: "Start HTTP Server with REST API",
|
||||
|
||||
// Connect all the wires, prepare services, run watchers, bind endpoints
|
||||
PreRunE: func(cmd *cobra.Command, args []string) error {
|
||||
s.waitFor(ctx, options.WaitFor(prefix))
|
||||
|
||||
if s.monitorOpt.Interval > 0 {
|
||||
go NewMonitor(int(s.monitorOpt.Interval / time.Second))
|
||||
}
|
||||
|
||||
return preRun(ctx)
|
||||
},
|
||||
|
||||
// Run the server
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
s.Serve(ctx)
|
||||
},
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (s *Server) MountRoutes(mm ...func(chi.Router)) {
|
||||
func (s *server) MountRoutes(mm ...func(chi.Router)) {
|
||||
s.endpoints = append(s.endpoints, mm...)
|
||||
}
|
||||
|
||||
func (s Server) Serve(ctx context.Context) {
|
||||
func (s server) Serve(ctx context.Context) {
|
||||
s.log.Info("Starting HTTP server with REST API", zap.String("address", s.httpOpt.Addr))
|
||||
|
||||
// configure resputil options
|
||||
@@ -98,7 +57,7 @@ func (s Server) Serve(ctx context.Context) {
|
||||
router := chi.NewRouter()
|
||||
|
||||
// Base middleware, CORS, RealIP, RequestID, context-logger
|
||||
router.Use(Base(s.log)...)
|
||||
router.Use(BaseMiddleware(s.log)...)
|
||||
|
||||
// Logging request if enabled
|
||||
if s.httpOpt.LogRequest {
|
||||
@@ -110,17 +69,17 @@ func (s Server) Serve(ctx context.Context) {
|
||||
router.Use(LogResponse)
|
||||
}
|
||||
|
||||
// Handle panic (sets 500 Server error headers)
|
||||
router.Use(HandlePanic)
|
||||
// Handle panic (sets 500 server error headers)
|
||||
router.Use(handlePanic)
|
||||
|
||||
// Reports error to Sentry if enabled
|
||||
if s.httpOpt.EnablePanicReporting {
|
||||
router.Use(Sentry())
|
||||
router.Use(sentryMiddleware())
|
||||
}
|
||||
|
||||
// Metrics tracking middleware
|
||||
if s.httpOpt.EnableMetrics {
|
||||
router.Use(Middleware(s.httpOpt.MetricsServiceLabel))
|
||||
router.Use(metricsMiddleware(s.httpOpt.MetricsServiceLabel))
|
||||
}
|
||||
|
||||
router.Group(func(r chi.Router) {
|
||||
@@ -135,11 +94,15 @@ func (s Server) Serve(ctx context.Context) {
|
||||
})
|
||||
|
||||
if s.httpOpt.EnableMetrics {
|
||||
Mount(router, s.httpOpt.MetricsUsername, s.httpOpt.MetricsPassword)
|
||||
metricsMount(router, s.httpOpt.MetricsUsername, s.httpOpt.MetricsPassword)
|
||||
}
|
||||
|
||||
if s.httpOpt.EnableDebugRoute {
|
||||
Debug(router)
|
||||
s.log.Debug("profiler: /__profiler", zap.Error(err))
|
||||
router.Mount("/__profiler", middleware.Profiler())
|
||||
|
||||
s.log.Debug("list of routes: /__routes", zap.Error(err))
|
||||
router.Get("/__routes", debugRoutes(router))
|
||||
}
|
||||
|
||||
if s.httpOpt.EnableVersionRoute {
|
||||
@@ -158,171 +121,5 @@ func (s Server) Serve(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
s.log.Info("HTTP server stopped", zap.Error(err))
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// waitFor sets up a simple status page, delays execution and probes services
|
||||
func (s Server) waitFor(ctx context.Context, opt *options.WaitForOpt) {
|
||||
var (
|
||||
services = opt.GetServices()
|
||||
)
|
||||
|
||||
if len(services) == 0 && opt.Delay == 0 {
|
||||
// Nothing to do here..
|
||||
return
|
||||
}
|
||||
|
||||
var (
|
||||
log = s.log.Named("wait-for")
|
||||
depChan = make(chan struct{})
|
||||
wg sync.WaitGroup
|
||||
serviceAddr string
|
||||
serviceURL *url.URL
|
||||
err error
|
||||
)
|
||||
|
||||
// Setup a simple HTTP server that will inform the impatent users
|
||||
listener, err := net.Listen("tcp", s.httpOpt.Addr)
|
||||
if err != nil {
|
||||
s.log.Error("Can not start server", zap.Error(err))
|
||||
os.Exit(1)
|
||||
}
|
||||
defer listener.Close()
|
||||
go func() {
|
||||
router := chi.NewRouter()
|
||||
router.Get("/*", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusPreconditionFailed)
|
||||
w.Write([]byte("waiting for services..."))
|
||||
})
|
||||
_ = http.Serve(listener, router)
|
||||
}()
|
||||
|
||||
if opt.Delay > 0 {
|
||||
s.log.Info("delaying", zap.Duration("delay", opt.Delay))
|
||||
|
||||
// First delay execution
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Debug("canceled")
|
||||
return
|
||||
case <-time.After(opt.Delay):
|
||||
// all good...
|
||||
}
|
||||
}
|
||||
|
||||
if len(services) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("waiting for services", zap.Strings("services", services))
|
||||
// Probe services
|
||||
wg.Add(len(services))
|
||||
go func() {
|
||||
|
||||
for _, service := range services {
|
||||
slog := log.With(zap.String("service", service))
|
||||
|
||||
go func(ctx context.Context, service string) {
|
||||
defer wg.Done()
|
||||
|
||||
if serviceAddr, serviceURL, err = s.resolveService(service); err != nil {
|
||||
log.Error("could not resolve service", zap.Error(err))
|
||||
}
|
||||
|
||||
for {
|
||||
ctx, cancelFn := context.WithTimeout(ctx, opt.ServicesProbeTimeout)
|
||||
defer cancelFn()
|
||||
|
||||
if serviceURL == nil {
|
||||
if err = s.probeService(ctx, serviceAddr); err != nil {
|
||||
slog.Warn("service probe failed", zap.Error(err))
|
||||
time.Sleep(opt.ServicesProbeInterval)
|
||||
continue
|
||||
}
|
||||
} else {
|
||||
if err = s.probeServiceURL(ctx, serviceURL); err != nil {
|
||||
slog.Warn("service URL probe failed", zap.Error(err))
|
||||
time.Sleep(opt.ServicesProbeInterval)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
slog.Debug("service ready")
|
||||
return
|
||||
}
|
||||
}(ctx, service)
|
||||
}
|
||||
wg.Wait()
|
||||
close(depChan)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Debug("canceled")
|
||||
return
|
||||
case <-depChan: // services are ready
|
||||
log.Debug("all services ready")
|
||||
return
|
||||
case <-time.After(opt.ServicesTimeout):
|
||||
log.Debug("services not ready")
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (s Server) resolveService(service string) (addr string, u *url.URL, err error) {
|
||||
addr = service
|
||||
|
||||
if strings.Contains(addr, "://") {
|
||||
// Is service an URL?
|
||||
u, err = url.Parse(addr)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
addr = u.Host
|
||||
|
||||
if u.Port() == "" {
|
||||
if u.Scheme == "https" {
|
||||
addr += ":443"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Default to port 80
|
||||
if !strings.Contains(addr, ":") {
|
||||
addr += ":80"
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (s Server) probeService(ctx context.Context, addr string) (err error) {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
dialer := net.Dialer{}
|
||||
_, err = dialer.DialContext(ctx, "tcp", addr)
|
||||
return
|
||||
}
|
||||
|
||||
func (s Server) probeServiceURL(ctx context.Context, u *url.URL) error {
|
||||
req, err := http.NewRequest("GET", u.String(), nil)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to assemble service request")
|
||||
}
|
||||
|
||||
rsp, err := http.DefaultClient.Do(req.WithContext(ctx))
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "service URL request failed")
|
||||
}
|
||||
|
||||
defer rsp.Body.Close()
|
||||
if rsp.StatusCode == http.StatusOK {
|
||||
return nil
|
||||
}
|
||||
|
||||
return errors.Errorf("service responded with unexpected status '%s'", rsp.Status)
|
||||
s.log.Info("Server stopped", zap.Error(err))
|
||||
}
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/pkg/errors"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// WaitFor sets up a simple status page, delays execution and probes services
|
||||
func (s server) WaitFor(ctx context.Context) {
|
||||
var (
|
||||
opt = s.waitForOpt
|
||||
services = opt.GetServices()
|
||||
)
|
||||
|
||||
if len(services) == 0 && opt.Delay == 0 {
|
||||
// Nothing to do here..
|
||||
return
|
||||
}
|
||||
|
||||
var (
|
||||
log = s.log.Named("wait-for")
|
||||
depChan = make(chan struct{})
|
||||
wg sync.WaitGroup
|
||||
serviceAddr string
|
||||
serviceURL *url.URL
|
||||
err error
|
||||
)
|
||||
|
||||
// Setup a simple HTTP server that will inform the impatient users
|
||||
listener, err := net.Listen("tcp", s.httpOpt.Addr)
|
||||
if err != nil {
|
||||
s.log.Error("Can not start server", zap.Error(err))
|
||||
os.Exit(1)
|
||||
}
|
||||
defer listener.Close()
|
||||
go func() {
|
||||
router := chi.NewRouter()
|
||||
router.Get("/*", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusPreconditionFailed)
|
||||
w.Write([]byte("waiting for services..."))
|
||||
})
|
||||
_ = http.Serve(listener, router)
|
||||
}()
|
||||
|
||||
if opt.Delay > 0 {
|
||||
s.log.Info("delaying", zap.Duration("delay", opt.Delay))
|
||||
|
||||
// First delay execution
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Debug("canceled")
|
||||
return
|
||||
case <-time.After(opt.Delay):
|
||||
// all good...
|
||||
}
|
||||
}
|
||||
|
||||
if len(services) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("waiting for services", zap.Strings("services", services))
|
||||
// Probe services
|
||||
wg.Add(len(services))
|
||||
go func() {
|
||||
|
||||
for _, service := range services {
|
||||
slog := log.With(zap.String("service", service))
|
||||
|
||||
go func(ctx context.Context, service string) {
|
||||
defer wg.Done()
|
||||
|
||||
if serviceAddr, serviceURL, err = s.resolveService(service); err != nil {
|
||||
log.Error("could not resolve service", zap.Error(err))
|
||||
}
|
||||
|
||||
for {
|
||||
ctx, cancelFn := context.WithTimeout(ctx, opt.ServicesProbeTimeout)
|
||||
defer cancelFn()
|
||||
|
||||
if serviceURL == nil {
|
||||
if err = s.probeService(ctx, serviceAddr); err != nil {
|
||||
slog.Warn("service probe failed", zap.Error(err))
|
||||
time.Sleep(opt.ServicesProbeInterval)
|
||||
continue
|
||||
}
|
||||
} else {
|
||||
if err = s.probeServiceURL(ctx, serviceURL); err != nil {
|
||||
slog.Warn("service URL probe failed", zap.Error(err))
|
||||
time.Sleep(opt.ServicesProbeInterval)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
slog.Debug("service ready")
|
||||
return
|
||||
}
|
||||
}(ctx, service)
|
||||
}
|
||||
wg.Wait()
|
||||
close(depChan)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Debug("canceled")
|
||||
return
|
||||
case <-depChan: // services are ready
|
||||
log.Debug("all services ready")
|
||||
return
|
||||
case <-time.After(opt.ServicesTimeout):
|
||||
log.Debug("services not ready")
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (s server) resolveService(service string) (addr string, u *url.URL, err error) {
|
||||
addr = service
|
||||
|
||||
if strings.Contains(addr, "://") {
|
||||
// Is service an URL?
|
||||
u, err = url.Parse(addr)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
addr = u.Host
|
||||
|
||||
if u.Port() == "" {
|
||||
if u.Scheme == "https" {
|
||||
addr += ":443"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Default to port 80
|
||||
if !strings.Contains(addr, ":") {
|
||||
addr += ":80"
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (s server) probeService(ctx context.Context, addr string) (err error) {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
dialer := net.Dialer{}
|
||||
_, err = dialer.DialContext(ctx, "tcp", addr)
|
||||
return
|
||||
}
|
||||
|
||||
func (s server) probeServiceURL(ctx context.Context, u *url.URL) error {
|
||||
req, err := http.NewRequest("GET", u.String(), nil)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to assemble service request")
|
||||
}
|
||||
|
||||
rsp, err := http.DefaultClient.Do(req.WithContext(ctx))
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "service URL request failed")
|
||||
}
|
||||
|
||||
defer rsp.Body.Close()
|
||||
if rsp.StatusCode == http.StatusOK {
|
||||
return nil
|
||||
}
|
||||
|
||||
return errors.Errorf("service responded with unexpected status '%s'", rsp.Status)
|
||||
}
|
||||
Reference in New Issue
Block a user