From 37234bf3f894edd727dc0abf8fff001b95af5a2e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 1 Sep 2020 21:58:57 -0700 Subject: [PATCH] filer store adds kv support can compile now, need to implement those unimplemented --- .../abstract_sql/abstract_sql_store_kv.go | 18 ++++++ weed/filer/cassandra/cassandra_store_kv.go | 18 ++++++ weed/filer/etcd/etcd_store_kv.go | 44 +++++++++++++++ weed/filer/filer.go | 2 - weed/filer/filerstore.go | 21 +++++++ weed/filer/leveldb/leveldb_store_kv.go | 39 +++++++++++++ weed/filer/leveldb2/leveldb2_store_kv.go | 56 +++++++++++++++++++ weed/filer/mongodb/mongodb_store_kv.go | 19 +++++++ weed/filer/redis/universal_redis_store_kv.go | 42 ++++++++++++++ weed/filer/redis2/universal_redis_store_kv.go | 42 ++++++++++++++ 10 files changed, 299 insertions(+), 2 deletions(-) create mode 100644 weed/filer/abstract_sql/abstract_sql_store_kv.go create mode 100644 weed/filer/cassandra/cassandra_store_kv.go create mode 100644 weed/filer/etcd/etcd_store_kv.go create mode 100644 weed/filer/leveldb/leveldb_store_kv.go create mode 100644 weed/filer/leveldb2/leveldb2_store_kv.go create mode 100644 weed/filer/mongodb/mongodb_store_kv.go create mode 100644 weed/filer/redis/universal_redis_store_kv.go create mode 100644 weed/filer/redis2/universal_redis_store_kv.go diff --git a/weed/filer/abstract_sql/abstract_sql_store_kv.go b/weed/filer/abstract_sql/abstract_sql_store_kv.go new file mode 100644 index 000000000..cfffc8918 --- /dev/null +++ b/weed/filer/abstract_sql/abstract_sql_store_kv.go @@ -0,0 +1,18 @@ +package abstract_sql + +import ( + "context" + "github.com/chrislusf/seaweedfs/weed/filer" +) + +func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + return filer.ErrKvNotImplemented +} + +func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + return nil, filer.ErrKvNotImplemented +} + +func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err error) { + return filer.ErrKvNotImplemented +} diff --git a/weed/filer/cassandra/cassandra_store_kv.go b/weed/filer/cassandra/cassandra_store_kv.go new file mode 100644 index 000000000..f7668746f --- /dev/null +++ b/weed/filer/cassandra/cassandra_store_kv.go @@ -0,0 +1,18 @@ +package cassandra + +import ( + "context" + "github.com/chrislusf/seaweedfs/weed/filer" +) + +func (store *CassandraStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + return filer.ErrKvNotImplemented +} + +func (store *CassandraStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + return nil, filer.ErrKvNotImplemented +} + +func (store *CassandraStore) KvDelete(ctx context.Context, key []byte) (err error) { + return filer.ErrKvNotImplemented +} diff --git a/weed/filer/etcd/etcd_store_kv.go b/weed/filer/etcd/etcd_store_kv.go new file mode 100644 index 000000000..a803a5834 --- /dev/null +++ b/weed/filer/etcd/etcd_store_kv.go @@ -0,0 +1,44 @@ +package etcd + +import ( + "context" + "fmt" + "github.com/chrislusf/seaweedfs/weed/filer" +) + +func (store *EtcdStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + + _, err = store.client.Put(ctx, string(key), string(value)) + + if err != nil { + return fmt.Errorf("kv put: %v", err) + } + + return nil +} + +func (store *EtcdStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + + resp, err := store.client.Get(ctx, string(key), nil) + + if err != nil { + return nil, fmt.Errorf("kv get: %v", err) + } + + if len(resp.Kvs) == 0 { + return nil, filer.ErrKvNotFound + } + + return resp.Kvs[0].Value, nil +} + +func (store *EtcdStore) KvDelete(ctx context.Context, key []byte) (err error) { + + _, err = store.client.Delete(ctx, string(key)) + + if err != nil { + return fmt.Errorf("kv delete: %v", err) + } + + return nil +} diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 71da1a080..16e86da01 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -2,7 +2,6 @@ package filer import ( "context" - "errors" "fmt" "os" "strings" @@ -25,7 +24,6 @@ const ( var ( OS_UID = uint32(os.Getuid()) OS_GID = uint32(os.Getgid()) - ErrUnsupportedListDirectoryPrefixed = errors.New("UNSUPPORTED") ) type Filer struct { diff --git a/weed/filer/filerstore.go b/weed/filer/filerstore.go index 48f7c99e4..518212437 100644 --- a/weed/filer/filerstore.go +++ b/weed/filer/filerstore.go @@ -2,6 +2,7 @@ package filer import ( "context" + "errors" "strings" "time" @@ -10,6 +11,12 @@ import ( "github.com/chrislusf/seaweedfs/weed/util" ) +var ( + ErrUnsupportedListDirectoryPrefixed = errors.New("unsupported directory prefix listing") + ErrKvNotImplemented = errors.New("kv not implemented yet") + ErrKvNotFound = errors.New("kv: not found") +) + type FilerStore interface { // GetName gets the name to locate the configuration in filer.toml file GetName() string @@ -28,6 +35,10 @@ type FilerStore interface { CommitTransaction(ctx context.Context) error RollbackTransaction(ctx context.Context) error + KvPut(ctx context.Context, key []byte, value []byte) (err error) + KvGet(ctx context.Context, key []byte) (value []byte, err error) + KvDelete(ctx context.Context, key []byte) (err error) + Shutdown() } @@ -206,3 +217,13 @@ func (fsw *FilerStoreWrapper) RollbackTransaction(ctx context.Context) error { func (fsw *FilerStoreWrapper) Shutdown() { fsw.ActualStore.Shutdown() } + +func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + return fsw.ActualStore.KvPut(ctx, key, value) +} +func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + return fsw.ActualStore.KvGet(ctx, key) +} +func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) { + return fsw.ActualStore.KvDelete(ctx, key) +} diff --git a/weed/filer/leveldb/leveldb_store_kv.go b/weed/filer/leveldb/leveldb_store_kv.go new file mode 100644 index 000000000..7fe1d3356 --- /dev/null +++ b/weed/filer/leveldb/leveldb_store_kv.go @@ -0,0 +1,39 @@ +package leveldb + +import ( + "context" + "fmt" +) + +func (store *LevelDBStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + + err = store.db.Put(key, value, nil) + + if err != nil { + return fmt.Errorf("kv put: %v", err) + } + + return nil +} + +func (store *LevelDBStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + + value, err = store.db.Get(key, nil) + + if err != nil { + return nil, fmt.Errorf("kv get: %v", err) + } + + return +} + +func (store *LevelDBStore) KvDelete(ctx context.Context, key []byte) (err error) { + + err = store.db.Delete(key, nil) + + if err != nil { + return fmt.Errorf("kv delete: %v", err) + } + + return nil +} diff --git a/weed/filer/leveldb2/leveldb2_store_kv.go b/weed/filer/leveldb2/leveldb2_store_kv.go new file mode 100644 index 000000000..b415d3c32 --- /dev/null +++ b/weed/filer/leveldb2/leveldb2_store_kv.go @@ -0,0 +1,56 @@ +package leveldb + +import ( + "context" + "fmt" + + "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/syndtr/goleveldb/leveldb" +) + +func (store *LevelDB2Store) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + + partitionId := bucketKvKey(key, store.dbCount) + + err = store.dbs[partitionId].Put(key, value, nil) + + if err != nil { + return fmt.Errorf("kv bucket %d put: %v", partitionId, err) + } + + return nil +} + +func (store *LevelDB2Store) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + + partitionId := bucketKvKey(key, store.dbCount) + + value, err = store.dbs[partitionId].Get(key, nil) + + if err == leveldb.ErrNotFound { + return nil, filer.ErrKvNotFound + } + + if err != nil { + return nil, fmt.Errorf("kv bucket %d get: %v", partitionId, err) + } + + return +} + +func (store *LevelDB2Store) KvDelete(ctx context.Context, key []byte) (err error) { + + partitionId := bucketKvKey(key, store.dbCount) + + err = store.dbs[partitionId].Delete(key, nil) + + if err != nil { + return fmt.Errorf("kv bucket %d delete: %v", partitionId, err) + } + + return nil +} + +func bucketKvKey(key []byte, dbCount int) (partitionId int) { + return int(key[len(key)-1]) % dbCount +} diff --git a/weed/filer/mongodb/mongodb_store_kv.go b/weed/filer/mongodb/mongodb_store_kv.go new file mode 100644 index 000000000..4e0a4b0f0 --- /dev/null +++ b/weed/filer/mongodb/mongodb_store_kv.go @@ -0,0 +1,19 @@ +package mongodb + +import ( + "context" + "github.com/chrislusf/seaweedfs/weed/filer" +) + + +func (store *MongodbStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + return filer.ErrKvNotImplemented +} + +func (store *MongodbStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + return nil, filer.ErrKvNotImplemented +} + +func (store *MongodbStore) KvDelete(ctx context.Context, key []byte) (err error) { + return filer.ErrKvNotImplemented +} diff --git a/weed/filer/redis/universal_redis_store_kv.go b/weed/filer/redis/universal_redis_store_kv.go new file mode 100644 index 000000000..0fc12c631 --- /dev/null +++ b/weed/filer/redis/universal_redis_store_kv.go @@ -0,0 +1,42 @@ +package redis + +import ( + "context" + "fmt" + + "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/go-redis/redis" +) + +func (store *UniversalRedisStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + + _, err = store.Client.Set(string(key), value, 0).Result() + + if err != nil { + return fmt.Errorf("kv put: %v", err) + } + + return nil +} + +func (store *UniversalRedisStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + + data, err := store.Client.Get(string(key)).Result() + + if err == redis.Nil { + return nil, filer.ErrKvNotFound + } + + return []byte(data), err +} + +func (store *UniversalRedisStore) KvDelete(ctx context.Context, key []byte) (err error) { + + _, err = store.Client.Del(string(key)).Result() + + if err != nil { + return fmt.Errorf("kv delete: %v", err) + } + + return nil +} diff --git a/weed/filer/redis2/universal_redis_store_kv.go b/weed/filer/redis2/universal_redis_store_kv.go new file mode 100644 index 000000000..658491ddf --- /dev/null +++ b/weed/filer/redis2/universal_redis_store_kv.go @@ -0,0 +1,42 @@ +package redis2 + +import ( + "context" + "fmt" + + "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/go-redis/redis" +) + +func (store *UniversalRedis2Store) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + + _, err = store.Client.Set(string(key), value, 0).Result() + + if err != nil { + return fmt.Errorf("kv put: %v", err) + } + + return nil +} + +func (store *UniversalRedis2Store) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + + data, err := store.Client.Get(string(key)).Result() + + if err == redis.Nil { + return nil, filer.ErrKvNotFound + } + + return []byte(data), err +} + +func (store *UniversalRedis2Store) KvDelete(ctx context.Context, key []byte) (err error) { + + _, err = store.Client.Del(string(key)).Result() + + if err != nil { + return fmt.Errorf("kv delete: %v", err) + } + + return nil +}