Files
summercms/modules/surf/summary.go
Jakub Zych 168d65cd1b feat(conga,dev): worker and scheduler reports, summer dev runs serve
- queue:work and schedule:run print a startup summary and a shutdown
  report on a terminal, log them as records otherwise, and return nil
  after a clean stop
- schedule:run --once ends with 'Ran N due entries (M failed)'
- summer dev runs bin/<app> serve with the forwarded global options,
  reports through Output, stops the old server before starting the new
  one and prints a dev report on Ctrl+C (exit 0)
- lagoon.DatabaseLabel describes a DSN without its secrets
- Long help and examples for queue:work, schedule:run, dev, docs:serve
2026-10-21 13:04:27 +02:00

357 lines
10 KiB
Go

package surf
import (
"context"
"errors"
"fmt"
"log/slog"
"net/http"
"strconv"
"strings"
"sync/atomic"
"time"
"git.golem15.com/golem15/summercms/modules/backpack"
"git.golem15.com/golem15/summercms/modules/bonfire"
"git.golem15.com/golem15/summercms/modules/conga"
"git.golem15.com/golem15/summercms/modules/cooler"
"git.golem15.com/golem15/summercms/modules/diary"
"git.golem15.com/golem15/summercms/modules/lagoon"
"git.golem15.com/golem15/summercms/modules/party"
)
// serveLog is the logger of serve's startup summary and shutdown report.
var serveLog = diary.For("serve")
// Messages of the records serve logs.
const (
summaryMessage = "startup summary"
reportMessage = "shutdown report"
listenPrefix = "listening on "
)
// requestCounter wraps the compiled handler and counts requests in flight
// and served, whatever http.access_log says, for the shutdown report.
type requestCounter struct {
next http.Handler
inFlight atomic.Int64
served atomic.Int64
}
func newRequestCounter(next http.Handler) *requestCounter {
return &requestCounter{next: next}
}
func (c *requestCounter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
c.inFlight.Add(1)
defer func() {
c.inFlight.Add(-1)
c.served.Add(1)
}()
c.next.ServeHTTP(w, req)
}
// pluginFact is one plugin's line of the verbose summary.
type pluginFact struct {
ID string
Requires []string
Pending int
}
// summaryFacts is what serve reports once it is listening. It holds no DSN,
// user, password, key or token: the database is its host, port and name.
type summaryFacts struct {
Header string
Environment string
Addr string
URL string
Admin string
AdminNotes []bannerLine
Database string
// Pending is the number of pending migrations; -1 when it could not
// be read.
Pending int
MigrateCmd string
Plugins []pluginFact
Queue string
Scheduler string
Mail string
Cache string
Realtime string
Search string
}
// factInputs are the values serve learned while booting.
type factInputs struct {
addr string
bin string
plugins []party.Plugin
queues []string
worker bool
adminPrefix string
mounted bool
admins adminPresence
pending []lagoon.MigrationRef
pendingErr error
version bonfire.Version
}
// buildFacts collects the summary from the app's config and in.
func buildFacts(app *backpack.App, in factInputs) summaryFacts {
name := strings.TrimSpace(configString(app, "app.name"))
if name == "" {
name = strings.TrimPrefix(in.bin, "bin/")
}
f := summaryFacts{
Header: in.version.Header(name),
Environment: "unknown",
Addr: in.addr,
URL: serveURL(in.addr),
Database: lagoon.DatabaseLabel(configString(app, "database.dsn")),
Pending: len(in.pending),
MigrateCmd: in.bin + " migrate",
Mail: driverFact(app, "mail.driver", "memory"),
Cache: driverFact(app, cooler.CacheDriverKey, "memory"),
Realtime: driverFact(app, "realtime.driver", "null"),
Search: driverFact(app, "search.driver", "null"),
}
if app != nil && app.Config != nil && app.Config.Environment() != "" {
f.Environment = app.Config.Environment()
}
if in.pendingErr != nil {
f.Pending = -1
}
f.Admin, f.AdminNotes = adminStatus(f.URL, in.adminPrefix, in.mounted, in.admins)
perPlugin := map[string]int{}
for _, ref := range in.pending {
perPlugin[ref.Plugin]++
}
for _, p := range in.plugins {
f.Plugins = append(f.Plugins, pluginFact{ID: p.ID(), Requires: p.Requires(), Pending: perPlugin[p.ID()]})
}
if in.worker {
f.Queue = strings.Join(in.queues, ", ") + " (worker in serve)"
f.Scheduler = "in serve"
} else {
f.Queue = "off in serve (queue.work_in_serve is false; run queue:work)"
f.Scheduler = "off in serve (run schedule:run)"
}
return f
}
func configString(app *backpack.App, key string) string {
if app == nil || app.Config == nil {
return ""
}
return app.Config.String(key)
}
// driverFact is the lower-cased driver named by key, or def when empty.
func driverFact(app *backpack.App, key, def string) string {
if v := strings.ToLower(strings.TrimSpace(configString(app, key))); v != "" {
return v
}
return def
}
// migrations is the migration line: "up to date", "N pending — run ..." or
// "unknown".
func (f summaryFacts) migrations() (string, bool) {
switch {
case f.Pending < 0:
return "unknown (the migration history could not be read)", true
case f.Pending == 0:
return "up to date", false
default:
return fmt.Sprintf("%d pending — run %s", f.Pending, f.MigrateCmd), true
}
}
func (f summaryFacts) pluginIDs() []string {
ids := make([]string, 0, len(f.Plugins))
for _, p := range f.Plugins {
ids = append(ids, p.ID)
}
return ids
}
// details is the block printed on a terminal; verbosity 1 or more adds one
// line per plugin with its required plugins and pending migrations.
func (f summaryFacts) details(verbosity int) []bonfire.Detail {
url := f.URL
if url == "" {
url = f.Addr
}
migrations, warn := f.migrations()
plugins := strconv.Itoa(len(f.Plugins))
if len(f.Plugins) > 0 {
plugins += " (" + strings.Join(f.pluginIDs(), ", ") + ")"
}
rows := []bonfire.Detail{
{Label: "Environment", Value: f.Environment},
{Label: "URL", Value: url},
{Label: "Admin", Value: f.Admin},
{Label: "Database", Value: f.Database},
{Label: "Migrations", Value: migrations, Warn: warn},
{Label: "Plugins", Value: plugins},
}
if verbosity >= 1 {
for _, p := range f.Plugins {
requires := "requires nothing"
if len(p.Requires) > 0 {
requires = "requires " + strings.Join(p.Requires, ", ")
}
rows = append(rows, bonfire.Detail{
Label: " " + p.ID,
Value: fmt.Sprintf("%s; %d pending migrations", requires, p.Pending),
Warn: p.Pending > 0,
})
}
}
return append(rows,
bonfire.Detail{Label: "Queue", Value: f.Queue},
bonfire.Detail{Label: "Scheduler", Value: f.Scheduler},
bonfire.Detail{Label: "Mail", Value: f.Mail},
bonfire.Detail{Label: "Cache", Value: f.Cache},
bonfire.Detail{Label: "Realtime", Value: f.Realtime},
bonfire.Detail{Label: "Search", Value: f.Search},
)
}
// attrs are the attributes of the "startup summary" record.
func (f summaryFacts) attrs() []slog.Attr {
return []slog.Attr{
slog.String("app", f.Header),
slog.String("environment", f.Environment),
slog.String("url", f.URL),
slog.String("admin", f.Admin),
slog.String("database", f.Database),
slog.Int("migrations_pending", f.Pending),
slog.Int("plugins", len(f.Plugins)),
slog.String("plugin_ids", strings.Join(f.pluginIDs(), ",")),
slog.String("queue", f.Queue),
slog.String("scheduler", f.Scheduler),
slog.String("mail", f.Mail),
slog.String("cache", f.Cache),
slog.String("realtime", f.Realtime),
slog.String("search", f.Search),
}
}
// printSummary prints the block and the admin notes through out.
func printSummary(out bonfire.Output, f summaryFacts) {
bonfire.Details(out, f.Header, f.details(bonfire.VerbosityOf(out)))
printBanner(out, f.AdminNotes)
}
// logSummary logs one INFO record "startup summary" with every fact, one
// whose message is exactly "listening on <addr>" (application smoke scripts
// wait for it) and the admin notes (warnings at WARN).
func logSummary(ctx context.Context, log *slog.Logger, f summaryFacts) {
log.LogAttrs(ctx, slog.LevelInfo, summaryMessage, f.attrs()...)
log.LogAttrs(ctx, slog.LevelInfo, listenPrefix+f.Addr, slog.String("url", f.URL))
for _, n := range f.AdminNotes {
level := slog.LevelInfo
if n.warn {
level = slog.LevelWarn
}
log.LogAttrs(ctx, level, n.text)
}
}
// shutdownReport is what serve did, printed and logged after a signal.
type shutdownReport struct {
// Draining is the number of requests in flight when the signal came.
Draining int64
// Aborted is the number still running at the drain deadline.
Aborted int64
Served int64
Worker *conga.WorkerStats
Uptime time.Duration
Errors []error
}
// Drained is the number of in-flight requests that finished in time.
func (r shutdownReport) Drained() int64 {
return max(0, r.Draining-r.Aborted)
}
// Err joins the drain errors; nil means a clean stop (exit 0).
func (r shutdownReport) Err() error {
return errors.Join(r.Errors...)
}
func (r shutdownReport) requestsLine() string {
return fmt.Sprintf("%d served, %d drained, %d aborted", r.Served, r.Drained(), r.Aborted)
}
func workerLine(s *conga.WorkerStats) string {
if s == nil {
return "not running in serve"
}
return fmt.Sprintf("%d done, %d failed, %d handed back to the queue", s.Done, s.Failed, s.HandedBack+s.InFlight)
}
func errorsLine(errs []error) string {
if len(errs) == 0 {
return "none"
}
parts := make([]string, 0, len(errs))
for _, e := range errs {
parts = append(parts, e.Error())
}
return strings.Join(parts, "; ")
}
// uptime rounds d for the report.
func uptime(d time.Duration) string {
switch {
case d >= time.Minute:
return d.Round(time.Second).String()
case d >= time.Second:
return d.Round(100 * time.Millisecond).String()
default:
return d.Round(time.Millisecond).String()
}
}
// details is the report as label/value lines.
func (r shutdownReport) details() []bonfire.Detail {
return []bonfire.Detail{
{Label: "Requests", Value: r.requestsLine(), Warn: r.Aborted > 0},
{Label: "Worker", Value: workerLine(r.Worker)},
{Label: "Uptime", Value: uptime(r.Uptime)},
{Label: "Errors", Value: errorsLine(r.Errors), Warn: len(r.Errors) > 0},
}
}
// attrs are the attributes of the "shutdown report" record.
func (r shutdownReport) attrs() []slog.Attr {
attrs := []slog.Attr{
slog.Int64("served", r.Served),
slog.Int64("drained", r.Drained()),
slog.Int64("aborted", r.Aborted),
slog.Int64("uptime_ms", r.Uptime.Milliseconds()),
}
if r.Worker != nil {
attrs = append(attrs,
slog.Int64("jobs_done", r.Worker.Done),
slog.Int64("jobs_failed", r.Worker.Failed),
slog.Int64("jobs_handed_back", r.Worker.HandedBack+r.Worker.InFlight),
)
}
if len(r.Errors) > 0 {
attrs = append(attrs, slog.String("errors", errorsLine(r.Errors)))
}
return attrs
}
// logReport logs the report as one record: INFO, or ERROR with drain errors.
func logReport(ctx context.Context, log *slog.Logger, r shutdownReport) {
level := slog.LevelInfo
if len(r.Errors) > 0 {
level = slog.LevelError
}
log.LogAttrs(ctx, level, reportMessage, r.attrs()...)
}