Discharge7443/fleet
1
0
Fork 0
Code Issues Pull Requests Actions 6 Packages Projects Releases Wiki Activity

slog migration: platform/mysql and related logic (#40072)

Browse Source 
<!-- Add the related story/sub-task/bug number, like Resolves #123, or
remove if NA -->
**Related issue:** Resolves #40054 

# Checklist for submitter

- [ ] Changes file added for user-visible changes in `changes/`,
`orbit/changes/` or `ee/fleetd-chrome/changes`.
  - already included in previous PR

## Testing

- [x] Added/updated automated tests
- [x] QA'd all new/changed functionality manually


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Refactor**
* Standardized logging infrastructure across the database and storage
layers for improved consistency and maintainability.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Victor Lyuboslavsky
2026-02-19 08:27:24 -06:00
committed by GitHub
parent 20049689ea
commit 6b3bb8a961
21 changed files with 112 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/fleet/serve.go
+1 -1
View File
@@ -1502,7 +1502,7 @@ the way that the Fleet server works.
}
// Host identify and conditional access SCEP feature only works if a private key has been set up
if len(config.Server.PrivateKey) > 0 {
hostIdentitySCEPDepot, err := mds.NewHostIdentitySCEPDepot(logger.With("component", "host-id-scep-depot"), &config)
hostIdentitySCEPDepot, err := mds.NewHostIdentitySCEPDepot(logger.SlogLogger().With("component", "host-id-scep-depot"), &config)
if err != nil {
initFatal(err, "setup host identity SCEP depot")
}
ee/server/integrationtest/hostidentity/suite.go
+1 -1
View File
@@ -57,7 +57,7 @@ func SetUpSuiteWithConfig(t *testing.T, uniqueTestName string, requireSignature
}
logger := logging.NewLogfmtLogger(os.Stdout)
hostIdentitySCEPDepot, err := ds.NewHostIdentitySCEPDepot(logger.With("component", "host-id-scep-depot"), &fleetCfg)
hostIdentitySCEPDepot, err := ds.NewHostIdentitySCEPDepot(logger.SlogLogger().With("component", "host-id-scep-depot"), &fleetCfg)
require.NoError(t, err)
users, server := service.RunServerForTestsWithServiceWithDS(t, ctx, ds, fleetSvc, &service.TestServerOpts{
License: license,
ee/server/service/hostidentity/depot/depot.go
+4 -4
View File
@@ -7,6 +7,7 @@ import (
"crypto/x509"
"errors"
"fmt"
"log/slog"
"math/big"
"time"
@@ -19,7 +20,6 @@ import (
"github.com/fleetdm/fleet/v4/server/mdm/assets"
"github.com/fleetdm/fleet/v4/server/mdm/scep/depot"
common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
"github.com/go-kit/log"
"github.com/jmoiron/sqlx"
)
@@ -29,14 +29,14 @@ const maxCommonNameLength = 255
type HostIdentitySCEPDepot struct {
db *sqlx.DB
ds fleet.Datastore
logger log.Logger
logger *slog.Logger
config *config.FleetConfig
}
var _ depot.Depot = (*HostIdentitySCEPDepot)(nil)
// NewHostIdentitySCEPDepot creates and returns a *HostIdentitySCEPDepot.
func NewHostIdentitySCEPDepot(db *sqlx.DB, ds fleet.Datastore, logger log.Logger, cfg *config.FleetConfig) (*HostIdentitySCEPDepot, error) {
func NewHostIdentitySCEPDepot(db *sqlx.DB, ds fleet.Datastore, logger *slog.Logger, cfg *config.FleetConfig) (*HostIdentitySCEPDepot, error) {
if err := db.Ping(); err != nil {
return nil, err
}
@@ -135,7 +135,7 @@ func (d *HostIdentitySCEPDepot) Put(name string, crt *x509.Certificate) error {
}
rowsAffected, _ := result.RowsAffected()
if rowsAffected > 0 {
d.logger.Log("msg", "revoked existing host identity certificate", "name", name)
d.logger.InfoContext(context.Background(), "revoked existing host identity certificate", "name", name)
}
_, err = tx.ExecContext(context.Background(), `
server/activity/internal/testutils/testutils.go
+1 -2
View File
@@ -9,7 +9,6 @@ import (
common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
mysql_testing_utils "github.com/fleetdm/fleet/v4/server/platform/mysql/testing_utils"
kitlog "github.com/go-kit/log"
"github.com/jmoiron/sqlx"
"github.com/stretchr/testify/require"
)
@@ -49,7 +48,7 @@ func (tdb *TestDB) Conns() *common_mysql.DBConnections {
// TruncateTables clears the tables used by activity bounded context.
func (tdb *TestDB) TruncateTables(t *testing.T) {
t.Helper()
mysql_testing_utils.TruncateTables(t, tdb.DB, kitlog.NewNopLogger(), nil, "host_activities", "activities", "hosts", "users")
mysql_testing_utils.TruncateTables(t, tdb.DB, tdb.Logger, nil, "host_activities", "activities", "hosts", "users")
}
// InsertUser creates a user in the database and returns the user ID.
server/datastore/mysql/android_device_test.go
+1 -1
View File
@@ -30,7 +30,7 @@ func TestAndroidDevices(t *testing.T) {
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
defer testing_utils.TruncateTables(t, ds.primary, ds.logger, nil)
defer testing_utils.TruncateTables(t, ds.primary, ds.logger.SlogLogger(), nil)
c.fn(t, ds)
})
server/datastore/mysql/android_enterprise_test.go
+1 -1
View File
@@ -24,7 +24,7 @@ func TestAndroidEnterprises(t *testing.T) {
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
defer testing_utils.TruncateTables(t, ds.primary, ds.logger, nil)
defer testing_utils.TruncateTables(t, ds.primary, ds.logger.SlogLogger(), nil)
c.fn(t, ds)
})
server/datastore/mysql/android_mysql.go
+2 -2
View File
@@ -45,9 +45,9 @@ func (ds *AndroidDatastore) Writer(_ context.Context) *sqlx.DB {
}
func (ds *AndroidDatastore) WithRetryTxx(ctx context.Context, fn common_mysql.TxFn) (err error) {
return common_mysql.WithRetryTxx(ctx, ds.Writer(ctx), fn, ds.logger)
return common_mysql.WithRetryTxx(ctx, ds.Writer(ctx), fn, ds.logger.SlogLogger())
}
func (ds *AndroidDatastore) WithTxx(ctx context.Context, fn common_mysql.TxFn) (err error) {
return common_mysql.WithTxx(ctx, ds.Writer(ctx), fn, ds.logger)
return common_mysql.WithTxx(ctx, ds.Writer(ctx), fn, ds.logger.SlogLogger())
}
server/datastore/mysql/host_identity_scep.go
+1 -1
View File
@@ -37,7 +37,7 @@ func (ds *Datastore) GetHostIdentityCertBySerialNumber(ctx context.Context, seri
func (ds *Datastore) UpdateHostIdentityCertHostIDBySerial(ctx context.Context, serialNumber uint64, hostID uint) error {
return common_mysql.WithRetryTxx(ctx, ds.writer(ctx), func(tx sqlx.ExtContext) error {
return updateHostIdentityCertHostIDBySerial(ctx, tx, hostID, serialNumber)
}, ds.logger)
}, ds.logger.SlogLogger())
}
func updateHostIdentityCertHostIDBySerial(ctx context.Context, tx sqlx.ExtContext, hostID uint, serialNumber uint64) error {
server/datastore/mysql/mysql.go
+5 -4
View File
@@ -6,6 +6,7 @@ import (
"database/sql"
"errors"
"fmt"
"log/slog"
"net"
"os"
"regexp"
@@ -182,7 +183,7 @@ func (ds *Datastore) NewSCEPDepot() (scep_depot.Depot, error) {
// NewHostIdentitySCEPDepot returns a scep_depot.Depot for host identity certs that uses the Datastore
// underlying MySQL writer *sql.DB.
func (ds *Datastore) NewHostIdentitySCEPDepot(logger log.Logger, cfg *config.FleetConfig) (scep_depot.Depot, error) {
func (ds *Datastore) NewHostIdentitySCEPDepot(logger *slog.Logger, cfg *config.FleetConfig) (scep_depot.Depot, error) {
return hostidscepdepot.NewHostIdentitySCEPDepot(ds.primary, ds, logger, cfg)
}
@@ -206,12 +207,12 @@ var (
)
func (ds *Datastore) withRetryTxx(ctx context.Context, fn common_mysql.TxFn) (err error) {
return common_mysql.WithRetryTxx(ctx, ds.writer(ctx), fn, ds.logger)
return common_mysql.WithRetryTxx(ctx, ds.writer(ctx), fn, ds.logger.SlogLogger())
}
// withTx provides a common way to commit/rollback a txFn
func (ds *Datastore) withTx(ctx context.Context, fn common_mysql.TxFn) (err error) {
return common_mysql.WithTxx(ctx, ds.writer(ctx), fn, ds.logger)
return common_mysql.WithTxx(ctx, ds.writer(ctx), fn, ds.logger.SlogLogger())
}
// withReadTx runs fn in a read-only transaction with a consistent snapshot of the DB
@@ -224,7 +225,7 @@ func (ds *Datastore) withReadTx(ctx context.Context, fn common_mysql.ReadTxFn) (
if !ok {
return ctxerr.New(ctx, "failed to cast reader to *sqlx.DB")
}
return common_mysql.WithReadOnlyTxx(ctx, readerDB, fn, ds.logger)
return common_mysql.WithReadOnlyTxx(ctx, readerDB, fn, ds.logger.SlogLogger())
}
// NewDBConnections creates database connections from config.
server/datastore/mysql/nanomdm_storage.go
+6 -25
View File
@@ -6,6 +6,7 @@ import (
"database/sql"
"errors"
"fmt"
"log/slog"
"strings"
"time"
@@ -18,10 +19,7 @@ import (
"github.com/fleetdm/fleet/v4/server/mdm/nanomdm/mdm"
nanomdm_mysql "github.com/fleetdm/fleet/v4/server/mdm/nanomdm/storage/mysql"
common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/jmoiron/sqlx"
nanomdm_log "github.com/micromdm/nanolib/log"
)
// lockConflictError indicates a lock command already exists for the host
@@ -53,33 +51,16 @@ type NanoMDMStorage struct {
*nanomdm_mysql.MySQLStorage
db *sqlx.DB
logger log.Logger
logger *slog.Logger
ds fleet.Datastore
}
type nanoMDMLogAdapter struct {
logger log.Logger
}
func (l nanoMDMLogAdapter) Info(args ...interface{}) {
level.Info(l.logger).Log(args...)
}
func (l nanoMDMLogAdapter) Debug(args ...interface{}) {
level.Debug(l.logger).Log(args...)
}
func (l nanoMDMLogAdapter) With(args ...interface{}) nanomdm_log.Logger {
wl := log.With(l.logger, args...)
return nanoMDMLogAdapter{logger: wl}
}
// NewMDMAppleMDMStorage returns a MySQL nanomdm storage that uses the Datastore
// underlying MySQL writer *sql.DB.
func (ds *Datastore) NewMDMAppleMDMStorage() (*NanoMDMStorage, error) {
s, err := nanomdm_mysql.New(
nanomdm_mysql.WithDB(ds.primary.DB),
nanomdm_mysql.WithLogger(nanoMDMLogAdapter{logger: ds.logger}),
nanomdm_mysql.WithLogger(ds.logger.SlogLogger()),
nanomdm_mysql.WithReaderFunc(ds.reader),
)
if err != nil {
@@ -88,7 +69,7 @@ func (ds *Datastore) NewMDMAppleMDMStorage() (*NanoMDMStorage, error) {
return &NanoMDMStorage{
MySQLStorage: s,
db: ds.primary,
logger: ds.logger,
logger: ds.logger.SlogLogger(),
ds: ds,
}, nil
}
@@ -99,7 +80,7 @@ func (ds *Datastore) NewMDMAppleMDMStorage() (*NanoMDMStorage, error) {
func (ds *Datastore) NewTestMDMAppleMDMStorage(asyncCap int, asyncInterval time.Duration) (*NanoMDMStorage, error) {
s, err := nanomdm_mysql.New(
nanomdm_mysql.WithDB(ds.primary.DB),
nanomdm_mysql.WithLogger(nanoMDMLogAdapter{logger: ds.logger}),
nanomdm_mysql.WithLogger(ds.logger.SlogLogger()),
nanomdm_mysql.WithReaderFunc(ds.reader),
nanomdm_mysql.WithAsyncLastSeen(asyncCap, asyncInterval),
)
@@ -109,7 +90,7 @@ func (ds *Datastore) NewTestMDMAppleMDMStorage(asyncCap int, asyncInterval time.
return &NanoMDMStorage{
MySQLStorage: s,
db: ds.primary,
logger: ds.logger,
logger: ds.logger.SlogLogger(),
ds: ds,
}, nil
}
server/datastore/mysql/nanomdm_storage_test.go
+2 -2
View File
@@ -3,6 +3,7 @@ package mysql
import (
"context"
"fmt"
"log/slog"
"sync"
"testing"
"time"
@@ -11,7 +12,6 @@ import (
"github.com/fleetdm/fleet/v4/server/mdm/nanomdm/mdm"
"github.com/fleetdm/fleet/v4/server/ptr"
"github.com/fleetdm/fleet/v4/server/test"
"github.com/go-kit/log"
"github.com/google/uuid"
"github.com/stretchr/testify/require"
)
@@ -267,7 +267,7 @@ func testEnqueueDeviceLockCommandRaceCondition(t *testing.T, ds *Datastore) {
// Create NanoMDMStorage
storage := &NanoMDMStorage{
db: ds.writer(ctx),
logger: log.NewNopLogger(),
logger: slog.New(slog.DiscardHandler),
ds: ds,
}
server/datastore/mysql/policies.go
+5 -5
View File
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
"log/slog"
"slices"
"sort"
"strings"
@@ -17,7 +18,6 @@ import (
"github.com/fleetdm/fleet/v4/server/fleet"
common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
"github.com/fleetdm/fleet/v4/server/ptr"
kitlog "github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/jmoiron/sqlx"
)
@@ -313,7 +313,7 @@ func (ds *Datastore) PolicyLite(ctx context.Context, id uint) (*fleet.PolicyLite
// Currently, SavePolicy does not allow updating the team of an existing policy.
func (ds *Datastore) SavePolicy(ctx context.Context, p *fleet.Policy, shouldRemoveAllPolicyMemberships bool, removePolicyStats bool) error {
if err := ds.withRetryTxx(ctx, func(tx sqlx.ExtContext) error {
return savePolicy(ctx, tx, ds.logger, p, shouldRemoveAllPolicyMemberships, removePolicyStats)
return savePolicy(ctx, tx, ds.logger.SlogLogger(), p, shouldRemoveAllPolicyMemberships, removePolicyStats)
}); err != nil {
return ctxerr.Wrap(ctx, err, "updating policy")
}
@@ -321,7 +321,7 @@ func (ds *Datastore) SavePolicy(ctx context.Context, p *fleet.Policy, shouldRemo
return nil
}
func savePolicy(ctx context.Context, db sqlx.ExtContext, logger kitlog.Logger, p *fleet.Policy, shouldRemoveAllPolicyMemberships bool, removePolicyStats bool) error {
func savePolicy(ctx context.Context, db sqlx.ExtContext, logger *slog.Logger, p *fleet.Policy, shouldRemoveAllPolicyMemberships bool, removePolicyStats bool) error {
if p.TeamID == nil && p.SoftwareInstallerID != nil {
return ctxerr.Wrap(ctx, errSoftwareTitleIDOnGlobalPolicy, "save policy")
}
@@ -474,7 +474,7 @@ func assertTeamMatches(ctx context.Context, db sqlx.QueryerContext, teamID uint,
func cleanupPolicy(
ctx context.Context, queryerContext sqlx.QueryerContext, extContext sqlx.ExtContext, policyID uint, policyPlatform string,
shouldRemoveAllPolicyMemberships bool,
removePolicyStats bool, logger kitlog.Logger,
removePolicyStats bool, logger *slog.Logger,
) error {
var err error
if shouldRemoveAllPolicyMemberships {
@@ -1452,7 +1452,7 @@ func (ds *Datastore) ApplyPolicySpecs(ctx context.Context, authorID uint, specs
// memberships that don't match current criteria).
if err = cleanupPolicy(
ctx, tx, tx, uint(lastID), spec.Platform, shouldRemoveAllPolicyMemberships, //nolint:gosec // dismiss G115
removePolicyStats, ds.logger,
removePolicyStats, ds.logger.SlogLogger(),
); err != nil {
return err
}
server/datastore/mysql/rdsauth/connector.go
+4 -4
View File
@@ -5,13 +5,13 @@ import (
"context"
"database/sql/driver"
"fmt"
"log/slog"
"strings"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/feature/rds/auth"
"github.com/fleetdm/fleet/v4/server/aws_common"
"github.com/fleetdm/fleet/v4/server/config"
"github.com/go-kit/log"
"github.com/go-sql-driver/mysql"
// Blank import registers the "rdsmysql" TLS config with pre-loaded AWS RDS CA certificates
_ "github.com/shogo82148/rdsmysql/v2"
@@ -73,7 +73,7 @@ func (g *iamAuthTokenGenerator) newToken(ctx context.Context) (string, error) {
type Connector struct {
baseDSN string
tokenGen *iamAuthTokenGenerator
logger log.Logger
logger *slog.Logger
}
// Connect implements driver.Connector
@@ -106,7 +106,7 @@ func (c *Connector) Driver() driver.Driver {
// NewConnectorFactory returns a factory function that creates IAM-authenticated
// database connectors. This factory can be injected into common_mysql.NewDB
// to enable IAM authentication without adding AWS dependencies to common_mysql.
func NewConnectorFactory(conf *config.MysqlConfig, host, port string) (func(dsn string, logger log.Logger) (driver.Connector, error), error) {
func NewConnectorFactory(conf *config.MysqlConfig, host, port string) (func(dsn string, logger *slog.Logger) (driver.Connector, error), error) {
tokenGen, err := newIAMAuthTokenGenerator(
host,
conf.Username,
@@ -119,7 +119,7 @@ func NewConnectorFactory(conf *config.MysqlConfig, host, port string) (func(dsn
return nil, fmt.Errorf("failed to create IAM token generator: %w", err)
}
return func(dsn string, logger log.Logger) (driver.Connector, error) {
return func(dsn string, logger *slog.Logger) (driver.Connector, error) {
return &Connector{
baseDSN: dsn,
tokenGen: tokenGen,
server/datastore/mysql/testing_utils.go
+1 -1
View File
@@ -469,7 +469,7 @@ func TruncateTables(t testing.TB, ds *Datastore, tables ...string) {
"osquery_options": true,
"software_categories": true,
}
testing_utils.TruncateTables(t, ds.writer(context.Background()), ds.logger, nonEmptyTables, tables...)
testing_utils.TruncateTables(t, ds.writer(context.Background()), ds.logger.SlogLogger(), nonEmptyTables, tables...)
}
// this is meant to be used for debugging/testing that statement uses an efficient
server/mdm/nanomdm/cli/cli.go
+46 -1
View File
@@ -2,8 +2,10 @@
package cli
import (
"context"
"errors"
"fmt"
"log/slog"
"strings"
"github.com/fleetdm/fleet/v4/server/mdm/nanomdm/storage"
@@ -101,9 +103,11 @@ func fileStorageConfig(dsn, options string) (*file.FileStorage, error) {
func mysqlStorageConfig(dsn, options string, logger log.Logger) (*mysql.MySQLStorage, error) {
logger = logger.With("storage", "mysql")
// mysql.WithLogger requires *slog.Logger; bridge the nanolib logger
slogLogger := slog.New(&nanoLibSlogHandler{logger: logger})
opts := []mysql.Option{
mysql.WithDSN(dsn),
mysql.WithLogger(logger),
mysql.WithLogger(slogLogger),
}
if options != "" {
for k, v := range splitOptions(options) {
@@ -135,3 +139,44 @@ func splitOptions(s string) map[string]string {
}
return out
}
// nanoLibSlogHandler adapts a nanolib/log.Logger to slog.Handler.
// This bridge exists because the standalone nanomdm CLI tools still use
// nanolib loggers, while the mysql storage backend now uses *slog.Logger.
type nanoLibSlogHandler struct {
logger log.Logger
attrs []slog.Attr
}
func (h *nanoLibSlogHandler) Enabled(_ context.Context, _ slog.Level) bool {
return true
}
func (h *nanoLibSlogHandler) Handle(_ context.Context, r slog.Record) error {
kvs := make([]any, 0, 2+2*len(h.attrs)+2*r.NumAttrs())
kvs = append(kvs, "msg", r.Message)
for _, a := range h.attrs {
kvs = append(kvs, a.Key, a.Value.Any())
}
r.Attrs(func(a slog.Attr) bool {
kvs = append(kvs, a.Key, a.Value.Any())
return true
})
if r.Level >= slog.LevelInfo {
h.logger.Info(kvs...)
} else {
h.logger.Debug(kvs...)
}
return nil
}
func (h *nanoLibSlogHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
newAttrs := make([]slog.Attr, len(h.attrs), len(h.attrs)+len(attrs))
copy(newAttrs, h.attrs)
newAttrs = append(newAttrs, attrs...)
return &nanoLibSlogHandler{logger: h.logger, attrs: newAttrs}
}
func (h *nanoLibSlogHandler) WithGroup(_ string) slog.Handler {
return h
}
server/mdm/nanomdm/storage/mysql/mysql.go
+11 -14
View File
@@ -7,6 +7,7 @@ import (
_ "embed"
"errors"
"fmt"
"log/slog"
"os"
"time"
@@ -15,8 +16,6 @@ import (
"github.com/fleetdm/fleet/v4/server/mdm/nanomdm/mdm"
common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
"github.com/jmoiron/sqlx"
"github.com/micromdm/nanolib/log"
"github.com/micromdm/nanolib/log/ctxlog"
)
// Schema holds the schema for the NanoMDM MySQL storage.
@@ -27,7 +26,7 @@ var Schema string
var ErrNoCert = errors.New("no certificate in MDM Request")
type MySQLStorage struct {
logger log.Logger
logger *slog.Logger
db *sql.DB
rm bool
asyncLastSeen *asyncLastSeen
@@ -38,7 +37,7 @@ type config struct {
driver string
dsn string
db *sql.DB
logger log.Logger
logger *slog.Logger
rm bool
asyncCap int
asyncInterval time.Duration
@@ -53,7 +52,7 @@ func WithReaderFunc(readerFunc func(ctx context.Context) fleet.DBReader) Option
}
}
func WithLogger(logger log.Logger) Option {
func WithLogger(logger *slog.Logger) Option {
return func(c *config) {
c.logger = logger
}
@@ -96,7 +95,7 @@ func New(opts ...Option) (*MySQLStorage, error) {
asyncLastSeenCap = 1000
)
cfg := &config{logger: log.NopLogger, driver: "mysql", asyncCap: asyncLastSeenCap, asyncInterval: asyncLastSeenFlushInterval}
cfg := &config{logger: slog.New(slog.DiscardHandler), driver: "mysql", asyncCap: asyncLastSeenCap, asyncInterval: asyncLastSeenFlushInterval}
for _, opt := range opts {
opt(cfg)
}
@@ -180,9 +179,7 @@ func (s *MySQLStorage) storeUserTokenUpdate(r *mdm.Request, msg *mdm.TokenUpdate
// there shouldn't be an Unlock Token on the user channel, but
// complain if there is to warn an admin
if len(msg.UnlockToken) > 0 {
ctxlog.Logger(r.Context, s.logger).Info(
"msg", "Unlock Token on user channel not stored",
)
s.logger.InfoContext(r.Context, "Unlock Token on user channel not stored")
}
_, err := s.db.ExecContext(
r.Context, `
@@ -334,20 +331,20 @@ func (s *MySQLStorage) updateLastSeenBatch(ctx context.Context, ids []string) {
stmt, args, err := sqlx.In(`UPDATE nano_enrollments SET last_seen_at = CURRENT_TIMESTAMP WHERE id IN (?)`, ids)
if err != nil {
s.logger.Info("msg", "error building nano_enrollments.last_seen_at sql", "err", err)
s.logger.ErrorContext(ctx, "error building nano_enrollments.last_seen_at sql", "err", err)
return
}
err = common_mysql.WithRetryTxx(ctx, sqlx.NewDb(s.db, ""), func(tx sqlx.ExtContext) error {
_, err := tx.ExecContext(ctx, stmt, args...)
return err
}, loggerWrapper{s.logger})
}, s.logger)
if err != nil {
s.logger.Info("msg", "error batch updating nano_enrollments.last_seen_at", "err", err)
s.logger.ErrorContext(ctx, "error batch updating nano_enrollments.last_seen_at", "err", err)
}
}
func (s *MySQLStorage) ExpandEmbeddedSecrets(_ context.Context, document string) (string, error) {
s.logger.Info("level", "error", "err", "MySQLStorage.ExpandEmbeddedSecrets not implemented")
func (s *MySQLStorage) ExpandEmbeddedSecrets(ctx context.Context, document string) (string, error) {
s.logger.ErrorContext(ctx, "MySQLStorage.ExpandEmbeddedSecrets not implemented")
return document, nil
}
server/mdm/nanomdm/storage/mysql/queue.go
+2 -12
View File
@@ -12,7 +12,6 @@ import (
common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
"github.com/google/uuid"
"github.com/jmoiron/sqlx"
"github.com/micromdm/nanolib/log"
)
func enqueue(ctx context.Context, tx sqlx.ExtContext, ids []string, cmd *mdm.CommandWithSubtype) error {
@@ -53,22 +52,13 @@ func enqueue(ctx context.Context, tx sqlx.ExtContext, ids []string, cmd *mdm.Com
return nil
}
type loggerWrapper struct {
logger log.Logger
}
func (l loggerWrapper) Log(keyvals ...interface{}) error {
l.logger.Info(keyvals...)
return nil
}
func (m *MySQLStorage) EnqueueCommand(ctx context.Context, ids []string, cmd *mdm.CommandWithSubtype) (map[string]error,
error) {
// We need to retry because this transaction may deadlock with updates to nano_enrollment.last_seen_at
// Deadlock seen in 2024/12/12 loadtest: https://docs.google.com/document/d/1-Q6qFTd7CDm-lh7MVRgpNlNNJijk6JZ4KO49R1fp80U
err := common_mysql.WithRetryTxx(ctx, sqlx.NewDb(m.db, ""), func(tx sqlx.ExtContext) error {
return enqueue(ctx, tx, ids, cmd)
}, loggerWrapper{m.logger})
}, m.logger)
return nil, err
}
@@ -270,7 +260,7 @@ func (m *MySQLStorage) BulkDeleteHostUserCommandsWithoutResults(ctx context.Cont
}
return common_mysql.WithRetryTxx(ctx, sqlx.NewDb(m.db, ""), func(tx sqlx.ExtContext) error {
return m.bulkDeleteHostUserCommandsWithoutResults(ctx, tx, commandToIDs)
}, loggerWrapper{m.logger})
}, m.logger)
}
func (m *MySQLStorage) bulkDeleteHostUserCommandsWithoutResults(ctx context.Context, tx sqlx.ExtContext,
server/platform/mysql/common.go
+8 -9
View File
@@ -5,12 +5,12 @@ import (
"database/sql"
"database/sql/driver"
"fmt"
"log/slog"
"net/url"
"time"
"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
"github.com/fleetdm/fleet/v4/server/platform/logging"
"github.com/go-kit/log"
"github.com/go-sql-driver/mysql"
"github.com/jmoiron/sqlx"
"github.com/ngrok/sqlmw"
@@ -19,7 +19,7 @@ import (
// ConnectorFactory creates a driver.Connector for custom database authentication.
// This allows injecting authentication mechanisms (like AWS IAM) without adding
// dependencies to this package.
type ConnectorFactory func(dsn string, logger log.Logger) (driver.Connector, error)
type ConnectorFactory func(dsn string, logger *slog.Logger) (driver.Connector, error)
// TestSQLMode combines ANSI mode components with MySQL 8 default strict modes for testing
// ANSI mode includes: REAL_AS_FLOAT, PIPES_AS_CONCAT, ANSI_QUOTES, IGNORE_SPACE, ONLY_FULL_GROUP_BY
@@ -83,7 +83,7 @@ func NewDB(conf *MysqlConfig, opts *DBOptions, otelDriverName string) (*sqlx.DB,
var db *sqlx.DB
if opts.ConnectorFactory != nil {
connector, err := opts.ConnectorFactory(dsn, opts.Logger)
connector, err := opts.ConnectorFactory(dsn, opts.Logger.SlogLogger())
if err != nil {
return nil, fmt.Errorf("failed to create connector: %w", err)
}
@@ -108,8 +108,7 @@ func NewDB(conf *MysqlConfig, opts *DBOptions, otelDriverName string) (*sqlx.DB,
break
}
interval := time.Duration(attempt) * time.Second
opts.Logger.Log("mysql", fmt.Sprintf(
"could not connect to db: %v, sleeping %v", dbError, interval))
opts.Logger.SlogLogger().WarnContext(context.Background(), "could not connect to db", "err", dbError, "sleep_interval", interval)
time.Sleep(interval)
}
@@ -164,7 +163,7 @@ func generateMysqlConnectionString(conf MysqlConfig) string {
return dsn
}
func WithTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) error {
func WithTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger *slog.Logger) error {
tx, err := db.BeginTxx(ctx, nil)
if err != nil {
return ctxerr.Wrap(ctx, err, "create transaction")
@@ -173,7 +172,7 @@ func WithTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) error
defer func() {
if p := recover(); p != nil {
if err := tx.Rollback(); err != nil {
logger.Log("err", err, "msg", "error encountered during transaction panic rollback")
logger.ErrorContext(ctx, "error encountered during transaction panic rollback", "err", err)
}
panic(p)
}
@@ -202,7 +201,7 @@ func WithTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) error
}
// WithReadOnlyTxx executes fn within an isolated, read-only transaction
func WithReadOnlyTxx(ctx context.Context, reader *sqlx.DB, fn ReadTxFn, logger log.Logger) error {
func WithReadOnlyTxx(ctx context.Context, reader *sqlx.DB, fn ReadTxFn, logger *slog.Logger) error {
tx, err := reader.BeginTxx(ctx, &sql.TxOptions{
ReadOnly: true,
Isolation: sql.LevelRepeatableRead,
@@ -214,7 +213,7 @@ func WithReadOnlyTxx(ctx context.Context, reader *sqlx.DB, fn ReadTxFn, logger l
defer func() {
if p := recover(); p != nil {
if err := tx.Rollback(); err != nil {
logger.Log("err", err, "msg", "error encountered during read-only transaction panic rollback")
logger.ErrorContext(ctx, "error encountered during read-only transaction panic rollback", "err", err)
}
panic(p)
}
server/platform/mysql/retry.go
+3 -3
View File
@@ -5,13 +5,13 @@ import (
"database/sql"
"errors"
"fmt"
"log/slog"
"sync"
"time"
"github.com/VividCortex/mysqlerr"
"github.com/cenkalti/backoff/v4"
"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
"github.com/go-kit/log"
"github.com/go-sql-driver/mysql"
"github.com/jmoiron/sqlx"
)
@@ -58,7 +58,7 @@ type TxFn func(tx sqlx.ExtContext) error
type ReadTxFn func(tx DBReadTx) error
// WithRetryTxx provides a common way to commit/rollback a txFn wrapped in a retry with exponential backoff
func WithRetryTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) error {
func WithRetryTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger *slog.Logger) error {
operation := func() error {
tx, err := db.BeginTxx(ctx, nil)
if err != nil {
@@ -68,7 +68,7 @@ func WithRetryTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger)
defer func() {
if p := recover(); p != nil {
if err := tx.Rollback(); err != nil {
logger.Log("err", err, "msg", "error encountered during transaction panic rollback")
logger.ErrorContext(ctx, "error encountered during transaction panic rollback", "err", err)
}
panic(p)
}
server/platform/mysql/retry_test.go
+5 -5
View File
@@ -2,12 +2,12 @@ package mysql
import (
"errors"
"log/slog"
"sync"
"sync/atomic"
"testing"
"github.com/DATA-DOG/go-sqlmock"
"github.com/go-kit/log"
gmysql "github.com/go-sql-driver/mysql"
"github.com/jmoiron/sqlx"
"github.com/stretchr/testify/assert"
@@ -76,7 +76,7 @@ func TestTransactionReadOnlyTriggersFatalError(t *testing.T) {
txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
return WithRetryTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
return readOnlyErr()
}, log.NewNopLogger())
}, slog.New(slog.DiscardHandler))
},
},
{
@@ -88,7 +88,7 @@ func TestTransactionReadOnlyTriggersFatalError(t *testing.T) {
txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
return WithRetryTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
return nil
}, log.NewNopLogger())
}, slog.New(slog.DiscardHandler))
},
},
{
@@ -100,7 +100,7 @@ func TestTransactionReadOnlyTriggersFatalError(t *testing.T) {
txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
return WithTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
return readOnlyErr()
}, log.NewNopLogger())
}, slog.New(slog.DiscardHandler))
},
},
{
@@ -112,7 +112,7 @@ func TestTransactionReadOnlyTriggersFatalError(t *testing.T) {
txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error {
return WithTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error {
return nil
}, log.NewNopLogger())
}, slog.New(slog.DiscardHandler))
},
},
}
server/platform/mysql/testing_utils/testing_utils.go
+2 -2
View File
@@ -3,6 +3,7 @@ package testing_utils
import (
"context"
"fmt"
"log/slog"
"os"
"os/exec"
"path/filepath"
@@ -11,7 +12,6 @@ import (
"testing"
common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
"github.com/go-kit/log"
"github.com/jmoiron/sqlx"
"github.com/stretchr/testify/require"
)
@@ -38,7 +38,7 @@ func getTestAddress() string {
// Note that the order is typically not important because FK checks are
// disabled while truncating. If no table is provided, all tables (except
// those that are seeded by the SQL schema file) are truncated.
func TruncateTables(t testing.TB, db *sqlx.DB, logger log.Logger, nonEmptyTables map[string]bool, tables ...string) {
func TruncateTables(t testing.TB, db *sqlx.DB, logger *slog.Logger, nonEmptyTables map[string]bool, tables ...string) {
// By setting DISABLE_TRUNCATE_TABLES a developer can troubleshoot tests
// by inspecting mysql tables.
if os.Getenv("DISABLE_TRUNCATE_TABLES") != "" {