seaweedfs/weed/server/master_grpc_server_volume.go

194 lines
5.3 KiB
Go
Raw Normal View History

package weed_server
import (
"context"
"fmt"
"github.com/chrislusf/seaweedfs/weed/storage/backend"
"github.com/chrislusf/raft"
2019-12-23 20:48:20 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
2019-12-23 20:48:20 +00:00
"github.com/chrislusf/seaweedfs/weed/storage/super_block"
"github.com/chrislusf/seaweedfs/weed/topology"
)
func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
if !ms.Topo.IsLeader() {
return nil, raft.NotLeaderError
}
resp := &master_pb.LookupVolumeResponse{}
volumeLocations := ms.lookupVolumeId(req.VolumeIds, req.Collection)
for _, result := range volumeLocations {
var locations []*master_pb.Location
for _, loc := range result.Locations {
locations = append(locations, &master_pb.Location{
Url: loc.Url,
PublicUrl: loc.PublicUrl,
})
}
resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{
VolumeId: result.VolumeId,
Locations: locations,
Error: result.Error,
})
}
return resp, nil
}
func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest) (*master_pb.AssignResponse, error) {
if !ms.Topo.IsLeader() {
return nil, raft.NotLeaderError
}
if req.Count == 0 {
req.Count = 1
}
if req.Replication == "" {
2019-06-23 10:08:27 +00:00
req.Replication = ms.option.DefaultReplicaPlacement
}
2019-12-23 20:48:20 +00:00
replicaPlacement, err := super_block.NewReplicaPlacementFromString(req.Replication)
if err != nil {
return nil, err
}
2019-04-19 04:43:36 +00:00
ttl, err := needle.ReadTTL(req.Ttl)
if err != nil {
return nil, err
}
option := &topology.VolumeGrowOption{
Collection: req.Collection,
ReplicaPlacement: replicaPlacement,
Ttl: ttl,
Prealloacte: ms.preallocateSize,
DataCenter: req.DataCenter,
Rack: req.Rack,
DataNode: req.DataNode,
2019-10-22 05:57:01 +00:00
MemoryMapMaxSizeMb: req.MemoryMapMaxSizeMb,
}
if !ms.Topo.HasWritableVolume(option) {
if ms.Topo.FreeSpace() <= 0 {
return nil, fmt.Errorf("No free volumes left!")
}
ms.vgLock.Lock()
if !ms.Topo.HasWritableVolume(option) {
if _, err = ms.vg.AutomaticGrowByType(option, ms.grpcDialOption, ms.Topo, int(req.WritableVolumeCount)); err != nil {
ms.vgLock.Unlock()
return nil, fmt.Errorf("Cannot grow volume group! %v", err)
}
}
ms.vgLock.Unlock()
}
fid, count, dn, err := ms.Topo.PickForWrite(req.Count, option)
if err != nil {
return nil, fmt.Errorf("%v", err)
}
return &master_pb.AssignResponse{
Fid: fid,
Url: dn.Url(),
PublicUrl: dn.PublicUrl,
Count: count,
2019-05-04 15:42:25 +00:00
Auth: string(security.GenJwt(ms.guard.SigningKey, ms.guard.ExpiresAfterSec, fid)),
}, nil
}
func (ms *MasterServer) Statistics(ctx context.Context, req *master_pb.StatisticsRequest) (*master_pb.StatisticsResponse, error) {
if !ms.Topo.IsLeader() {
return nil, raft.NotLeaderError
}
if req.Replication == "" {
2019-06-23 10:08:27 +00:00
req.Replication = ms.option.DefaultReplicaPlacement
}
2019-12-23 20:48:20 +00:00
replicaPlacement, err := super_block.NewReplicaPlacementFromString(req.Replication)
if err != nil {
return nil, err
}
2019-04-19 04:43:36 +00:00
ttl, err := needle.ReadTTL(req.Ttl)
if err != nil {
return nil, err
}
volumeLayout := ms.Topo.GetVolumeLayout(req.Collection, replicaPlacement, ttl)
stats := volumeLayout.Stats()
totalSize := ms.Topo.GetMaxVolumeCount() * int64(ms.option.VolumeSizeLimitMB) * 1024 * 1024
resp := &master_pb.StatisticsResponse{
TotalSize: uint64(totalSize),
UsedSize: stats.UsedSize,
FileCount: stats.FileCount,
}
return resp, nil
}
2019-03-18 03:27:08 +00:00
func (ms *MasterServer) VolumeList(ctx context.Context, req *master_pb.VolumeListRequest) (*master_pb.VolumeListResponse, error) {
if !ms.Topo.IsLeader() {
return nil, raft.NotLeaderError
}
resp := &master_pb.VolumeListResponse{
2019-05-06 04:17:23 +00:00
TopologyInfo: ms.Topo.ToTopologyInfo(),
2019-06-23 10:08:27 +00:00
VolumeSizeLimitMb: uint64(ms.option.VolumeSizeLimitMB),
2019-03-18 03:27:08 +00:00
}
return resp, nil
}
func (ms *MasterServer) LookupEcVolume(ctx context.Context, req *master_pb.LookupEcVolumeRequest) (*master_pb.LookupEcVolumeResponse, error) {
if !ms.Topo.IsLeader() {
return nil, raft.NotLeaderError
}
resp := &master_pb.LookupEcVolumeResponse{}
ecLocations, found := ms.Topo.LookupEcShards(needle.VolumeId(req.VolumeId))
if !found {
return resp, fmt.Errorf("ec volume %d not found", req.VolumeId)
}
resp.VolumeId = req.VolumeId
for shardId, shardLocations := range ecLocations.Locations {
var locations []*master_pb.Location
for _, dn := range shardLocations {
locations = append(locations, &master_pb.Location{
Url: string(dn.Id()),
PublicUrl: dn.PublicUrl,
})
}
resp.ShardIdLocations = append(resp.ShardIdLocations, &master_pb.LookupEcVolumeResponse_EcShardIdLocation{
ShardId: uint32(shardId),
Locations: locations,
})
}
return resp, nil
}
2019-06-23 22:29:49 +00:00
func (ms *MasterServer) GetMasterConfiguration(ctx context.Context, req *master_pb.GetMasterConfigurationRequest) (*master_pb.GetMasterConfigurationResponse, error) {
resp := &master_pb.GetMasterConfigurationResponse{
MetricsAddress: ms.option.MetricsAddress,
MetricsIntervalSeconds: uint32(ms.option.MetricsIntervalSec),
StorageBackends: backend.ToPbStorageBackends(),
DefaultReplication: ms.option.DefaultReplicaPlacement,
2019-06-23 22:29:49 +00:00
}
return resp, nil
}