Read queue depths with Find, not Scan (closes #234) #237
@@ -24,11 +24,24 @@ func NewTestDatabase(db *gorm.DB) *Database {
|
|||||||
// NewTestWebhookDBManager creates a WebhookDBManager backed by the given
|
// NewTestWebhookDBManager creates a WebhookDBManager backed by the given
|
||||||
// data directory. Intended for use in tests without the fx lifecycle.
|
// data directory. Intended for use in tests without the fx lifecycle.
|
||||||
func NewTestWebhookDBManager(dataDir string) *WebhookDBManager {
|
func NewTestWebhookDBManager(dataDir string) *WebhookDBManager {
|
||||||
return &WebhookDBManager{
|
return NewTestWebhookDBManagerWithLogger(
|
||||||
dataDir: dataDir,
|
dataDir,
|
||||||
log: slog.New(slog.NewTextHandler(
|
slog.New(slog.NewTextHandler(
|
||||||
os.Stderr,
|
os.Stderr,
|
||||||
&slog.HandlerOptions{Level: slog.LevelDebug},
|
&slog.HandlerOptions{Level: slog.LevelDebug},
|
||||||
)),
|
)),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewTestWebhookDBManagerWithLogger is NewTestWebhookDBManager with the
|
||||||
|
// logger supplied by the caller. The per-webhook databases this manager
|
||||||
|
// opens hand that logger to gormlog, so a test that needs to see the SQL
|
||||||
|
// the service emits can capture it.
|
||||||
|
func NewTestWebhookDBManagerWithLogger(
|
||||||
|
dataDir string, log *slog.Logger,
|
||||||
|
) *WebhookDBManager {
|
||||||
|
return &WebhookDBManager{
|
||||||
|
dataDir: dataDir,
|
||||||
|
log: log,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -95,6 +95,8 @@ func (e *Engine) sampleQueueDepths(ctx context.Context) {
|
|||||||
// targetTypesByID maps every configured target id to its type. The
|
// targetTypesByID maps every configured target id to its type. The
|
||||||
// deliveries live in the per-webhook databases but carry only a
|
// deliveries live in the per-webhook databases but carry only a
|
||||||
// target id, so the type label has to come from the main database.
|
// target id, so the type label has to come from the main database.
|
||||||
|
//
|
||||||
|
// Find rather than Scan: see sampleWebhookQueueDepths.
|
||||||
func (e *Engine) targetTypesByID() (
|
func (e *Engine) targetTypesByID() (
|
||||||
map[string]database.TargetType, error,
|
map[string]database.TargetType, error,
|
||||||
) {
|
) {
|
||||||
@@ -106,7 +108,7 @@ func (e *Engine) targetTypesByID() (
|
|||||||
err := e.database.DB().
|
err := e.database.DB().
|
||||||
Model(&database.Target{}).
|
Model(&database.Target{}).
|
||||||
Select("id", "type").
|
Select("id", "type").
|
||||||
Scan(&rows).Error
|
Find(&rows).Error
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("loading targets: %w", err)
|
return nil, fmt.Errorf("loading targets: %w", err)
|
||||||
}
|
}
|
||||||
@@ -128,6 +130,13 @@ func (e *Engine) targetTypesByID() (
|
|||||||
// folds that into the unknown series rather than dropping it: a
|
// folds that into the unknown series rather than dropping it: a
|
||||||
// backlog stuck behind a deleted target is a backlog that still needs
|
// backlog stuck behind a deleted target is a backlog that still needs
|
||||||
// to be alertable.
|
// to be alertable.
|
||||||
|
//
|
||||||
|
// The aggregate is read with Find, not Scan. (*gorm.DB).Scan swaps
|
||||||
|
// GORM's own trace recorder in for the logging adapter, and that
|
||||||
|
// recorder does not implement gorm.ParamsFilter, so the statement
|
||||||
|
// reaches the log with its bound values interpolated — here, the
|
||||||
|
// status list. Find goes through the normal query callback, which is
|
||||||
|
// filtered. See internal/gormlog and its scan_guard_test.go.
|
||||||
func (e *Engine) sampleWebhookQueueDepths(
|
func (e *Engine) sampleWebhookQueueDepths(
|
||||||
webhookID string,
|
webhookID string,
|
||||||
types map[string]database.TargetType,
|
types map[string]database.TargetType,
|
||||||
@@ -158,7 +167,7 @@ func (e *Engine) sampleWebhookQueueDepths(
|
|||||||
database.DeliveryStatusRetrying,
|
database.DeliveryStatusRetrying,
|
||||||
}).
|
}).
|
||||||
Group("target_id, status").
|
Group("target_id, status").
|
||||||
Scan(&rows).Error
|
Find(&rows).Error
|
||||||
if err != nil {
|
if err != nil {
|
||||||
e.log.Error(
|
e.log.Error(
|
||||||
"queue depth sample: "+
|
"queue depth sample: "+
|
||||||
|
|||||||
179
internal/delivery/queue_depth_gormlog_test.go
Normal file
179
internal/delivery/queue_depth_gormlog_test.go
Normal file
@@ -0,0 +1,179 @@
|
|||||||
|
package delivery_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/google/uuid"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"gorm.io/driver/sqlite"
|
||||||
|
"gorm.io/gorm"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
|
"sneak.berlin/go/webhooker/internal/delivery"
|
||||||
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
|
)
|
||||||
|
|
||||||
|
// qdAggregateMarker identifies the queue-depth aggregate in the
|
||||||
|
// captured SQL. It is the one statement in this test that binds
|
||||||
|
// anything, and the raw count() expression appears in no other.
|
||||||
|
const qdAggregateMarker = "count(*)"
|
||||||
|
|
||||||
|
// qdSyncBuf collects log output from whichever goroutine GORM writes
|
||||||
|
// on.
|
||||||
|
type qdSyncBuf struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
b bytes.Buffer
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *qdSyncBuf) Write(p []byte) (int, error) {
|
||||||
|
q.mu.Lock()
|
||||||
|
defer q.mu.Unlock()
|
||||||
|
|
||||||
|
return q.b.Write(p)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *qdSyncBuf) String() string {
|
||||||
|
q.mu.Lock()
|
||||||
|
defer q.mu.Unlock()
|
||||||
|
|
||||||
|
return q.b.String()
|
||||||
|
}
|
||||||
|
|
||||||
|
// qdMainDB opens a main database whose GORM logger is the service's
|
||||||
|
// adapter, writing through log.
|
||||||
|
func qdMainDB(t *testing.T, log *slog.Logger) *gorm.DB {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
dsn := fmt.Sprintf(
|
||||||
|
"file:%s?cache=shared&mode=rwc",
|
||||||
|
filepath.Join(t.TempDir(), "main-gormlog.db"),
|
||||||
|
)
|
||||||
|
|
||||||
|
sqlDB, err := sql.Open("sqlite", dsn)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||||
|
|
||||||
|
db, err := gorm.Open(
|
||||||
|
sqlite.Dialector{Conn: sqlDB},
|
||||||
|
&gorm.Config{Logger: gormlog.New(log)},
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
require.NoError(t, db.AutoMigrate(
|
||||||
|
&database.Webhook{},
|
||||||
|
&database.Target{},
|
||||||
|
))
|
||||||
|
|
||||||
|
return db
|
||||||
|
}
|
||||||
|
|
||||||
|
// qdLinesContaining returns every captured line carrying marker.
|
||||||
|
func qdLinesContaining(out, marker string) []string {
|
||||||
|
var found []string
|
||||||
|
|
||||||
|
for line := range strings.SplitSeq(out, "\n") {
|
||||||
|
if strings.Contains(line, marker) {
|
||||||
|
found = append(found, line)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return found
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestQueueDepthSample_LogsNoBoundValue holds the queue-depth sampler
|
||||||
|
// to the values-off property internal/gormlog exists to provide.
|
||||||
|
//
|
||||||
|
// The aggregate binds the delivery status list. Read with
|
||||||
|
// (*gorm.DB).Scan it was logged with those values interpolated, because
|
||||||
|
// Scan records the statement through GORM's own traceRecorder, which
|
||||||
|
// does not implement gorm.ParamsFilter. Read with Find it goes through
|
||||||
|
// the normal query callback and the adapter's filter applies. Restore
|
||||||
|
// the Scan call in queue_depth.go and this fails on the status literals
|
||||||
|
// below; scan_guard_test.go catches the same regression statically.
|
||||||
|
func TestQueueDepthSample_LogsNoBoundValue(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
buf := &qdSyncBuf{}
|
||||||
|
log := slog.New(slog.NewTextHandler(
|
||||||
|
buf, &slog.HandlerOptions{Level: slog.LevelDebug},
|
||||||
|
))
|
||||||
|
|
||||||
|
mainDB := qdMainDB(t, log)
|
||||||
|
dbMgr := database.NewTestWebhookDBManagerWithLogger(
|
||||||
|
t.TempDir(), log,
|
||||||
|
)
|
||||||
|
|
||||||
|
webhookID := uuid.New().String()
|
||||||
|
webhookDB := iSeedWebhookDB(t, dbMgr, webhookID)
|
||||||
|
|
||||||
|
iCreateWebhook(t, mainDB, webhookID, "queue-depth-gormlog")
|
||||||
|
|
||||||
|
targetID := uuid.New().String()
|
||||||
|
|
||||||
|
iCreateTarget(t, mainDB, targetID, webhookID,
|
||||||
|
"queue-depth-gormlog-target", database.TargetTypeHTTP,
|
||||||
|
iHTTPConfig("https://example.com/hook"), 3,
|
||||||
|
)
|
||||||
|
|
||||||
|
event := iSeedEvent(
|
||||||
|
t, webhookDB, webhookID, `{"queued":true}`,
|
||||||
|
)
|
||||||
|
|
||||||
|
iSeedDelivery(
|
||||||
|
t, webhookDB, event.ID, targetID,
|
||||||
|
database.DeliveryStatusPending,
|
||||||
|
)
|
||||||
|
iSeedDelivery(
|
||||||
|
t, webhookDB, event.ID, targetID,
|
||||||
|
database.DeliveryStatusRetrying,
|
||||||
|
)
|
||||||
|
|
||||||
|
engine := delivery.NewTestEngineWithDB(
|
||||||
|
database.NewTestDatabase(mainDB),
|
||||||
|
dbMgr,
|
||||||
|
log,
|
||||||
|
&http.Client{Timeout: 5 * time.Second},
|
||||||
|
2,
|
||||||
|
)
|
||||||
|
|
||||||
|
engine.ExportSampleQueueDepths(context.Background())
|
||||||
|
|
||||||
|
out := buf.String()
|
||||||
|
|
||||||
|
lines := qdLinesContaining(out, qdAggregateMarker)
|
||||||
|
require.NotEmpty(
|
||||||
|
t, lines,
|
||||||
|
"the queue-depth aggregate was never logged, so the "+
|
||||||
|
"assertions below are vacuous",
|
||||||
|
)
|
||||||
|
|
||||||
|
for _, line := range lines {
|
||||||
|
assert.Contains(
|
||||||
|
t, line, "?",
|
||||||
|
"the aggregate was logged without its placeholders: %s",
|
||||||
|
line,
|
||||||
|
)
|
||||||
|
|
||||||
|
for _, status := range []database.DeliveryStatus{
|
||||||
|
database.DeliveryStatusPending,
|
||||||
|
database.DeliveryStatusRetrying,
|
||||||
|
} {
|
||||||
|
assert.NotContains(
|
||||||
|
t, line, string(status),
|
||||||
|
"a bound status value was interpolated into the "+
|
||||||
|
"logged statement: %s", line,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user