You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

194 lines
6.3 KiB

4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
3 years ago
4 years ago
  1. package command
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/pb"
  6. "github.com/golang/protobuf/jsonpb"
  7. jsoniter "github.com/json-iterator/go"
  8. "github.com/olivere/elastic/v7"
  9. "os"
  10. "path/filepath"
  11. "strings"
  12. "time"
  13. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  14. "github.com/chrislusf/seaweedfs/weed/security"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. )
  17. func init() {
  18. cmdFilerMetaTail.Run = runFilerMetaTail // break init cycle
  19. }
  20. var cmdFilerMetaTail = &Command{
  21. UsageLine: "filer.meta.tail [-filer=localhost:8888] [-pathPrefix=/]",
  22. Short: "see continuous changes on a filer",
  23. Long: `See continuous changes on a filer.
  24. weed filer.meta.tail -timeAgo=30h | grep truncate
  25. weed filer.meta.tail -timeAgo=30h | jq .
  26. weed filer.meta.tail -timeAgo=30h | jq .eventNotification.newEntry.name
  27. weed filer.meta.tail -timeAgo=30h -es=http://<elasticSearchServerHost>:<port> -es.index=seaweedfs
  28. `,
  29. }
  30. var (
  31. tailFiler = cmdFilerMetaTail.Flag.String("filer", "localhost:8888", "filer hostname:port")
  32. tailTarget = cmdFilerMetaTail.Flag.String("pathPrefix", "/", "path to a folder or common prefix for the folders or files on filer")
  33. tailStart = cmdFilerMetaTail.Flag.Duration("timeAgo", 0, "start time before now. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"")
  34. tailPattern = cmdFilerMetaTail.Flag.String("pattern", "", "full path or just filename pattern, ex: \"/home/?opher\", \"*.pdf\", see https://golang.org/pkg/path/filepath/#Match ")
  35. esServers = cmdFilerMetaTail.Flag.String("es", "", "comma-separated elastic servers http://<host:port>")
  36. esIndex = cmdFilerMetaTail.Flag.String("es.index", "seaweedfs", "ES index name")
  37. )
  38. func runFilerMetaTail(cmd *Command, args []string) bool {
  39. util.LoadConfiguration("security", false)
  40. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
  41. var filterFunc func(dir, fname string) bool
  42. if *tailPattern != "" {
  43. if strings.Contains(*tailPattern, "/") {
  44. println("watch path pattern", *tailPattern)
  45. filterFunc = func(dir, fname string) bool {
  46. matched, err := filepath.Match(*tailPattern, dir+"/"+fname)
  47. if err != nil {
  48. fmt.Printf("error: %v", err)
  49. }
  50. return matched
  51. }
  52. } else {
  53. println("watch file pattern", *tailPattern)
  54. filterFunc = func(dir, fname string) bool {
  55. matched, err := filepath.Match(*tailPattern, fname)
  56. if err != nil {
  57. fmt.Printf("error: %v", err)
  58. }
  59. return matched
  60. }
  61. }
  62. }
  63. shouldPrint := func(resp *filer_pb.SubscribeMetadataResponse) bool {
  64. if resp.EventNotification.OldEntry == nil && resp.EventNotification.NewEntry == nil {
  65. return false
  66. }
  67. if filterFunc == nil {
  68. return true
  69. }
  70. if resp.EventNotification.OldEntry != nil && filterFunc(resp.Directory, resp.EventNotification.OldEntry.Name) {
  71. return true
  72. }
  73. if resp.EventNotification.NewEntry != nil && filterFunc(resp.EventNotification.NewParentPath, resp.EventNotification.NewEntry.Name) {
  74. return true
  75. }
  76. return false
  77. }
  78. jsonpbMarshaler := jsonpb.Marshaler{
  79. EmitDefaults: false,
  80. }
  81. eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
  82. jsonpbMarshaler.Marshal(os.Stdout, resp)
  83. fmt.Fprintln(os.Stdout)
  84. return nil
  85. }
  86. if *esServers != "" {
  87. var err error
  88. eachEntryFunc, err = sendToElasticSearchFunc(*esServers, *esIndex)
  89. if err != nil {
  90. fmt.Printf("create elastic search client to %s: %+v\n", *esServers, err)
  91. return false
  92. }
  93. }
  94. tailErr := pb.FollowMetadata(pb.ServerAddress(*tailFiler), grpcDialOption, "tail",
  95. *tailTarget, nil, time.Now().Add(-*tailStart).UnixNano(), 0,
  96. func(resp *filer_pb.SubscribeMetadataResponse) error {
  97. if !shouldPrint(resp) {
  98. return nil
  99. }
  100. if err := eachEntryFunc(resp); err != nil {
  101. return err
  102. }
  103. return nil
  104. }, false)
  105. if tailErr != nil {
  106. fmt.Printf("tail %s: %v\n", *tailFiler, tailErr)
  107. }
  108. return true
  109. }
  110. type EsDocument struct {
  111. Dir string `json:"dir,omitempty"`
  112. Name string `json:"name,omitempty"`
  113. IsDirectory bool `json:"isDir,omitempty"`
  114. Size uint64 `json:"size,omitempty"`
  115. Uid uint32 `json:"uid,omitempty"`
  116. Gid uint32 `json:"gid,omitempty"`
  117. UserName string `json:"userName,omitempty"`
  118. Collection string `json:"collection,omitempty"`
  119. Crtime int64 `json:"crtime,omitempty"`
  120. Mtime int64 `json:"mtime,omitempty"`
  121. Mime string `json:"mime,omitempty"`
  122. }
  123. func toEsEntry(event *filer_pb.EventNotification) (*EsDocument, string) {
  124. entry := event.NewEntry
  125. dir, name := event.NewParentPath, entry.Name
  126. id := util.Md5String([]byte(util.NewFullPath(dir, name)))
  127. esEntry := &EsDocument{
  128. Dir: dir,
  129. Name: name,
  130. IsDirectory: entry.IsDirectory,
  131. Size: entry.Attributes.FileSize,
  132. Uid: entry.Attributes.Uid,
  133. Gid: entry.Attributes.Gid,
  134. UserName: entry.Attributes.UserName,
  135. Collection: entry.Attributes.Collection,
  136. Crtime: entry.Attributes.Crtime,
  137. Mtime: entry.Attributes.Mtime,
  138. Mime: entry.Attributes.Mime,
  139. }
  140. return esEntry, id
  141. }
  142. func sendToElasticSearchFunc(servers string, esIndex string) (func(resp *filer_pb.SubscribeMetadataResponse) error, error) {
  143. options := []elastic.ClientOptionFunc{}
  144. options = append(options, elastic.SetURL(strings.Split(servers, ",")...))
  145. options = append(options, elastic.SetSniff(false))
  146. options = append(options, elastic.SetHealthcheck(false))
  147. client, err := elastic.NewClient(options...)
  148. if err != nil {
  149. return nil, err
  150. }
  151. return func(resp *filer_pb.SubscribeMetadataResponse) error {
  152. event := resp.EventNotification
  153. if event.OldEntry != nil &&
  154. (event.NewEntry == nil || resp.Directory != event.NewParentPath || event.OldEntry.Name != event.NewEntry.Name) {
  155. // delete or not update the same file
  156. dir, name := resp.Directory, event.OldEntry.Name
  157. id := util.Md5String([]byte(util.NewFullPath(dir, name)))
  158. println("delete", id)
  159. _, err := client.Delete().Index(esIndex).Id(id).Do(context.Background())
  160. return err
  161. }
  162. if event.NewEntry != nil {
  163. // add a new file or update the same file
  164. esEntry, id := toEsEntry(event)
  165. value, err := jsoniter.Marshal(esEntry)
  166. if err != nil {
  167. return err
  168. }
  169. println(string(value))
  170. _, err = client.Index().Index(esIndex).Id(id).BodyJson(string(value)).Do(context.Background())
  171. return err
  172. }
  173. return nil
  174. }, nil
  175. }