1
0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-12-30 04:57:55 +08:00
seaweedfs/weed/filer/elastic/v7/elastic_store.go

319 lines
9.5 KiB
Go
Raw Normal View History

//go:build elastic
// +build elastic
2020-09-03 16:34:58 +08:00
package elastic
import (
"context"
"fmt"
"math"
"strings"
jsoniter "github.com/json-iterator/go"
elastic "github.com/olivere/elastic/v7"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
weed_util "github.com/seaweedfs/seaweedfs/weed/util"
2020-09-03 16:34:58 +08:00
)
var (
indexType = "_doc"
indexPrefix = ".seaweedfs_"
indexKV = ".seaweedfs_kv_entries"
kvMappings = ` {
2020-09-10 15:59:16 +08:00
"mappings": {
"enabled": false,
"properties": {
"Value":{
"type": "binary"
}
}
}
}`
2020-09-03 16:34:58 +08:00
)
type ESEntry struct {
ParentId string `json:"ParentId"`
2020-09-03 17:05:26 +08:00
Entry *filer.Entry
2020-09-03 16:34:58 +08:00
}
type ESKVEntry struct {
2020-09-10 15:59:16 +08:00
Value []byte `json:"Value"`
}
2020-09-03 16:34:58 +08:00
func init() {
2020-09-03 17:05:26 +08:00
filer.Stores = append(filer.Stores, &ElasticStore{})
2020-09-03 16:34:58 +08:00
}
type ElasticStore struct {
client *elastic.Client
maxPageSize int
}
func (store *ElasticStore) GetName() string {
return "elastic7"
}
2020-09-10 14:22:07 +08:00
2020-09-03 16:34:58 +08:00
func (store *ElasticStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
options := []elastic.ClientOptionFunc{}
servers := configuration.GetStringSlice(prefix + "servers")
options = append(options, elastic.SetURL(servers...))
2020-09-10 14:22:07 +08:00
username := configuration.GetString(prefix + "username")
password := configuration.GetString(prefix + "password")
if username != "" && password != "" {
options = append(options, elastic.SetBasicAuth(username, password))
}
options = append(options, elastic.SetSniff(configuration.GetBool(prefix+"sniff_enabled")))
options = append(options, elastic.SetHealthcheck(configuration.GetBool(prefix+"healthcheck_enabled")))
store.maxPageSize = configuration.GetInt(prefix + "index.max_result_window")
if store.maxPageSize <= 0 {
store.maxPageSize = 10000
}
2020-09-10 16:30:15 +08:00
glog.Infof("filer store elastic endpoints: %v.", servers)
return store.initialize(options)
}
func (store *ElasticStore) initialize(options []elastic.ClientOptionFunc) (err error) {
ctx := context.Background()
store.client, err = elastic.NewClient(options...)
if err != nil {
2024-11-22 00:52:37 +08:00
return fmt.Errorf("init elastic %v", err)
}
if ok, err := store.client.IndexExists(indexKV).Do(ctx); err == nil && !ok {
_, err = store.client.CreateIndex(indexKV).Body(kvMappings).Do(ctx)
if err != nil {
2024-11-22 00:52:37 +08:00
return fmt.Errorf("create index(%s) %v", indexKV, err)
}
}
return nil
2020-09-10 14:22:07 +08:00
}
2020-09-03 16:34:58 +08:00
func (store *ElasticStore) BeginTransaction(ctx context.Context) (context.Context, error) {
return ctx, nil
}
func (store *ElasticStore) CommitTransaction(ctx context.Context) error {
return nil
}
func (store *ElasticStore) RollbackTransaction(ctx context.Context) error {
return nil
}
2020-09-10 14:22:07 +08:00
2021-01-16 15:56:24 +08:00
func (store *ElasticStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
2020-09-03 17:05:26 +08:00
}
func (store *ElasticStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
index := getIndex(entry.FullPath, false)
2020-09-03 16:34:58 +08:00
dir, _ := entry.FullPath.DirAndName()
2020-09-10 16:11:18 +08:00
id := weed_util.Md5String([]byte(entry.FullPath))
2020-09-03 16:34:58 +08:00
esEntry := &ESEntry{
2020-09-10 16:11:18 +08:00
ParentId: weed_util.Md5String([]byte(dir)),
2020-09-03 16:34:58 +08:00
Entry: entry,
}
value, err := jsoniter.Marshal(esEntry)
if err != nil {
glog.Errorf("insert entry(%s) %v.", string(entry.FullPath), err)
2024-11-22 00:52:37 +08:00
return fmt.Errorf("insert entry marshal %v", err)
2020-09-03 16:34:58 +08:00
}
_, err = store.client.Index().
Index(index).
Type(indexType).
Id(id).
BodyJson(string(value)).
Do(ctx)
2020-09-03 16:34:58 +08:00
if err != nil {
glog.Errorf("insert entry(%s) %v.", string(entry.FullPath), err)
2024-11-22 00:52:37 +08:00
return fmt.Errorf("insert entry %v", err)
2020-09-03 16:34:58 +08:00
}
return nil
}
2020-09-10 14:22:07 +08:00
2020-09-03 17:05:26 +08:00
func (store *ElasticStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
2020-09-03 16:34:58 +08:00
return store.InsertEntry(ctx, entry)
}
2020-09-10 14:22:07 +08:00
2020-09-03 17:05:26 +08:00
func (store *ElasticStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
index := getIndex(fullpath, false)
2020-09-10 16:11:18 +08:00
id := weed_util.Md5String([]byte(fullpath))
2020-09-03 16:34:58 +08:00
searchResult, err := store.client.Get().
Index(index).
Type(indexType).
Id(id).
Do(ctx)
2020-09-03 16:34:58 +08:00
if elastic.IsNotFound(err) {
return nil, filer_pb.ErrNotFound
}
if searchResult != nil && searchResult.Found {
esEntry := &ESEntry{
ParentId: "",
2020-09-03 17:05:26 +08:00
Entry: &filer.Entry{},
2020-09-03 16:34:58 +08:00
}
err := jsoniter.Unmarshal(searchResult.Source, esEntry)
return esEntry.Entry, err
}
glog.Errorf("find entry(%s),%v.", string(fullpath), err)
return nil, filer_pb.ErrNotFound
}
2020-09-10 14:22:07 +08:00
2020-09-03 16:34:58 +08:00
func (store *ElasticStore) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) {
index := getIndex(fullpath, false)
2020-09-10 16:11:18 +08:00
id := weed_util.Md5String([]byte(fullpath))
strFullpath := string(fullpath)
// A top-level subdirectory refers to an Elasticsearch index.
// If we delete an entry at the top level, we should attempt to delete the corresponding Elasticsearch index.
if strings.Count(strFullpath, "/") == 1 {
entry, err2 := store.FindEntry(ctx, fullpath)
if err2 == nil && entry.IsDirectory() {
bucketIndex := indexPrefix + strFullpath[1:]
store.deleteIndex(ctx, bucketIndex)
}
2020-09-03 16:34:58 +08:00
}
return store.deleteEntry(ctx, index, id)
2020-09-03 16:34:58 +08:00
}
2020-09-10 14:22:07 +08:00
func (store *ElasticStore) deleteIndex(ctx context.Context, index string) (err error) {
deleteResult, err := store.client.DeleteIndex(index).Do(ctx)
2020-09-03 16:34:58 +08:00
if elastic.IsNotFound(err) || (err == nil && deleteResult.Acknowledged) {
return nil
}
glog.Errorf("delete index(%s) %v.", index, err)
return err
}
2020-09-10 14:22:07 +08:00
func (store *ElasticStore) deleteEntry(ctx context.Context, index, id string) (err error) {
2020-09-03 16:34:58 +08:00
deleteResult, err := store.client.Delete().
Index(index).
Type(indexType).
Id(id).
Do(ctx)
2020-09-03 16:34:58 +08:00
if err == nil {
if deleteResult.Result == "deleted" || deleteResult.Result == "not_found" {
return nil
}
}
glog.Errorf("delete entry(index:%s,_id:%s) %v.", index, id, err)
2024-11-22 00:52:37 +08:00
return fmt.Errorf("delete entry %v", err)
2020-09-03 16:34:58 +08:00
}
2020-09-10 14:22:07 +08:00
func (store *ElasticStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
2021-01-16 15:56:24 +08:00
_, err = store.ListDirectoryEntries(ctx, fullpath, "", false, math.MaxInt32, func(entry *filer.Entry) bool {
if err := store.DeleteEntry(ctx, entry.FullPath); err != nil {
glog.Errorf("elastic delete %s: %v.", entry.FullPath, err)
return false
2020-09-03 16:34:58 +08:00
}
2021-01-16 15:56:24 +08:00
return true
})
return
2020-09-03 16:34:58 +08:00
}
2021-01-16 15:56:24 +08:00
func (store *ElasticStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return store.listDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
2020-09-03 16:34:58 +08:00
}
func (store *ElasticStore) listDirectoryEntries(
2021-01-16 15:56:24 +08:00
ctx context.Context, fullpath weed_util.FullPath, startFileName string, inclusive bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
2020-09-03 16:34:58 +08:00
first := true
index := getIndex(fullpath, true)
2020-09-03 16:34:58 +08:00
nextStart := ""
2020-09-10 16:11:18 +08:00
parentId := weed_util.Md5String([]byte(fullpath))
2021-01-16 15:56:24 +08:00
if _, err = store.client.Refresh(index).Do(ctx); err != nil {
2020-09-03 16:34:58 +08:00
if elastic.IsNotFound(err) {
store.client.CreateIndex(index).Do(ctx)
2021-01-16 15:56:24 +08:00
return
2020-09-03 16:34:58 +08:00
}
}
for {
result := &elastic.SearchResult{}
if (startFileName == "" && first) || inclusive {
if result, err = store.search(ctx, index, parentId); err != nil {
2020-09-03 16:34:58 +08:00
glog.Errorf("search (%s,%s,%t,%d) %v.", string(fullpath), startFileName, inclusive, limit, err)
2021-01-16 15:56:24 +08:00
return
2020-09-03 16:34:58 +08:00
}
} else {
fullPath := string(fullpath) + "/" + startFileName
if !first {
fullPath = nextStart
}
2020-09-10 16:11:18 +08:00
after := weed_util.Md5String([]byte(fullPath))
if result, err = store.searchAfter(ctx, index, parentId, after); err != nil {
2020-09-03 16:34:58 +08:00
glog.Errorf("searchAfter (%s,%s,%t,%d) %v.", string(fullpath), startFileName, inclusive, limit, err)
2021-01-16 15:56:24 +08:00
return
2020-09-03 16:34:58 +08:00
}
}
first = false
for _, hit := range result.Hits.Hits {
esEntry := &ESEntry{
ParentId: "",
2020-09-03 17:05:26 +08:00
Entry: &filer.Entry{},
2020-09-03 16:34:58 +08:00
}
if err := jsoniter.Unmarshal(hit.Source, esEntry); err == nil {
limit--
if limit < 0 {
2021-01-16 15:56:24 +08:00
return lastFileName, nil
2020-09-03 16:34:58 +08:00
}
nextStart = string(esEntry.Entry.FullPath)
fileName := esEntry.Entry.FullPath.Name()
2020-09-03 16:34:58 +08:00
if fileName == startFileName && !inclusive {
continue
}
2021-01-16 15:56:24 +08:00
if !eachEntryFunc(esEntry.Entry) {
break
}
lastFileName = fileName
2020-09-03 16:34:58 +08:00
}
}
if len(result.Hits.Hits) < store.maxPageSize {
break
}
2020-09-03 16:34:58 +08:00
}
2021-01-16 15:56:24 +08:00
return
2020-09-03 16:34:58 +08:00
}
func (store *ElasticStore) search(ctx context.Context, index, parentId string) (result *elastic.SearchResult, err error) {
if count, err := store.client.Count(index).Do(ctx); err == nil && count == 0 {
2020-09-03 16:34:58 +08:00
return &elastic.SearchResult{
Hits: &elastic.SearchHits{
Hits: make([]*elastic.SearchHit, 0)},
}, nil
}
queryResult, err := store.client.Search().
Index(index).
Query(elastic.NewMatchQuery("ParentId", parentId)).
Size(store.maxPageSize).
Sort("_id", false).
Do(ctx)
2020-09-03 16:34:58 +08:00
return queryResult, err
}
func (store *ElasticStore) searchAfter(ctx context.Context, index, parentId, after string) (result *elastic.SearchResult, err error) {
2020-09-03 16:34:58 +08:00
queryResult, err := store.client.Search().
Index(index).
Query(elastic.NewMatchQuery("ParentId", parentId)).
SearchAfter(after).
Size(store.maxPageSize).
Sort("_id", false).
Do(ctx)
2020-09-03 16:34:58 +08:00
return queryResult, err
}
func (store *ElasticStore) Shutdown() {
store.client.Stop()
}
func getIndex(fullpath weed_util.FullPath, isDirectory bool) string {
2020-09-03 16:34:58 +08:00
path := strings.Split(string(fullpath), "/")
if isDirectory && len(path) >= 2 {
return indexPrefix + strings.ToLower(path[1])
2020-09-03 16:34:58 +08:00
}
if len(path) > 2 {
return indexPrefix + strings.ToLower(path[1])
}
if len(path) == 2 {
return indexPrefix
2020-09-03 16:34:58 +08:00
}
return ""
}