From a7922912fe0f6a4da0245f427c84e844c67876b9 Mon Sep 17 00:00:00 2001 From: Dana Axinte <53751979+dana-axinte@users.noreply.github.com> Date: Tue, 20 May 2025 11:48:47 +0100 Subject: [PATCH] SecretsManager: Introduce secrets database wrapper (#105472) SecretsManager: Introduce secret database wrapper Co-authored-by: PoorlyDefinedBehaviour Co-authored-by: Leandro Deveikis Co-authored-by: Matheus Macabu --- .../apis/secret/contracts/database.go | 20 ++++ pkg/server/wire.go | 4 + pkg/storage/secret/database/database.go | 98 +++++++++++++++++++ pkg/storage/unified/sql/backend.go | 6 +- 4 files changed, 125 insertions(+), 3 deletions(-) create mode 100644 pkg/registry/apis/secret/contracts/database.go create mode 100644 pkg/storage/secret/database/database.go diff --git a/pkg/registry/apis/secret/contracts/database.go b/pkg/registry/apis/secret/contracts/database.go new file mode 100644 index 00000000000..305855c6453 --- /dev/null +++ b/pkg/registry/apis/secret/contracts/database.go @@ -0,0 +1,20 @@ +package contracts + +import ( + "context" + "database/sql" +) + +type Database interface { + DriverName() string + Transaction(ctx context.Context, f func(context.Context) error) error + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) + QueryContext(ctx context.Context, query string, args ...any) (Rows, error) +} + +type Rows interface { + Close() error + Next() bool + Scan(dest ...any) error + Err() error +} diff --git a/pkg/server/wire.go b/pkg/server/wire.go index 07110b74088..ff862100b67 100644 --- a/pkg/server/wire.go +++ b/pkg/server/wire.go @@ -41,6 +41,7 @@ import ( apiregistry "github.com/grafana/grafana/pkg/registry/apis" "github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy" "github.com/grafana/grafana/pkg/registry/apis/provisioning/repository/github" + secretcontracts "github.com/grafana/grafana/pkg/registry/apis/secret/contracts" secretdecrypt "github.com/grafana/grafana/pkg/registry/apis/secret/decrypt" appregistry "github.com/grafana/grafana/pkg/registry/apps" "github.com/grafana/grafana/pkg/services/accesscontrol" @@ -163,6 +164,7 @@ import ( "github.com/grafana/grafana/pkg/services/user/userimpl" "github.com/grafana/grafana/pkg/setting" legacydualwrite "github.com/grafana/grafana/pkg/storage/legacysql/dualwrite" + secretdatabase "github.com/grafana/grafana/pkg/storage/secret/database" secretmetadata "github.com/grafana/grafana/pkg/storage/secret/metadata" "github.com/grafana/grafana/pkg/storage/unified/resource" unifiedsearch "github.com/grafana/grafana/pkg/storage/unified/search" @@ -417,6 +419,8 @@ var wireBasicSet = wire.NewSet( // Secrets Manager secretmetadata.ProvideSecureValueMetadataStorage, secretmetadata.ProvideKeeperMetadataStorage, + secretdatabase.ProvideDatabase, + wire.Bind(new(secretcontracts.Database), new(*secretdatabase.Database)), secretdecrypt.ProvideDecryptAuthorizer, secretdecrypt.ProvideDecryptAllowList, // Unified storage diff --git a/pkg/storage/secret/database/database.go b/pkg/storage/secret/database/database.go new file mode 100644 index 00000000000..8997a24ee9e --- /dev/null +++ b/pkg/storage/secret/database/database.go @@ -0,0 +1,98 @@ +package database + +import ( + "context" + "database/sql" + "errors" + + "github.com/jmoiron/sqlx" + + "github.com/grafana/grafana/pkg/infra/db" + "github.com/grafana/grafana/pkg/registry/apis/secret/contracts" + "github.com/grafana/grafana/pkg/util/xorm" +) + +// contextSessionTxKey is the key used to store the transaction in the context. +type contextSessionTxKey struct{} + +// Implements contracts.Database +type Database struct { + dbType string + sqlx *sqlx.DB + + // Keep the xorm.Engine instance and its references alive until the apiserver is shut down. + // This is only needed because the xorm.Engine calls a runtime.SetFinalizer, in a RAII-like pattern to close the DB, + // when the engine is garbage collected. Normally, this will only ever happen when the server shuts down. + // Ref: pkg/util/xorm/xorm.go:118 (it seems to be a relic from the xorm codebase that was copied over). + // In single tenant Grafana, there are many other services and references to the xorm.Engine, so it never gets GC'd. + // At some point in the future if we migrate everything away from it, we need to revisit how we set up the DB opening. + // However, with the multi-tenant apiserver, we are no longer using the xorm.Engine directly for our DB queries. + // We only use it to bootstrap the database and run migrations. + // Instead, we use a pointer to *sql.DB directly, and that is created from *xorm.Engine -> *core.DB (also xorm) -> *sql.DB. + // The GC notices that the xorm.Engine is no longer referenced, and calls the finalizer to close the DB, because we + // only reference the pointer to *sql.DB. Here we tie the lifetime of the xorm.Engine to the Database we use for queries. + engine *xorm.Engine +} + +func ProvideDatabase(db db.DB) *Database { + engine := db.GetEngine() + + return &Database{ + dbType: string(db.GetDBType()), + sqlx: sqlx.NewDb(engine.DB().DB, db.GetDialect().DriverName()), + engine: engine, + } +} + +func (db *Database) DriverName() string { + return db.dbType +} + +func (db *Database) Transaction(ctx context.Context, callback func(context.Context) error) error { + txCtx := ctx + + // If another transaction is already open, we just use that one instead of nesting. + sqlxTx, ok := txCtx.Value(contextSessionTxKey{}).(*sqlx.Tx) + if sqlxTx != nil && ok { + // We are already in a transaction, so we don't commit or rollback, let the outermost transaction do it. + return callback(txCtx) + } + + tx, err := db.sqlx.Beginx() + if err != nil { + return err + } + + sqlxTx = tx + + // Save it in the context so the transaction can be reused in case it is nested. + txCtx = context.WithValue(ctx, contextSessionTxKey{}, sqlxTx) + + if err := callback(txCtx); err != nil { + if rbErr := sqlxTx.Rollback(); rbErr != nil { + return errors.Join(err, rbErr) + } + + return err + } + + return sqlxTx.Commit() +} + +func (db *Database) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) { + // If another transaction is already open, we just use that one instead of nesting. + if tx, ok := ctx.Value(contextSessionTxKey{}).(*sqlx.Tx); tx != nil && ok { + return tx.ExecContext(ctx, db.sqlx.Rebind(query), args...) + } + + return db.sqlx.ExecContext(ctx, db.sqlx.Rebind(query), args...) +} + +func (db *Database) QueryContext(ctx context.Context, query string, args ...any) (contracts.Rows, error) { + // If another transaction is already open, we just use that one instead of nesting. + if tx, ok := ctx.Value(contextSessionTxKey{}).(*sqlx.Tx); tx != nil && ok { + return tx.QueryContext(ctx, db.sqlx.Rebind(query), args...) + } + + return db.sqlx.QueryContext(ctx, db.sqlx.Rebind(query), args...) +} diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index 5856940a11c..720f4f24322 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -343,7 +343,7 @@ func (b *backend) create(ctx context.Context, event resource.WriteEvent) (int64, Folder: folder, GUID: guid, }); err != nil { - if isRowAlreadyExistsError(err) { + if IsRowAlreadyExistsError(err) { return guid, resource.ErrResourceAlreadyExists } return guid, fmt.Errorf("insert into resource: %w", err) @@ -387,8 +387,8 @@ func (b *backend) create(ctx context.Context, event resource.WriteEvent) (int64, return rv, nil } -// isRowAlreadyExistsError checks if the error is the result of the row inserted already existing. -func isRowAlreadyExistsError(err error) bool { +// IsRowAlreadyExistsError checks if the error is the result of the row inserted already existing. +func IsRowAlreadyExistsError(err error) bool { var sqlite sqlite3.Error if errors.As(err, &sqlite) { return sqlite.ExtendedCode == sqlite3.ErrConstraintUnique