forked from TrueCloudLab/frostfs-node
[#1437] blobovnicza: Fix contextcheck linter
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
parent
08182bd6f1
commit
ea190dd425
22 changed files with 82 additions and 81 deletions
|
@ -27,7 +27,7 @@ func openBlobovnicza(cmd *cobra.Command) *blobovnicza.Blobovnicza {
|
||||||
blobovnicza.WithPath(vPath),
|
blobovnicza.WithPath(vPath),
|
||||||
blobovnicza.WithReadOnly(true),
|
blobovnicza.WithReadOnly(true),
|
||||||
)
|
)
|
||||||
common.ExitOnErr(cmd, blz.Open())
|
common.ExitOnErr(cmd, blz.Open(cmd.Context()))
|
||||||
|
|
||||||
return blz
|
return blz
|
||||||
}
|
}
|
||||||
|
|
|
@ -69,7 +69,7 @@ func TestBlobovnicza(t *testing.T) {
|
||||||
defer os.Remove(p)
|
defer os.Remove(p)
|
||||||
|
|
||||||
// open Blobovnicza
|
// open Blobovnicza
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
|
|
||||||
// initialize Blobovnicza
|
// initialize Blobovnicza
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
|
@ -16,7 +16,7 @@ import (
|
||||||
//
|
//
|
||||||
// If the database file does not exist, it will be created automatically.
|
// If the database file does not exist, it will be created automatically.
|
||||||
// If blobovnicza is already open, does nothing.
|
// If blobovnicza is already open, does nothing.
|
||||||
func (b *Blobovnicza) Open() error {
|
func (b *Blobovnicza) Open(ctx context.Context) error {
|
||||||
b.controlMtx.Lock()
|
b.controlMtx.Lock()
|
||||||
defer b.controlMtx.Unlock()
|
defer b.controlMtx.Unlock()
|
||||||
|
|
||||||
|
@ -24,7 +24,7 @@ func (b *Blobovnicza) Open() error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
b.log.Debug(context.Background(), logs.BlobovniczaCreatingDirectoryForBoltDB,
|
b.log.Debug(ctx, logs.BlobovniczaCreatingDirectoryForBoltDB,
|
||||||
zap.String("path", b.path),
|
zap.String("path", b.path),
|
||||||
zap.Bool("ro", b.boltOptions.ReadOnly),
|
zap.Bool("ro", b.boltOptions.ReadOnly),
|
||||||
)
|
)
|
||||||
|
@ -38,7 +38,7 @@ func (b *Blobovnicza) Open() error {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
b.log.Debug(context.Background(), logs.BlobovniczaOpeningBoltDB,
|
b.log.Debug(ctx, logs.BlobovniczaOpeningBoltDB,
|
||||||
zap.String("path", b.path),
|
zap.String("path", b.path),
|
||||||
zap.Stringer("permissions", b.perm),
|
zap.Stringer("permissions", b.perm),
|
||||||
)
|
)
|
||||||
|
|
|
@ -26,7 +26,7 @@ func TestBlobovnicza_Get(t *testing.T) {
|
||||||
WithObjectSizeLimit(szLimit),
|
WithObjectSizeLimit(szLimit),
|
||||||
)
|
)
|
||||||
|
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -15,7 +15,7 @@ import (
|
||||||
func TestBlobovniczaIterate(t *testing.T) {
|
func TestBlobovniczaIterate(t *testing.T) {
|
||||||
filename := filepath.Join(t.TempDir(), "blob")
|
filename := filepath.Join(t.TempDir(), "blob")
|
||||||
b := New(WithPath(filename))
|
b := New(WithPath(filename))
|
||||||
require.NoError(t, b.Open())
|
require.NoError(t, b.Open(context.Background()))
|
||||||
require.NoError(t, b.Init())
|
require.NoError(t, b.Init())
|
||||||
|
|
||||||
data := [][]byte{{0, 1, 2, 3}, {5, 6, 7, 8}}
|
data := [][]byte{{0, 1, 2, 3}, {5, 6, 7, 8}}
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
package blobovniczatree
|
package blobovniczatree
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
|
@ -53,8 +54,8 @@ func newActiveDBManager(dbManager *dbManager, rootPath string) *activeDBManager
|
||||||
|
|
||||||
// GetOpenedActiveDBForLevel returns active DB for level.
|
// GetOpenedActiveDBForLevel returns active DB for level.
|
||||||
// DB must be closed after use.
|
// DB must be closed after use.
|
||||||
func (m *activeDBManager) GetOpenedActiveDBForLevel(lvlPath string) (*activeDB, error) {
|
func (m *activeDBManager) GetOpenedActiveDBForLevel(ctx context.Context, lvlPath string) (*activeDB, error) {
|
||||||
activeDB, err := m.getCurrentActiveIfOk(lvlPath)
|
activeDB, err := m.getCurrentActiveIfOk(ctx, lvlPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -62,7 +63,7 @@ func (m *activeDBManager) GetOpenedActiveDBForLevel(lvlPath string) (*activeDB,
|
||||||
return activeDB, nil
|
return activeDB, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
return m.updateAndGetActive(lvlPath)
|
return m.updateAndGetActive(ctx, lvlPath)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *activeDBManager) Open() {
|
func (m *activeDBManager) Open() {
|
||||||
|
@ -83,7 +84,7 @@ func (m *activeDBManager) Close() {
|
||||||
m.closed = true
|
m.closed = true
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *activeDBManager) getCurrentActiveIfOk(lvlPath string) (*activeDB, error) {
|
func (m *activeDBManager) getCurrentActiveIfOk(ctx context.Context, lvlPath string) (*activeDB, error) {
|
||||||
m.levelToActiveDBGuard.RLock()
|
m.levelToActiveDBGuard.RLock()
|
||||||
defer m.levelToActiveDBGuard.RUnlock()
|
defer m.levelToActiveDBGuard.RUnlock()
|
||||||
|
|
||||||
|
@ -96,7 +97,7 @@ func (m *activeDBManager) getCurrentActiveIfOk(lvlPath string) (*activeDB, error
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
blz, err := db.Open() // open db for usage, will be closed on activeDB.Close()
|
blz, err := db.Open(ctx) // open db for usage, will be closed on activeDB.Close()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -112,11 +113,11 @@ func (m *activeDBManager) getCurrentActiveIfOk(lvlPath string) (*activeDB, error
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error) {
|
func (m *activeDBManager) updateAndGetActive(ctx context.Context, lvlPath string) (*activeDB, error) {
|
||||||
m.levelLock.Lock(lvlPath)
|
m.levelLock.Lock(lvlPath)
|
||||||
defer m.levelLock.Unlock(lvlPath)
|
defer m.levelLock.Unlock(lvlPath)
|
||||||
|
|
||||||
current, err := m.getCurrentActiveIfOk(lvlPath)
|
current, err := m.getCurrentActiveIfOk(ctx, lvlPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -124,7 +125,7 @@ func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error)
|
||||||
return current, nil
|
return current, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
nextShDB, err := m.getNextSharedDB(lvlPath)
|
nextShDB, err := m.getNextSharedDB(ctx, lvlPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -133,7 +134,7 @@ func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error)
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
blz, err := nextShDB.Open() // open db for client, client must call Close() after usage
|
blz, err := nextShDB.Open(ctx) // open db for client, client must call Close() after usage
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -143,7 +144,7 @@ func (m *activeDBManager) updateAndGetActive(lvlPath string) (*activeDB, error)
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *activeDBManager) getNextSharedDB(lvlPath string) (*sharedDB, error) {
|
func (m *activeDBManager) getNextSharedDB(ctx context.Context, lvlPath string) (*sharedDB, error) {
|
||||||
var nextActiveDBIdx uint64
|
var nextActiveDBIdx uint64
|
||||||
hasActive, currentIdx := m.hasActiveDB(lvlPath)
|
hasActive, currentIdx := m.hasActiveDB(lvlPath)
|
||||||
if hasActive {
|
if hasActive {
|
||||||
|
@ -160,7 +161,7 @@ func (m *activeDBManager) getNextSharedDB(lvlPath string) (*sharedDB, error) {
|
||||||
|
|
||||||
path := filepath.Join(lvlPath, u64ToHexStringExt(nextActiveDBIdx))
|
path := filepath.Join(lvlPath, u64ToHexStringExt(nextActiveDBIdx))
|
||||||
next := m.dbManager.GetByPath(path)
|
next := m.dbManager.GetByPath(path)
|
||||||
_, err := next.Open() // open db to hold active DB open, will be closed if db is full, after m.replace or by activeDBManager.Close()
|
_, err := next.Open(ctx) // open db to hold active DB open, will be closed if db is full, after m.replace or by activeDBManager.Close()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -81,12 +81,12 @@ func (c *dbCache) Close() {
|
||||||
c.closed = true
|
c.closed = true
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *dbCache) GetOrCreate(path string) *sharedDB {
|
func (c *dbCache) GetOrCreate(ctx context.Context, path string) *sharedDB {
|
||||||
value := c.getExisted(path)
|
value := c.getExisted(path)
|
||||||
if value != nil {
|
if value != nil {
|
||||||
return value
|
return value
|
||||||
}
|
}
|
||||||
return c.create(path)
|
return c.create(ctx, path)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *dbCache) EvictAndMarkNonCached(path string) {
|
func (c *dbCache) EvictAndMarkNonCached(path string) {
|
||||||
|
@ -122,7 +122,7 @@ func (c *dbCache) getExisted(path string) *sharedDB {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *dbCache) create(path string) *sharedDB {
|
func (c *dbCache) create(ctx context.Context, path string) *sharedDB {
|
||||||
c.pathLock.Lock(path)
|
c.pathLock.Lock(path)
|
||||||
defer c.pathLock.Unlock(path)
|
defer c.pathLock.Unlock(path)
|
||||||
|
|
||||||
|
@ -133,7 +133,7 @@ func (c *dbCache) create(path string) *sharedDB {
|
||||||
|
|
||||||
value = c.dbManager.GetByPath(path)
|
value = c.dbManager.GetByPath(path)
|
||||||
|
|
||||||
_, err := value.Open() // open db to hold reference, closed by evictedDB.Close() or if cache closed
|
_, err := value.Open(ctx) // open db to hold reference, closed by evictedDB.Close() or if cache closed
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return value
|
return value
|
||||||
}
|
}
|
||||||
|
|
|
@ -46,7 +46,7 @@ func (b *Blobovniczas) initializeDBs(ctx context.Context) error {
|
||||||
eg.Go(func() error {
|
eg.Go(func() error {
|
||||||
p = strings.TrimSuffix(p, rebuildSuffix)
|
p = strings.TrimSuffix(p, rebuildSuffix)
|
||||||
shBlz := b.getBlobovniczaWithoutCaching(p)
|
shBlz := b.getBlobovniczaWithoutCaching(p)
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(egCtx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -91,8 +91,8 @@ func (b *Blobovniczas) Close() error {
|
||||||
// returns blobovnicza with path p
|
// returns blobovnicza with path p
|
||||||
//
|
//
|
||||||
// If blobovnicza is already cached, instance from cache is returned w/o changes.
|
// If blobovnicza is already cached, instance from cache is returned w/o changes.
|
||||||
func (b *Blobovniczas) getBlobovnicza(p string) *sharedDB {
|
func (b *Blobovniczas) getBlobovnicza(ctx context.Context, p string) *sharedDB {
|
||||||
return b.dbCache.GetOrCreate(p)
|
return b.dbCache.GetOrCreate(ctx, p)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *Blobovniczas) getBlobovniczaWithoutCaching(p string) *sharedDB {
|
func (b *Blobovniczas) getBlobovniczaWithoutCaching(p string) *sharedDB {
|
||||||
|
|
|
@ -16,13 +16,13 @@ func (b *Blobovniczas) ObjectsCount(ctx context.Context) (uint64, error) {
|
||||||
b.metrics.ObjectsCount(time.Since(startedAt), success)
|
b.metrics.ObjectsCount(time.Since(startedAt), success)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
_, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.ObjectsCount")
|
ctx, span := tracing.StartSpanFromContext(ctx, "Blobovniczas.ObjectsCount")
|
||||||
defer span.End()
|
defer span.End()
|
||||||
|
|
||||||
var result uint64
|
var result uint64
|
||||||
err := b.iterateExistingDBPaths(ctx, func(p string) (bool, error) {
|
err := b.iterateExistingDBPaths(ctx, func(p string) (bool, error) {
|
||||||
shDB := b.getBlobovniczaWithoutCaching(p)
|
shDB := b.getBlobovniczaWithoutCaching(p)
|
||||||
blz, err := shDB.Open()
|
blz, err := shDB.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return true, err
|
return true, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -61,8 +61,8 @@ func (b *Blobovniczas) Delete(ctx context.Context, prm common.DeletePrm) (res co
|
||||||
|
|
||||||
if prm.StorageID != nil {
|
if prm.StorageID != nil {
|
||||||
id := NewIDFromBytes(prm.StorageID)
|
id := NewIDFromBytes(prm.StorageID)
|
||||||
shBlz := b.getBlobovnicza(id.Path())
|
shBlz := b.getBlobovnicza(ctx, id.Path())
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return res, err
|
return res, err
|
||||||
}
|
}
|
||||||
|
@ -109,8 +109,8 @@ func (b *Blobovniczas) Delete(ctx context.Context, prm common.DeletePrm) (res co
|
||||||
//
|
//
|
||||||
// returns no error if object was removed from some blobovnicza of the same level.
|
// returns no error if object was removed from some blobovnicza of the same level.
|
||||||
func (b *Blobovniczas) deleteObjectFromLevel(ctx context.Context, prm blobovnicza.DeletePrm, blzPath string) (common.DeleteRes, error) {
|
func (b *Blobovniczas) deleteObjectFromLevel(ctx context.Context, prm blobovnicza.DeletePrm, blzPath string) (common.DeleteRes, error) {
|
||||||
shBlz := b.getBlobovnicza(blzPath)
|
shBlz := b.getBlobovnicza(ctx, blzPath)
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return common.DeleteRes{}, err
|
return common.DeleteRes{}, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -37,8 +37,8 @@ func (b *Blobovniczas) Exists(ctx context.Context, prm common.ExistsPrm) (common
|
||||||
|
|
||||||
if prm.StorageID != nil {
|
if prm.StorageID != nil {
|
||||||
id := NewIDFromBytes(prm.StorageID)
|
id := NewIDFromBytes(prm.StorageID)
|
||||||
shBlz := b.getBlobovnicza(id.Path())
|
shBlz := b.getBlobovnicza(ctx, id.Path())
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return common.ExistsRes{}, err
|
return common.ExistsRes{}, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -48,8 +48,8 @@ func (b *Blobovniczas) Get(ctx context.Context, prm common.GetPrm) (res common.G
|
||||||
|
|
||||||
if prm.StorageID != nil {
|
if prm.StorageID != nil {
|
||||||
id := NewIDFromBytes(prm.StorageID)
|
id := NewIDFromBytes(prm.StorageID)
|
||||||
shBlz := b.getBlobovnicza(id.Path())
|
shBlz := b.getBlobovnicza(ctx, id.Path())
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return res, err
|
return res, err
|
||||||
}
|
}
|
||||||
|
@ -95,8 +95,8 @@ func (b *Blobovniczas) Get(ctx context.Context, prm common.GetPrm) (res common.G
|
||||||
// returns error if object could not be read from any blobovnicza of the same level.
|
// returns error if object could not be read from any blobovnicza of the same level.
|
||||||
func (b *Blobovniczas) getObjectFromLevel(ctx context.Context, prm blobovnicza.GetPrm, blzPath string) (common.GetRes, error) {
|
func (b *Blobovniczas) getObjectFromLevel(ctx context.Context, prm blobovnicza.GetPrm, blzPath string) (common.GetRes, error) {
|
||||||
// open blobovnicza (cached inside)
|
// open blobovnicza (cached inside)
|
||||||
shBlz := b.getBlobovnicza(blzPath)
|
shBlz := b.getBlobovnicza(ctx, blzPath)
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return common.GetRes{}, err
|
return common.GetRes{}, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -47,8 +47,8 @@ func (b *Blobovniczas) GetRange(ctx context.Context, prm common.GetRangePrm) (re
|
||||||
|
|
||||||
if prm.StorageID != nil {
|
if prm.StorageID != nil {
|
||||||
id := NewIDFromBytes(prm.StorageID)
|
id := NewIDFromBytes(prm.StorageID)
|
||||||
shBlz := b.getBlobovnicza(id.Path())
|
shBlz := b.getBlobovnicza(ctx, id.Path())
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return common.GetRangeRes{}, err
|
return common.GetRangeRes{}, err
|
||||||
}
|
}
|
||||||
|
@ -103,8 +103,8 @@ func (b *Blobovniczas) GetRange(ctx context.Context, prm common.GetRangePrm) (re
|
||||||
// returns error if object could not be read from any blobovnicza of the same level.
|
// returns error if object could not be read from any blobovnicza of the same level.
|
||||||
func (b *Blobovniczas) getRangeFromLevel(ctx context.Context, prm common.GetRangePrm, blzPath string) (common.GetRangeRes, error) {
|
func (b *Blobovniczas) getRangeFromLevel(ctx context.Context, prm common.GetRangePrm, blzPath string) (common.GetRangeRes, error) {
|
||||||
// open blobovnicza (cached inside)
|
// open blobovnicza (cached inside)
|
||||||
shBlz := b.getBlobovnicza(blzPath)
|
shBlz := b.getBlobovnicza(ctx, blzPath)
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return common.GetRangeRes{}, err
|
return common.GetRangeRes{}, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -72,8 +72,8 @@ func (b *Blobovniczas) Iterate(ctx context.Context, prm common.IteratePrm) (comm
|
||||||
// iterator over all Blobovniczas in unsorted order. Break on f's error return.
|
// iterator over all Blobovniczas in unsorted order. Break on f's error return.
|
||||||
func (b *Blobovniczas) iterateBlobovniczas(ctx context.Context, ignoreErrors bool, f func(string, *blobovnicza.Blobovnicza) error) error {
|
func (b *Blobovniczas) iterateBlobovniczas(ctx context.Context, ignoreErrors bool, f func(string, *blobovnicza.Blobovnicza) error) error {
|
||||||
return b.iterateExistingDBPaths(ctx, func(p string) (bool, error) {
|
return b.iterateExistingDBPaths(ctx, func(p string) (bool, error) {
|
||||||
shBlz := b.getBlobovnicza(p)
|
shBlz := b.getBlobovnicza(ctx, p)
|
||||||
blz, err := shBlz.Open()
|
blz, err := shBlz.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if ignoreErrors {
|
if ignoreErrors {
|
||||||
b.log.Warn(ctx, logs.BlobstorErrorOccurredDuringTheIteration,
|
b.log.Warn(ctx, logs.BlobstorErrorOccurredDuringTheIteration,
|
||||||
|
|
|
@ -49,7 +49,7 @@ func newSharedDB(options []blobovnicza.Option, path string, readOnly bool,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *sharedDB) Open() (*blobovnicza.Blobovnicza, error) {
|
func (b *sharedDB) Open(ctx context.Context) (*blobovnicza.Blobovnicza, error) {
|
||||||
if b.closedFlag.Load() {
|
if b.closedFlag.Load() {
|
||||||
return nil, errClosed
|
return nil, errClosed
|
||||||
}
|
}
|
||||||
|
@ -68,7 +68,7 @@ func (b *sharedDB) Open() (*blobovnicza.Blobovnicza, error) {
|
||||||
blobovnicza.WithMetrics(b.metrics),
|
blobovnicza.WithMetrics(b.metrics),
|
||||||
)...)
|
)...)
|
||||||
|
|
||||||
if err := blz.Open(); err != nil {
|
if err := blz.Open(ctx); err != nil {
|
||||||
return nil, fmt.Errorf("could not open blobovnicza %s: %w", b.path, err)
|
return nil, fmt.Errorf("could not open blobovnicza %s: %w", b.path, err)
|
||||||
}
|
}
|
||||||
if err := blz.Init(); err != nil {
|
if err := blz.Init(); err != nil {
|
||||||
|
|
|
@ -77,7 +77,7 @@ type putIterator struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (i *putIterator) iterate(ctx context.Context, lvlPath string) (bool, error) {
|
func (i *putIterator) iterate(ctx context.Context, lvlPath string) (bool, error) {
|
||||||
active, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(lvlPath)
|
active, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(ctx, lvlPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if !isLogical(err) {
|
if !isLogical(err) {
|
||||||
i.B.reportError(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, err)
|
i.B.reportError(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, err)
|
||||||
|
|
|
@ -165,7 +165,7 @@ func (b *Blobovniczas) selectDBsDoNotMatchFillPercent(ctx context.Context, targe
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
path := filepath.Join(lvlPath, e.Name())
|
path := filepath.Join(lvlPath, e.Name())
|
||||||
resettlementRequired, err := b.rebuildBySize(path, target)
|
resettlementRequired, err := b.rebuildBySize(ctx, path, target)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
@ -180,9 +180,9 @@ func (b *Blobovniczas) selectDBsDoNotMatchFillPercent(ctx context.Context, targe
|
||||||
return result, nil
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *Blobovniczas) rebuildBySize(path string, targetFillPercent int) (bool, error) {
|
func (b *Blobovniczas) rebuildBySize(ctx context.Context, path string, targetFillPercent int) (bool, error) {
|
||||||
shDB := b.getBlobovnicza(path)
|
shDB := b.getBlobovnicza(ctx, path)
|
||||||
blz, err := shDB.Open()
|
blz, err := shDB.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
@ -196,8 +196,8 @@ func (b *Blobovniczas) rebuildBySize(path string, targetFillPercent int) (bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *Blobovniczas) rebuildDB(ctx context.Context, path string, meta common.MetaStorage, limiter common.ConcurrentWorkersLimiter) (uint64, error) {
|
func (b *Blobovniczas) rebuildDB(ctx context.Context, path string, meta common.MetaStorage, limiter common.ConcurrentWorkersLimiter) (uint64, error) {
|
||||||
shDB := b.getBlobovnicza(path)
|
shDB := b.getBlobovnicza(ctx, path)
|
||||||
blz, err := shDB.Open()
|
blz, err := shDB.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
}
|
}
|
||||||
|
@ -365,8 +365,8 @@ func (b *Blobovniczas) completeIncompletedMove(ctx context.Context, metaStore co
|
||||||
err := b.iterateIncompletedRebuildDBPaths(ctx, func(s string) (bool, error) {
|
err := b.iterateIncompletedRebuildDBPaths(ctx, func(s string) (bool, error) {
|
||||||
rebuildTmpFilePath := s
|
rebuildTmpFilePath := s
|
||||||
s = strings.TrimSuffix(s, rebuildSuffix)
|
s = strings.TrimSuffix(s, rebuildSuffix)
|
||||||
shDB := b.getBlobovnicza(s)
|
shDB := b.getBlobovnicza(ctx, s)
|
||||||
blz, err := shDB.Open()
|
blz, err := shDB.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return true, err
|
return true, err
|
||||||
}
|
}
|
||||||
|
@ -398,8 +398,8 @@ func (b *Blobovniczas) completeIncompletedMove(ctx context.Context, metaStore co
|
||||||
func (b *Blobovniczas) performMove(ctx context.Context, source *blobovnicza.Blobovnicza, sourcePath string,
|
func (b *Blobovniczas) performMove(ctx context.Context, source *blobovnicza.Blobovnicza, sourcePath string,
|
||||||
move blobovnicza.MoveInfo, metaStore common.MetaStorage,
|
move blobovnicza.MoveInfo, metaStore common.MetaStorage,
|
||||||
) error {
|
) error {
|
||||||
targetDB := b.getBlobovnicza(NewIDFromBytes(move.TargetStorageID).Path())
|
targetDB := b.getBlobovnicza(ctx, NewIDFromBytes(move.TargetStorageID).Path())
|
||||||
target, err := targetDB.Open()
|
target, err := targetDB.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -477,7 +477,7 @@ type moveIterator struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (i *moveIterator) tryMoveToLvl(ctx context.Context, lvlPath string) (bool, error) {
|
func (i *moveIterator) tryMoveToLvl(ctx context.Context, lvlPath string) (bool, error) {
|
||||||
target, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(lvlPath)
|
target, err := i.B.activeDBManager.GetOpenedActiveDBForLevel(ctx, lvlPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if !isLogical(err) {
|
if !isLogical(err) {
|
||||||
i.B.reportError(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, err)
|
i.B.reportError(logs.BlobovniczatreeCouldNotGetActiveBlobovnicza, err)
|
||||||
|
|
|
@ -35,7 +35,7 @@ func testRebuildFailoverOnlyMoveInfoSaved(t *testing.T) {
|
||||||
dir := t.TempDir()
|
dir := t.TempDir()
|
||||||
|
|
||||||
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
||||||
obj := blobstortest.NewObject(1024)
|
obj := blobstortest.NewObject(1024)
|
||||||
|
@ -65,7 +65,7 @@ func testRebuildFailoverObjectSavedToTarget(t *testing.T) {
|
||||||
dir := t.TempDir()
|
dir := t.TempDir()
|
||||||
|
|
||||||
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
||||||
obj := blobstortest.NewObject(1024)
|
obj := blobstortest.NewObject(1024)
|
||||||
|
@ -89,7 +89,7 @@ func testRebuildFailoverObjectSavedToTarget(t *testing.T) {
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
|
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
||||||
_, err = blz.Put(context.Background(), pPrm)
|
_, err = blz.Put(context.Background(), pPrm)
|
||||||
|
@ -105,7 +105,7 @@ func testRebuildFailoverObjectDeletedFromSource(t *testing.T) {
|
||||||
dir := t.TempDir()
|
dir := t.TempDir()
|
||||||
|
|
||||||
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
||||||
obj := blobstortest.NewObject(1024)
|
obj := blobstortest.NewObject(1024)
|
||||||
|
@ -123,7 +123,7 @@ func testRebuildFailoverObjectDeletedFromSource(t *testing.T) {
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
|
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
||||||
var pPrm blobovnicza.PutPrm
|
var pPrm blobovnicza.PutPrm
|
||||||
|
@ -173,7 +173,7 @@ func testRebuildFailoverValidate(t *testing.T, dir string, obj *objectSDK.Object
|
||||||
require.NoError(t, b.Close())
|
require.NoError(t, b.Close())
|
||||||
|
|
||||||
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
blz := blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "1.db")))
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
||||||
moveInfo, err := blz.ListMoveInfo(context.Background())
|
moveInfo, err := blz.ListMoveInfo(context.Background())
|
||||||
|
@ -188,7 +188,7 @@ func testRebuildFailoverValidate(t *testing.T, dir string, obj *objectSDK.Object
|
||||||
require.NoError(t, blz.Close())
|
require.NoError(t, blz.Close())
|
||||||
|
|
||||||
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
|
blz = blobovnicza.New(blobovnicza.WithPath(filepath.Join(dir, "0", "0", "0.db")))
|
||||||
require.NoError(t, blz.Open())
|
require.NoError(t, blz.Open(context.Background()))
|
||||||
require.NoError(t, blz.Init())
|
require.NoError(t, blz.Init())
|
||||||
|
|
||||||
moveInfo, err = blz.ListMoveInfo(context.Background())
|
moveInfo, err = blz.ListMoveInfo(context.Background())
|
||||||
|
|
|
@ -269,8 +269,8 @@ type containerSource struct {
|
||||||
|
|
||||||
func (s *containerSource) IsContainerAvailable(ctx context.Context, id cid.ID) (bool, error) {
|
func (s *containerSource) IsContainerAvailable(ctx context.Context, id cid.ID) (bool, error) {
|
||||||
select {
|
select {
|
||||||
case <-context.Background().Done():
|
case <-ctx.Done():
|
||||||
return false, context.Background().Err()
|
return false, ctx.Err()
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -389,7 +389,7 @@ func (c *Client) Wait(ctx context.Context, n uint32) error {
|
||||||
|
|
||||||
height, err = c.rpcActor.GetBlockCount()
|
height, err = c.rpcActor.GetBlockCount()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.logger.Error(context.Background(), logs.ClientCantGetBlockchainHeight,
|
c.logger.Error(ctx, logs.ClientCantGetBlockchainHeight,
|
||||||
zap.String("error", err.Error()))
|
zap.String("error", err.Error()))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
@ -403,7 +403,7 @@ func (c *Client) Wait(ctx context.Context, n uint32) error {
|
||||||
|
|
||||||
newHeight, err = c.rpcActor.GetBlockCount()
|
newHeight, err = c.rpcActor.GetBlockCount()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.logger.Error(context.Background(), logs.ClientCantGetBlockchainHeight243,
|
c.logger.Error(ctx, logs.ClientCantGetBlockchainHeight243,
|
||||||
zap.String("error", err.Error()))
|
zap.String("error", err.Error()))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
|
@ -269,7 +269,7 @@ loop:
|
||||||
continue loop
|
continue loop
|
||||||
}
|
}
|
||||||
|
|
||||||
l.handleNotifyEvent(notifyEvent)
|
l.handleNotifyEvent(ctx, notifyEvent)
|
||||||
case notaryEvent, ok := <-chs.NotaryRequestsCh:
|
case notaryEvent, ok := <-chs.NotaryRequestsCh:
|
||||||
if !ok {
|
if !ok {
|
||||||
l.log.Warn(ctx, logs.EventStopEventListenerByNotaryChannel)
|
l.log.Warn(ctx, logs.EventStopEventListenerByNotaryChannel)
|
||||||
|
@ -316,16 +316,16 @@ func (l *listener) handleNotaryEvent(notaryEvent *result.NotaryRequestEvent) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *listener) handleNotifyEvent(notifyEvent *state.ContainedNotificationEvent) {
|
func (l *listener) handleNotifyEvent(ctx context.Context, notifyEvent *state.ContainedNotificationEvent) {
|
||||||
if err := l.pool.Submit(func() {
|
if err := l.pool.Submit(func() {
|
||||||
l.parseAndHandleNotification(notifyEvent)
|
l.parseAndHandleNotification(ctx, notifyEvent)
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
l.log.Warn(context.Background(), logs.EventListenerWorkerPoolDrained,
|
l.log.Warn(ctx, logs.EventListenerWorkerPoolDrained,
|
||||||
zap.Int("capacity", l.pool.Cap()))
|
zap.Int("capacity", l.pool.Cap()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotificationEvent) {
|
func (l *listener) parseAndHandleNotification(ctx context.Context, notifyEvent *state.ContainedNotificationEvent) {
|
||||||
log := l.log.With(
|
log := l.log.With(
|
||||||
zap.String("script hash LE", notifyEvent.ScriptHash.StringLE()),
|
zap.String("script hash LE", notifyEvent.ScriptHash.StringLE()),
|
||||||
)
|
)
|
||||||
|
@ -347,7 +347,7 @@ func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotifi
|
||||||
l.mtx.RUnlock()
|
l.mtx.RUnlock()
|
||||||
|
|
||||||
if !ok {
|
if !ok {
|
||||||
log.Debug(context.Background(), logs.EventEventParserNotSet)
|
log.Debug(ctx, logs.EventEventParserNotSet)
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
@ -355,7 +355,7 @@ func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotifi
|
||||||
// parse the notification event
|
// parse the notification event
|
||||||
event, err := parser(notifyEvent)
|
event, err := parser(notifyEvent)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn(context.Background(), logs.EventCouldNotParseNotificationEvent,
|
log.Warn(ctx, logs.EventCouldNotParseNotificationEvent,
|
||||||
zap.String("error", err.Error()),
|
zap.String("error", err.Error()),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -368,7 +368,7 @@ func (l *listener) parseAndHandleNotification(notifyEvent *state.ContainedNotifi
|
||||||
l.mtx.RUnlock()
|
l.mtx.RUnlock()
|
||||||
|
|
||||||
if len(handlers) == 0 {
|
if len(handlers) == 0 {
|
||||||
log.Info(context.Background(), logs.EventNotificationHandlersForParsedNotificationEventWereNotRegistered,
|
log.Info(ctx, logs.EventNotificationHandlersForParsedNotificationEventWereNotRegistered,
|
||||||
zap.Any("event", event),
|
zap.Any("event", event),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
@ -254,7 +254,7 @@ func (s *subscriber) switchEndpoint(ctx context.Context, finishCh chan<- bool) b
|
||||||
s.Lock()
|
s.Lock()
|
||||||
chs := newSubChannels()
|
chs := newSubChannels()
|
||||||
go func() {
|
go func() {
|
||||||
finishCh <- s.restoreSubscriptions(chs.NotifyChan, chs.BlockChan, chs.NotaryChan)
|
finishCh <- s.restoreSubscriptions(ctx, chs.NotifyChan, chs.BlockChan, chs.NotaryChan)
|
||||||
}()
|
}()
|
||||||
s.current = chs
|
s.current = chs
|
||||||
s.Unlock()
|
s.Unlock()
|
||||||
|
@ -295,7 +295,7 @@ drainloop:
|
||||||
|
|
||||||
// restoreSubscriptions restores subscriptions according to
|
// restoreSubscriptions restores subscriptions according to
|
||||||
// cached information about them.
|
// cached information about them.
|
||||||
func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotificationEvent,
|
func (s *subscriber) restoreSubscriptions(ctx context.Context, notifCh chan<- *state.ContainedNotificationEvent,
|
||||||
blCh chan<- *block.Block, notaryCh chan<- *result.NotaryRequestEvent,
|
blCh chan<- *block.Block, notaryCh chan<- *result.NotaryRequestEvent,
|
||||||
) bool {
|
) bool {
|
||||||
var err error
|
var err error
|
||||||
|
@ -304,7 +304,7 @@ func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotific
|
||||||
if s.subscribedToNewBlocks {
|
if s.subscribedToNewBlocks {
|
||||||
_, err = s.client.ReceiveBlocks(blCh)
|
_, err = s.client.ReceiveBlocks(blCh)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.log.Error(context.Background(), logs.ClientCouldNotRestoreBlockSubscriptionAfterRPCSwitch, zap.Error(err))
|
s.log.Error(ctx, logs.ClientCouldNotRestoreBlockSubscriptionAfterRPCSwitch, zap.Error(err))
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -313,7 +313,7 @@ func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotific
|
||||||
for contract := range s.subscribedEvents {
|
for contract := range s.subscribedEvents {
|
||||||
_, err = s.client.ReceiveExecutionNotifications(contract, notifCh)
|
_, err = s.client.ReceiveExecutionNotifications(contract, notifCh)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.log.Error(context.Background(), logs.ClientCouldNotRestoreNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
|
s.log.Error(ctx, logs.ClientCouldNotRestoreNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -322,7 +322,7 @@ func (s *subscriber) restoreSubscriptions(notifCh chan<- *state.ContainedNotific
|
||||||
for signer := range s.subscribedNotaryEvents {
|
for signer := range s.subscribedNotaryEvents {
|
||||||
_, err = s.client.ReceiveNotaryRequests(signer, notaryCh)
|
_, err = s.client.ReceiveNotaryRequests(signer, notaryCh)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.log.Error(context.Background(), logs.ClientCouldNotRestoreNotaryNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
|
s.log.Error(ctx, logs.ClientCouldNotRestoreNotaryNotificationSubscriptionAfterRPCSwitch, zap.Error(err))
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue