Browse Source

refactor

pull/2241/head
Chris Lu 3 years ago
parent
commit
270770d7d7
  1. 43
      weed/pb/volume_server.proto
  2. 1192
      weed/pb/volume_server_pb/volume_server.pb.go
  3. 38
      weed/server/volume_grpc_read_write.go
  4. 49
      weed/server/volume_grpc_remote.go

43
weed/pb/volume_server.proto

@ -56,8 +56,6 @@ service VolumeServer {
} }
rpc WriteNeedleBlob (WriteNeedleBlobRequest) returns (WriteNeedleBlobResponse) { rpc WriteNeedleBlob (WriteNeedleBlobRequest) returns (WriteNeedleBlobResponse) {
} }
rpc FetchAndWriteNeedle (FetchAndWriteNeedleRequest) returns (FetchAndWriteNeedleResponse) {
}
rpc VolumeTailSender (VolumeTailSenderRequest) returns (stream VolumeTailSenderResponse) { rpc VolumeTailSender (VolumeTailSenderRequest) returns (stream VolumeTailSenderResponse) {
} }
@ -95,6 +93,10 @@ service VolumeServer {
rpc VolumeServerLeave (VolumeServerLeaveRequest) returns (VolumeServerLeaveResponse) { rpc VolumeServerLeave (VolumeServerLeaveRequest) returns (VolumeServerLeaveResponse) {
} }
// remote storage
rpc FetchAndWriteNeedle (FetchAndWriteNeedleRequest) returns (FetchAndWriteNeedleResponse) {
}
// <experimental> query // <experimental> query
rpc Query (QueryRequest) returns (stream QueriedStripe) { rpc Query (QueryRequest) returns (stream QueriedStripe) {
} }
@ -278,23 +280,6 @@ message WriteNeedleBlobRequest {
} }
message WriteNeedleBlobResponse { message WriteNeedleBlobResponse {
} }
message FetchAndWriteNeedleRequest {
uint32 volume_id = 1;
uint64 needle_id = 2;
int64 offset = 3;
int64 size = 4;
// remote info
string remote_type = 5;
string remote_name = 6;
string s3_access_key = 8;
string s3_secret_key = 9;
string s3_region = 10;
string s3_endpoint = 11;
string remote_bucket = 12;
string remote_key = 13;
}
message FetchAndWriteNeedleResponse {
}
message VolumeTailSenderRequest { message VolumeTailSenderRequest {
uint32 volume_id = 1; uint32 volume_id = 1;
@ -445,6 +430,7 @@ message VolumeInfo {
string replication = 3; string replication = 3;
} }
// tiered storage
message VolumeTierMoveDatToRemoteRequest { message VolumeTierMoveDatToRemoteRequest {
uint32 volume_id = 1; uint32 volume_id = 1;
string collection = 2; string collection = 2;
@ -479,6 +465,25 @@ message VolumeServerLeaveRequest {
message VolumeServerLeaveResponse { message VolumeServerLeaveResponse {
} }
// remote storage
message FetchAndWriteNeedleRequest {
uint32 volume_id = 1;
uint64 needle_id = 2;
int64 offset = 3;
int64 size = 4;
// remote conf
string remote_type = 5;
string remote_name = 6;
string s3_access_key = 8;
string s3_secret_key = 9;
string s3_region = 10;
string s3_endpoint = 11;
string remote_bucket = 12;
string remote_key = 13;
}
message FetchAndWriteNeedleResponse {
}
// select on volume servers // select on volume servers
message QueryRequest { message QueryRequest {
repeated string selections = 1; repeated string selections = 1;

1192
weed/pb/volume_server_pb/volume_server.pb.go
File diff suppressed because it is too large
View File

38
weed/server/volume_grpc_read_write.go

@ -38,41 +38,3 @@ func (vs *VolumeServer) WriteNeedleBlob(ctx context.Context, req *volume_server_
return resp, nil return resp, nil
} }
func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_server_pb.FetchAndWriteNeedleRequest) (resp *volume_server_pb.FetchAndWriteNeedleResponse, err error) {
resp = &volume_server_pb.FetchAndWriteNeedleResponse{}
v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
if v == nil {
return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
}
remoteConf := &filer_pb.RemoteConf{
Type: req.RemoteType,
Name: req.RemoteName,
S3AccessKey: req.S3AccessKey,
S3SecretKey: req.S3SecretKey,
S3Region: req.S3Region,
S3Endpoint: req.S3Endpoint,
}
client, getClientErr := remote_storage.GetRemoteStorage(remoteConf)
if getClientErr != nil {
return nil, fmt.Errorf("get remote client: %v", getClientErr)
}
remoteStorageLocation := &filer_pb.RemoteStorageLocation{
Name: req.RemoteName,
Bucket: req.RemoteBucket,
Path: req.RemoteKey,
}
data, ReadRemoteErr := client.ReadFile(remoteStorageLocation, req.Offset, req.Size)
if ReadRemoteErr != nil {
return nil, fmt.Errorf("read from remote %+v: %v", remoteStorageLocation, ReadRemoteErr)
}
if err = v.WriteNeedleBlob(types.NeedleId(req.NeedleId), data, types.Size(req.Size)); err != nil {
return nil, fmt.Errorf("write blob needle %d size %d: %v", req.NeedleId, req.Size, err)
}
return resp, nil
}

49
weed/server/volume_grpc_remote.go

@ -0,0 +1,49 @@
package weed_server
import (
"context"
"fmt"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
"github.com/chrislusf/seaweedfs/weed/remote_storage"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
"github.com/chrislusf/seaweedfs/weed/storage/types"
)
func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_server_pb.FetchAndWriteNeedleRequest) (resp *volume_server_pb.FetchAndWriteNeedleResponse, err error) {
resp = &volume_server_pb.FetchAndWriteNeedleResponse{}
v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
if v == nil {
return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
}
remoteConf := &filer_pb.RemoteConf{
Type: req.RemoteType,
Name: req.RemoteName,
S3AccessKey: req.S3AccessKey,
S3SecretKey: req.S3SecretKey,
S3Region: req.S3Region,
S3Endpoint: req.S3Endpoint,
}
client, getClientErr := remote_storage.GetRemoteStorage(remoteConf)
if getClientErr != nil {
return nil, fmt.Errorf("get remote client: %v", getClientErr)
}
remoteStorageLocation := &filer_pb.RemoteStorageLocation{
Name: req.RemoteName,
Bucket: req.RemoteBucket,
Path: req.RemoteKey,
}
data, ReadRemoteErr := client.ReadFile(remoteStorageLocation, req.Offset, req.Size)
if ReadRemoteErr != nil {
return nil, fmt.Errorf("read from remote %+v: %v", remoteStorageLocation, ReadRemoteErr)
}
if err = v.WriteNeedleBlob(types.NeedleId(req.NeedleId), data, types.Size(req.Size)); err != nil {
return nil, fmt.Errorf("write blob needle %d size %d: %v", req.NeedleId, req.Size, err)
}
return resp, nil
}
Loading…
Cancel
Save