gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

ES backended filer support kv ops.

Browse source
This commit is contained in:
ruitao.liu 2020-09-04 15:40:13 +08:00
parent 0ddcc2a8f5
commit 83080b5e03
3 changed files with 61 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
go.mod
View file

@ -27,7 +27,6 @@ require (
github.com/go-sql-driver/mysql v1.5.0
github.com/gocql/gocql v0.0.0-20190829130954-e163eff7a8c6
github.com/gogo/protobuf v1.2.2-0.20190730201129-28a6bbf47e48 // indirect
github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
github.com/golang/protobuf v1.4.2
github.com/google/btree v1.0.0
github.com/google/uuid v1.1.1

2
weed/command/scaffold.go
View file

@ -177,7 +177,7 @@ database = "seaweedfs"
[elastic7]
enabled = false
servers = "http://localhost:9200"
# increase the value is recommend, both here and in elastic cluster configuration
# increase the value is recommend, be sure the value in Elastic is greater or equal here
index.max_result_window = 10000
`

63
weed/filer/elastic/v7/elastic_store.go
View file

@ -18,6 +18,7 @@ import (
var (
indexType = "_doc"
indexPrefix = ".seaweedfs_"
indexKV = ".seaweedfs_kv_entries"
)
type ESEntry struct {
@ -34,6 +35,11 @@ type ElasticStore struct {
maxPageSize int
}
type ESKVEntry struct {
Key string `json:Key`
Value string `json:Value`
}
func (store *ElasticStore) GetName() string {
return "elastic7"
}
@ -66,15 +72,66 @@ func (store *ElasticStore) CommitTransaction(ctx context.Context) error {
func (store *ElasticStore) RollbackTransaction(ctx context.Context) error {
return nil
}
func (store *ElasticStore) KvDelete(ctx context.Context, key []byte) (err error) {
return filer.ErrKvNotImplemented
id := fmt.Sprintf("%x", md5.Sum(key))
deleteResult, err := store.client.Delete().
Index(indexKV).
Type(indexType).
Id(id).
Do(context.Background())
if err == nil {
if deleteResult.Result == "deleted" || deleteResult.Result == "not_found" {
return nil
}
}
glog.Errorf("delete key(id:%s) %v.", string(key), err)
return fmt.Errorf("delete key %v.", err)
}
func (store *ElasticStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
return []byte(""), filer.ErrKvNotImplemented
id := fmt.Sprintf("%x", md5.Sum(key))
searchResult, err := store.client.Get().
Index(indexKV).
Type(indexType).
Id(id).
Do(context.Background())
if elastic.IsNotFound(err) {
return nil, filer_pb.ErrNotFound
}
if searchResult != nil && searchResult.Found {
esEntry := &ESKVEntry{}
if err := jsoniter.Unmarshal(searchResult.Source, esEntry); err == nil {
return []byte(esEntry.Value), nil
}
}
glog.Errorf("find key(%s),%v.", string(key), err)
return nil, filer_pb.ErrNotFound
}
func (store *ElasticStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
return filer.ErrKvNotImplemented
id := fmt.Sprintf("%x", md5.Sum(key))
esEntry := &ESKVEntry{
string(key),
string(value),
}
val, err := jsoniter.Marshal(esEntry)
if err != nil {
glog.Errorf("insert key(%s) %v.", string(key), err)
return fmt.Errorf("insert key %v.", err)
}
_, err = store.client.Index().
Index(indexKV).
Type(indexType).
Id(id).
BodyJson(string(val)).
Do(context.Background())
if err != nil {
return fmt.Errorf("kv put: %v", err)
}
return nil
}
func (store *ElasticStore) ListDirectoryPrefixedEntries(ctx context.Context, fullpath weed_util.FullPath, startFileName string, inclusive bool, limit int, prefix string) (entries []*filer.Entry, err error) {
return nil, filer.ErrUnsupportedListDirectoryPrefixed
}