seaweedfs/weed/command/filer_meta_tail_elastic.go

//go:build elastic
// +build elastic

package command

import (
	"context"
	jsoniter "github.com/json-iterator/go"
	elastic "github.com/olivere/elastic/v7"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/util"
	"strings"
)

type EsDocument struct {
	Dir         string `json:"dir,omitempty"`
	Name        string `json:"name,omitempty"`
	IsDirectory bool   `json:"isDir,omitempty"`
	Size        uint64 `json:"size,omitempty"`
	Uid         uint32 `json:"uid,omitempty"`
	Gid         uint32 `json:"gid,omitempty"`
	UserName    string `json:"userName,omitempty"`
	Crtime      int64  `json:"crtime,omitempty"`
	Mtime       int64  `json:"mtime,omitempty"`
	Mime        string `json:"mime,omitempty"`
}

func toEsEntry(event *filer_pb.EventNotification) (*EsDocument, string) {
	entry := event.NewEntry
	dir, name := event.NewParentPath, entry.Name
	id := util.Md5String([]byte(util.NewFullPath(dir, name)))
	esEntry := &EsDocument{
		Dir:         dir,
		Name:        name,
		IsDirectory: entry.IsDirectory,
		Size:        entry.Attributes.FileSize,
		Uid:         entry.Attributes.Uid,
		Gid:         entry.Attributes.Gid,
		UserName:    entry.Attributes.UserName,
		Crtime:      entry.Attributes.Crtime,
		Mtime:       entry.Attributes.Mtime,
		Mime:        entry.Attributes.Mime,
	}
	return esEntry, id
}

func sendToElasticSearchFunc(servers string, esIndex string) (func(resp *filer_pb.SubscribeMetadataResponse) error, error) {
	options := []elastic.ClientOptionFunc{}
	options = append(options, elastic.SetURL(strings.Split(servers, ",")...))
	options = append(options, elastic.SetSniff(false))
	options = append(options, elastic.SetHealthcheck(false))
	client, err := elastic.NewClient(options...)
	if err != nil {
		return nil, err
	}
	return func(resp *filer_pb.SubscribeMetadataResponse) error {
		event := resp.EventNotification
		if event.OldEntry != nil &&
			(event.NewEntry == nil || resp.Directory != event.NewParentPath || event.OldEntry.Name != event.NewEntry.Name) {
			// delete or not update the same file
			dir, name := resp.Directory, event.OldEntry.Name
			id := util.Md5String([]byte(util.NewFullPath(dir, name)))
			println("delete", id)
			_, err := client.Delete().Index(esIndex).Id(id).Do(context.Background())
			return err
		}
		if event.NewEntry != nil {
			// add a new file or update the same file
			esEntry, id := toEsEntry(event)
			value, err := jsoniter.Marshal(esEntry)
			if err != nil {
				return err
			}
			println(string(value))
			_, err = client.Index().Index(esIndex).Id(id).BodyJson(string(value)).Do(context.Background())
			return err
		}
		return nil
	}, nil
}