tail_volume.go 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889
  1. package operation
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/pb"
  6. "io"
  7. "google.golang.org/grpc"
  8. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  9. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  10. )
  11. func TailVolume(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid needle.VolumeId, sinceNs uint64, timeoutSeconds int, fn func(n *needle.Needle) error) error {
  12. // find volume location, replication, ttl info
  13. lookup, err := LookupVolumeId(masterFn, grpcDialOption, vid.String())
  14. if err != nil {
  15. return fmt.Errorf("look up volume %d: %v", vid, err)
  16. }
  17. if len(lookup.Locations) == 0 {
  18. return fmt.Errorf("unable to locate volume %d", vid)
  19. }
  20. volumeServer := lookup.Locations[0].ServerAddress()
  21. return TailVolumeFromSource(volumeServer, grpcDialOption, vid, sinceNs, timeoutSeconds, fn)
  22. }
  23. func TailVolumeFromSource(volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, vid needle.VolumeId, sinceNs uint64, idleTimeoutSeconds int, fn func(n *needle.Needle) error) error {
  24. return WithVolumeServerClient(true, volumeServer, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  25. ctx, cancel := context.WithCancel(context.Background())
  26. defer cancel()
  27. stream, err := client.VolumeTailSender(ctx, &volume_server_pb.VolumeTailSenderRequest{
  28. VolumeId: uint32(vid),
  29. SinceNs: sinceNs,
  30. IdleTimeoutSeconds: uint32(idleTimeoutSeconds),
  31. })
  32. if err != nil {
  33. return err
  34. }
  35. for {
  36. resp, recvErr := stream.Recv()
  37. if recvErr != nil {
  38. if recvErr == io.EOF {
  39. break
  40. } else {
  41. return recvErr
  42. }
  43. }
  44. needleHeader := resp.NeedleHeader
  45. needleBody := resp.NeedleBody
  46. if len(needleHeader) == 0 {
  47. continue
  48. }
  49. for !resp.IsLastChunk {
  50. resp, recvErr = stream.Recv()
  51. if recvErr != nil {
  52. if recvErr == io.EOF {
  53. break
  54. } else {
  55. return recvErr
  56. }
  57. }
  58. needleBody = append(needleBody, resp.NeedleBody...)
  59. }
  60. n := new(needle.Needle)
  61. n.ParseNeedleHeader(needleHeader)
  62. err = n.ReadNeedleBodyBytes(needleBody, needle.CurrentVersion)
  63. if err != nil {
  64. return err
  65. }
  66. err = fn(n)
  67. if err != nil {
  68. return err
  69. }
  70. }
  71. return nil
  72. })
  73. }