filer_meta_tail.go 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126
  1. package command
  2. import (
  3. "fmt"
  4. "github.com/seaweedfs/seaweedfs/weed/filer"
  5. "github.com/seaweedfs/seaweedfs/weed/pb"
  6. "os"
  7. "path/filepath"
  8. "strings"
  9. "time"
  10. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  11. "github.com/seaweedfs/seaweedfs/weed/security"
  12. "github.com/seaweedfs/seaweedfs/weed/util"
  13. )
  14. func init() {
  15. cmdFilerMetaTail.Run = runFilerMetaTail // break init cycle
  16. }
  17. var cmdFilerMetaTail = &Command{
  18. UsageLine: "filer.meta.tail [-filer=localhost:8888] [-pathPrefix=/]",
  19. Short: "see continuous changes on a filer",
  20. Long: `See continuous changes on a filer.
  21. weed filer.meta.tail -timeAgo=30h | grep truncate
  22. weed filer.meta.tail -timeAgo=30h | jq .
  23. weed filer.meta.tail -timeAgo=30h -untilTimeAgo=20h | jq .
  24. weed filer.meta.tail -timeAgo=30h | jq .eventNotification.newEntry.name
  25. weed filer.meta.tail -timeAgo=30h -es=http://<elasticSearchServerHost>:<port> -es.index=seaweedfs
  26. `,
  27. }
  28. var (
  29. tailFiler = cmdFilerMetaTail.Flag.String("filer", "localhost:8888", "filer hostname:port")
  30. tailTarget = cmdFilerMetaTail.Flag.String("pathPrefix", "/", "path to a folder or common prefix for the folders or files on filer")
  31. 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\"")
  32. tailStop = cmdFilerMetaTail.Flag.Duration("untilTimeAgo", 0, "read until this time ago. \"300ms\", \"1.5h\" or \"2h45m\". Valid time units are \"ns\", \"us\" (or \"µs\"), \"ms\", \"s\", \"m\", \"h\"")
  33. tailPattern = cmdFilerMetaTail.Flag.String("pattern", "", "full path or just filename pattern, ex: \"/home/?opher\", \"*.pdf\", see https://golang.org/pkg/path/filepath/#Match ")
  34. esServers = cmdFilerMetaTail.Flag.String("es", "", "comma-separated elastic servers http://<host:port>")
  35. esIndex = cmdFilerMetaTail.Flag.String("es.index", "seaweedfs", "ES index name")
  36. )
  37. func runFilerMetaTail(cmd *Command, args []string) bool {
  38. util.LoadConfiguration("security", false)
  39. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client")
  40. clientId := util.RandomInt32()
  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 filer_pb.IsEmpty(resp) {
  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. eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
  79. filer.ProtoToText(os.Stdout, resp)
  80. fmt.Fprintln(os.Stdout)
  81. return nil
  82. }
  83. if *esServers != "" {
  84. var err error
  85. eachEntryFunc, err = sendToElasticSearchFunc(*esServers, *esIndex)
  86. if err != nil {
  87. fmt.Printf("create elastic search client to %s: %+v\n", *esServers, err)
  88. return false
  89. }
  90. }
  91. var untilTsNs int64
  92. if *tailStop != 0 {
  93. untilTsNs = time.Now().Add(-*tailStop).UnixNano()
  94. }
  95. tailErr := pb.FollowMetadata(pb.ServerAddress(*tailFiler), grpcDialOption, "tail", clientId, 0, *tailTarget, nil,
  96. time.Now().Add(-*tailStart).UnixNano(), untilTsNs, 0, 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. }, pb.TrivialOnError)
  105. if tailErr != nil {
  106. fmt.Printf("tail %s: %v\n", *tailFiler, tailErr)
  107. }
  108. return true
  109. }