xk6-frostfs/internal/registry/obj_registry.go
Evgenii Stratonikov 22d7996f79 [#42] registry: Optimize ObjectInfo marshaling
1. Get rid of JSON in the database.
2. Store `CreatedAt` as int64. It decreases JSON marshaling time by
   about ~25% with no changes for native scheme.

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
2023-05-15 16:08:26 +03:00

128 lines
2.7 KiB
Go

package registry
import (
"context"
"encoding/binary"
"errors"
"os"
"time"
"go.etcd.io/bbolt"
)
type ObjRegistry struct {
ctx context.Context
cancel context.CancelFunc
boltDB *bbolt.DB
}
const (
// Indicates that an object was created, but its data wasn't verified yet.
statusCreated = "created"
)
const bucketName = "_object"
// NewObjRegistry creates a new instance of object registry that stores information
// about objects in the specified bolt database. As registry uses read-write
// connection to the database, there may be only one instance of object registry
// per database file at a time.
func NewObjRegistry(ctx context.Context, dbFilePath string) *ObjRegistry {
options := bbolt.Options{Timeout: 100 * time.Millisecond}
boltDB, err := bbolt.Open(dbFilePath, os.ModePerm, &options)
if err != nil {
panic(err)
}
ctx, cancel := context.WithCancel(ctx)
objRepository := &ObjRegistry{
ctx: ctx,
cancel: cancel,
boltDB: boltDB,
}
return objRepository
}
func (o *ObjRegistry) AddObject(cid, oid, s3Bucket, s3Key, payloadHash string) error {
return o.boltDB.Update(func(tx *bbolt.Tx) error {
b, err := tx.CreateBucketIfNotExists([]byte(bucketName))
if err != nil {
return err
}
id, err := b.NextSequence()
if err != nil {
return err
}
object := ObjectInfo{
Id: id,
CreatedAt: time.Now().UTC().Unix(),
CID: cid,
OID: oid,
S3Bucket: s3Bucket,
S3Key: s3Key,
PayloadHash: payloadHash,
Status: statusCreated,
}
objBytes, err := object.Marshal()
if err != nil {
return err
}
return b.Put(encodeId(id), objBytes)
})
}
func (o *ObjRegistry) SetObjectStatus(id uint64, newStatus string) error {
return o.boltDB.Update(func(tx *bbolt.Tx) error {
b, err := tx.CreateBucketIfNotExists([]byte(bucketName))
if err != nil {
return err
}
objBytes := b.Get(encodeId(id))
if objBytes == nil {
return errors.New("object doesn't exist")
}
obj := new(ObjectInfo)
if err := obj.Unmarshal(objBytes); err != nil {
return err
}
obj.Status = newStatus
objBytes, err = obj.Marshal()
if err != nil {
return err
}
return b.Put(encodeId(id), objBytes)
})
}
func (o *ObjRegistry) DeleteObject(id uint64) error {
return o.boltDB.Update(func(tx *bbolt.Tx) error {
b, err := tx.CreateBucketIfNotExists([]byte(bucketName))
if err != nil {
return err
}
return b.Delete(encodeId(id))
})
}
func (o *ObjRegistry) Close() error {
o.cancel()
return o.boltDB.Close()
}
func encodeId(id uint64) []byte {
idBytes := make([]byte, 8)
binary.BigEndian.PutUint64(idBytes, id)
return idBytes
}
func decodeId(idBytes []byte) uint64 {
return binary.BigEndian.Uint64(idBytes)
}