Files
felhom.eu/hub/internal/store/timeddriver.go
T

198 lines
5.6 KiB
Go

package store
import (
"context"
"database/sql"
"database/sql/driver"
"log"
"reflect"
"strings"
"sync/atomic"
"time"
)
// R-1346 (2026-10-30): the per-statement timing. The access log names a slow REQUEST; this names the slow STATEMENT
// inside it. Every Exec, Query (until its rows are closed) and BEGIN that takes longer than slowStatementAfter is
// logged once as
//
// [WARN] slow statement: 1234ms (over 1s): SELECT DISTINCT controller_version FROM reports WHERE …
//
// — the SQL text only (whitespace folded, at most 300 characters), NEVER the bound values: they can be customer data.
// A wrapper around modernc's driver, registered once as "sqlite-timed"; store.New opens it. Pinned by
// TestTimedDriver_ASlowStatementIsNamed and TestTimedDriver_NeverLogsValues.
const timedDriverName = "sqlite-timed"
// slowStatementAfter is the threshold (nanoseconds); SetSlowStatementThreshold changes it for a test or a probe.
var slowStatementAfter atomic.Int64
// SetSlowStatementThreshold changes the threshold (a test, the R-1346 probe). Not called by the hub itself.
func SetSlowStatementThreshold(d time.Duration) { slowStatementAfter.Store(int64(d)) }
// statementLogger is where the line goes (the store's logger, set by New; the standard logger until then).
var statementLogger atomic.Pointer[log.Logger]
func init() {
slowStatementAfter.Store(int64(time.Second))
// The SAME driver value modernc registered as "sqlite" (its connection hooks live on it); sql.Open connects nothing.
db, err := sql.Open("sqlite", "")
if err != nil {
panic("store: the sqlite driver is not registered: " + err.Error())
}
sql.Register(timedDriverName, &timedDriver{inner: db.Driver()})
_ = db.Close()
}
func logSlowStatement(kind, query string, took time.Duration) {
after := time.Duration(slowStatementAfter.Load())
if took < after {
return
}
q := strings.Join(strings.Fields(query), " ")
if len(q) > 300 {
q = q[:300] + "…"
}
if kind != "" {
q = kind + " " + q
}
l := statementLogger.Load()
if l == nil {
l = log.Default()
}
l.Printf("[WARN] slow statement: %dms (over %s): %s", took.Milliseconds(), after, q)
}
type timedDriver struct{ inner driver.Driver }
func (d *timedDriver) Open(name string) (driver.Conn, error) {
c, err := d.inner.Open(name)
if err != nil {
return nil, err
}
return &timedConn{Conn: c}, nil
}
type timedConn struct{ driver.Conn }
func (c *timedConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
ex, ok := c.Conn.(driver.ExecerContext)
if !ok {
return nil, driver.ErrSkip
}
began := time.Now()
r, err := ex.ExecContext(ctx, query, args)
logSlowStatement("", query, time.Since(began))
return r, err
}
func (c *timedConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
q, ok := c.Conn.(driver.QueryerContext)
if !ok {
return nil, driver.ErrSkip
}
began := time.Now()
r, err := q.QueryContext(ctx, query, args)
if err != nil {
logSlowStatement("", query, time.Since(began))
return nil, err
}
return &timedRows{Rows: r, query: query, began: began}, nil
}
func (c *timedConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
if p, ok := c.Conn.(driver.ConnPrepareContext); ok {
return p.PrepareContext(ctx, query)
}
return c.Conn.Prepare(query)
}
// BeginTx: with _txlock=immediate the BEGIN itself waits for the write lock — a slow one is a writer queue, named so.
func (c *timedConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
began := time.Now()
var (
tx driver.Tx
err error
)
if b, ok := c.Conn.(driver.ConnBeginTx); ok {
tx, err = b.BeginTx(ctx, opts)
} else {
tx, err = c.Conn.Begin() //nolint:staticcheck // the fallback database/sql itself uses
}
logSlowStatement("", "BEGIN (waiting for the write lock)", time.Since(began))
return tx, err
}
func (c *timedConn) Ping(ctx context.Context) error {
if p, ok := c.Conn.(driver.Pinger); ok {
return p.Ping(ctx)
}
return nil
}
func (c *timedConn) ResetSession(ctx context.Context) error {
if r, ok := c.Conn.(driver.SessionResetter); ok {
return r.ResetSession(ctx)
}
return nil
}
func (c *timedConn) IsValid() bool {
if v, ok := c.Conn.(driver.Validator); ok {
return v.IsValid()
}
return true
}
// timedRows times a query until its rows are closed — a GROUP BY or a DISTINCT does its work before the first row, a
// plain scan while the rows are read; both count.
type timedRows struct {
driver.Rows
query string
began time.Time
closed bool
}
func (r *timedRows) Close() error {
err := r.Rows.Close()
if !r.closed {
r.closed = true
logSlowStatement("", r.query, time.Since(r.began))
}
return err
}
func (r *timedRows) ColumnTypeDatabaseTypeName(i int) string {
if c, ok := r.Rows.(driver.RowsColumnTypeDatabaseTypeName); ok {
return c.ColumnTypeDatabaseTypeName(i)
}
return ""
}
func (r *timedRows) ColumnTypeLength(i int) (int64, bool) {
if c, ok := r.Rows.(driver.RowsColumnTypeLength); ok {
return c.ColumnTypeLength(i)
}
return 0, false
}
func (r *timedRows) ColumnTypeNullable(i int) (bool, bool) {
if c, ok := r.Rows.(driver.RowsColumnTypeNullable); ok {
return c.ColumnTypeNullable(i)
}
return false, false
}
func (r *timedRows) ColumnTypePrecisionScale(i int) (int64, int64, bool) {
if c, ok := r.Rows.(driver.RowsColumnTypePrecisionScale); ok {
return c.ColumnTypePrecisionScale(i)
}
return 0, 0, false
}
func (r *timedRows) ColumnTypeScanType(i int) reflect.Type {
if c, ok := r.Rows.(driver.RowsColumnTypeScanType); ok {
return c.ColumnTypeScanType(i)
}
return reflect.TypeOf(new(any)).Elem()
}