JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite: pass ctx into satellitedb.Open

Browse Source 
Opening a database requires ctx, this is first step to passing ctx
to the appropriate level.

Change-Id: Ic303e69f868ef3449ae36377937a29670cf635e2
This commit is contained in:
Egon Elbre 2020-10-28 15:48:31 +02:00
parent ed1f6d7973
commit 9b2e00a38b
13 changed files with 31 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/satellite/admin.go
View File

@ -26,7 +26,7 @@ func cmdAdminRun(cmd *cobra.Command, args []string) (err error) {
return errs.New("Failed to load identity: %+v", err) return errs.New("Failed to load identity: %+v", err)
} }
db, err := satellitedb.New(log.Named("db"), runCfg.Database, satellitedb.Options{ db, err := satellitedb.Open(ctx, log.Named("db"), runCfg.Database, satellitedb.Options{
APIKeysLRUOptions: runCfg.APIKeysLRUOptions(), APIKeysLRUOptions: runCfg.APIKeysLRUOptions(),
}) })
if err != nil { if err != nil {

2
cmd/satellite/api.go
View File

@ -31,7 +31,7 @@ func cmdAPIRun(cmd *cobra.Command, args []string) (err error) {
return errs.New("Failed to load identity: %+v", err) return errs.New("Failed to load identity: %+v", err)
} }
db, err := satellitedb.New(log.Named("db"), runCfg.Database, satellitedb.Options{ db, err := satellitedb.Open(ctx, log.Named("db"), runCfg.Database, satellitedb.Options{
APIKeysLRUOptions: runCfg.APIKeysLRUOptions(), APIKeysLRUOptions: runCfg.APIKeysLRUOptions(),
RevocationLRUOptions: runCfg.RevocationLRUOptions(), RevocationLRUOptions: runCfg.RevocationLRUOptions(),
}) })

8
cmd/satellite/billing.go
View File

@ -21,10 +21,10 @@ import (
"storj.io/storj/satellite/satellitedb/dbx" "storj.io/storj/satellite/satellitedb/dbx"
) )
func runBillingCmd(cmdFunc func(*stripecoinpayments.Service, *dbx.DB) error) error { func runBillingCmd(ctx context.Context, cmdFunc func(context.Context, *stripecoinpayments.Service, *dbx.DB) error) error {
// Open SatelliteDB for the Payment Service // Open SatelliteDB for the Payment Service
logger := zap.L() logger := zap.L()
db, err := satellitedb.New(logger.Named("db"), runCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, logger.Named("db"), runCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("error connecting to master database on satellite: %+v", err) return errs.New("error connecting to master database on satellite: %+v", err)
} }
@ -55,7 +55,7 @@ func runBillingCmd(cmdFunc func(*stripecoinpayments.Service, *dbx.DB) error) err
return err return err
} }
return cmdFunc(payments, dbxDB) return cmdFunc(ctx, payments, dbxDB)
} }
func setupPayments(log *zap.Logger, db satellite.DB) (*stripecoinpayments.Service, error) { func setupPayments(log *zap.Logger, db satellite.DB) (*stripecoinpayments.Service, error) {
@ -124,7 +124,7 @@ type userData struct {
// generateStripeCustomers creates missing stripe-customers for users in our database. // generateStripeCustomers creates missing stripe-customers for users in our database.
func generateStripeCustomers(ctx context.Context) (err error) { func generateStripeCustomers(ctx context.Context) (err error) {
return runBillingCmd(func(payments *stripecoinpayments.Service, dbxDB *dbx.DB) error { return runBillingCmd(ctx, func(ctx context.Context, payments *stripecoinpayments.Service, dbxDB *dbx.DB) error {
accounts := payments.Accounts() accounts := payments.Accounts()
rows, err := dbxDB.Query(ctx, "SELECT id, email FROM users WHERE id NOT IN (SELECT user_id from stripe_customers) AND users.status=1") rows, err := dbxDB.Query(ctx, "SELECT id, email FROM users WHERE id NOT IN (SELECT user_id from stripe_customers) AND users.status=1")

6
cmd/satellite/compensation.go
View File

@ -32,7 +32,7 @@ func generateInvoicesCSV(ctx context.Context, period compensation.Period, out io
WithheldPercents: generateInvoicesCfg.Compensation.WithheldPercents, WithheldPercents: generateInvoicesCfg.Compensation.WithheldPercents,
} }
db, err := satellitedb.New(zap.L().Named("db"), generateInvoicesCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, zap.L().Named("db"), generateInvoicesCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("error connecting to master database on satellite: %+v", err) return errs.New("error connecting to master database on satellite: %+v", err)
} }
@ -141,7 +141,7 @@ func recordPeriod(ctx context.Context, paystubsCSV, paymentsCSV string) (int, in
return 0, 0, err return 0, 0, err
} }
db, err := satellitedb.New(zap.L().Named("db"), recordPeriodCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, zap.L().Named("db"), recordPeriodCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return 0, 0, errs.New("error connecting to master database on satellite: %+v", err) return 0, 0, errs.New("error connecting to master database on satellite: %+v", err)
} }
@ -165,7 +165,7 @@ func recordOneOffPayments(ctx context.Context, paymentsCSV string) (int, error)
return 0, err return 0, err
} }
db, err := satellitedb.New(zap.L().Named("db"), recordOneOffPaymentsCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, zap.L().Named("db"), recordOneOffPaymentsCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return 0, errs.New("error connecting to master database on satellite: %+v", err) return 0, errs.New("error connecting to master database on satellite: %+v", err)
} }

2
cmd/satellite/gc.go
View File

@ -28,7 +28,7 @@ func cmdGCRun(cmd *cobra.Command, args []string) (err error) {
return errs.New("Failed to load identity: %+v", err) return errs.New("Failed to load identity: %+v", err)
} }
db, err := satellitedb.New(log.Named("db"), runCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, log.Named("db"), runCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("Error starting master database on satellite GC: %+v", err) return errs.New("Error starting master database on satellite GC: %+v", err)
} }

