[#9999] writecache: Move DB related code to upgrade.go
Some checks failed
DCO action / DCO (pull_request) Successful in 52s
Build / Build Components (pull_request) Failing after 1m8s
Vulncheck / Vulncheck (pull_request) Failing after 1m15s
Tests and linters / Run gofumpt (pull_request) Successful in 1m39s
Tests and linters / Staticcheck (pull_request) Failing after 1m43s
Pre-commit hooks / Pre-commit (pull_request) Successful in 1m52s
Tests and linters / gopls check (pull_request) Failing after 1m50s
Tests and linters / Lint (pull_request) Failing after 2m21s
Tests and linters / Tests (pull_request) Failing after 2m27s
Tests and linters / Tests with -race (pull_request) Failing after 2m27s

This is done to drop this file in the future.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2024-09-11 17:08:52 +03:00
parent d1c44de01c
commit 3b1795d22f
4 changed files with 110 additions and 108 deletions

View file

@ -1,12 +1,8 @@
package writecache
import (
"bytes"
"context"
"errors"
"fmt"
"os"
"path/filepath"
"time"
"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-sdk-go/client"
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/trace"
"go.uber.org/zap"
@ -232,83 +226,3 @@ func (c *cache) Flush(ctx context.Context, ignoreErrors, seal bool) error {
func (c *cache) flush(ctx context.Context, ignoreErrors bool) error {
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
}

View file

@ -16,8 +16,6 @@ import (
"go.uber.org/zap"
)
const dbName = "small.bolt"
func (c *cache) openStore(mod mode.ComponentMode) error {
err := util.MkdirAllX(c.path, os.ModePerm)
if err != nil {

View 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,
})
}

View file

@ -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,
})
}