volume_tailer.go 1.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  1. package main
  2. import (
  3. "flag"
  4. "github.com/seaweedfs/seaweedfs/weed/pb"
  5. "log"
  6. "time"
  7. "github.com/seaweedfs/seaweedfs/weed/operation"
  8. "github.com/seaweedfs/seaweedfs/weed/security"
  9. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  10. util2 "github.com/seaweedfs/seaweedfs/weed/util"
  11. "golang.org/x/tools/godoc/util"
  12. )
  13. var (
  14. master = flag.String("master", "localhost:9333", "master server host and port")
  15. volumeId = flag.Int("volumeId", -1, "a volume id")
  16. rewindDuration = flag.Duration("rewind", -1, "rewind back in time. -1 means from the first entry. 0 means from now.")
  17. timeoutSeconds = flag.Int("timeoutSeconds", 0, "disconnect if no activity after these seconds")
  18. showTextFile = flag.Bool("showTextFile", false, "display textual file content")
  19. )
  20. func main() {
  21. flag.Parse()
  22. util2.LoadConfiguration("security", false)
  23. grpcDialOption := security.LoadClientTLS(util2.GetViper(), "grpc.client")
  24. vid := needle.VolumeId(*volumeId)
  25. var sinceTimeNs int64
  26. if *rewindDuration == 0 {
  27. sinceTimeNs = time.Now().UnixNano()
  28. } else if *rewindDuration == -1 {
  29. sinceTimeNs = 0
  30. } else if *rewindDuration > 0 {
  31. sinceTimeNs = time.Now().Add(-*rewindDuration).UnixNano()
  32. }
  33. err := operation.TailVolume(func() pb.ServerAddress { return pb.ServerAddress(*master) }, grpcDialOption, vid, uint64(sinceTimeNs), *timeoutSeconds, func(n *needle.Needle) (err error) {
  34. if n.Size == 0 {
  35. println("-", n.String())
  36. return nil
  37. } else {
  38. println("+", n.String())
  39. }
  40. if *showTextFile {
  41. data := n.Data
  42. if n.IsCompressed() {
  43. if data, err = util2.DecompressData(data); err != nil {
  44. return err
  45. }
  46. }
  47. if util.IsText(data) {
  48. println(string(data))
  49. }
  50. println("-", n.String(), "compressed", n.IsCompressed(), "original size", len(data))
  51. }
  52. return nil
  53. })
  54. if err != nil {
  55. log.Printf("Error VolumeTailSender volume %d: %v", vid, err)
  56. }
  57. }