-
Notifications
You must be signed in to change notification settings - Fork 393
/
database.go
398 lines (327 loc) · 11.5 KB
/
database.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package satellitedb
import (
"context"
"sync"
"github.com/zeebo/errs"
"go.uber.org/zap"
"storj.io/common/dbutil"
"storj.io/common/dbutil/pgutil"
"storj.io/common/lrucache"
"storj.io/common/tagsql"
"storj.io/storj/private/logging"
"storj.io/storj/private/migrate"
"storj.io/storj/satellite"
"storj.io/storj/satellite/accounting"
"storj.io/storj/satellite/attribution"
"storj.io/storj/satellite/audit"
"storj.io/storj/satellite/buckets"
"storj.io/storj/satellite/compensation"
"storj.io/storj/satellite/console"
"storj.io/storj/satellite/nodeapiversion"
"storj.io/storj/satellite/nodeevents"
"storj.io/storj/satellite/oidc"
"storj.io/storj/satellite/orders"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/payments/billing"
"storj.io/storj/satellite/payments/storjscan"
"storj.io/storj/satellite/payments/stripe"
"storj.io/storj/satellite/repair/queue"
"storj.io/storj/satellite/reputation"
"storj.io/storj/satellite/revocation"
"storj.io/storj/satellite/satellitedb/dbx"
"storj.io/storj/satellite/snopayouts"
)
// Error is the default satellitedb errs class.
var Error = errs.Class("satellitedb")
type satelliteDBCollection struct {
dbs map[string]*satelliteDB
}
// satelliteDB combines access to different database tables with a record
// of the db driver, db implementation, and db source URL.
type satelliteDB struct {
*dbx.DB
migrationDB tagsql.DB
opts Options
log *zap.Logger
driver string
impl dbutil.Implementation
source string
consoleDBOnce sync.Once
consoleDB *ConsoleDB
revocationDBOnce sync.Once
revocationDB *revocationDB
}
// Options includes options for how a satelliteDB runs.
type Options struct {
ApplicationName string
APIKeysLRUOptions lrucache.Options
RevocationLRUOptions lrucache.Options
// How many storage node rollups to save/read in one batch.
SaveRollupBatchSize int
ReadRollupBatchSize int
}
var _ dbx.DBMethods = &satelliteDB{}
var safelyPartitionableDBs = map[string]bool{
// WARNING: only list additional db names here after they have been
// validated to be safely partitionable and that they do not do
// cross-db queries.
"repairqueue": true,
"nodeevents": true,
"verifyqueue": true,
"reverifyqueue": true,
}
// Open creates instance of satellite.DB.
func Open(ctx context.Context, log *zap.Logger, databaseURL string, opts Options) (rv satellite.DB, err error) {
dbMapping, err := dbutil.ParseDBMapping(databaseURL)
if err != nil {
return nil, err
}
dbc := &satelliteDBCollection{dbs: map[string]*satelliteDB{}}
defer func() {
if err != nil {
err = errs.Combine(err, dbc.Close())
}
}()
for key, val := range dbMapping {
db, err := open(ctx, log, val, opts, key)
if err != nil {
return nil, err
}
dbc.dbs[key] = db
}
return dbc, nil
}
func open(ctx context.Context, log *zap.Logger, databaseURL string, opts Options, override string) (*satelliteDB, error) {
driver, source, impl, err := dbutil.SplitConnStr(databaseURL)
if err != nil {
return nil, err
}
if impl != dbutil.Postgres && impl != dbutil.Cockroach {
return nil, Error.New("unsupported driver %q", driver)
}
source, err = pgutil.CheckApplicationName(source, opts.ApplicationName)
if err != nil {
return nil, err
}
dbxDB, err := dbx.Open(driver, source)
if err != nil {
return nil, Error.New("failed opening database via DBX at %q: %v",
source, err)
}
if log.Level() == zap.DebugLevel {
log.Debug("Connected to:", zap.String("db source", logging.Redacted(source)))
}
name := "satellitedb"
if override != "" {
name += ":" + override
}
dbutil.Configure(ctx, dbxDB.DB, name, mon)
core := &satelliteDB{
DB: dbxDB,
opts: opts,
log: log,
driver: driver,
impl: impl,
source: source,
}
core.migrationDB = core
return core, nil
}
func (dbc *satelliteDBCollection) getByName(name string) *satelliteDB {
if safelyPartitionableDBs[name] {
if db, exists := dbc.dbs[name]; exists {
return db
}
}
return dbc.dbs[""]
}
// PeerIdentities returns a storage for peer identities.
func (dbc *satelliteDBCollection) PeerIdentities() overlay.PeerIdentities {
return &peerIdentities{db: dbc.getByName("peeridentities")}
}
// Attribution is a getter for value attribution repository.
func (dbc *satelliteDBCollection) Attribution() attribution.DB {
return &attributionDB{db: dbc.getByName("attribution")}
}
// OverlayCache is a getter for overlay cache repository.
func (dbc *satelliteDBCollection) OverlayCache() overlay.DB {
return &overlaycache{db: dbc.getByName("overlaycache")}
}
// NodeEvents is a getter for node events repository.
func (dbc *satelliteDBCollection) NodeEvents() nodeevents.DB {
return &nodeEvents{db: dbc.getByName("nodeevents")}
}
// Reputation is a getter for overlay cache repository.
func (dbc *satelliteDBCollection) Reputation() reputation.DB {
return &reputations{db: dbc.getByName("reputations")}
}
// RepairQueue is a getter for RepairQueue repository.
func (dbc *satelliteDBCollection) RepairQueue() queue.RepairQueue {
return &repairQueue{db: dbc.getByName("repairqueue")}
}
// VerifyQueue is a getter for VerifyQueue database.
func (dbc *satelliteDBCollection) VerifyQueue() audit.VerifyQueue {
return &verifyQueue{db: dbc.getByName("verifyqueue")}
}
// ReverifyQueue is a getter for ReverifyQueue database.
func (dbc *satelliteDBCollection) ReverifyQueue() audit.ReverifyQueue {
return &reverifyQueue{db: dbc.getByName("reverifyqueue")}
}
// StoragenodeAccounting returns database for tracking storagenode usage.
func (dbc *satelliteDBCollection) StoragenodeAccounting() accounting.StoragenodeAccounting {
return &StoragenodeAccounting{db: dbc.getByName("storagenodeaccounting")}
}
// ProjectAccounting returns database for tracking project data use.
func (dbc *satelliteDBCollection) ProjectAccounting() accounting.ProjectAccounting {
return &ProjectAccounting{db: dbc.getByName("projectaccounting")}
}
// Revocation returns the database to deal with macaroon revocation.
func (dbc *satelliteDBCollection) Revocation() revocation.DB {
db := dbc.getByName("revocation")
db.revocationDBOnce.Do(func() {
options := db.opts.RevocationLRUOptions
options.Name = "satellitedb-revocations"
db.revocationDB = &revocationDB{
db: db,
lru: lrucache.NewOf[bool](options),
methods: db,
}
})
return db.revocationDB
}
// Console returns database for storing users, projects and api keys.
func (dbc *satelliteDBCollection) Console() console.DB {
db := dbc.getByName("console")
db.consoleDBOnce.Do(func() {
db.consoleDB = &ConsoleDB{
apikeysLRUOptions: db.opts.APIKeysLRUOptions,
db: db,
methods: db,
apikeysOnce: new(sync.Once),
}
})
return db.consoleDB
}
// OIDC returns the database for storing OAuth and OIDC information.
func (dbc *satelliteDBCollection) OIDC() oidc.DB {
db := dbc.getByName("oidc")
return oidc.NewDB(db.DB)
}
// Orders returns database for storing orders.
func (dbc *satelliteDBCollection) Orders() orders.DB {
db := dbc.getByName("orders")
return &ordersDB{db: db}
}
// Containment returns database for storing pending audit info.
// It does all of its work by way of the ReverifyQueue.
func (dbc *satelliteDBCollection) Containment() audit.Containment {
return &containment{reverifyQueue: dbc.ReverifyQueue()}
}
// StripeCoinPayments returns database for stripecoinpayments.
func (dbc *satelliteDBCollection) StripeCoinPayments() stripe.DB {
return &stripeCoinPaymentsDB{db: dbc.getByName("stripecoinpayments")}
}
// Billing returns database for billing and payment transactions.
func (dbc *satelliteDBCollection) Billing() billing.TransactionsDB {
return &billingDB{db: dbc.getByName("billing")}
}
// Wallets returns database for storjscan wallets.
func (dbc *satelliteDBCollection) Wallets() storjscan.WalletsDB {
return &storjscanWalletsDB{db: dbc.getByName("storjscan")}
}
// SNOPayouts returns database for storagenode payStubs and payments info.
func (dbc *satelliteDBCollection) SNOPayouts() snopayouts.DB {
return &snopayoutsDB{db: dbc.getByName("snopayouts")}
}
// Compensation returns database for storage node compensation.
func (dbc *satelliteDBCollection) Compensation() compensation.DB {
return &compensationDB{db: dbc.getByName("compensation")}
}
// NodeAPIVersion returns database for storage node api version lower bounds.
func (dbc *satelliteDBCollection) NodeAPIVersion() nodeapiversion.DB {
return &nodeAPIVersionDB{db: dbc.getByName("nodeapiversion")}
}
// Buckets returns database for interacting with buckets.
func (dbc *satelliteDBCollection) Buckets() buckets.DB {
return &bucketsDB{db: dbc.getByName("buckets")}
}
// StorjscanPayments returns database for storjscan payments.
func (dbc *satelliteDBCollection) StorjscanPayments() storjscan.PaymentsDB {
return &storjscanPayments{db: dbc.getByName("storjscan_payments")}
}
// CheckVersion confirms all databases are at the desired version.
func (dbc *satelliteDBCollection) CheckVersion(ctx context.Context) error {
var eg errs.Group
for _, db := range dbc.dbs {
eg.Add(db.CheckVersion(ctx))
}
return eg.Err()
}
// MigrateToLatest migrates all databases to the latest version.
func (dbc *satelliteDBCollection) MigrateToLatest(ctx context.Context) error {
var eg errs.Group
for _, db := range dbc.dbs {
eg.Add(db.MigrateToLatest(ctx))
}
return eg.Err()
}
// Close closes all satellite dbs.
func (dbc *satelliteDBCollection) Close() error {
var eg errs.Group
for _, db := range dbc.dbs {
eg.Add(db.Close())
}
return eg.Err()
}
// Testing provides access to testing facilities. These should not be used in production code.
func (db *satelliteDB) Testing() satellite.TestingDB {
return &satelliteDBTesting{satelliteDB: db}
}
type satelliteDBTesting struct{ *satelliteDB }
// RawDB returns the underlying database connection to the primary database.
func (db *satelliteDBTesting) RawDB() tagsql.DB {
return db.satelliteDB.DB
}
// Schema returns the full schema for the database.
func (db *satelliteDBTesting) Schema() string {
return db.satelliteDB.Schema()
}
// ProductionMigration returns the primary migration.
func (db *satelliteDBTesting) ProductionMigration() *migrate.Migration {
return db.satelliteDB.ProductionMigration()
}
// TestMigration returns the migration used for tests.
func (db *satelliteDBTesting) TestMigration() *migrate.Migration {
return db.satelliteDB.TestMigration()
}
// Testing provides access to testing facilities. These should not be used in production code.
func (dbc *satelliteDBCollection) Testing() satellite.TestingDB {
return &satelliteDBCollectionTesting{satelliteDBCollection: dbc}
}
type satelliteDBCollectionTesting struct{ *satelliteDBCollection }
// RawDB returns the underlying database connection to the primary database.
func (dbc *satelliteDBCollectionTesting) RawDB() tagsql.DB {
return dbc.getByName("").DB.DB
}
// Schema returns the full schema for the database.
func (dbc *satelliteDBCollectionTesting) Schema() string {
return dbc.getByName("").Schema()
}
// MigrateToLatest initializes the database for testplanet.
func (dbc *satelliteDBCollectionTesting) TestMigrateToLatest(ctx context.Context) error {
var eg errs.Group
for _, db := range dbc.dbs {
eg.Add(db.Testing().TestMigrateToLatest(ctx))
}
return eg.Err()
}
// ProductionMigration returns the primary migration.
func (dbc *satelliteDBCollectionTesting) ProductionMigration() *migrate.Migration {
return dbc.getByName("").ProductionMigration()
}
// TestMigration returns the migration used for tests.
func (dbc *satelliteDBCollectionTesting) TestMigration() *migrate.Migration {
return dbc.getByName("").TestMigration()
}