gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into filer_mongodb

Browse Source 
# Conflicts:
#	go.mod
#	go.sum
#	weed/server/filer_server.go
This commit is contained in:
bukton
2020-04-19 00:21:45 +07:00
parent 6234ea441b 11f5a6d913
commit 290c6b7f01
101 changed files with 3019 additions and 1123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
weed/server/filer_grpc_server.go
View File

@@ -217,6 +217,38 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr
return &filer_pb.UpdateEntryResponse{}, err
}
func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendToEntryRequest) (*filer_pb.AppendToEntryResponse, error) {
fullpath := util.NewFullPath(req.Directory, req.EntryName)
var offset int64 = 0
entry, err := fs.filer.FindEntry(ctx, util.FullPath(fullpath))
if err == filer_pb.ErrNotFound {
entry = &filer2.Entry{
FullPath: fullpath,
Attr: filer2.Attr{
Crtime: time.Now(),
Mtime: time.Now(),
Mode: os.FileMode(0644),
Uid: OS_UID,
Gid: OS_GID,
},
}
} else {
offset = int64(filer2.TotalSize(entry.Chunks))
}
for _, chunk := range req.Chunks {
chunk.Offset = offset
offset += int64(chunk.Size)
}
entry.Chunks = append(entry.Chunks, req.Chunks...)
err = fs.filer.CreateEntry(context.Background(), entry, false)
return &filer_pb.AppendToEntryResponse{}, err
}
func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (resp *filer_pb.DeleteEntryResponse, err error) {
err = fs.filer.DeleteEntryMetaAndData(ctx, util.JoinPath(req.Directory, req.Name), req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData)
resp = &filer_pb.DeleteEntryResponse{}
@@ -232,7 +264,7 @@ func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVol
if req.TtlSec > 0 {
ttlStr = strconv.Itoa(int(req.TtlSec))
}
collection, replication := fs.detectCollection(req.ParentPath, req.Collection, req.Replication)
collection, replication, _ := fs.detectCollection(req.ParentPath, req.Collection, req.Replication)
var altRequest *operation.VolumeAssignRequest
@@ -327,7 +359,6 @@ func (fs *FilerServer) GetFilerConfiguration(ctx context.Context, req *filer_pb.
Replication: fs.option.DefaultReplication,
MaxMb: uint32(fs.option.MaxMB),
DirBuckets: fs.filer.DirBucketsPath,
DirQueues: fs.filer.DirQueuesPath,
Cipher: fs.filer.Cipher,
}, nil
}

8
weed/server/filer_grpc_server_listen.go
View File

@@ -4,12 +4,13 @@ import (
"strings"
"time"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
)
func (fs *FilerServer) ListenForEvents(req *filer_pb.ListenForEventsRequest, stream filer_pb.SeaweedFiler_ListenForEventsServer) error {
func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error {
peerAddress := findClientAddress(stream.Context(), 0)
@@ -37,7 +38,7 @@ func (fs *FilerServer) ListenForEvents(req *filer_pb.ListenForEventsRequest, str
fullpath := util.Join(dirPath, entryName)
// skip on filer internal meta logs
if strings.HasPrefix(fullpath, "/.meta") {
if strings.HasPrefix(fullpath, filer2.SystemLogDir) {
return nil
}
@@ -45,7 +46,7 @@ func (fs *FilerServer) ListenForEvents(req *filer_pb.ListenForEventsRequest, str
return nil
}
message := &filer_pb.FullEventNotification{
message := &filer_pb.SubscribeMetadataResponse{
Directory: dirPath,
EventNotification: eventNotification,
}
@@ -64,7 +65,6 @@ func (fs *FilerServer) ListenForEvents(req *filer_pb.ListenForEventsRequest, str
fs.listenersLock.Unlock()
}
return nil
}
func (fs *FilerServer) addClient(clientType string, clientAddress string) (clientName string) {

28
weed/server/filer_grpc_server_rename.go
View File

@@ -44,12 +44,19 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
}
func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer2.Entry, newParent util.FullPath, newName string, events *MoveEvents) error {
if entry.IsDirectory() {
if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName, events); err != nil {
return err
if err := fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, events, func() error {
if entry.IsDirectory() {
if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName, events); err != nil {
return err
}
}
return nil
}); err != nil {
return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(entry.Name()), newParent.Child(newName), err)
}
return fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, events)
return nil
}
func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, entry *filer2.Entry, newParent util.FullPath, newName string, events *MoveEvents) error {
@@ -85,7 +92,8 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.
return nil
}
func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer2.Entry, newParent util.FullPath, newName string, events *MoveEvents) error {
func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer2.Entry, newParent util.FullPath, newName string, events *MoveEvents,
moveFolderSubEntries func() error) error {
oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
@@ -107,6 +115,14 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
return createErr
}
events.newEntries = append(events.newEntries, newEntry)
if moveFolderSubEntries != nil {
if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
return moveChildrenErr
}
}
// delete old entry
deleteErr := fs.filer.DeleteEntryMetaAndData(ctx, oldPath, false, false, false)
if deleteErr != nil {
@@ -114,7 +130,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
}
events.oldEntries = append(events.oldEntries, entry)
events.newEntries = append(events.newEntries, newEntry)
return nil
}

