1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399
|
package badger
import (
"bytes"
"encoding/binary"
"strings"
"github.com/dgraph-io/badger"
"github.com/dgraph-io/badger/options"
"github.com/pkg/errors"
"github.com/smallstep/nosql/database"
)
// DB is a wrapper over *badger.DB,
type DB struct {
db *badger.DB
}
// Open opens or creates a BoltDB database in the given path.
func (db *DB) Open(dir string, opt ...database.Option) (err error) {
opts := &database.Options{}
for _, o := range opt {
if err := o(opts); err != nil {
return err
}
}
bo := badger.DefaultOptions(dir)
// Set the Table and Value LoadingMode - default is MemoryMap. Low memory/RAM
// systems may want to use FileIO.
switch strings.ToLower(opts.BadgerFileLoadingMode) {
case "", database.BadgerMemoryMap, "memorymap":
bo.TableLoadingMode = options.MemoryMap
bo.ValueLogLoadingMode = options.MemoryMap
case database.BadgerFileIO:
bo.TableLoadingMode = options.FileIO
bo.ValueLogLoadingMode = options.FileIO
default:
return badger.ErrInvalidLoadingMode
}
if opts.ValueDir != "" {
bo.ValueDir = opts.ValueDir
} else {
bo.ValueDir = dir
}
db.db, err = badger.Open(bo)
return errors.Wrap(err, "error opening Badger database")
}
// Close closes the DB database.
func (db *DB) Close() error {
return errors.Wrap(db.db.Close(), "error closing Badger database")
}
// CreateTable creates a token element with the 'bucket' prefix so that such
// that their appears to be a table.
func (db *DB) CreateTable(bucket []byte) error {
bk, err := badgerEncode(bucket)
if err != nil {
return err
}
return db.db.Update(func(txn *badger.Txn) error {
return errors.Wrapf(txn.Set(bk, []byte{}), "failed to create %s/", bucket)
})
}
// DeleteTable deletes a root or embedded bucket. Returns an error if the
// bucket cannot be found or if the key represents a non-bucket value.
func (db *DB) DeleteTable(bucket []byte) error {
var tableExists bool
prefix, err := badgerEncode(bucket)
if err != nil {
return err
}
deleteKeys := func(keysForDelete [][]byte) error {
if err := db.db.Update(func(txn *badger.Txn) error {
for _, key := range keysForDelete {
tableExists = true
if err := txn.Delete(key); err != nil {
return errors.Wrapf(err, "error deleting key %s", key)
}
}
return nil
}); err != nil {
return errors.Wrapf(err, "update failed")
}
return nil
}
collectSize := 1000
err = db.db.View(func(txn *badger.Txn) error {
opts := badger.DefaultIteratorOptions
opts.AllVersions = false
opts.PrefetchValues = false
it := txn.NewIterator(opts)
defer it.Close()
keysForDelete := make([][]byte, collectSize)
keysCollected := 0
for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
key := it.Item().KeyCopy(nil)
keysForDelete[keysCollected] = key
keysCollected++
if keysCollected == collectSize {
if err := deleteKeys(keysForDelete); err != nil {
return err
}
keysCollected = 0
}
}
if keysCollected > 0 {
if err := deleteKeys(keysForDelete[:keysCollected]); err != nil {
return err
}
}
if !tableExists {
return errors.Wrapf(database.ErrNotFound, "table %s does not exist", bucket)
}
return nil
})
return err
}
// badgerGet is a helper for the Get method.
func badgerGet(txn *badger.Txn, key []byte) ([]byte, error) {
item, err := txn.Get(key)
switch {
case err == badger.ErrKeyNotFound:
return nil, errors.Wrapf(database.ErrNotFound, "key %s not found", key)
case err != nil:
return nil, errors.Wrapf(err, "failed to get key %s", key)
default:
val, err := item.ValueCopy(nil)
if err != nil {
return nil, errors.Wrap(err, "error accessing value returned by database")
}
return val, nil
}
}
// Get returns the value stored in the given bucked and key.
func (db *DB) Get(bucket, key []byte) (ret []byte, err error) {
bk, err := toBadgerKey(bucket, key)
if err != nil {
return nil, errors.Wrapf(err, "error converting %s/%s to badgerKey", bucket, key)
}
err = db.db.View(func(txn *badger.Txn) error {
ret, err = badgerGet(txn, bk)
return err
})
return
}
// Set stores the given value on bucket and key.
func (db *DB) Set(bucket, key, value []byte) error {
bk, err := toBadgerKey(bucket, key)
if err != nil {
return errors.Wrapf(err, "error converting %s/%s to badgerKey", bucket, key)
}
return db.db.Update(func(txn *badger.Txn) error {
return errors.Wrapf(txn.Set(bk, value), "failed to set %s/%s", bucket, key)
})
}
// Del deletes the value stored in the given bucked and key.
func (db *DB) Del(bucket, key []byte) error {
bk, err := toBadgerKey(bucket, key)
if err != nil {
return errors.Wrapf(err, "error converting %s/%s to badgerKey", bucket, key)
}
return db.db.Update(func(txn *badger.Txn) error {
return errors.Wrapf(txn.Delete(bk), "failed to delete %s/%s", bucket, key)
})
}
// List returns the full list of entries in a bucket.
func (db *DB) List(bucket []byte) ([]*database.Entry, error) {
var (
entries []*database.Entry
tableExists bool
)
err := db.db.View(func(txn *badger.Txn) error {
it := txn.NewIterator(badger.DefaultIteratorOptions)
defer it.Close()
prefix, err := badgerEncode(bucket)
if err != nil {
return err
}
for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
tableExists = true
item := it.Item()
bk := item.KeyCopy(nil)
if isBadgerTable(bk) {
continue
}
_bucket, key, err := fromBadgerKey(bk)
if err != nil {
return errors.Wrapf(err, "error converting from badgerKey %s", bk)
}
if !bytes.Equal(_bucket, bucket) {
return errors.Errorf("bucket names do not match; want %v, but got %v",
bucket, _bucket)
}
v, err := item.ValueCopy(nil)
if err != nil {
return errors.Wrap(err, "error retrieving contents from database value")
}
entries = append(entries, &database.Entry{
Bucket: _bucket,
Key: key,
Value: v,
})
}
if !tableExists {
return errors.Wrapf(database.ErrNotFound, "bucket %s not found", bucket)
}
return nil
})
return entries, err
}
// CmpAndSwap modifies the value at the given bucket and key (to newValue)
// only if the existing (current) value matches oldValue.
func (db *DB) CmpAndSwap(bucket, key, oldValue, newValue []byte) ([]byte, bool, error) {
bk, err := toBadgerKey(bucket, key)
if err != nil {
return nil, false, err
}
badgerTxn := db.db.NewTransaction(true)
defer badgerTxn.Discard()
val, swapped, err := cmpAndSwap(badgerTxn, bk, oldValue, newValue)
switch {
case err != nil:
return nil, false, err
case swapped:
if err := badgerTxn.Commit(); err != nil {
return nil, false, errors.Wrapf(err, "failed to commit badger transaction")
}
return val, swapped, nil
default:
return val, swapped, err
}
}
func cmpAndSwap(badgerTxn *badger.Txn, bk, oldValue, newValue []byte) ([]byte, bool, error) {
current, err := badgerGet(badgerTxn, bk)
// If value does not exist but expected is not nil, then return w/out swapping.
if err != nil && !database.IsErrNotFound(err) {
return nil, false, err
}
if !bytes.Equal(current, oldValue) {
return current, false, nil
}
if err := badgerTxn.Set(bk, newValue); err != nil {
return current, false, errors.Wrapf(err, "failed to set %s", bk)
}
return newValue, true, nil
}
// Update performs multiple commands on one read-write transaction.
func (db *DB) Update(txn *database.Tx) error {
return db.db.Update(func(badgerTxn *badger.Txn) (err error) {
for _, q := range txn.Operations {
switch q.Cmd {
case database.CreateTable:
if err = db.CreateTable(q.Bucket); err != nil {
return err
}
continue
case database.DeleteTable:
if err = db.DeleteTable(q.Bucket); err != nil {
return err
}
continue
}
bk, err := toBadgerKey(q.Bucket, q.Key)
if err != nil {
return err
}
switch q.Cmd {
case database.Get:
if q.Result, err = badgerGet(badgerTxn, bk); err != nil {
return errors.Wrapf(err, "failed to get %s/%s", q.Bucket, q.Key)
}
case database.Set:
if err := badgerTxn.Set(bk, q.Value); err != nil {
return errors.Wrapf(err, "failed to set %s/%s", q.Bucket, q.Key)
}
case database.Delete:
if err = badgerTxn.Delete(bk); err != nil {
return errors.Wrapf(err, "failed to delete %s/%s", q.Bucket, q.Key)
}
case database.CmpAndSwap:
q.Result, q.Swapped, err = cmpAndSwap(badgerTxn, bk, q.CmpValue, q.Value)
if err != nil {
return errors.Wrapf(err, "failed to CmpAndSwap %s/%s", q.Bucket, q.Key)
}
case database.CmpOrRollback:
return database.ErrOpNotSupported
default:
return database.ErrOpNotSupported
}
}
return nil
})
}
// toBadgerKey returns the Badger database key using the following algorithm:
// First 2 bytes are the length of the bucket/table name in little endian format,
// followed by the bucket/table name,
// followed by 2 bytes representing the length of the key in little endian format,
// followed by the key.
func toBadgerKey(bucket, key []byte) ([]byte, error) {
first, err := badgerEncode(bucket)
if err != nil {
return nil, err
}
second, err := badgerEncode(key)
if err != nil {
return nil, err
}
return append(first, second...), nil
}
// isBadgerTable returns True if the slice is a badgerTable token, false otherwise.
// badgerTable means that the slice contains only the [size|value] of one section
// of a badgerKey and no remainder. A badgerKey is [buket|key], while a badgerTable
// is only the bucket section.
func isBadgerTable(bk []byte) bool {
if k, rest := parseBadgerEncode(bk); len(k) > 0 && len(rest) == 0 {
return true
}
return false
}
// fromBadgerKey returns the bucket and key encoded in a BadgerKey.
// See documentation for toBadgerKey.
func fromBadgerKey(bk []byte) ([]byte, []byte, error) {
bucket, rest := parseBadgerEncode(bk)
if len(bucket) == 0 || len(rest) == 0 {
return nil, nil, errors.Errorf("invalid badger key: %v", bk)
}
key, rest2 := parseBadgerEncode(rest)
if len(key) == 0 || len(rest2) != 0 {
return nil, nil, errors.Errorf("invalid badger key: %v", bk)
}
return bucket, key, nil
}
// badgerEncode encodes a byte slice into a section of a BadgerKey.
// See documentation for toBadgerKey.
func badgerEncode(val []byte) ([]byte, error) {
l := len(val)
switch {
case l == 0:
return nil, errors.Errorf("input cannot be empty")
case l > 65535:
return nil, errors.Errorf("length of input cannot be greater than 65535")
default:
lb := new(bytes.Buffer)
if err := binary.Write(lb, binary.LittleEndian, uint16(l)); err != nil {
return nil, errors.Wrap(err, "error doing binary Write")
}
return append(lb.Bytes(), val...), nil
}
}
func parseBadgerEncode(bk []byte) (value, rest []byte) {
var (
keyLen uint16
start = uint16(2)
length = uint16(len(bk))
)
if uint16(len(bk)) < start {
return nil, bk
}
// First 2 bytes stores the length of the value.
if err := binary.Read(bytes.NewReader(bk[:2]), binary.LittleEndian, &keyLen); err != nil {
return nil, bk
}
end := start + keyLen
switch {
case length < end:
return nil, bk
case length == end:
return bk[start:end], nil
default:
return bk[start:end], bk[end:]
}
}
|