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.

117 lines
3.5 KiB

6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "time"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/chrislusf/seaweedfs/weed/operation"
  8. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  9. "github.com/chrislusf/seaweedfs/weed/storage"
  10. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  11. )
  12. func (vs *VolumeServer) VolumeTailSender(req *volume_server_pb.VolumeTailSenderRequest, stream volume_server_pb.VolumeServer_VolumeTailSenderServer) error {
  13. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  14. if v == nil {
  15. return fmt.Errorf("not found volume id %d", req.VolumeId)
  16. }
  17. defer glog.V(1).Infof("tailing volume %d finished", v.Id)
  18. lastTimestampNs := req.SinceNs
  19. drainingSeconds := req.IdleTimeoutSeconds
  20. for {
  21. lastProcessedTimestampNs, err := sendNeedlesSince(stream, v, lastTimestampNs)
  22. if err != nil {
  23. glog.Infof("sendNeedlesSince: %v", err)
  24. return fmt.Errorf("streamFollow: %v", err)
  25. }
  26. time.Sleep(2 * time.Second)
  27. if req.IdleTimeoutSeconds == 0 {
  28. lastTimestampNs = lastProcessedTimestampNs
  29. continue
  30. }
  31. if lastProcessedTimestampNs == lastTimestampNs {
  32. drainingSeconds--
  33. if drainingSeconds <= 0 {
  34. return nil
  35. }
  36. glog.V(1).Infof("tailing volume %d drains requests with %d seconds remaining", v.Id, drainingSeconds)
  37. } else {
  38. lastTimestampNs = lastProcessedTimestampNs
  39. drainingSeconds = req.IdleTimeoutSeconds
  40. glog.V(1).Infof("tailing volume %d resets draining wait time to %d seconds", v.Id, drainingSeconds)
  41. }
  42. }
  43. }
  44. func sendNeedlesSince(stream volume_server_pb.VolumeServer_VolumeTailSenderServer, v *storage.Volume, lastTimestampNs uint64) (lastProcessedTimestampNs uint64, err error) {
  45. foundOffset, isLastOne, err := v.BinarySearchByAppendAtNs(lastTimestampNs)
  46. if err != nil {
  47. return 0, fmt.Errorf("fail to locate by appendAtNs %d: %s", lastTimestampNs, err)
  48. }
  49. // log.Printf("reading ts %d offset %d isLast %v", lastTimestampNs, foundOffset, isLastOne)
  50. if isLastOne {
  51. // need to heart beat to the client to ensure the connection health
  52. sendErr := stream.Send(&volume_server_pb.VolumeTailSenderResponse{IsLastChunk: true})
  53. return lastTimestampNs, sendErr
  54. }
  55. err = storage.ScanVolumeFileNeedleFrom(v.Version(), v.DataFile(), foundOffset.ToAcutalOffset(), func(needleHeader, needleBody []byte, needleAppendAtNs uint64) error {
  56. isLastChunk := false
  57. // need to send body by chunks
  58. for i := 0; i < len(needleBody); i += BufferSizeLimit {
  59. stopOffset := i + BufferSizeLimit
  60. if stopOffset >= len(needleBody) {
  61. isLastChunk = true
  62. stopOffset = len(needleBody)
  63. }
  64. sendErr := stream.Send(&volume_server_pb.VolumeTailSenderResponse{
  65. NeedleHeader: needleHeader,
  66. NeedleBody: needleBody[i:stopOffset],
  67. IsLastChunk: isLastChunk,
  68. })
  69. if sendErr != nil {
  70. return sendErr
  71. }
  72. }
  73. lastProcessedTimestampNs = needleAppendAtNs
  74. return nil
  75. })
  76. return
  77. }
  78. func (vs *VolumeServer) VolumeTailReceiver(ctx context.Context, req *volume_server_pb.VolumeTailReceiverRequest) (*volume_server_pb.VolumeTailReceiverResponse, error) {
  79. resp := &volume_server_pb.VolumeTailReceiverResponse{}
  80. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  81. if v == nil {
  82. return resp, fmt.Errorf("receiver not found volume id %d", req.VolumeId)
  83. }
  84. defer glog.V(1).Infof("receive tailing volume %d finished", v.Id)
  85. return resp, operation.TailVolumeFromSource(req.SourceVolumeServer, vs.grpcDialOption, v.Id, req.SinceNs, int(req.IdleTimeoutSeconds), func(n *needle.Needle) error {
  86. _, _, err := vs.store.WriteVolumeNeedle(v.Id, n)
  87. return err
  88. })
  89. }