use grpc and jwt

weed/operation/assign_file_id.go

@@ -3,14 +3,11 @@ package operation
 import (
 	"context"
 	"fmt"
-	"strings"
-
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"google.golang.org/grpc"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/security"
-	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 type VolumeAssignRequest struct {
@@ -96,18 +93,28 @@ func Assign(masterFn GetMasterFn, grpcDialOption grpc.DialOption, primaryRequest
 	return ret, lastError
 }
 
-func LookupJwt(master string, fileId string) security.EncodedJwt {
+func LookupJwt(master string, grpcDialOption grpc.DialOption, fileId string) (token security.EncodedJwt) {
 
-	tokenStr := ""
+	WithMasterServerClient(master, grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 
-	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:]
+		resp, grpcErr := masterClient.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{
+			VolumeOrFileIds: []string{fileId},
+		})
+		if grpcErr != nil {
+			return grpcErr
 		}
-	}
 
-	return security.EncodedJwt(tokenStr)
+		if len(resp.VolumeIdLocations) == 0 {
+			return nil
+		}
+
+		token = security.EncodedJwt(resp.VolumeIdLocations[0].Auth)
+
+		return nil
+
+	})
+
+	return
 }
 
 type StorageOption struct {

weed/operation/chunked_file.go

@@ -40,13 +40,14 @@ type ChunkManifest struct {
 
 // seekable chunked file reader
 type ChunkedFileReader struct {
-	totalSize int64
-	chunkList []*ChunkInfo
-	master    string
-	pos       int64
-	pr        *io.PipeReader
-	pw        *io.PipeWriter
-	mutex     sync.Mutex
+	totalSize      int64
+	chunkList      []*ChunkInfo
+	master         string
+	pos            int64
+	pr             *io.PipeReader
+	pw             *io.PipeWriter
+	mutex          sync.Mutex
+	grpcDialOption grpc.DialOption
 }
 
 func (s ChunkList) Len() int           { return len(s) }
@@ -92,7 +93,7 @@ func (cm *ChunkManifest) DeleteChunks(masterFn GetMasterFn, usePublicUrl bool, g
 	return nil
 }
 
-func readChunkNeedle(fileUrl string, w io.Writer, offset int64) (written int64, e error) {
+func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (written int64, e error) {
 	req, err := http.NewRequest("GET", fileUrl, nil)
 	if err != nil {
 		return written, err
@@ -126,16 +127,17 @@ func readChunkNeedle(fileUrl string, w io.Writer, offset int64) (written int64,
 	return io.Copy(w, resp.Body)
 }
 
-func NewChunkedFileReader(chunkList []*ChunkInfo, master string) *ChunkedFileReader {
+func NewChunkedFileReader(chunkList []*ChunkInfo, master string, grpcDialOption grpc.DialOption) *ChunkedFileReader {
 	var totalSize int64
 	for _, chunk := range chunkList {
 		totalSize += chunk.Size
 	}
 	sort.Sort(ChunkList(chunkList))
 	return &ChunkedFileReader{
-		totalSize: totalSize,
-		chunkList: chunkList,
-		master:    master,
+		totalSize:      totalSize,
+		chunkList:      chunkList,
+		master:         master,
+		grpcDialOption: grpcDialOption,
 	}
 }
 
@@ -174,13 +176,13 @@ func (cf *ChunkedFileReader) WriteTo(w io.Writer) (n int64, err error) {
 	for ; chunkIndex < len(cf.chunkList); chunkIndex++ {
 		ci := cf.chunkList[chunkIndex]
 		// if we need read date from local volume server first?
-		fileUrl, lookupError := LookupFileId(func() string {
+		fileUrl, jwt, lookupError := LookupFileId(func() string {
 			return cf.master
-		}, ci.Fid)
+		}, cf.grpcDialOption, ci.Fid)
 		if lookupError != nil {
 			return n, lookupError
 		}
-		if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset); e != nil {
+		if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset, jwt); e != nil {
 			return n, e
 		} else {
 			n += wn

weed/operation/delete_content.go

@@ -30,7 +30,7 @@ func ParseFileId(fid string) (vid string, key_cookie string, err error) {
 // DeleteFiles batch deletes a list of fileIds
 func DeleteFiles(masterFn GetMasterFn, usePublicUrl bool, grpcDialOption grpc.DialOption, fileIds []string) ([]*volume_server_pb.DeleteResult, error) {
 
-	lookupFunc := func(vids []string) (results map[string]LookupResult, err error) {
+	lookupFunc := func(vids []string) (results map[string]*LookupResult, err error) {
 		results, err = LookupVolumeIds(masterFn, grpcDialOption, vids)
 		if err == nil && usePublicUrl {
 			for _, result := range results {
@@ -46,7 +46,7 @@ func DeleteFiles(masterFn GetMasterFn, usePublicUrl bool, grpcDialOption grpc.Di
 
 }
 
-func DeleteFilesWithLookupVolumeId(grpcDialOption grpc.DialOption, fileIds []string, lookupFunc func(vid []string) (map[string]LookupResult, error)) ([]*volume_server_pb.DeleteResult, error) {
+func DeleteFilesWithLookupVolumeId(grpcDialOption grpc.DialOption, fileIds []string, lookupFunc func(vid []string) (map[string]*LookupResult, error)) ([]*volume_server_pb.DeleteResult, error) {
 
 	var ret []*volume_server_pb.DeleteResult
 

weed/operation/lookup.go

@@ -2,17 +2,14 @@ package operation
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"google.golang.org/grpc"
 	"math/rand"
-	"net/url"
 	"strings"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
-	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 type Location struct {
@@ -20,63 +17,33 @@ type Location struct {
 	PublicUrl string `json:"publicUrl,omitempty"`
 }
 type LookupResult struct {
-	VolumeId  string     `json:"volumeId,omitempty"`
-	Locations []Location `json:"locations,omitempty"`
-	Error     string     `json:"error,omitempty"`
+	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("VolumeId:%s, Locations:%v, Error:%s", lr.VolumeId, lr.Locations, lr.Error)
+	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 Lookup(masterFn GetMasterFn, vid string) (ret *LookupResult, err error) {
-	locations, cache_err := vc.Get(vid)
-	if cache_err != nil {
-		if ret, err = do_lookup(masterFn, vid); err == nil {
-			vc.Set(vid, ret.Locations, 10*time.Minute)
-		}
-	} else {
-		ret = &LookupResult{VolumeId: vid, Locations: locations}
-	}
-	return
-}
-
-func do_lookup(masterFn GetMasterFn, vid string) (*LookupResult, error) {
-	values := make(url.Values)
-	values.Add("volumeId", vid)
-	server := masterFn()
-	jsonBlob, err := util.Post("http://"+server+"/dir/lookup", values)
-	if err != nil {
-		return nil, err
-	}
-	var ret LookupResult
-	err = json.Unmarshal(jsonBlob, &ret)
-	if err != nil {
-		return nil, err
-	}
-	if ret.Error != "" {
-		return nil, errors.New(ret.Error)
-	}
-	return &ret, nil
-}
-
-func LookupFileId(masterFn GetMasterFn, fileId string) (fullUrl string, err error) {
+func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrl string, jwt string, err error) {
 	parts := strings.Split(fileId, ",")
 	if len(parts) != 2 {
-		return "", errors.New("Invalid fileId " + fileId)
+		return "", jwt, errors.New("Invalid fileId " + fileId)
 	}
-	lookup, lookupError := Lookup(masterFn, parts[0])
+	lookup, lookupError := LookupVolumeId(masterFn, grpcDialOption, parts[0])
 	if lookupError != nil {
-		return "", lookupError
+		return "", jwt, lookupError
 	}
 	if len(lookup.Locations) == 0 {
-		return "", errors.New("File Not Found")
+		return "", jwt, errors.New("File Not Found")
 	}
-	return "http://" + lookup.Locations[rand.Intn(len(lookup.Locations))].Url + "/" + fileId, nil
+	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) {
@@ -93,7 +60,7 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 	for _, vid := range vids {
 		locations, cacheErr := vc.Get(vid)
 		if cacheErr == nil {
-			ret[vid] = &LookupResult{VolumeId: vid, Locations: locations}
+			ret[vid] = &LookupResult{VolumeOrFileId: vid, Locations: locations}
 		} else {
 			unknown_vids = append(unknown_vids, vid)
 		}
@@ -108,7 +75,7 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 	err := WithMasterServerClient(masterFn(), grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
 
 		req := &master_pb.LookupVolumeRequest{
-			VolumeIds: unknown_vids,
+			VolumeOrFileIds: unknown_vids,
 		}
 		resp, grpcErr := masterClient.LookupVolume(context.Background(), req)
 		if grpcErr != nil {
@@ -125,12 +92,13 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 				})
 			}
 			if vidLocations.Error != "" {
-				vc.Set(vidLocations.VolumeId, locations, 10*time.Minute)
+				vc.Set(vidLocations.VolumeOrFileId, locations, 10*time.Minute)
 			}
-			ret[vidLocations.VolumeId] = &LookupResult{
-				VolumeId:  vidLocations.VolumeId,
-				Locations: locations,
-				Error:     vidLocations.Error,
+			ret[vidLocations.VolumeOrFileId] = &LookupResult{
+				VolumeOrFileId: vidLocations.VolumeOrFileId,
+				Locations:      locations,
+				Jwt:            vidLocations.Auth,
+				Error:          vidLocations.Error,
 			}
 		}