seaweedfs/weed/operation/tail_volume.go

79 lines
1.7 KiB
Go
Raw Normal View History

2019-04-19 04:43:36 +00:00
package operation
2019-04-19 04:17:43 +00:00
import (
"context"
"fmt"
"io"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
2019-04-19 04:43:36 +00:00
"github.com/chrislusf/seaweedfs/weed/storage/needle"
2019-04-19 04:17:43 +00:00
"google.golang.org/grpc"
)
2019-04-19 04:43:36 +00:00
func TailVolume(master string, grpcDialOption grpc.DialOption, vid needle.VolumeId, sinceNs uint64, timeoutSeconds int, fn func(n *needle.Needle) error) error {
2019-04-19 04:17:43 +00:00
// find volume location, replication, ttl info
2019-04-19 04:43:36 +00:00
lookup, err := Lookup(master, vid.String())
2019-04-19 04:17:43 +00:00
if err != nil {
return fmt.Errorf("look up volume %d: %v", vid, err)
}
if len(lookup.Locations) == 0 {
return fmt.Errorf("unable to locate volume %d", vid)
}
volumeServer := lookup.Locations[0].Url
2019-04-19 04:43:36 +00:00
return WithVolumeServerClient(volumeServer, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
2019-04-19 04:17:43 +00:00
stream, err := client.VolumeTail(context.Background(), &volume_server_pb.VolumeTailRequest{
VolumeId: uint32(vid),
SinceNs: sinceNs,
DrainingSeconds: uint32(timeoutSeconds),
})
if err != nil {
return err
}
for {
resp, recvErr := stream.Recv()
if recvErr != nil {
if recvErr == io.EOF {
break
} else {
return recvErr
}
}
needleHeader := resp.NeedleHeader
needleBody := resp.NeedleBody
if len(needleHeader) == 0 {
continue
}
for !resp.IsLastChunk {
resp, recvErr = stream.Recv()
if recvErr != nil {
if recvErr == io.EOF {
break
} else {
return recvErr
}
}
needleBody = append(needleBody, resp.NeedleBody...)
}
2019-04-19 04:43:36 +00:00
n := new(needle.Needle)
2019-04-19 04:17:43 +00:00
n.ParseNeedleHeader(needleHeader)
2019-04-19 04:43:36 +00:00
n.ReadNeedleBodyBytes(needleBody, needle.CurrentVersion)
2019-04-19 04:17:43 +00:00
err = fn(n)
if err != nil {
return err
}
}
return nil
})
}