frostfs-node/pkg/ape/chainbase/boltdb.go

330 lines
8.8 KiB
Go

package chainbase
import (
"context"
"errors"
"fmt"
"path/filepath"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
"git.frostfs.info/TrueCloudLab/policy-engine/pkg/chain"
policyengine "git.frostfs.info/TrueCloudLab/policy-engine/pkg/engine"
"github.com/nspcc-dev/neo-go/pkg/util/slice"
"go.etcd.io/bbolt"
)
type boltLocalOverrideStorage struct {
*cfg
db *bbolt.DB
}
var chainBucket = []byte{0}
var (
// ErrRootBucketNotFound signals the database has not been properly initialized.
ErrRootBucketNotFound = logicerr.New("root bucket not found")
ErrGlobalNamespaceBucketNotFound = logicerr.New("global namespace bucket not found")
ErrTargetTypeBucketNotFound = logicerr.New("target type bucket not found")
ErrTargetNameBucketNotFound = logicerr.New("target name bucket not found")
ErrBucketNotContainsChainID = logicerr.New("chain id not found in bucket")
errChainIDIsNotSet = errors.New("chain ID is not set")
)
// NewBoltLocalOverrideDatabase returns storage wrapper for storing access policy engine
// local overrides.
//
// chain storage (chainBucket):
// -> global namespace bucket (nBucket):
// --> target bucket (tBucket)
// ---> target name (resource) bucket (rBucket):
//
// | Key | Value |
// x---------------------x-------------------x
// | chain id (string) | serialized chain |
// x---------------------x-------------------x
//
//nolint:godot
func NewBoltLocalOverrideDatabase(opts ...Option) LocalOverrideDatabase {
c := defaultCfg()
for i := range opts {
opts[i](c)
}
return &boltLocalOverrideStorage{
cfg: c,
}
}
func (cs *boltLocalOverrideStorage) Init() error {
return cs.db.Update(func(tx *bbolt.Tx) error {
_, err := tx.CreateBucketIfNotExists(chainBucket)
return err
})
}
func (cs *boltLocalOverrideStorage) Open(context.Context) error {
err := util.MkdirAllX(filepath.Dir(cs.path), cs.perm)
if err != nil {
return fmt.Errorf("can't create dir %s for the chain DB: %w", cs.path, err)
}
opts := *bbolt.DefaultOptions
opts.NoSync = cs.noSync
opts.Timeout = 100 * time.Millisecond
cs.db, err = bbolt.Open(cs.path, cs.perm, &opts)
if err != nil {
return fmt.Errorf("can't open the chain DB: %w", err)
}
cs.db.MaxBatchSize = cs.maxBatchSize
cs.db.MaxBatchDelay = cs.maxBatchDelay
return nil
}
func (cs *boltLocalOverrideStorage) Close() error {
var err error
if cs.db != nil {
err = cs.db.Close()
}
return err
}
func getTypeBucket(tx *bbolt.Tx, name chain.Name, target policyengine.Target) (*bbolt.Bucket, error) {
cbucket := tx.Bucket(chainBucket)
if cbucket == nil {
return nil, ErrRootBucketNotFound
}
nbucket := cbucket.Bucket([]byte(name))
if nbucket == nil {
return nil, fmt.Errorf("%w: %w: %s", policyengine.ErrChainNotFound, ErrGlobalNamespaceBucketNotFound, name)
}
return nbucket.Bucket([]byte{byte(target.Type)}), nil
}
func normalizeTargetName(target *policyengine.Target) {
if target.Type == policyengine.Namespace && target.Name == "" {
target.Name = "root"
}
}
func getTargetBucket(tx *bbolt.Tx, name chain.Name, target policyengine.Target) (*bbolt.Bucket, error) {
typeBucket, err := getTypeBucket(tx, name, target)
if err != nil {
return nil, err
}
if typeBucket == nil {
return nil, fmt.Errorf("%w: %w: %c", policyengine.ErrChainNotFound, ErrTargetTypeBucketNotFound, target.Type)
}
normalizeTargetName(&target)
rbucket := typeBucket.Bucket([]byte(target.Name))
if rbucket == nil {
return nil, fmt.Errorf("%w: %w: %s", policyengine.ErrChainNotFound, ErrTargetNameBucketNotFound, target.Name)
}
return rbucket, nil
}
func getTargetBucketCreateIfEmpty(tx *bbolt.Tx, name chain.Name, target policyengine.Target) (*bbolt.Bucket, error) {
cbucket := tx.Bucket(chainBucket)
if cbucket == nil {
return nil, ErrRootBucketNotFound
}
nbucket := cbucket.Bucket([]byte(name))
if nbucket == nil {
var err error
nbucket, err = cbucket.CreateBucket([]byte(name))
if err != nil {
return nil, fmt.Errorf("could not create a bucket for the global chain name %s: %w", name, err)
}
}
typeBucket := nbucket.Bucket([]byte{byte(target.Type)})
if typeBucket == nil {
var err error
typeBucket, err = nbucket.CreateBucket([]byte{byte(target.Type)})
if err != nil {
return nil, fmt.Errorf("could not create a bucket for the target type '%c': %w", target.Type, err)
}
}
normalizeTargetName(&target)
rbucket := typeBucket.Bucket([]byte(target.Name))
if rbucket == nil {
var err error
rbucket, err = typeBucket.CreateBucket([]byte(target.Name))
if err != nil {
return nil, fmt.Errorf("could not create a bucket for the target name %s: %w", target.Name, err)
}
}
return rbucket, nil
}
func (cs *boltLocalOverrideStorage) AddOverride(name chain.Name, target policyengine.Target, c *chain.Chain) (chain.ID, error) {
if len(c.ID) == 0 {
return chain.ID{}, errChainIDIsNotSet
}
serializedChain := c.Bytes()
err := cs.db.Update(func(tx *bbolt.Tx) error {
rbuck, err := getTargetBucketCreateIfEmpty(tx, name, target)
if err != nil {
return err
}
return rbuck.Put([]byte(c.ID), serializedChain)
})
return c.ID, err
}
func (cs *boltLocalOverrideStorage) GetOverride(name chain.Name, target policyengine.Target, chainID chain.ID) (*chain.Chain, error) {
var serializedChain []byte
if err := cs.db.View(func(tx *bbolt.Tx) error {
rbuck, err := getTargetBucket(tx, name, target)
if err != nil {
return err
}
serializedChain = rbuck.Get([]byte(chainID))
if serializedChain == nil {
return fmt.Errorf("%w: %w: %s", policyengine.ErrChainNotFound, ErrBucketNotContainsChainID, chainID)
}
serializedChain = slice.Copy(serializedChain)
return nil
}); err != nil {
return nil, err
}
c := &chain.Chain{}
if err := c.DecodeBytes(serializedChain); err != nil {
return nil, err
}
return c, nil
}
func (cs *boltLocalOverrideStorage) RemoveOverride(name chain.Name, target policyengine.Target, chainID chain.ID) error {
return cs.db.Update(func(tx *bbolt.Tx) error {
rbuck, err := getTargetBucket(tx, name, target)
if err != nil {
return err
}
return rbuck.Delete([]byte(chainID))
})
}
func (cs *boltLocalOverrideStorage) RemoveOverridesByTarget(name chain.Name, target policyengine.Target) error {
return cs.db.Update(func(tx *bbolt.Tx) error {
typeBucket, err := getTypeBucket(tx, name, target)
if err != nil {
return err
}
normalizeTargetName(&target)
return typeBucket.DeleteBucket([]byte(target.Name))
})
}
func (cs *boltLocalOverrideStorage) ListOverrides(name chain.Name, target policyengine.Target) ([]*chain.Chain, error) {
var serializedChains [][]byte
var serializedChain []byte
if err := cs.db.View(func(tx *bbolt.Tx) error {
rbuck, err := getTargetBucket(tx, name, target)
if err != nil {
return err
}
return rbuck.ForEach(func(_, v []byte) error {
serializedChain = slice.Copy(v)
serializedChains = append(serializedChains, serializedChain)
return nil
})
}); err != nil {
if errors.Is(err, policyengine.ErrChainNotFound) {
return []*chain.Chain{}, nil
}
return nil, err
}
chains := make([]*chain.Chain, 0, len(serializedChains))
for _, serializedChain = range serializedChains {
c := &chain.Chain{}
if err := c.DecodeBytes(serializedChain); err != nil {
return nil, err
}
chains = append(chains, c)
}
return chains, nil
}
func (cs *boltLocalOverrideStorage) DropAllOverrides(name chain.Name) error {
return cs.db.Update(func(tx *bbolt.Tx) error {
cbucket := tx.Bucket(chainBucket)
if cbucket == nil {
return ErrRootBucketNotFound
}
nbucket := cbucket.Bucket([]byte(name))
if nbucket == nil {
return fmt.Errorf("%w: %w: global namespace %s", policyengine.ErrChainNotFound, ErrGlobalNamespaceBucketNotFound, name)
}
return tx.DeleteBucket([]byte(name))
})
}
func (cs *boltLocalOverrideStorage) ListOverrideDefinedTargets(name chain.Name) ([]policyengine.Target, error) {
var targets []policyengine.Target
if err := cs.db.View(func(tx *bbolt.Tx) error {
var err error
targets, err = getTargets(tx, name)
if err != nil {
return err
}
return nil
}); err != nil {
return nil, err
}
return targets, nil
}
func getTargets(tx *bbolt.Tx, name chain.Name) ([]policyengine.Target, error) {
var targets []policyengine.Target
cbucket := tx.Bucket(chainBucket)
if cbucket == nil {
return nil, ErrRootBucketNotFound
}
nbucket := cbucket.Bucket([]byte(name))
if nbucket == nil {
return nil, fmt.Errorf("%w: %w: %s", policyengine.ErrChainNotFound, ErrGlobalNamespaceBucketNotFound, name)
}
if err := nbucket.ForEachBucket(func(k []byte) error {
ttype := policyengine.TargetType(k[0])
if err := nbucket.Bucket(k).ForEachBucket(func(k []byte) error {
targets = append(targets, policyengine.Target{
Type: ttype,
Name: string(slice.Copy(k)),
})
return nil
}); err != nil {
return err
}
return nil
}); err != nil {
return nil, err
}
return targets, nil
}