feat(conga,lagoon): River and GORM log through diary components
- River's logger is diary.For("river") (warn by default), conga's own lines diary.For("conga"); an app-published logger still wins
- GORM logs through logger.NewSlogLogger(diary.For("db")): warn level, record-not-found ignored, parameterized SQL traces only at verbosity 3
- Smoke tests TestRiverLoggerUsesDiarySmoke and TestGormLoggerUsesDiarySmoke
This commit is contained in:
@@ -38,6 +38,7 @@ Use cases:
|
||||
- Scheduled commands: every worker carries one River periodic job per `pact.HasSchedule` entry, in plugin activation order then declaration order, with the id `<plugin id>[<index>]:<command>`. Only the elected leader enqueues. Each run is a `conga.ScheduledCommandArgs` job on `conga.QueueScheduled` with one attempt (an interrupted run is not retried; the next period runs normally) and unique by args within its cadence period, so a leader failover cannot double-enqueue a period. The worker runs a job only when its entry exists in the compiled schedule and its command and arguments match that entry exactly, so a forged `river_job` row cannot run an arbitrary command. An unregistered command, or an app with no published `bonfire.Catalog`, is logged at Warn (`schedule: command not registered; skipping`) and skipped without failing the worker or other entries. Command output is logged line by line at Info with a `command` attribute; failures are logged with the duration.
|
||||
- Framework schedule entries: before the plugin entries, every worker carries the framework's own entries from `lagoon.FrameworkSchedule`, listed under the plugin id `lagoon.FrameworkScheduleID`. Today that is the daily `deferred:purge` at `database.deferred_bindings.purge_at` (default `03:00`), with the id `summercms.lagoon[0]:deferred:purge`; it runs, appears in `schedule:run --once` and is protected by the compiled entry table exactly like a plugin entry. An empty `purge_at` removes it, and a malformed one fails the worker start.
|
||||
- Wall-clock schedules: `conga.Daily` fires at the next `hh:mm` in its location and `conga.Every` at the next multiple of its interval since local midnight, so a restart never delays a daily run by up to a day the way `river.PeriodicInterval(24h)` would. On a DST day `conga.Daily` keeps the wall-clock time. A schedule entry with an empty command, a zero cadence, a daily time out of range, or an interval under one second or not dividing 24h fails the worker start with an error naming the plugin id and entry index.
|
||||
- Logging: River logs through [diary](../diary/README.md) as component `river`, which the framework default `log.components.river: warn` keeps at warn, and conga's own lines (scheduled runs, job failures) log as component `conga`. An application that publishes its own `*slog.Logger` on the `backpack.App` gets both instead.
|
||||
- Commands: `conga.RuntimeCommands` adds `queue:work`, `schedule:run` and `queue:clear` to the application binary. `schedule:run` runs a scheduler-only worker in its own process, and `schedule:run --once` runs the entries due in the current minute without River, for system cron.
|
||||
|
||||
## Usage
|
||||
@@ -228,7 +229,7 @@ queues:
|
||||
|
||||
## Dependencies
|
||||
|
||||
- SummerCMS modules: [backpack](../backpack/README.md), [bonfire](../bonfire/README.md) (commands and the `bonfire.Catalog` scheduled runs call), [bouncer](../bouncer/README.md) (the dispatching principal), [lagoon](../lagoon/README.md) (the shared pool, `lagoon.JobsTable` and the migrations that create River's schema and `summer_jobs`), [pact](../pact/README.md), [party](../party/README.md).
|
||||
- SummerCMS modules: [backpack](../backpack/README.md), [bonfire](../bonfire/README.md) (commands and the `bonfire.Catalog` scheduled runs call), [bouncer](../bouncer/README.md) (the dispatching principal), [diary](../diary/README.md) (the river and conga component loggers), [lagoon](../lagoon/README.md) (the shared pool, `lagoon.JobsTable` and the migrations that create River's schema and `summer_jobs`), [pact](../pact/README.md), [party](../party/README.md).
|
||||
- Third-party: `github.com/riverqueue/river` v0.47.0 with its `riverdriver/riverdatabasesql` and `rivertype` modules. River is the Postgres job queue the project stack names: it gives transactional inserts on the shared `*sql.DB`, retries, stuck-job rescue, leader election for periodic work and `LISTEN`/`NOTIFY` wake-ups. `github.com/jackc/pgx/v5/pgxpool` opens the listener pool; `gorm.io/gorm` writes the record rows.
|
||||
|
||||
## Testing
|
||||
@@ -237,4 +238,4 @@ queues:
|
||||
go test ./modules/conga/...
|
||||
```
|
||||
|
||||
The tests start a `postgres:16-alpine` container through testcontainers-go and migrate a fresh database per test with `lagoon.Migrate`, so they need a running Docker daemon. `TestListenPickupLatency` sets a 30-second poll interval and checks that a job committed by a separate client is picked up in under one second, while a poll-only worker does not pick it up within two. `TestScheduleRunsCommand` checks that an `Every(1s)` entry runs its command through a periodic job and that an unregistered command is skipped with a Warn log. `go test -short ./modules/conga/...` skips the database tests.
|
||||
The tests start a `postgres:16-alpine` container through testcontainers-go and migrate a fresh database per test with `lagoon.Migrate`, so they need a running Docker daemon. `TestListenPickupLatency` sets a 30-second poll interval and checks that a job committed by a separate client is picked up in under one second, while a poll-only worker does not pick it up within two. `TestScheduleRunsCommand` checks that an `Every(1s)` entry runs its command through a periodic job and that an unregistered command is skipped with a Warn log. `go test -short ./modules/conga/...` skips the database tests; `TestRiverLoggerUsesDiarySmoke` runs without them and checks that River logs as component `river` unless the application publishes a logger.
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
"git.golem15.com/golem15/summercms/modules/diary"
|
||||
"github.com/riverqueue/river"
|
||||
"github.com/riverqueue/river/riverdriver/riverdatabasesql"
|
||||
)
|
||||
@@ -113,11 +114,37 @@ func newInsertClient(sqlDB *sql.DB, s settings, log *slog.Logger) (*river.Client
|
||||
return river.NewClient(riverdatabasesql.New(sqlDB), baseConfig(s, log))
|
||||
}
|
||||
|
||||
func loggerFromApp(app *backpack.App) *slog.Logger {
|
||||
// moduleLog is the logger of conga's own lines (component conga).
|
||||
var moduleLog = diary.For("conga")
|
||||
|
||||
// riverLog is River's logger (component river). The framework default
|
||||
// log.components.river: warn keeps River's info lines out of the console.
|
||||
var riverLog = diary.For("river")
|
||||
|
||||
// publishedLogger returns the *slog.Logger the application published, if any.
|
||||
func publishedLogger(app *backpack.App) (*slog.Logger, bool) {
|
||||
if app != nil {
|
||||
if log, ok := app.Lookup[*slog.Logger](); ok && log != nil {
|
||||
return log
|
||||
return log, true
|
||||
}
|
||||
}
|
||||
return slog.Default()
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// loggerFromApp is the logger of conga's own lines: the logger the
|
||||
// application published, else the conga component logger.
|
||||
func loggerFromApp(app *backpack.App) *slog.Logger {
|
||||
if log, ok := publishedLogger(app); ok {
|
||||
return log
|
||||
}
|
||||
return moduleLog
|
||||
}
|
||||
|
||||
// riverLogger is the logger handed to River: the logger the application
|
||||
// published, else diary.For("river").
|
||||
func riverLogger(app *backpack.App) *slog.Logger {
|
||||
if log, ok := publishedLogger(app); ok {
|
||||
return log
|
||||
}
|
||||
return riverLog
|
||||
}
|
||||
|
||||
@@ -514,7 +514,7 @@ func (m *Manager) insertOnlyLocked() (*river.Client[*sql.Tx], error) {
|
||||
if !ok || sqlDB == nil {
|
||||
return nil, ErrNoDatabase
|
||||
}
|
||||
c, err := newInsertClient(sqlDB, settingsFromApp(m.app), loggerFromApp(m.app))
|
||||
c, err := newInsertClient(sqlDB, settingsFromApp(m.app), riverLogger(m.app))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("conga: river client: %w", err)
|
||||
}
|
||||
|
||||
45
modules/conga/logger_smoke_test.go
Normal file
45
modules/conga/logger_smoke_test.go
Normal file
@@ -0,0 +1,45 @@
|
||||
package conga
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/backpack"
|
||||
"git.golem15.com/golem15/summercms/modules/bonfire"
|
||||
"git.golem15.com/golem15/summercms/modules/compass"
|
||||
"git.golem15.com/golem15/summercms/modules/diary"
|
||||
)
|
||||
|
||||
// TestRiverLoggerUsesDiarySmoke: River logs through the diary component
|
||||
// river unless the application publishes its own logger, and conga's own
|
||||
// lines use component conga.
|
||||
func TestRiverLoggerUsesDiarySmoke(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
t.Cleanup(diary.Reset)
|
||||
diary.Bootstrap(&buf, diary.Options{Color: bonfire.ColorNever})
|
||||
|
||||
riverLogger(nil).Warn("probe")
|
||||
if !strings.Contains(buf.String(), "component=river") {
|
||||
t.Fatalf("River record lacks component=river:\n%s", buf.String())
|
||||
}
|
||||
loggerFromApp(nil).Warn("own line")
|
||||
if !strings.Contains(buf.String(), "component=conga") {
|
||||
t.Fatalf("conga record lacks component=conga:\n%s", buf.String())
|
||||
}
|
||||
|
||||
var capture bytes.Buffer
|
||||
app := backpack.New(&compass.Config{})
|
||||
if err := app.Publish(slog.New(slog.NewTextHandler(&capture, nil))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
buf.Reset()
|
||||
riverLogger(app).Warn("published probe")
|
||||
if !strings.Contains(capture.String(), "published probe") {
|
||||
t.Fatalf("published logger did not receive the River record:\n%s", capture.String())
|
||||
}
|
||||
if strings.Contains(buf.String(), "published probe") {
|
||||
t.Fatalf("River record reached diary although the app published a logger:\n%s", buf.String())
|
||||
}
|
||||
}
|
||||
@@ -81,7 +81,7 @@ func StartWorker(ctx context.Context, app *backpack.App, plugins []party.Plugin,
|
||||
return nil, ErrNoDatabase
|
||||
}
|
||||
s := settingsFromApp(app)
|
||||
log := loggerFromApp(app)
|
||||
log := riverLogger(app)
|
||||
periodic, table, err := periodicJobs(app, plugins)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
Reference in New Issue
Block a user