forked from TrueCloudLab/frostfs-node
258 lines
7 KiB
Go
258 lines
7 KiB
Go
package chainbase
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"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")
|
|
)
|
|
|
|
// 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 getTargetBucket(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)
|
|
}
|
|
|
|
typeBucket := nbucket.Bucket([]byte{byte(target.Type)})
|
|
if typeBucket == nil {
|
|
return nil, fmt.Errorf("%w: %w: %c", policyengine.ErrChainNotFound, ErrTargetTypeBucketNotFound, target.Type)
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
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 c.ID == "" {
|
|
return "", fmt.Errorf("chain ID is not set")
|
|
}
|
|
|
|
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 := json.Unmarshal(serializedChain, c); 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) 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 := json.Unmarshal(serializedChain, c); 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))
|
|
})
|
|
}
|