@@ 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")
}
@@ 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.")
}