2
cmd/satellite/gracefulexit.go
View File

@ -27,7 +27,7 @@ import (
// generateGracefulExitCSV creates a report with graceful exit data for exiting or exited nodes in a given period. // generateGracefulExitCSV creates a report with graceful exit data for exiting or exited nodes in a given period.
func generateGracefulExitCSV(ctx context.Context, completed bool, start time.Time, end time.Time, output io.Writer) error { func generateGracefulExitCSV(ctx context.Context, completed bool, start time.Time, end time.Time, output io.Writer) error {
db, err := satellitedb.New(zap.L().Named("db"), gracefulExitCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, zap.L().Named("db"), gracefulExitCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("error connecting to master database on satellite: %+v", err) return errs.New("error connecting to master database on satellite: %+v", err)
} }

17
cmd/satellite/main.go
View File

@ -325,7 +325,7 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
return errs.New("Failed to load identity: %+v", err) return errs.New("Failed to load identity: %+v", err)
} }
db, err := satellitedb.New(log.Named("db"), runCfg.Database, satellitedb.Options{ db, err := satellitedb.Open(ctx, log.Named("db"), runCfg.Database, satellitedb.Options{
ReportedRollupsReadBatchSize: runCfg.Orders.SettlementBatchSize, ReportedRollupsReadBatchSize: runCfg.Orders.SettlementBatchSize,
}) })
if err != nil { if err != nil {
@ -399,7 +399,7 @@ func cmdMigrationRun(cmd *cobra.Command, args []string) (err error) {
ctx, _ := process.Ctx(cmd) ctx, _ := process.Ctx(cmd)
log := zap.L() log := zap.L()
db, err := satellitedb.New(log.Named("migration"), runCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, log.Named("migration"), runCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("Error creating new master database connection for satellitedb migration: %+v", err) return errs.New("Error creating new master database connection for satellitedb migration: %+v", err)
} }
@ -447,9 +447,10 @@ func cmdSetup(cmd *cobra.Command, args []string) (err error) {
} }
func cmdQDiag(cmd *cobra.Command, args []string) (err error) { func cmdQDiag(cmd *cobra.Command, args []string) (err error) {
ctx, _ := process.Ctx(cmd)
// open the master db // open the master db
database, err := satellitedb.New(zap.L().Named("db"), qdiagCfg.Database, satellitedb.Options{}) database, err := satellitedb.Open(ctx, zap.L().Named("db"), qdiagCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("error connecting to master database on satellite: %+v", err) return errs.New("error connecting to master database on satellite: %+v", err)
} }
@ -637,7 +638,7 @@ func cmdPrepareCustomerInvoiceRecords(cmd *cobra.Command, args []string) (err er
return errs.New("invalid period specified: %v", err) return errs.New("invalid period specified: %v", err)
} }
return runBillingCmd(func(payments *stripecoinpayments.Service, _ *dbx.DB) error { return runBillingCmd(ctx, func(ctx context.Context, payments *stripecoinpayments.Service, _ *dbx.DB) error {
return payments.PrepareInvoiceProjectRecords(ctx, period) return payments.PrepareInvoiceProjectRecords(ctx, period)
}) })
} }
@ -650,7 +651,7 @@ func cmdCreateCustomerInvoiceItems(cmd *cobra.Command, args []string) (err error
return errs.New("invalid period specified: %v", err) return errs.New("invalid period specified: %v", err)
} }
return runBillingCmd(func(payments *stripecoinpayments.Service, _ *dbx.DB) error { return runBillingCmd(ctx, func(ctx context.Context, payments *stripecoinpayments.Service, _ *dbx.DB) error {
return payments.InvoiceApplyProjectRecords(ctx, period) return payments.InvoiceApplyProjectRecords(ctx, period)
}) })
} }
@ -663,7 +664,7 @@ func cmdCreateCustomerInvoiceCoupons(cmd *cobra.Command, args []string) (err err
return errs.New("invalid period specified: %v", err) return errs.New("invalid period specified: %v", err)
} }
return runBillingCmd(func(payments *stripecoinpayments.Service, _ *dbx.DB) error { return runBillingCmd(ctx, func(ctx context.Context, payments *stripecoinpayments.Service, _ *dbx.DB) error {
return payments.InvoiceApplyCoupons(ctx, period) return payments.InvoiceApplyCoupons(ctx, period)
}) })
} }
@ -676,7 +677,7 @@ func cmdCreateCustomerInvoices(cmd *cobra.Command, args []string) (err error) {
return errs.New("invalid period specified: %v", err) return errs.New("invalid period specified: %v", err)
} }
return runBillingCmd(func(payments *stripecoinpayments.Service, _ *dbx.DB) error { return runBillingCmd(ctx, func(ctx context.Context, payments *stripecoinpayments.Service, _ *dbx.DB) error {
return payments.CreateInvoices(ctx, period) return payments.CreateInvoices(ctx, period)
}) })
} }
@ -684,7 +685,7 @@ func cmdCreateCustomerInvoices(cmd *cobra.Command, args []string) (err error) {
func cmdFinalizeCustomerInvoices(cmd *cobra.Command, args []string) (err error) { func cmdFinalizeCustomerInvoices(cmd *cobra.Command, args []string) (err error) {
ctx, _ := process.Ctx(cmd) ctx, _ := process.Ctx(cmd)
return runBillingCmd(func(payments *stripecoinpayments.Service, _ *dbx.DB) error { return runBillingCmd(ctx, func(ctx context.Context, payments *stripecoinpayments.Service, _ *dbx.DB) error {
return payments.FinalizeInvoices(ctx) return payments.FinalizeInvoices(ctx)
}) })
} }

2
cmd/satellite/repairer.go
View File

@ -30,7 +30,7 @@ func cmdRepairerRun(cmd *cobra.Command, args []string) (err error) {
return errs.New("Failed to load identity: %+v", err) return errs.New("Failed to load identity: %+v", err)
} }
db, err := satellitedb.New(log.Named("db"), runCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, log.Named("db"), runCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("Error starting master database: %+v", err) return errs.New("Error starting master database: %+v", err)
} }

