stream read multiple volumes in a volume server

unmaintained/stream_read_volume/stream_read_volume.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/operation"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/chrislusf/seaweedfs/weed/security"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"google.golang.org/grpc"
+	"io"
+)
+
+var (
+	volumeServer       = flag.String("volumeServer", "localhost:8080", "a volume server")
+	volumeId         = flag.Int("volumeId", -1, "a volume id to stream read")
+	grpcDialOption   grpc.DialOption
+)
+
+func main() {
+	flag.Parse()
+
+	util.LoadConfiguration("security", false)
+	grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
+
+	vid := uint32(*volumeId)
+
+	eachNeedleFunc := func(resp *volume_server_pb.ReadAllNeedlesResponse) error {
+		fmt.Printf("%d,%x%08x %d\n", resp.VolumeId, resp.NeedleId, resp.Cookie, len(resp.NeedleBlob))
+		return nil
+	}
+
+	err := operation.WithVolumeServerClient(pb.ServerAddress(*volumeServer), grpcDialOption, func(vs volume_server_pb.VolumeServerClient) error {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		copyFileClient, err := vs.ReadAllNeedles(ctx, &volume_server_pb.ReadAllNeedlesRequest{
+			VolumeIds:           []uint32{vid},
+		})
+		if err != nil {
+			return err
+		}
+		for {
+			resp, err := copyFileClient.Recv()
+			if errors.Is(err, io.EOF) {
+				break
+			}
+			if err != nil {
+				return err
+			}
+			if err = eachNeedleFunc(resp); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		fmt.Printf("read %s: %v\n", *volumeServer, err)
+	}
+
+}
+

weed/pb/volume_server.proto

@@ -287,11 +287,13 @@ message WriteNeedleBlobResponse {
 }
 
 message ReadAllNeedlesRequest {
-    uint32 volume_id = 1;
+    repeated uint32 volume_ids = 1;
 }
 message ReadAllNeedlesResponse {
-    uint64 needle_id = 1;
-    bytes needle_blob = 2;
+    uint32 volume_id = 1;
+    uint64 needle_id = 2;
+    uint32 cookie = 3;
+    bytes needle_blob = 5;
 }
 
 message VolumeTailSenderRequest {

weed/server/volume_grpc_read_all.go

@@ -10,13 +10,23 @@ import (
 
 func (vs *VolumeServer) ReadAllNeedles(req *volume_server_pb.ReadAllNeedlesRequest, stream volume_server_pb.VolumeServer_ReadAllNeedlesServer) (err error) {
 
-	v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
+	for _, vid := range req.VolumeIds {
+		if err := vs.streaReadOneVolume(needle.VolumeId(vid), stream, err); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (vs *VolumeServer) streaReadOneVolume(vid needle.VolumeId, stream volume_server_pb.VolumeServer_ReadAllNeedlesServer, err error) error {
+	v := vs.store.GetVolume(vid)
 	if v == nil {
-		return fmt.Errorf("not found volume id %d", req.VolumeId)
+		return fmt.Errorf("not found volume id %d", vid)
 	}
 
 	scanner := &VolumeFileScanner4ReadAll{
 		stream: stream,
+		v:      v,
 	}
 
 	offset := int64(v.SuperBlock.BlockSize())
@@ -24,11 +34,11 @@ func (vs *VolumeServer) ReadAllNeedles(req *volume_server_pb.ReadAllNeedlesReque
 	err = storage.ScanVolumeFileFrom(v.Version(), v.DataBackend, offset, scanner)
 
 	return err
-
 }
 
 type VolumeFileScanner4ReadAll struct {
 	stream volume_server_pb.VolumeServer_ReadAllNeedlesServer
+	v      *storage.Volume
 }
 
 func (scanner *VolumeFileScanner4ReadAll) VisitSuperBlock(superBlock super_block.SuperBlock) error {
@@ -42,7 +52,9 @@ func (scanner *VolumeFileScanner4ReadAll) ReadNeedleBody() bool {
 func (scanner *VolumeFileScanner4ReadAll) VisitNeedle(n *needle.Needle, offset int64, needleHeader, needleBody []byte) error {
 
 	sendErr := scanner.stream.Send(&volume_server_pb.ReadAllNeedlesResponse{
+		VolumeId:   uint32(scanner.v.Id),
 		NeedleId:   uint64(n.Id),
+		Cookie:     uint32(n.Cookie),
 		NeedleBlob: n.Data,
 	})
 	if sendErr != nil {