12
weed/server/filer_server.go
View File

@@ -24,6 +24,7 @@ import (
_ "github.com/chrislusf/seaweedfs/weed/filer2/mysql"
_ "github.com/chrislusf/seaweedfs/weed/filer2/postgres"
_ "github.com/chrislusf/seaweedfs/weed/filer2/redis"
_ "github.com/chrislusf/seaweedfs/weed/filer2/redis2"
_ "github.com/chrislusf/seaweedfs/weed/filer2/mongodb"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/notification"
@@ -73,7 +74,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
glog.Fatal("master list is required!")
}
fs.filer = filer2.NewFiler(option.Masters, fs.grpcDialOption, option.Port+10000, fs.notifyMetaListeners)
fs.filer = filer2.NewFiler(option.Masters, fs.grpcDialOption, option.Port+10000, option.Collection, option.DefaultReplication, fs.notifyMetaListeners)
fs.filer.Cipher = option.Cipher
maybeStartMetrics(fs, option)
@@ -92,10 +93,9 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
util.LoadConfiguration("notification", false)
fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete")
v.Set("filer.option.buckets_folder", "/buckets")
v.Set("filer.option.queues_folder", "/queues")
fs.filer.DirBucketsPath = v.GetString("filer.option.buckets_folder")
fs.filer.DirQueuesPath = v.GetString("filer.option.queues_folder")
v.SetDefault("filer.options.buckets_folder", "/buckets")
fs.filer.DirBucketsPath = v.GetString("filer.options.buckets_folder")
fs.filer.FsyncBuckets = v.GetStringSlice("filer.options.buckets_fsync")
fs.filer.LoadConfiguration(v)
notification.LoadConfiguration(v, "notification.")
@@ -108,7 +108,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
readonlyMux.HandleFunc("/", fs.readonlyFilerHandler)
}
fs.filer.LoadBuckets(fs.filer.DirBucketsPath)
fs.filer.LoadBuckets()
util.OnInterrupt(func() {
fs.filer.Shutdown()

21
weed/server/filer_server_handlers_read.go
View File

@@ -8,6 +8,7 @@ import (
"path/filepath"
"strconv"
"strings"
"time"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
@@ -78,8 +79,26 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request,
w.Header().Set("Content-Type", mimeType)
}
// if modified since
if !entry.Attr.Mtime.IsZero() {
w.Header().Set("Last-Modified", entry.Attr.Mtime.UTC().Format(http.TimeFormat))
if r.Header.Get("If-Modified-Since") != "" {
if t, parseError := time.Parse(http.TimeFormat, r.Header.Get("If-Modified-Since")); parseError == nil {
if t.After(entry.Attr.Mtime) {
w.WriteHeader(http.StatusNotModified)
return
}
}
}
}
// set etag
setEtag(w, filer2.ETag(entry.Chunks))
etag := filer2.ETagEntry(entry)
if inm := r.Header.Get("If-None-Match"); inm == "\""+etag+"\"" {
w.WriteHeader(http.StatusNotModified)
return
}
setEtag(w, etag)
if r.Method == "HEAD" {
w.Header().Set("Content-Length", strconv.FormatInt(int64(filer2.TotalSize(entry.Chunks)), 10))

42
weed/server/filer_server_handlers_write.go
View File

@@ -40,7 +40,7 @@ type FilerPostResult struct {
Url string `json:"url,omitempty"`
}
func (fs *FilerServer) assignNewFileInfo(w http.ResponseWriter, r *http.Request, replication, collection, dataCenter, ttlString string) (fileId, urlLocation string, auth security.EncodedJwt, err error) {
func (fs *FilerServer) assignNewFileInfo(w http.ResponseWriter, r *http.Request, replication, collection, dataCenter, ttlString string, fsync bool) (fileId, urlLocation string, auth security.EncodedJwt, err error) {
stats.FilerRequestCounter.WithLabelValues("assign").Inc()
start := time.Now()
@@ -73,6 +73,9 @@ func (fs *FilerServer) assignNewFileInfo(w http.ResponseWriter, r *http.Request,
}
fileId = assignResult.Fid
urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid
if fsync {
urlLocation += "?fsync=true"
}
auth = assignResult.Auth
return
}
@@ -82,7 +85,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
ctx := context.Background()
query := r.URL.Query()
collection, replication := fs.detectCollection(r.RequestURI, query.Get("collection"), query.Get("replication"))
collection, replication, fsync := fs.detectCollection(r.RequestURI, query.Get("collection"), query.Get("replication"))
dataCenter := query.Get("dataCenter")
if dataCenter == "" {
dataCenter = fs.option.DataCenter
@@ -96,12 +99,12 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
ttlSeconds = int32(ttl.Minutes()) * 60
}
if autoChunked := fs.autoChunk(ctx, w, r, replication, collection, dataCenter, ttlSeconds, ttlString); autoChunked {
if autoChunked := fs.autoChunk(ctx, w, r, replication, collection, dataCenter, ttlSeconds, ttlString, fsync); autoChunked {
return
}
if fs.option.Cipher {
reply, err := fs.encrypt(ctx, w, r, replication, collection, dataCenter, ttlSeconds, ttlString)
reply, err := fs.encrypt(ctx, w, r, replication, collection, dataCenter, ttlSeconds, ttlString, fsync)
if err != nil {
writeJsonError(w, r, http.StatusInternalServerError, err)
} else if reply != nil {
@@ -111,7 +114,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
return
}
fileId, urlLocation, auth, err := fs.assignNewFileInfo(w, r, replication, collection, dataCenter, ttlString)
fileId, urlLocation, auth, err := fs.assignNewFileInfo(w, r, replication, collection, dataCenter, ttlString, fsync)
if err != nil || fileId == "" || urlLocation == "" {
glog.V(0).Infof("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, collection, dataCenter)
@@ -122,12 +125,12 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
glog.V(4).Infof("write %s to %v", r.URL.Path, urlLocation)
u, _ := url.Parse(urlLocation)
ret, err := fs.uploadToVolumeServer(r, u, auth, w, fileId)
ret, md5value, err := fs.uploadToVolumeServer(r, u, auth, w, fileId)
if err != nil {
return
}
if err = fs.updateFilerStore(ctx, r, w, replication, collection, ret, fileId, ttlSeconds); err != nil {
if err = fs.updateFilerStore(ctx, r, w, replication, collection, ret, md5value, fileId, ttlSeconds); err != nil {
return
}
@@ -144,8 +147,8 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
}
// update metadata in filer store
func (fs *FilerServer) updateFilerStore(ctx context.Context, r *http.Request, w http.ResponseWriter,
replication string, collection string, ret *operation.UploadResult, fileId string, ttlSeconds int32) (err error) {
func (fs *FilerServer) updateFilerStore(ctx context.Context, r *http.Request, w http.ResponseWriter, replication string,
collection string, ret *operation.UploadResult, md5value []byte, fileId string, ttlSeconds int32) (err error) {
stats.FilerRequestCounter.WithLabelValues("postStoreWrite").Inc()
start := time.Now()
@@ -186,6 +189,7 @@ func (fs *FilerServer) updateFilerStore(ctx context.Context, r *http.Request, w
Collection: collection,
TtlSec: ttlSeconds,
Mime: ret.Mime,
Md5: md5value,
},
Chunks: []*filer_pb.FileChunk{{
FileId: fileId,
@@ -212,15 +216,20 @@ func (fs *FilerServer) updateFilerStore(ctx context.Context, r *http.Request, w
}
// send request to volume server
func (fs *FilerServer) uploadToVolumeServer(r *http.Request, u *url.URL, auth security.EncodedJwt, w http.ResponseWriter, fileId string) (ret *operation.UploadResult, err error) {
func (fs *FilerServer) uploadToVolumeServer(r *http.Request, u *url.URL, auth security.EncodedJwt, w http.ResponseWriter, fileId string) (ret *operation.UploadResult, md5value []byte, err error) {
stats.FilerRequestCounter.WithLabelValues("postUpload").Inc()
start := time.Now()
defer func() { stats.FilerRequestHistogram.WithLabelValues("postUpload").Observe(time.Since(start).Seconds()) }()
ret = &operation.UploadResult{}
hash := md5.New()
var body = ioutil.NopCloser(io.TeeReader(r.Body, hash))
md5Hash := md5.New()
body := r.Body
if r.Method == "PUT" {
// only PUT or large chunked files has Md5 in attributes
body = ioutil.NopCloser(io.TeeReader(r.Body, md5Hash))
}
request := &http.Request{
Method: r.Method,
@@ -285,7 +294,10 @@ func (fs *FilerServer) uploadToVolumeServer(r *http.Request, u *url.URL, auth se
}
}
// use filer calculated md5 ETag, instead of the volume server crc ETag
ret.ETag = fmt.Sprintf("%x", hash.Sum(nil))
if r.Method == "PUT" {
md5value = md5Hash.Sum(nil)
}
ret.ETag = getEtag(resp)
return
}
@@ -318,7 +330,7 @@ func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNoContent)
}
func (fs *FilerServer) detectCollection(requestURI, qCollection, qReplication string) (collection, replication string) {
func (fs *FilerServer) detectCollection(requestURI, qCollection, qReplication string) (collection, replication string, fsync bool) {
// default
collection = fs.option.Collection
replication = fs.option.DefaultReplication
@@ -341,7 +353,7 @@ func (fs *FilerServer) detectCollection(requestURI, qCollection, qReplication st
if t > 0 {
collection = bucketAndObjectKey[:t]
}
replication = fs.filer.ReadBucketOption(collection)
replication, fsync = fs.filer.ReadBucketOption(collection)
}
return

16
weed/server/filer_server_handlers_write_autochunk.go
View File

@@ -2,7 +2,9 @@ package weed_server
import (
"context"
"crypto/md5"
"io"
"io/ioutil"
"net/http"
"path"
"strconv"
@@ -19,7 +21,7 @@ import (
)
func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request,
replication string, collection string, dataCenter string, ttlSec int32, ttlString string) bool {
replication string, collection string, dataCenter string, ttlSec int32, ttlString string, fsync bool) bool {
if r.Method != "POST" {
glog.V(4).Infoln("AutoChunking not supported for method", r.Method)
return false
@@ -55,7 +57,7 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
return false
}
reply, err := fs.doAutoChunk(ctx, w, r, contentLength, chunkSize, replication, collection, dataCenter, ttlSec, ttlString)
reply, err := fs.doAutoChunk(ctx, w, r, contentLength, chunkSize, replication, collection, dataCenter, ttlSec, ttlString, fsync)
if err != nil {
writeJsonError(w, r, http.StatusInternalServerError, err)
} else if reply != nil {
@@ -65,7 +67,7 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
}
func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request,
contentLength int64, chunkSize int32, replication string, collection string, dataCenter string, ttlSec int32, ttlString string) (filerResult *FilerPostResult, replyerr error) {
contentLength int64, chunkSize int32, replication string, collection string, dataCenter string, ttlSec int32, ttlString string, fsync bool) (filerResult *FilerPostResult, replyerr error) {
stats.FilerRequestCounter.WithLabelValues("postAutoChunk").Inc()
start := time.Now()
@@ -91,13 +93,16 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
var fileChunks []*filer_pb.FileChunk
md5Hash := md5.New()
var partReader = ioutil.NopCloser(io.TeeReader(part1, md5Hash))
chunkOffset := int64(0)
for chunkOffset < contentLength {
limitedReader := io.LimitReader(part1, int64(chunkSize))
limitedReader := io.LimitReader(partReader, int64(chunkSize))
// assign one file id for one chunk
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(w, r, replication, collection, dataCenter, ttlString)
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(w, r, replication, collection, dataCenter, ttlString, fsync)
if assignErr != nil {
return nil, assignErr
}
@@ -157,6 +162,7 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
Collection: collection,
TtlSec: ttlSec,
Mime: contentType,
Md5: md5Hash.Sum(nil),
},
Chunks: fileChunks,
}

4
weed/server/filer_server_handlers_write_cipher.go
View File

@@ -17,9 +17,9 @@ import (
// handling single chunk POST or PUT upload
func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *http.Request,
replication string, collection string, dataCenter string, ttlSeconds int32, ttlString string) (filerResult *FilerPostResult, err error) {
replication string, collection string, dataCenter string, ttlSeconds int32, ttlString string, fsync bool) (filerResult *FilerPostResult, err error) {
fileId, urlLocation, auth, err := fs.assignNewFileInfo(w, r, replication, collection, dataCenter, ttlString)
fileId, urlLocation, auth, err := fs.assignNewFileInfo(w, r, replication, collection, dataCenter, ttlString, fsync)
if err != nil || fileId == "" || urlLocation == "" {
return nil, fmt.Errorf("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, collection, dataCenter)

14
weed/server/master_grpc_server.go
View File

@@ -11,6 +11,7 @@ import (
"google.golang.org/grpc/peer"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/storage/backend"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
@@ -190,6 +191,18 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
peerAddress := findClientAddress(stream.Context(), req.GrpcPort)
// only one shell can be connected at any time
if req.Name == pb.AdminShellClient {
if ms.currentAdminShellClient == "" {
ms.currentAdminShellClient = peerAddress
defer func() {
ms.currentAdminShellClient = ""
}()
} else {
return fmt.Errorf("only one concurrent shell allowed, but another shell is already connected from %s", peerAddress)
}
}
stopChan := make(chan bool)
clientName, messageChan := ms.addClient(req.Name, peerAddress)
@@ -230,7 +243,6 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
}
}
return nil
}
func (ms *MasterServer) informNewLeader(stream master_pb.Seaweed_KeepConnectedServer) error {

11
weed/server/master_server.go
View File

@@ -64,6 +64,8 @@ type MasterServer struct {
grpcDialOption grpc.DialOption
MasterClient *wdclient.MasterClient
currentAdminShellClient string
}
func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *MasterServer {
@@ -197,8 +199,8 @@ func (ms *MasterServer) startAdminScripts() {
v.SetDefault("master.maintenance.sleep_minutes", 17)
sleepMinutes := v.GetInt("master.maintenance.sleep_minutes")
v.SetDefault("master.filer.default_filer_url", "http://localhost:8888/")
filerURL := v.GetString("master.filer.default_filer_url")
v.SetDefault("master.filer.default", "localhost:8888")
filerHostPort := v.GetString("master.filer.default")
scriptLines := strings.Split(adminScripts, "\n")
@@ -208,9 +210,10 @@ func (ms *MasterServer) startAdminScripts() {
shellOptions.GrpcDialOption = security.LoadClientTLS(v, "grpc.master")
shellOptions.Masters = &masterAddress
shellOptions.FilerHost, shellOptions.FilerPort, shellOptions.Directory, err = util.ParseFilerUrl(filerURL)
shellOptions.FilerHost, shellOptions.FilerPort, err = util.ParseHostPort(filerHostPort)
shellOptions.Directory = "/"
if err != nil {
glog.V(0).Infof("failed to parse master.filer.default_filer_urll=%s : %v\n", filerURL, err)
glog.V(0).Infof("failed to parse master.filer.default = %s : %v\n", filerHostPort, err)
return
}

23
weed/server/msg_broker_grpc_server.go
View File

@@ -1,23 +0,0 @@
package weed_server
import (
"context"
"github.com/chrislusf/seaweedfs/weed/pb/queue_pb"
)
func (broker *MessageBroker) ConfigureTopic(context.Context, *queue_pb.ConfigureTopicRequest) (*queue_pb.ConfigureTopicResponse, error) {
panic("implement me")
}
func (broker *MessageBroker) DeleteTopic(context.Context, *queue_pb.DeleteTopicRequest) (*queue_pb.DeleteTopicResponse, error) {
panic("implement me")
}
func (broker *MessageBroker) StreamWrite(queue_pb.SeaweedQueue_StreamWriteServer) error {
panic("implement me")
}
func (broker *MessageBroker) StreamRead(*queue_pb.ReadMessageRequest, queue_pb.SeaweedQueue_StreamReadServer) error {
panic("implement me")
}

121
weed/server/msg_broker_server.go
View File

@@ -1,121 +0,0 @@
package weed_server
import (
"context"
"fmt"
"time"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util"
)
type MessageBrokerOption struct {
Filers []string
DefaultReplication string
MaxMB int
Port int
}
type MessageBroker struct {
option *MessageBrokerOption
grpcDialOption grpc.DialOption
}
func NewMessageBroker(option *MessageBrokerOption) (messageBroker *MessageBroker, err error) {
messageBroker = &MessageBroker{
option: option,
grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.msg_broker"),
}
go messageBroker.loopForEver()
return messageBroker, nil
}
func (broker *MessageBroker) loopForEver() {
for {
broker.checkPeers()
time.Sleep(3 * time.Second)
}
}
func (broker *MessageBroker) checkPeers() {
// contact a filer about masters
var masters []string
for _, filer := range broker.option.Filers {
err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return err
}
masters = append(masters, resp.Masters...)
return nil
})
if err != nil {
fmt.Printf("failed to read masters from %+v: %v\n", broker.option.Filers, err)
return
}
}
// contact each masters for filers
var filers []string
for _, master := range masters {
err := broker.withMasterClient(master, func(client master_pb.SeaweedClient) error {
resp, err := client.ListMasterClients(context.Background(), &master_pb.ListMasterClientsRequest{
ClientType: "filer",
})
if err != nil {
return err
}
fmt.Printf("filers: %+v\n", resp.GrpcAddresses)
filers = append(filers, resp.GrpcAddresses...)
return nil
})
if err != nil {
fmt.Printf("failed to list filers: %v\n", err)
return
}
}
// contact each filer about brokers
for _, filer := range filers {
err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil {
return err
}
masters = append(masters, resp.Masters...)
return nil
})
if err != nil {
fmt.Printf("failed to read masters from %+v: %v\n", broker.option.Filers, err)
return
}
}
}
func (broker *MessageBroker) withFilerClient(filer string, fn func(filer_pb.SeaweedFilerClient) error) error {
return pb.WithFilerClient(filer, broker.grpcDialOption, fn)
}
func (broker *MessageBroker) withMasterClient(master string, fn func(client master_pb.SeaweedClient) error) error {
return pb.WithMasterClient(master, broker.grpcDialOption, func(client master_pb.SeaweedClient) error {
return fn(client)
})
}

2
weed/server/volume_grpc_tail.go
View File

@@ -90,7 +90,7 @@ func (vs *VolumeServer) VolumeTailReceiver(ctx context.Context, req *volume_serv
defer glog.V(1).Infof("receive tailing volume %d finished", v.Id)
return resp, operation.TailVolumeFromSource(req.SourceVolumeServer, vs.grpcDialOption, v.Id, req.SinceNs, int(req.IdleTimeoutSeconds), func(n *needle.Needle) error {
_, err := vs.store.WriteVolumeNeedle(v.Id, n)
_, err := vs.store.WriteVolumeNeedle(v.Id, n, false)
return err
})

8
weed/server/volume_server_handlers_write.go
View File

@@ -166,3 +166,11 @@ func setEtag(w http.ResponseWriter, etag string) {
}
}
}
func getEtag(resp *http.Response) (etag string) {
etag = resp.Header.Get("ETag")
if strings.HasPrefix(etag, "\"") && strings.HasSuffix(etag, "\"") {
return etag[1 : len(etag)-1]
}
return
}

10
weed/server/volume_server_ui/templates.go
View File

@@ -151,9 +151,9 @@ var StatusTpl = template.Must(template.New("status").Funcs(funcMap).Parse(`<!DOC
<tr>
<td><code>{{ .Id }}</code></td>
<td>{{ .Collection }}</td>
<td>{{ .Size }} Bytes</td>
<td>{{ bytesToHumanReadble .Size }}</td>
<td>{{ .FileCount }}</td>
<td>{{ .DeleteCount }} / {{.DeletedByteCount}} Bytes</td>
<td>{{ .DeleteCount }} / {{bytesToHumanReadble .DeletedByteCount}}</td>
<td>{{ .Ttl }}</td>
<td>{{ .ReadOnly }}</td>
</tr>
@@ -181,9 +181,9 @@ var StatusTpl = template.Must(template.New("status").Funcs(funcMap).Parse(`<!DOC
<tr>
<td><code>{{ .Id }}</code></td>
<td>{{ .Collection }}</td>
<td>{{ .Size }} Bytes</td>
<td>{{ bytesToHumanReadble .Size }}</td>
<td>{{ .FileCount }}</td>
<td>{{ .DeleteCount }} / {{.DeletedByteCount}} Bytes</td>
<td>{{ .DeleteCount }} / {{bytesToHumanReadble .DeletedByteCount}}</td>
<td>{{ .RemoteStorageName }}</td>
<td>{{ .RemoteStorageKey }}</td>
</tr>
@@ -209,7 +209,7 @@ var StatusTpl = template.Must(template.New("status").Funcs(funcMap).Parse(`<!DOC
<tr>
<td><code>{{ .VolumeId }}</code></td>
<td>{{ .Collection }}</td>
<td>{{ .ShardSize }} Bytes</td>
<td>{{ bytesToHumanReadble .ShardSize }}</td>
<td>{{ .ShardIdList }}</td>
<td>{{ .CreatedAt.Format "02 Jan 06 15:04 -0700" }}</td>
</tr>

13
weed/server/webdav_server.go
View File

@@ -16,8 +16,8 @@ import (
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/pb_cache"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
@@ -34,6 +34,8 @@ type WebDavOption struct {
Uid uint32
Gid uint32
Cipher bool
CacheDir string
CacheSizeMB int64
}
type WebDavServer struct {
@@ -67,7 +69,7 @@ type WebDavFileSystem struct {
secret security.SigningKey
filer *filer2.Filer
grpcDialOption grpc.DialOption
chunkCache *pb_cache.ChunkCache
chunkCache *chunk_cache.ChunkCache
}
type FileInfo struct {
@@ -96,9 +98,14 @@ type WebDavFile struct {
}
func NewWebDavFileSystem(option *WebDavOption) (webdav.FileSystem, error) {
chunkCache := chunk_cache.NewChunkCache(256, option.CacheDir, option.CacheSizeMB)
util.OnInterrupt(func() {
chunkCache.Shutdown()
})
return &WebDavFileSystem{
option: option,
chunkCache: pb_cache.NewChunkCache(1000),
chunkCache: chunkCache,
}, nil
}