forked from TrueCloudLab/frostfs-node
[#1367] writecache: Move DB related code to upgrade.go
This is done to drop this file in the future. Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
parent
3b236160a6
commit
f345fe9a58
5 changed files with 111 additions and 112 deletions
|
@ -45,10 +45,7 @@ const (
|
||||||
defaultMaxCacheSize = 1 << 30 // 1 GiB
|
defaultMaxCacheSize = 1 << 30 // 1 GiB
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var dummyCanceler context.CancelFunc = func() {}
|
||||||
defaultBucket = []byte{0}
|
|
||||||
dummyCanceler context.CancelFunc = func() {}
|
|
||||||
)
|
|
||||||
|
|
||||||
// New creates new writecache instance.
|
// New creates new writecache instance.
|
||||||
func New(opts ...Option) Cache {
|
func New(opts ...Option) Cache {
|
||||||
|
|
|
@ -1,12 +1,8 @@
|
||||||
package writecache
|
package writecache
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
|
||||||
|
@ -20,8 +16,6 @@ import (
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
|
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
|
||||||
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
|
||||||
"go.etcd.io/bbolt"
|
|
||||||
"go.opentelemetry.io/otel/attribute"
|
"go.opentelemetry.io/otel/attribute"
|
||||||
"go.opentelemetry.io/otel/trace"
|
"go.opentelemetry.io/otel/trace"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
|
@ -239,83 +233,3 @@ func (c *cache) Flush(ctx context.Context, ignoreErrors, seal bool) error {
|
||||||
func (c *cache) flush(ctx context.Context, ignoreErrors bool) error {
|
func (c *cache) flush(ctx context.Context, ignoreErrors bool) error {
|
||||||
return c.flushFSTree(ctx, ignoreErrors)
|
return c.flushFSTree(ctx, ignoreErrors)
|
||||||
}
|
}
|
||||||
|
|
||||||
type batchItem struct {
|
|
||||||
data []byte
|
|
||||||
address string
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *cache) flushAndDropBBoltDB(ctx context.Context) error {
|
|
||||||
_, err := os.Stat(filepath.Join(c.path, dbName))
|
|
||||||
if err != nil && os.IsNotExist(err) {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("could not check write-cache database existence: %w", err)
|
|
||||||
}
|
|
||||||
db, err := OpenDB(c.path, true, os.OpenFile)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("could not open write-cache database: %w", err)
|
|
||||||
}
|
|
||||||
defer func() {
|
|
||||||
_ = db.Close()
|
|
||||||
}()
|
|
||||||
|
|
||||||
var last string
|
|
||||||
for {
|
|
||||||
batch, err := c.readNextDBBatch(db, last)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if len(batch) == 0 {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
for _, item := range batch {
|
|
||||||
var obj objectSDK.Object
|
|
||||||
if err := obj.Unmarshal(item.data); err != nil {
|
|
||||||
return fmt.Errorf("unmarshal object from database: %w", err)
|
|
||||||
}
|
|
||||||
if err := c.flushObject(ctx, &obj, item.data, StorageTypeDB); err != nil {
|
|
||||||
return fmt.Errorf("flush object from database: %w", err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
last = batch[len(batch)-1].address
|
|
||||||
}
|
|
||||||
if err := db.Close(); err != nil {
|
|
||||||
return fmt.Errorf("close write-cache database: %w", err)
|
|
||||||
}
|
|
||||||
if err := os.Remove(filepath.Join(c.path, dbName)); err != nil {
|
|
||||||
return fmt.Errorf("remove write-cache database: %w", err)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *cache) readNextDBBatch(db *bbolt.DB, last string) ([]batchItem, error) {
|
|
||||||
const batchSize = 100
|
|
||||||
var batch []batchItem
|
|
||||||
err := db.View(func(tx *bbolt.Tx) error {
|
|
||||||
var addr oid.Address
|
|
||||||
|
|
||||||
b := tx.Bucket(defaultBucket)
|
|
||||||
cs := b.Cursor()
|
|
||||||
for k, data := cs.Seek([]byte(last)); k != nil; k, data = cs.Next() {
|
|
||||||
sa := string(k)
|
|
||||||
if sa == last {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if err := addr.DecodeString(sa); err != nil {
|
|
||||||
return fmt.Errorf("decode address from database: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
batch = append(batch, batchItem{data: bytes.Clone(data), address: sa})
|
|
||||||
if len(batch) == batchSize {
|
|
||||||
return errIterationCompleted
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
if err == nil || errors.Is(err, errIterationCompleted) {
|
|
||||||
return batch, nil
|
|
||||||
}
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
|
@ -16,8 +16,6 @@ import (
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
const dbName = "small.bolt"
|
|
||||||
|
|
||||||
func (c *cache) openStore(mod mode.ComponentMode) error {
|
func (c *cache) openStore(mod mode.ComponentMode) error {
|
||||||
err := util.MkdirAllX(c.path, os.ModePerm)
|
err := util.MkdirAllX(c.path, os.ModePerm)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
110
pkg/local_object_storage/writecache/upgrade.go
Normal file
110
pkg/local_object_storage/writecache/upgrade.go
Normal file
|
@ -0,0 +1,110 @@
|
||||||
|
package writecache
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io/fs"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
|
"go.etcd.io/bbolt"
|
||||||
|
)
|
||||||
|
|
||||||
|
const dbName = "small.bolt"
|
||||||
|
|
||||||
|
var defaultBucket = []byte{0}
|
||||||
|
|
||||||
|
func (c *cache) flushAndDropBBoltDB(ctx context.Context) error {
|
||||||
|
_, err := os.Stat(filepath.Join(c.path, dbName))
|
||||||
|
if err != nil && os.IsNotExist(err) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("could not check write-cache database existence: %w", err)
|
||||||
|
}
|
||||||
|
db, err := OpenDB(c.path, true, os.OpenFile)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("could not open write-cache database: %w", err)
|
||||||
|
}
|
||||||
|
defer func() {
|
||||||
|
_ = db.Close()
|
||||||
|
}()
|
||||||
|
|
||||||
|
var last string
|
||||||
|
for {
|
||||||
|
batch, err := c.readNextDBBatch(db, last)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if len(batch) == 0 {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
for _, item := range batch {
|
||||||
|
var obj objectSDK.Object
|
||||||
|
if err := obj.Unmarshal(item.data); err != nil {
|
||||||
|
return fmt.Errorf("unmarshal object from database: %w", err)
|
||||||
|
}
|
||||||
|
if err := c.flushObject(ctx, &obj, item.data, StorageTypeDB); err != nil {
|
||||||
|
return fmt.Errorf("flush object from database: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
last = batch[len(batch)-1].address
|
||||||
|
}
|
||||||
|
if err := db.Close(); err != nil {
|
||||||
|
return fmt.Errorf("close write-cache database: %w", err)
|
||||||
|
}
|
||||||
|
if err := os.Remove(filepath.Join(c.path, dbName)); err != nil {
|
||||||
|
return fmt.Errorf("remove write-cache database: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type batchItem struct {
|
||||||
|
data []byte
|
||||||
|
address string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *cache) readNextDBBatch(db *bbolt.DB, last string) ([]batchItem, error) {
|
||||||
|
const batchSize = 100
|
||||||
|
var batch []batchItem
|
||||||
|
err := db.View(func(tx *bbolt.Tx) error {
|
||||||
|
var addr oid.Address
|
||||||
|
|
||||||
|
b := tx.Bucket(defaultBucket)
|
||||||
|
cs := b.Cursor()
|
||||||
|
for k, data := cs.Seek([]byte(last)); k != nil; k, data = cs.Next() {
|
||||||
|
sa := string(k)
|
||||||
|
if sa == last {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err := addr.DecodeString(sa); err != nil {
|
||||||
|
return fmt.Errorf("decode address from database: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
batch = append(batch, batchItem{data: bytes.Clone(data), address: sa})
|
||||||
|
if len(batch) == batchSize {
|
||||||
|
return errIterationCompleted
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err == nil || errors.Is(err, errIterationCompleted) {
|
||||||
|
return batch, nil
|
||||||
|
}
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// OpenDB opens BoltDB instance for write-cache. Opens in read-only mode if ro is true.
|
||||||
|
func OpenDB(p string, ro bool, openFile func(string, int, fs.FileMode) (*os.File, error)) (*bbolt.DB, error) {
|
||||||
|
return bbolt.Open(filepath.Join(p, dbName), os.ModePerm, &bbolt.Options{
|
||||||
|
NoFreelistSync: true,
|
||||||
|
ReadOnly: ro,
|
||||||
|
Timeout: 100 * time.Millisecond,
|
||||||
|
OpenFile: openFile,
|
||||||
|
})
|
||||||
|
}
|
|
@ -1,20 +0,0 @@
|
||||||
package writecache
|
|
||||||
|
|
||||||
import (
|
|
||||||
"io/fs"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"go.etcd.io/bbolt"
|
|
||||||
)
|
|
||||||
|
|
||||||
// OpenDB opens BoltDB instance for write-cache. Opens in read-only mode if ro is true.
|
|
||||||
func OpenDB(p string, ro bool, openFile func(string, int, fs.FileMode) (*os.File, error)) (*bbolt.DB, error) {
|
|
||||||
return bbolt.Open(filepath.Join(p, dbName), os.ModePerm, &bbolt.Options{
|
|
||||||
NoFreelistSync: true,
|
|
||||||
ReadOnly: ro,
|
|
||||||
Timeout: 100 * time.Millisecond,
|
|
||||||
OpenFile: openFile,
|
|
||||||
})
|
|
||||||
}
|
|
Loading…
Reference in a new issue