remote.cache supports replication

weed/pb/volume_server.proto

@@ -475,6 +475,12 @@ message FetchAndWriteNeedleRequest {
     uint32 cookie = 3;
     int64 offset = 4;
     int64 size = 5;
+    message Replica {
+        string url = 1;
+        string public_url = 2;
+    }
+    repeated Replica replicas = 6;
+    string auth = 7;
     // remote conf
     remote_pb.RemoteConf remote_conf = 15;
     remote_pb.RemoteStorageLocation remote_location = 16;

weed/server/filer_grpc_server_remote.go

@@ -111,6 +111,14 @@ func (fs *FilerServer) DownloadToLocal(ctx context.Context, req *filer_pb.Downlo
 				return
 			}
 
+			var replicas []*volume_server_pb.FetchAndWriteNeedleRequest_Replica
+			for _, r := range assignResult.Replicas {
+				replicas = append(replicas, &volume_server_pb.FetchAndWriteNeedleRequest_Replica{
+					Url:       r.Url,
+					PublicUrl: r.PublicUrl,
+				})
+			}
+
 			// tell filer to tell volume server to download into needles
 			err = operation.WithVolumeServerClient(assignResult.Url, fs.grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
 				_, fetchAndWriteErr := volumeServerClient.FetchAndWriteNeedle(context.Background(), &volume_server_pb.FetchAndWriteNeedleRequest{
@@ -119,6 +127,8 @@ func (fs *FilerServer) DownloadToLocal(ctx context.Context, req *filer_pb.Downlo
 					Cookie:     uint32(fileId.Cookie),
 					Offset:     localOffset,
 					Size:       size,
+					Replicas:   replicas,
+					Auth:       string(assignResult.Auth),
 					RemoteConf: storageConf,
 					RemoteLocation: &remote_pb.RemoteStorageLocation{
 						Name:   remoteStorageMountedLocation.Name,

weed/server/volume_grpc_remote.go

@@ -3,10 +3,13 @@ package weed_server
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/chrislusf/seaweedfs/weed/remote_storage"
+	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
+	"sync"
 )
 
 func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_server_pb.FetchAndWriteNeedleRequest) (resp *volume_server_pb.FetchAndWriteNeedleResponse, err error) {
@@ -30,16 +33,48 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 		return nil, fmt.Errorf("read from remote %+v: %v", remoteStorageLocation, ReadRemoteErr)
 	}
 
-	n := new(needle.Needle)
-	n.Id = types.NeedleId(req.NeedleId)
-	n.Cookie = types.Cookie(req.Cookie)
-	n.Data, n.DataSize = data, uint32(len(data))
-	// copied from *Needle.prepareWriteBuffer()
-	n.Size = 4 + types.Size(n.DataSize) + 1
-	n.Checksum = needle.NewCRC(n.Data)
-	if _, err = vs.store.WriteVolumeNeedle(v.Id, n, true, false); err != nil {
-		return nil, fmt.Errorf("write needle %d size %d: %v", req.NeedleId, req.Size, err)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		n := new(needle.Needle)
+		n.Id = types.NeedleId(req.NeedleId)
+		n.Cookie = types.Cookie(req.Cookie)
+		n.Data, n.DataSize = data, uint32(len(data))
+		// copied from *Needle.prepareWriteBuffer()
+		n.Size = 4 + types.Size(n.DataSize) + 1
+		n.Checksum = needle.NewCRC(n.Data)
+		if _, localWriteErr := vs.store.WriteVolumeNeedle(v.Id, n, true, false); localWriteErr != nil {
+			if err == nil {
+				err = fmt.Errorf("local write needle %d size %d: %v", req.NeedleId, req.Size, err)
+			}
+		}
+	}()
+	if len(req.Replicas)>0{
+		fileId := needle.NewFileId(v.Id, req.NeedleId, req.Cookie)
+		for _, replica := range req.Replicas {
+			wg.Add(1)
+			go func(targetVolumeServer string) {
+				defer wg.Done()
+				uploadOption := &operation.UploadOption{
+					UploadUrl:         fmt.Sprintf("http://%s/%s?type=replicate", targetVolumeServer, fileId.String()),
+					Filename:          "",
+					Cipher:            false,
+					IsInputCompressed: false,
+					MimeType:          "",
+					PairMap:           nil,
+					Jwt:               security.EncodedJwt(req.Auth),
+				}
+				if _, replicaWriteErr := operation.UploadData(data, uploadOption); replicaWriteErr != nil {
+					if err == nil {
+						err = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, err)
+					}
+				}
+			}(replica.Url)
+		}
 	}
 
-	return resp, nil
+	wg.Wait()
+
+	return resp, err
 }