|
@ -1,7 +1,6 @@ |
|
|
package filersink |
|
|
package filersink |
|
|
|
|
|
|
|
|
import ( |
|
|
import ( |
|
|
"context" |
|
|
|
|
|
"fmt" |
|
|
"fmt" |
|
|
"github.com/seaweedfs/seaweedfs/weed/util" |
|
|
"github.com/seaweedfs/seaweedfs/weed/util" |
|
|
"sync" |
|
|
"sync" |
|
@ -12,7 +11,6 @@ import ( |
|
|
"github.com/seaweedfs/seaweedfs/weed/operation" |
|
|
"github.com/seaweedfs/seaweedfs/weed/operation" |
|
|
"github.com/seaweedfs/seaweedfs/weed/pb" |
|
|
"github.com/seaweedfs/seaweedfs/weed/pb" |
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|
|
"github.com/seaweedfs/seaweedfs/weed/security" |
|
|
|
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path string) (replicatedChunks []*filer_pb.FileChunk, err error) { |
|
|
func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path string) (replicatedChunks []*filer_pb.FileChunk, err error) { |
|
@ -67,12 +65,9 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string) |
|
|
} |
|
|
} |
|
|
defer util.CloseResponse(resp) |
|
|
defer util.CloseResponse(resp) |
|
|
|
|
|
|
|
|
var host string |
|
|
|
|
|
var auth security.EncodedJwt |
|
|
|
|
|
|
|
|
|
|
|
if err := fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { |
|
|
|
|
|
return util.Retry("assignVolume", func() error { |
|
|
|
|
|
request := &filer_pb.AssignVolumeRequest{ |
|
|
|
|
|
|
|
|
fileId, uploadResult, err, _ := operation.UploadWithRetry( |
|
|
|
|
|
fs, |
|
|
|
|
|
&filer_pb.AssignVolumeRequest{ |
|
|
Count: 1, |
|
|
Count: 1, |
|
|
Replication: fs.replication, |
|
|
Replication: fs.replication, |
|
|
Collection: fs.collection, |
|
|
Collection: fs.collection, |
|
@ -80,49 +75,31 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string) |
|
|
DataCenter: fs.dataCenter, |
|
|
DataCenter: fs.dataCenter, |
|
|
DiskType: fs.diskType, |
|
|
DiskType: fs.diskType, |
|
|
Path: path, |
|
|
Path: path, |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
resp, err := client.AssignVolume(context.Background(), request) |
|
|
|
|
|
if err != nil { |
|
|
|
|
|
glog.V(0).Infof("assign volume failure %v: %v", request, err) |
|
|
|
|
|
return err |
|
|
|
|
|
} |
|
|
|
|
|
if resp.Error != "" { |
|
|
|
|
|
return fmt.Errorf("assign volume failure %v: %v", request, resp.Error) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
fileId, host, auth = resp.FileId, resp.Location.Url, security.EncodedJwt(resp.Auth) |
|
|
|
|
|
|
|
|
|
|
|
return nil |
|
|
|
|
|
}) |
|
|
|
|
|
}); err != nil { |
|
|
|
|
|
return "", fmt.Errorf("filerGrpcAddress assign volume: %v", err) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
fileUrl := fmt.Sprintf("http://%s/%s", host, fileId) |
|
|
|
|
|
if fs.writeChunkByFiler { |
|
|
|
|
|
fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header) |
|
|
|
|
|
|
|
|
|
|
|
// fetch data as is, regardless whether it is encrypted or not
|
|
|
|
|
|
uploadOption := &operation.UploadOption{ |
|
|
|
|
|
UploadUrl: fileUrl, |
|
|
|
|
|
|
|
|
}, |
|
|
|
|
|
&operation.UploadOption{ |
|
|
Filename: filename, |
|
|
Filename: filename, |
|
|
Cipher: false, |
|
|
Cipher: false, |
|
|
IsInputCompressed: "gzip" == header.Get("Content-Encoding"), |
|
|
IsInputCompressed: "gzip" == header.Get("Content-Encoding"), |
|
|
MimeType: header.Get("Content-Type"), |
|
|
MimeType: header.Get("Content-Type"), |
|
|
PairMap: nil, |
|
|
PairMap: nil, |
|
|
Jwt: auth, |
|
|
|
|
|
|
|
|
}, |
|
|
|
|
|
func(host, fileId string) string { |
|
|
|
|
|
fileUrl := fmt.Sprintf("http://%s/%s", host, fileId) |
|
|
|
|
|
if fs.writeChunkByFiler { |
|
|
|
|
|
fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId) |
|
|
} |
|
|
} |
|
|
uploadResult, err, _ := operation.Upload(resp.Body, uploadOption) |
|
|
|
|
|
|
|
|
glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header) |
|
|
|
|
|
return fileUrl |
|
|
|
|
|
}, |
|
|
|
|
|
resp.Body, |
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
if err != nil { |
|
|
if err != nil { |
|
|
glog.V(0).Infof("upload source data %v to %s: %v", sourceChunk.GetFileIdString(), fileUrl, err) |
|
|
|
|
|
|
|
|
glog.V(0).Infof("upload source data %v: %v", sourceChunk.GetFileIdString(), err) |
|
|
return "", fmt.Errorf("upload data: %v", err) |
|
|
return "", fmt.Errorf("upload data: %v", err) |
|
|
} |
|
|
} |
|
|
if uploadResult.Error != "" { |
|
|
if uploadResult.Error != "" { |
|
|
glog.V(0).Infof("upload failure %v to %s: %v", filename, fileUrl, err) |
|
|
|
|
|
|
|
|
glog.V(0).Infof("upload failure %v: %v", filename, err) |
|
|
return "", fmt.Errorf("upload result: %v", uploadResult.Error) |
|
|
return "", fmt.Errorf("upload result: %v", uploadResult.Error) |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|