seaweedfs/weed/operation/lookup.go
chrislu 9f9ef1340c use streaming mode for long poll grpc calls
streaming mode would create separate grpc connections for each call.
this is to ensure the long poll connections are properly closed.
2021-12-26 00:15:03 -08:00

122 lines
3.3 KiB
Go

package operation
import (
"context"
"errors"
"fmt"
"github.com/chrislusf/seaweedfs/weed/pb"
"google.golang.org/grpc"
"math/rand"
"strings"
"time"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
)
type Location struct {
Url string `json:"url,omitempty"`
PublicUrl string `json:"publicUrl,omitempty"`
GrpcPort int `json:"grpcPort,omitempty"`
}
func (l *Location) ServerAddress() pb.ServerAddress {
return pb.NewServerAddressWithGrpcPort(l.Url, l.GrpcPort)
}
type LookupResult struct {
VolumeOrFileId string `json:"volumeOrFileId,omitempty"`
Locations []Location `json:"locations,omitempty"`
Jwt string `json:"jwt,omitempty"`
Error string `json:"error,omitempty"`
}
func (lr *LookupResult) String() string {
return fmt.Sprintf("VolumeOrFileId:%s, Locations:%v, Error:%s", lr.VolumeOrFileId, lr.Locations, lr.Error)
}
var (
vc VidCache // caching of volume locations, re-check if after 10 minutes
)
func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrl string, jwt string, err error) {
parts := strings.Split(fileId, ",")
if len(parts) != 2 {
return "", jwt, errors.New("Invalid fileId " + fileId)
}
lookup, lookupError := LookupVolumeId(masterFn, grpcDialOption, parts[0])
if lookupError != nil {
return "", jwt, lookupError
}
if len(lookup.Locations) == 0 {
return "", jwt, errors.New("File Not Found")
}
return "http://" + lookup.Locations[rand.Intn(len(lookup.Locations))].Url + "/" + fileId, lookup.Jwt, nil
}
func LookupVolumeId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid string) (*LookupResult, error) {
results, err := LookupVolumeIds(masterFn, grpcDialOption, []string{vid})
return results[vid], err
}
// LookupVolumeIds find volume locations by cache and actual lookup
func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids []string) (map[string]*LookupResult, error) {
ret := make(map[string]*LookupResult)
var unknown_vids []string
//check vid cache first
for _, vid := range vids {
locations, cacheErr := vc.Get(vid)
if cacheErr == nil {
ret[vid] = &LookupResult{VolumeOrFileId: vid, Locations: locations}
} else {
unknown_vids = append(unknown_vids, vid)
}
}
//return success if all volume ids are known
if len(unknown_vids) == 0 {
return ret, nil
}
//only query unknown_vids
err := WithMasterServerClient(false, masterFn(), grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
req := &master_pb.LookupVolumeRequest{
VolumeOrFileIds: unknown_vids,
}
resp, grpcErr := masterClient.LookupVolume(context.Background(), req)
if grpcErr != nil {
return grpcErr
}
//set newly checked vids to cache
for _, vidLocations := range resp.VolumeIdLocations {
var locations []Location
for _, loc := range vidLocations.Locations {
locations = append(locations, Location{
Url: loc.Url,
PublicUrl: loc.PublicUrl,
GrpcPort: int(loc.GrpcPort),
})
}
if vidLocations.Error != "" {
vc.Set(vidLocations.VolumeOrFileId, locations, 10*time.Minute)
}
ret[vidLocations.VolumeOrFileId] = &LookupResult{
VolumeOrFileId: vidLocations.VolumeOrFileId,
Locations: locations,
Jwt: vidLocations.Auth,
Error: vidLocations.Error,
}
}
return nil
})
if err != nil {
return nil, err
}
return ret, nil
}