2
cmd/satellite/reports/attribution.go
View File

@ -32,7 +32,7 @@ var headers = []string{
// GenerateAttributionCSV creates a report with. // GenerateAttributionCSV creates a report with.
func GenerateAttributionCSV(ctx context.Context, database string, partnerID uuid.UUID, start time.Time, end time.Time, output io.Writer) error { func GenerateAttributionCSV(ctx context.Context, database string, partnerID uuid.UUID, start time.Time, end time.Time, output io.Writer) error {
log := zap.L().Named("db") log := zap.L().Named("db")
db, err := satellitedb.New(log, database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, log, database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("error connecting to master database on satellite: %+v", err) return errs.New("error connecting to master database on satellite: %+v", err)
} }

2
cmd/satellite/usage.go
View File

@ -21,7 +21,7 @@ import (
// generateNodeUsageCSV creates a report with node usage data for all nodes in a given period which can be used for payments. // generateNodeUsageCSV creates a report with node usage data for all nodes in a given period which can be used for payments.
func generateNodeUsageCSV(ctx context.Context, start time.Time, end time.Time, output io.Writer) error { func generateNodeUsageCSV(ctx context.Context, start time.Time, end time.Time, output io.Writer) error {
db, err := satellitedb.New(zap.L().Named("db"), nodeUsageCfg.Database, satellitedb.Options{}) db, err := satellitedb.Open(ctx, zap.L().Named("db"), nodeUsageCfg.Database, satellitedb.Options{})
if err != nil { if err != nil {
return errs.New("error connecting to master database on satellite: %+v", err) return errs.New("error connecting to master database on satellite: %+v", err)
} }

5
satellite/satellitedb/database.go
View File

@ -4,6 +4,7 @@
package satellitedb package satellitedb
import ( import (
"context"
"sync" "sync"
"github.com/zeebo/errs" "github.com/zeebo/errs"
@ -70,8 +71,8 @@ type Options struct {
var _ dbx.DBMethods = &satelliteDB{} var _ dbx.DBMethods = &satelliteDB{}
// New creates instance of database supports postgres. // Open creates instance of database supports postgres.
func New(log *zap.Logger, databaseURL string, opts Options) (satellite.DB, error) { func Open(ctx context.Context, log *zap.Logger, databaseURL string, opts Options) (satellite.DB, error) {
driver, source, implementation, err := dbutil.SplitConnStr(databaseURL) driver, source, implementation, err := dbutil.SplitConnStr(databaseURL)
if err != nil { if err != nil {
return nil, err return nil, err

4
satellite/satellitedb/migrate_test.go
View File

@ -155,7 +155,7 @@ func migrateTest(t *testing.T, connStr string) {
defer func() { require.NoError(t, tempDB.Close()) }() defer func() { require.NoError(t, tempDB.Close()) }()
// create a new satellitedb connection // create a new satellitedb connection
db, err := satellitedb.New(log, tempDB.ConnStr, satellitedb.Options{}) db, err := satellitedb.Open(ctx, log, tempDB.ConnStr, satellitedb.Options{})
require.NoError(t, err) require.NoError(t, err)
defer func() { require.NoError(t, db.Close()) }() defer func() { require.NoError(t, db.Close()) }()
@ -247,7 +247,7 @@ func benchmarkSetup(b *testing.B, connStr string, merged bool) {
defer func() { require.NoError(b, tempDB.Close()) }() defer func() { require.NoError(b, tempDB.Close()) }()
// create a new satellitedb connection // create a new satellitedb connection
db, err := satellitedb.New(log, tempDB.ConnStr, satellitedb.Options{}) db, err := satellitedb.Open(ctx, log, tempDB.ConnStr, satellitedb.Options{})
require.NoError(b, err) require.NoError(b, err)
defer func() { require.NoError(b, db.Close()) }() defer func() { require.NoError(b, db.Close()) }()

6
satellite/satellitedb/satellitedbtest/run.go
View File

@ -119,13 +119,13 @@ func CreateMasterDB(ctx context.Context, log *zap.Logger, name string, category
return nil, err return nil, err
} }
return CreateMasterDBOnTopOf(log, tempDB) return CreateMasterDBOnTopOf(ctx, log, tempDB)
} }
// CreateMasterDBOnTopOf creates a new satellite database on top of an already existing // CreateMasterDBOnTopOf creates a new satellite database on top of an already existing
// temporary database. // temporary database.
func CreateMasterDBOnTopOf(log *zap.Logger, tempDB *dbutil.TempDatabase) (db satellite.DB, err error) { func CreateMasterDBOnTopOf(ctx context.Context, log *zap.Logger, tempDB *dbutil.TempDatabase) (db satellite.DB, err error) {
masterDB, err := satellitedb.New(log.Named("db"), tempDB.ConnStr, satellitedb.Options{}) masterDB, err := satellitedb.Open(ctx, log.Named("db"), tempDB.ConnStr, satellitedb.Options{})
return &tempMasterDB{DB: masterDB, tempDB: tempDB}, err return &tempMasterDB{DB: masterDB, tempDB: tempDB}, err
} }