seaweedfs/weed/operation/assign_file_id.go

149 lines
3.9 KiB
Go
Raw Normal View History

package operation
import (
2018-11-23 08:26:15 +00:00
"context"
"fmt"
"strings"
2020-11-16 00:58:48 +00:00
"github.com/chrislusf/seaweedfs/weed/storage/needle"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
2019-02-15 08:09:19 +00:00
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util"
)
2016-06-26 02:50:18 +00:00
type VolumeAssignRequest struct {
Count uint64
Replication string
Collection string
Ttl string
2020-12-16 17:14:05 +00:00
DiskType string
DataCenter string
Rack string
DataNode string
WritableVolumeCount uint32
2016-06-26 02:50:18 +00:00
}
type AssignResult struct {
2019-02-15 08:09:19 +00:00
Fid string `json:"fid,omitempty"`
Url string `json:"url,omitempty"`
PublicUrl string `json:"publicUrl,omitempty"`
Count uint64 `json:"count,omitempty"`
Error string `json:"error,omitempty"`
Auth security.EncodedJwt `json:"auth,omitempty"`
}
func Assign(masterFn GetMasterFn, grpcDialOption grpc.DialOption, primaryRequest *VolumeAssignRequest, alternativeRequests ...*VolumeAssignRequest) (*AssignResult, error) {
var requests []*VolumeAssignRequest
requests = append(requests, primaryRequest)
requests = append(requests, alternativeRequests...)
2016-06-23 03:19:09 +00:00
var lastError error
ret := &AssignResult{}
for i, request := range requests {
if request == nil {
continue
}
lastError = WithMasterServerClient(masterFn(), grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
req := &master_pb.AssignRequest{
2020-11-17 08:36:21 +00:00
Count: request.Count,
Replication: request.Replication,
Collection: request.Collection,
Ttl: request.Ttl,
2020-12-16 17:14:05 +00:00
DiskType: request.DiskType,
2020-11-17 08:36:21 +00:00
DataCenter: request.DataCenter,
Rack: request.Rack,
DataNode: request.DataNode,
WritableVolumeCount: request.WritableVolumeCount,
}
resp, grpcErr := masterClient.Assign(context.Background(), req)
if grpcErr != nil {
return grpcErr
}
ret.Count = resp.Count
ret.Fid = resp.Fid
ret.Url = resp.Url
ret.PublicUrl = resp.PublicUrl
ret.Error = resp.Error
2019-02-15 08:09:19 +00:00
ret.Auth = security.EncodedJwt(resp.Auth)
return nil
})
if lastError != nil {
continue
}
if ret.Count <= 0 {
lastError = fmt.Errorf("assign failure %d: %v", i+1, ret.Error)
continue
}
2021-04-14 10:29:28 +00:00
break
}
return ret, lastError
}
2019-02-15 08:09:19 +00:00
func LookupJwt(master string, fileId string) security.EncodedJwt {
tokenStr := ""
if h, e := util.Head(fmt.Sprintf("http://%s/dir/lookup?fileId=%s", master, fileId)); e == nil {
bearer := h.Get("Authorization")
if len(bearer) > 7 && strings.ToUpper(bearer[0:6]) == "BEARER" {
tokenStr = bearer[7:]
2019-02-15 08:09:19 +00:00
}
}
return security.EncodedJwt(tokenStr)
}
2020-11-16 00:58:48 +00:00
type StorageOption struct {
Replication string
2020-12-16 17:14:05 +00:00
DiskType string
Collection string
DataCenter string
Rack string
TtlSeconds int32
Fsync bool
VolumeGrowthCount uint32
2020-11-16 00:58:48 +00:00
}
func (so *StorageOption) TtlString() string {
return needle.SecondsToTTL(so.TtlSeconds)
}
func (so *StorageOption) ToAssignRequests(count int) (ar *VolumeAssignRequest, altRequest *VolumeAssignRequest) {
ar = &VolumeAssignRequest{
Count: uint64(count),
Replication: so.Replication,
Collection: so.Collection,
Ttl: so.TtlString(),
2020-12-16 17:14:05 +00:00
DiskType: so.DiskType,
DataCenter: so.DataCenter,
Rack: so.Rack,
WritableVolumeCount: so.VolumeGrowthCount,
2020-11-16 00:58:48 +00:00
}
if so.DataCenter != "" || so.Rack != "" {
altRequest = &VolumeAssignRequest{
Count: uint64(count),
Replication: so.Replication,
Collection: so.Collection,
Ttl: so.TtlString(),
2020-12-16 17:14:05 +00:00
DiskType: so.DiskType,
DataCenter: "",
Rack: "",
WritableVolumeCount: so.VolumeGrowthCount,
2020-11-16 00:58:48 +00:00
}
}
return
}