refactor(dirpush): implement DirAware storage functionality with new sharedkv implementations and configuration.

Author: sendya

conf/conf.go

@@ -44,8 +44,9 @@ type Server struct {
 }
 
 type ServerPProf struct {
-	Username string `json:"username" yaml:"username"`
-	Password string `json:"password" yaml:"password"`
+	AllowLocalRequest bool   `json:"allow_local_request" yaml:"allow_local_request"`
+	Username          string `json:"username" yaml:"username"`
+	Password          string `json:"password" yaml:"password"`
 }
 
 type ServerAccessLog struct {
@@ -76,9 +77,20 @@ type Storage struct {
 	EvictionPolicy  string    `json:"eviction_policy" yaml:"eviction_policy"`
 	SelectionPolicy string    `json:"selection_policy" yaml:"selection_policy"`
 	SliceSize       uint64    `json:"slice_size" yaml:"slice_size"`
+	DirAware        *DirAware `json:"diraware" yaml:"diraware"`
 	Buckets         []*Bucket `json:"buckets" yaml:"buckets"`
 }
 
+func (r *Storage) FillDefault() {
+	if r.DirAware == nil {
+		r.DirAware = &DirAware{
+			Enabled:   true,
+			StorePath: "/tmp/.diraware",
+			AutoClear: true,
+		}
+	}
+}
+
 type Bucket struct {
 	Path           string         `json:"path" yaml:"path"`                         // local path or ?
 	Driver         string         `json:"driver" yaml:"driver"`                     // native, custom-driver
@@ -91,6 +103,12 @@ type Bucket struct {
 	DBConfig       map[string]any `json:"db_config" yaml:"db_config"`               // custom db config
 }
 
+type DirAware struct {
+	Enabled   bool   `json:"enabled" yaml:"enabled"`       // 目录推送标记删除功能开关
+	StorePath string `json:"store_path" yaml:"store_path"` // 推送任务储存路径(建议SSD)
+	AutoClear bool   `json:"auto_clear" yaml:"auto_clear"` // 自动清理过期任务(凌晨2点左右执行)
+}
+
 type Plugin struct {
 	Name    string         `json:"name" yaml:"name"`
 	Options map[string]any `json:"options" yaml:"options"`

config.example.yaml

@@ -93,6 +93,10 @@ storage:
   eviction_policy: fifo # fifo, lru, lfu
   selection_policy: hashring # hashring, roundrobin
   slice_size: 1048576 # 1MB
+  diraware:
+    enabled: true # 默认 true
+    store_path: /cache1/.diraware
+    auto_clear: true 
   buckets:
     - path: /cache1
       type: normal

main.go

@@ -37,7 +37,6 @@ import (
 	"github.com/omalloc/tavern/proxy"
 	"github.com/omalloc/tavern/server"
 	"github.com/omalloc/tavern/storage"
-	"github.com/omalloc/tavern/storage/marked"
 )
 
 var (
@@ -114,11 +113,10 @@ func newApp(bc *conf.Bootstrap, logger log.Logger) (*kratos.App, error) {
 	}
 
 	// init storage
-	store, err := storage.New(bc.Storage, log.GetLogger())
+	store, err := storage.New(bc.Storage, logger)
 	if err != nil {
 		log.Fatalf("failed to initialize storage: %v", err)
 	}
-	store = marked.WrapStorage(store, marked.NewSharedKVChecker(store.SharedKV()))
 	storage.SetDefault(store)
 
 	// init upstream

storage/bucket/disk/disk.go

@@ -69,7 +69,7 @@ func New(config *conf.Bucket, sharedkv storage.SharedKV) (storage.Bucket, error)
 	db, err := indexdb.Create(config.DBType,
 		indexdb.NewOption(config.DBPath, indexdb.WithType("pebble"), indexdb.WithDBConfig(config.DBConfig)))
 	if err != nil {
-		log.Errorf("failed to create %s indexdb %v", config.DBType, err)
+		log.Errorf("failed to create %s(%s) indexdb %v", config.DBType, config.DBPath, err)
 		return nil, err
 	}
 	bucket.indexdb = db

storage/marked/bucket.go storage/diraware/bucket.gostorage/marked/bucket.go renamed to storage/diraware/bucket.go

@@ -1,4 +1,4 @@
-package marked
+package diraware
 
 import (
 	"context"

storage/marked/marked.go storage/diraware/diraware.gostorage/marked/marked.go renamed to storage/diraware/diraware.go

@@ -1,4 +1,4 @@
-package marked
+package diraware
 
 import (
 	"context"
@@ -37,7 +37,7 @@ func WithAutoClear(clear bool) SharedKVOption {
 	}
 }
 
-func NewSharedKVChecker(kv storagev1.SharedKV, opts ...SharedKVOption) Checker {
+func NewChecker(kv storagev1.SharedKV, opts ...SharedKVOption) Checker {
 	c := &checker{
 		KV:        kv,
 		pathtrie:  pathtrie.NewPathTrie[string, int64](),

storage/marked/wrap_storage.go storage/diraware/storage.gostorage/marked/wrap_storage.go renamed to storage/diraware/storage.go

@@ -1,4 +1,4 @@
-package marked
+package diraware
 
 import (
 	"context"
@@ -14,9 +14,9 @@ type Checker interface {
 	TrieAdd(ctx context.Context, storePath string)
 }
 
-// WrapStorage wraps a storage with push-mark logic.
+// New wraps a storage with push-mark logic.
 // If checker is nil, returns the original storage.
-func WrapStorage(base storagev1.Storage, checker Checker) storagev1.Storage {
+func New(base storagev1.Storage, checker Checker) storagev1.Storage {
 	if base == nil || checker == nil {
 		return base
 	}

storage/sharedkv/memkv.go

@@ -0,0 +1,24 @@
+package sharedkv
+
+import (
+	"github.com/cockroachdb/pebble/v2"
+	"github.com/cockroachdb/pebble/v2/vfs"
+	"github.com/omalloc/tavern/api/defined/v1/storage"
+	"github.com/omalloc/tavern/contrib/log"
+)
+
+// NewMemSharedKV create a new memory kv store
+func NewMemSharedKV() storage.SharedKV {
+	opts := &pebble.Options{
+		FS:         vfs.NewMem(),
+		DisableWAL: true,
+		Logger:     log.NewHelper(log.NewFilter(log.GetLogger(), log.FilterLevel(log.LevelWarn))),
+	}
+
+	db, err := newNoneKV("", opts)
+	if err != nil {
+		panic(err)
+	}
+
+	return db
+}

storage/sharedkv/kv_pebble.go storage/sharedkv/nonekv.gostorage/sharedkv/kv_pebble.go renamed to storage/sharedkv/nonekv.go

@@ -6,22 +6,20 @@ import (
 	"errors"
 
 	"github.com/cockroachdb/pebble/v2"
-	"github.com/cockroachdb/pebble/v2/vfs"
 	"github.com/omalloc/tavern/api/defined/v1/storage"
-	"github.com/omalloc/tavern/contrib/log"
 )
 
-var _ storage.SharedKV = (*memSharedKV)(nil)
+var _ storage.SharedKV = (*noneSharedKV)(nil)
 
-type memSharedKV struct {
+type noneSharedKV struct {
 	db *pebble.DB
 }
 
-func (r *memSharedKV) Close() error {
+func (r *noneSharedKV) Close() error {
 	return r.db.Close()
 }
 
-func (r *memSharedKV) Get(_ context.Context, key []byte) ([]byte, error) {
+func (r *noneSharedKV) Get(_ context.Context, key []byte) ([]byte, error) {
 	val, c, err := r.db.Get(key)
 	if err != nil {
 		if errors.Is(err, pebble.ErrNotFound) {
@@ -35,11 +33,11 @@ func (r *memSharedKV) Get(_ context.Context, key []byte) ([]byte, error) {
 	return val, nil
 }
 
-func (r *memSharedKV) Set(_ context.Context, key []byte, val []byte) error {
+func (r *noneSharedKV) Set(_ context.Context, key []byte, val []byte) error {
 	return r.db.Set(key, val, pebble.NoSync)
 }
 
-func (r *memSharedKV) Incr(_ context.Context, key []byte, delta uint32) (uint32, error) {
+func (r *noneSharedKV) Incr(_ context.Context, key []byte, delta uint32) (uint32, error) {
 	batch := r.db.NewIndexedBatch()
 	defer func() { _ = batch.Close() }()
 
@@ -70,7 +68,7 @@ func (r *memSharedKV) Incr(_ context.Context, key []byte, delta uint32) (uint32,
 	return counter, nil
 }
 
-func (r *memSharedKV) Decr(_ context.Context, key []byte, delta uint32) (uint32, error) {
+func (r *noneSharedKV) Decr(_ context.Context, key []byte, delta uint32) (uint32, error) {
 	batch := r.db.NewIndexedBatch()
 	defer func() { _ = batch.Close() }()
 
@@ -105,7 +103,7 @@ func (r *memSharedKV) Decr(_ context.Context, key []byte, delta uint32) (uint32,
 	return counter, nil
 }
 
-func (r *memSharedKV) GetCounter(_ context.Context, key []byte) (uint32, error) {
+func (r *noneSharedKV) GetCounter(_ context.Context, key []byte) (uint32, error) {
 	val, closer, err := r.db.Get(key)
 	if err != nil {
 		return 0, err
@@ -115,19 +113,19 @@ func (r *memSharedKV) GetCounter(_ context.Context, key []byte) (uint32, error)
 	return binary.BigEndian.Uint32(val), nil
 }
 
-func (r *memSharedKV) Delete(_ context.Context, key []byte) error {
+func (r *noneSharedKV) Delete(_ context.Context, key []byte) error {
 	return r.db.Delete(key, pebble.NoSync)
 }
 
-func (r *memSharedKV) DropPrefix(ctx context.Context, prefix []byte) error {
+func (r *noneSharedKV) DropPrefix(ctx context.Context, prefix []byte) error {
 	end := make([]byte, len(prefix))
 	copy(end, prefix)
 	end[len(end)-1]++
 
 	return r.db.DeleteRange(prefix, end, pebble.NoSync)
 }
 
-func (r *memSharedKV) Iterate(ctx context.Context, f func(key []byte, val []byte) error) error {
+func (r *noneSharedKV) Iterate(ctx context.Context, f func(key []byte, val []byte) error) error {
 	iter, err := r.db.NewIterWithContext(ctx, &pebble.IterOptions{})
 	if err != nil {
 		return err
@@ -148,7 +146,7 @@ func (r *memSharedKV) Iterate(ctx context.Context, f func(key []byte, val []byte
 	return nil
 }
 
-func (r *memSharedKV) IteratePrefix(ctx context.Context, prefix []byte, f func(key []byte, val []byte) error) error {
+func (r *noneSharedKV) IteratePrefix(ctx context.Context, prefix []byte, f func(key []byte, val []byte) error) error {
 	iter, err := r.db.NewIterWithContext(ctx, &pebble.IterOptions{
 		LowerBound: prefix,
 		UpperBound: keyUpperBound(prefix),
@@ -184,17 +182,14 @@ func keyUpperBound(b []byte) []byte {
 	return nil // no upper-bound
 }
 
-func NewMemSharedKV() storage.SharedKV {
-	db, err := pebble.Open("", &pebble.Options{
-		FS:     vfs.NewMem(),
-		Logger: log.NewHelper(log.NewFilter(log.GetLogger(), log.FilterLevel(log.LevelWarn))),
-	})
+func newNoneKV(storePath string, opts *pebble.Options) (storage.SharedKV, error) {
+	db, err := pebble.Open(storePath, opts)
 	if err != nil {
-		panic(err)
+		return nil, err
 	}
 
-	r := &memSharedKV{
+	r := &noneSharedKV{
 		db: db,
 	}
-	return r
+	return r, nil
 }

storage/sharedkv/storekv.go

@@ -0,0 +1,20 @@
+package sharedkv
+
+import (
+	"github.com/cockroachdb/pebble/v2"
+	"github.com/omalloc/tavern/api/defined/v1/storage"
+	"github.com/omalloc/tavern/contrib/log"
+)
+
+// NewStoreSharedKV create a new store kv store
+func NewStoreSharedKV(storePath string) storage.SharedKV {
+	db, err := newNoneKV(storePath, &pebble.Options{
+		DisableWAL: true,
+		Logger:     log.NewHelper(log.NewFilter(log.GetLogger(), log.FilterLevel(log.LevelWarn))),
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	return db
+}