From e72ffa65af4d3a0af12ab3b17a5095109c1009ac Mon Sep 17 00:00:00 2001 From: Drew DeVault Date: Wed, 7 Oct 2020 10:24:19 -0400 Subject: [PATCH] server: incorporate queue management --- server/email.go | 117 +++++++++++++++++++++++------------------------ server/server.go | 59 +++++++++++++++++++++--- 2 files changed, 111 insertions(+), 65 deletions(-) diff --git a/server/email.go b/server/email.go index 55d65b4a08e8d8be9454db7b89168d9df5951934..5728b27bcdff6a2b99b31a585f8a3a596b69d11f 100644 --- a/server/email.go +++ b/server/email.go @@ -9,70 +9,70 @@ import ( "runtime" "github.com/99designs/gqlgen/graphql" - "github.com/vaughan0/go-ini" gomail "gopkg.in/mail.v2" "git.sr.ht/~sircmpwn/core-go/auth" + "git.sr.ht/~sircmpwn/core-go/config" "git.sr.ht/~sircmpwn/core-go/email" ) // Provides a graphql.RecoverFunc which will print the stack trace, and if // debug mode is not enabled, email it to the administrator. -func EmailRecover(config ini.File, debug bool, srv string) graphql.RecoverFunc { - return func(ctx context.Context, _origErr interface{}) error { - var ( - ok bool - origErr error - ) - if origErr, ok = _origErr.(error); !ok { - log.Printf("Unexpected error in recover: %v\n", origErr) - return fmt.Errorf("internal system error") - } - - if errors.Is(origErr, context.Canceled) { - return origErr - } - - if errors.Is(origErr, context.DeadlineExceeded) { - return origErr - } - - if origErr.Error() == "pq: canceling statement due to user request" { - return origErr - } - - stack := make([]byte, 32768) // 32 KiB - i := runtime.Stack(stack, false) - log.Println(origErr.Error()) - log.Println(string(stack[:i])) - if debug { - return fmt.Errorf("internal system error") - } - - to, ok := config.Get("mail", "error-to") - if !ok { - return fmt.Errorf("internal system error") - } - from, _ := config.Get("mail", "error-from") - - m := gomail.NewMessage() - sender, err := mail.ParseAddress(from) - if err != nil { - log.Fatalf("Failed to parse sender address") - } - m.SetAddressHeader("From", sender.Address, sender.Name) - recipient, err := mail.ParseAddress(to) - if err != nil { - log.Fatalf("Failed to parse recipient address") - } - m.SetAddressHeader("To", recipient.Address, recipient.Name) - m.SetHeader("Subject", fmt.Sprintf( - "[%s] GraphQL query error: %v", srv, origErr)) - - quser := auth.ForContext(ctx) - octx := graphql.GetOperationContext(ctx) - - m.SetBody("text/plain", fmt.Sprintf(`Error occured processing GraphQL request: +func emailRecover(ctx context.Context, _origErr interface{}) error { + var ( + ok bool + origErr error + ) + if origErr, ok = _origErr.(error); !ok { + log.Printf("Unexpected error in recover: %v\n", origErr) + return fmt.Errorf("internal system error") + } + + if errors.Is(origErr, context.Canceled) { + return origErr + } + + if errors.Is(origErr, context.DeadlineExceeded) { + return origErr + } + + if origErr.Error() == "pq: canceling statement due to user request" { + return origErr + } + + stack := make([]byte, 32768) // 32 KiB + i := runtime.Stack(stack, false) + log.Println(origErr.Error()) + log.Println(string(stack[:i])) + if config.Debug { + return fmt.Errorf("internal system error") + } + + conf := config.ForContext(ctx) + to, ok := conf.Get("mail", "error-to") + if !ok { + return fmt.Errorf("internal system error") + } + from, _ := conf.Get("mail", "error-from") + + m := gomail.NewMessage() + sender, err := mail.ParseAddress(from) + if err != nil { + log.Fatalf("Failed to parse sender address") + } + m.SetAddressHeader("From", sender.Address, sender.Name) + recipient, err := mail.ParseAddress(to) + if err != nil { + log.Fatalf("Failed to parse recipient address") + } + m.SetAddressHeader("To", recipient.Address, recipient.Name) + m.SetHeader("Subject", fmt.Sprintf( + "[%s] GraphQL query error: %v", config.ServiceName(ctx), origErr)) + + quser := auth.ForContext(ctx) + octx := graphql.GetOperationContext(ctx) + + m.SetBody("text/plain", fmt.Sprintf(`Error occured processing GraphQL request: %v @@ -84,7 +84,6 @@ The following stack trace was produced: %s`, origErr, quser.Username, quser.Email, octx.RawQuery, string(stack[:i]))) - email.Enqueue(ctx, m) - return fmt.Errorf("internal system error") - } + email.Enqueue(ctx, m) + return fmt.Errorf("internal system error") } diff --git a/server/server.go b/server/server.go index de4e946d7c6a5a1c34e6eb021f36431de993d552..766a7ca079104eac3cc319aecca8ceabc2128575 100644 --- a/server/server.go +++ b/server/server.go @@ -1,11 +1,14 @@ package server import ( + "context" "database/sql" "fmt" "log" "net" "net/http" + "os" + "os/signal" "strconv" "time" @@ -20,6 +23,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/vaughan0/go-ini" + "git.sr.ht/~sircmpwn/dowork" "git.sr.ht/~sircmpwn/core-go/auth" "git.sr.ht/~sircmpwn/core-go/config" @@ -44,6 +48,7 @@ type Server struct { router chi.Router schema graphql.ExecutableSchema service string + queues []*work.Queue } func NewServer(service string, conf ini.File) *Server { @@ -59,6 +64,11 @@ func (server *Server) Router() chi.Router { return server.router } +func (server *Server) Get(pattern string, handlerFn http.HandlerFunc) *Server { + server.router.Get(pattern, handlerFn) + return server +} + func (server *Server) WithSchema(schema graphql.ExecutableSchema) *Server { server.schema = schema @@ -76,11 +86,9 @@ func (server *Server) WithSchema(schema graphql.ExecutableSchema) *Server { complexity = 250 } - // TODO: Remove config parameter from EmailRecover - rec := EmailRecover(server.conf, config.Debug, server.service) srv := handler.GraphQL(schema, handler.ComplexityLimit(complexity), - handler.RecoverFunc(rec)) + handler.RecoverFunc(emailRecover)) server.router.Handle("/query", srv) if config.Debug { @@ -151,11 +159,50 @@ func (server *Server) WithMiddleware( return server } -func (server *Server) MakeServer() (*http.Server, net.Listener) { - listen, err := net.Listen("tcp", config.Addr) +func (server *Server) WithQueues(queues ...*work.Queue) *Server { + server.queues = append(server.queues, queues...) + for _, queue := range queues { + queue.Start(context.Background()) + } + return server +} + +func (server *Server) Run() { + qlisten, err := net.Listen("tcp", config.Addr) if err != nil { panic(err) } log.Printf("Running on %s", config.Addr) - return &http.Server{Handler: server.router}, listen + qserver := &http.Server{Handler: server.router} + go qserver.Serve(qlisten) + + mux := &http.ServeMux{} + mux.Handle("/metrics", promhttp.Handler()) + pserver := &http.Server{Handler: mux} + plisten, err := net.Listen("tcp", ":0") + if err != nil { + panic(err) + } + log.Printf("Prometheus listening on :%d", plisten.Addr().(*net.TCPAddr).Port) + go pserver.Serve(plisten) + + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt) + <-sig + signal.Reset(os.Interrupt) + log.Println("SIGINT caught, initiating warm shutdown") + log.Println("SIGINT again to terminate immediately and drop pending requests & tasks") + + log.Println("Terminating server...") + ctx, cancel := context.WithDeadline(context.Background(), + time.Now().Add(30 * time.Second)) + qserver.Shutdown(ctx) + cancel() + + log.Println("Terminating work queues...") + log.Printf("Progress available via Prometheus stats on port %d", + plisten.Addr().(*net.TCPAddr).Port) + work.Join(server.queues...) + qserver.Close() + log.Println("Terminating process.") }