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

merge conflicts

Browse Source
This commit is contained in:
Chris Lu
2015-02-07 15:35:28 -08:00
parent 714ccb6e2b
commit f7998f8652
21 changed files with 271 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
go/weed/benchmark.go
View File

@@ -16,6 +16,7 @@ import (
"github.com/chrislusf/weed-fs/go/glog"
"github.com/chrislusf/weed-fs/go/operation"
"github.com/chrislusf/weed-fs/go/security"
"github.com/chrislusf/weed-fs/go/util"
)
@@ -32,6 +33,7 @@ type BenchmarkOptions struct {
collection *string
cpuprofile *string
maxCpu *int
secretKey *string
vid2server map[string]string //cache for vid locations
}
@@ -56,6 +58,7 @@ func init() {
b.collection = cmdBenchmark.Flag.String("collection", "benchmark", "write data to this collection")
b.cpuprofile = cmdBenchmark.Flag.String("cpuprofile", "", "cpu profile output file")
b.maxCpu = cmdBenchmark.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
b.secretKey = cmdBenchmark.Flag.String("secure.secret", "", "secret to encrypt Json Web Token(JWT)")
b.vid2server = make(map[string]string)
sharedBytes = make([]byte, 1024)
}
@@ -181,6 +184,8 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
defer wait.Done()
delayedDeleteChan := make(chan *delayedFile, 100)
var waitForDeletions sync.WaitGroup
secret := security.Secret(*b.secretKey)
for i := 0; i < 7; i++ {
waitForDeletions.Add(1)
go func() {
@@ -189,7 +194,8 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
if df.enterTime.After(time.Now()) {
time.Sleep(df.enterTime.Sub(time.Now()))
}
if e := util.Delete("http://" + df.fp.Server + "/" + df.fp.Fid); e == nil {
if e := util.Delete("http://"+df.fp.Server+"/"+df.fp.Fid,
security.GenJwt(secret, df.fp.Fid)); e == nil {
s.completed++
} else {
s.failed++
@@ -204,7 +210,7 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stat) {
fp := &operation.FilePart{Reader: &FakeReader{id: uint64(id), size: fileSize}, FileSize: fileSize}
if assignResult, err := operation.Assign(*b.server, 1, "", *b.collection, ""); err == nil {
fp.Server, fp.Fid, fp.Collection = assignResult.Url, assignResult.Fid, *b.collection
if _, err := fp.Upload(0, *b.server); err == nil {
if _, err := fp.Upload(0, *b.server, secret); err == nil {
if rand.Intn(100) < *b.deletePercentage {
s.total++
delayedDeleteChan <- &delayedFile{time.Now().Add(time.Second), fp}

4
go/weed/filer.go
View File

@@ -22,6 +22,7 @@ type FilerOptions struct {
defaultReplicaPlacement *string
dir *string
redirectOnRead *bool
secretKey *string
cassandra_server *string
cassandra_keyspace *string
redis_server *string
@@ -40,6 +41,8 @@ func init() {
f.cassandra_keyspace = cmdFiler.Flag.String("cassandra.keyspace", "seaweed", "keyspace of the cassandra server")
f.redis_server = cmdFiler.Flag.String("redis.server", "", "host:port of the redis server, e.g., 127.0.0.1:6379")
f.redis_database = cmdFiler.Flag.Int("redis.database", 0, "the database on the redis server")
f.secretKey = cmdFiler.Flag.String("secure.secret", "", "secret to encrypt Json Web Token(JWT)")
}
var cmdFiler = &Command{
@@ -73,6 +76,7 @@ func runFiler(cmd *Command, args []string) bool {
r := http.NewServeMux()
_, nfs_err := weed_server.NewFilerServer(r, *f.port, *f.master, *f.dir, *f.collection,
*f.defaultReplicaPlacement, *f.redirectOnRead,
*f.secretKey,
*f.cassandra_server, *f.cassandra_keyspace,
*f.redis_server, *f.redis_database,
)

2
go/weed/master.go
View File

@@ -41,7 +41,7 @@ var (
mMaxCpu = cmdMaster.Flag.Int("maxCpu", 0, "maximum number of CPUs. 0 means all available CPUs")
garbageThreshold = cmdMaster.Flag.String("garbageThreshold", "0.3", "threshold to vacuum and reclaim spaces")
masterWhiteListOption = cmdMaster.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
masterSecureKey = cmdMaster.Flag.String("secure.key", "", "secret key to check permission")
masterSecureKey = cmdMaster.Flag.String("secure.secret", "", "secret to encrypt Json Web Token(JWT)")
masterWhiteList []string
)

5
go/weed/server.go
View File

@@ -56,7 +56,7 @@ var (
serverRack = cmdServer.Flag.String("rack", "", "current volume server's rack name")
serverWhiteListOption = cmdServer.Flag.String("whiteList", "", "comma separated Ip addresses having write permission. No limit if empty.")
serverPeers = cmdServer.Flag.String("master.peers", "", "other master nodes in comma separated ip:masterPort list")
serverSecureKey = cmdServer.Flag.String("secure.key", "", "secret key to ensure authenticated access")
serverSecureKey = cmdServer.Flag.String("secure.secret", "", "secret to encrypt Json Web Token(JWT)")
serverGarbageThreshold = cmdServer.Flag.String("garbageThreshold", "0.3", "threshold to vacuum and reclaim spaces")
masterPort = cmdServer.Flag.Int("master.port", 9333, "master server http listen port")
masterMetaFolder = cmdServer.Flag.String("master.dir", "", "data directory to store meta data, default to same as -dir specified")
@@ -86,10 +86,10 @@ func init() {
filerOptions.cassandra_keyspace = cmdServer.Flag.String("filer.cassandra.keyspace", "seaweed", "keyspace of the cassandra server")
filerOptions.redis_server = cmdServer.Flag.String("filer.redis.server", "", "host:port of the redis server, e.g., 127.0.0.1:6379")
filerOptions.redis_database = cmdServer.Flag.Int("filer.redis.database", 0, "the database on the redis server")
}
func runServer(cmd *Command, args []string) bool {
filerOptions.secretKey = serverSecureKey
if *serverOptions.cpuprofile != "" {
f, err := os.Create(*serverOptions.cpuprofile)
if err != nil {
@@ -162,6 +162,7 @@ func runServer(cmd *Command, args []string) bool {
r := http.NewServeMux()
_, nfs_err := weed_server.NewFilerServer(r, *filerOptions.port, *filerOptions.master, *filerOptions.dir, *filerOptions.collection,
*filerOptions.defaultReplicaPlacement, *filerOptions.redirectOnRead,
*filerOptions.secretKey,
"", "",
"", 0,
)

22
go/weed/upload.go
View File

@@ -7,6 +7,7 @@ import (
"path/filepath"
"github.com/chrislusf/weed-fs/go/operation"
"github.com/chrislusf/weed-fs/go/security"
)
var (
@@ -15,6 +16,7 @@ var (
uploadDir *string
uploadTtl *string
include *string
uploadSecretKey *string
maxMB *int
)
@@ -28,13 +30,14 @@ func init() {
uploadCollection = cmdUpload.Flag.String("collection", "", "optional collection name")
uploadTtl = cmdUpload.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
maxMB = cmdUpload.Flag.Int("maxMB", 0, "split files larger than the limit")
uploadSecretKey = cmdUpload.Flag.String("secure.secret", "", "secret to encrypt Json Web Token(JWT)")
}
var cmdUpload = &Command{
UsageLine: "upload -server=localhost:9333 file1 [file2 file3]\n upload -server=localhost:9333 -dir=one_directory -include=*.pdf",
Short: "upload one or a list of files",
Long: `upload one or a list of files, or batch upload one whole folder recursively.
If uploading a list of files:
It uses consecutive file keys for the list of files.
e.g. If the file1 uses key k, file2 can be read via k_1
@@ -42,18 +45,19 @@ var cmdUpload = &Command{
If uploading a whole folder recursively:
All files under the folder and subfolders will be uploaded, each with its own file key.
Optional parameter "-include" allows you to specify the file name patterns.
If any file has a ".gz" extension, the content are considered gzipped already, and will be stored as is.
This can save volume server's gzipped processing and allow customizable gzip compression level.
The file name will strip out ".gz" and stored. For example, "jquery.js.gz" will be stored as "jquery.js".
If "maxMB" is set to a positive number, files larger than it would be split into chunks and uploaded separatedly.
The list of file ids of those chunks would be stored in an additional chunk, and this additional chunk's file id would be returned.
If "maxMB" is set to a positive number, files larger than it would be split into chunks and uploaded separatedly.
The list of file ids of those chunks would be stored in an additional chunk, and this additional chunk's file id would be returned.
`,
}
func runUpload(cmd *Command, args []string) bool {
secret := security.Secret(*uploadSecretKey)
if len(cmdUpload.Flag.Args()) == 0 {
if *uploadDir == "" {
return false
@@ -70,7 +74,9 @@ func runUpload(cmd *Command, args []string) bool {
if e != nil {
return e
}
results, e := operation.SubmitFiles(*server, parts, *uploadReplication, *uploadCollection, *uploadTtl, *maxMB)
results, e := operation.SubmitFiles(*server, parts,
*uploadReplication, *uploadCollection,
*uploadTtl, *maxMB, secret)
bytes, _ := json.Marshal(results)
fmt.Println(string(bytes))
if e != nil {
@@ -87,7 +93,9 @@ func runUpload(cmd *Command, args []string) bool {
if e != nil {
fmt.Println(e.Error())
}
results, _ := operation.SubmitFiles(*server, parts, *uploadReplication, *uploadCollection, *uploadTtl, *maxMB)
results, _ := operation.SubmitFiles(*server, parts,
*uploadReplication, *uploadCollection,
*uploadTtl, *maxMB, secret)
bytes, _ := json.Marshal(results)
fmt.Println(string(bytes))
}

4
go/weed/weed_server/common.go
View File

@@ -12,6 +12,7 @@ import (
"github.com/chrislusf/weed-fs/go/glog"
"github.com/chrislusf/weed-fs/go/operation"
"github.com/chrislusf/weed-fs/go/security"
"github.com/chrislusf/weed-fs/go/stats"
"github.com/chrislusf/weed-fs/go/storage"
"github.com/chrislusf/weed-fs/go/util"
@@ -75,6 +76,7 @@ func debug(params ...interface{}) {
}
func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl string) {
jwt := security.GetJwt(r)
m := make(map[string]interface{})
if r.Method != "POST" {
writeJsonError(w, r, http.StatusMethodNotAllowed, errors.New("Only submit via POST!"))
@@ -102,7 +104,7 @@ func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterUrl st
}
debug("upload file to store", url)
uploadResult, err := operation.Upload(url, fname, bytes.NewReader(data), isGzipped, mimeType)
uploadResult, err := operation.Upload(url, fname, bytes.NewReader(data), isGzipped, mimeType, jwt)
if err != nil {
writeJsonError(w, r, http.StatusInternalServerError, err)
return

7
go/weed/weed_server/filer_server.go
View File

@@ -10,6 +10,7 @@ import (
"github.com/chrislusf/weed-fs/go/filer/flat_namespace"
"github.com/chrislusf/weed-fs/go/filer/redis_store"
"github.com/chrislusf/weed-fs/go/glog"
"github.com/chrislusf/weed-fs/go/security"
)
type FilerServer struct {
@@ -18,11 +19,13 @@ type FilerServer struct {
collection string
defaultReplication string
redirectOnRead bool
secret security.Secret
filer filer.Filer
}
func NewFilerServer(r *http.ServeMux, port int, master string, dir string, collection string,
replication string, redirectOnRead bool,
secret string,
cassandra_server string, cassandra_keyspace string,
redis_server string, redis_database int,
) (fs *FilerServer, err error) {
@@ -56,3 +59,7 @@ func NewFilerServer(r *http.ServeMux, port int, master string, dir string, colle
return fs, nil
}
func (fs *FilerServer) jwt(fileId string) security.EncodedJwt {
return security.GenJwt(fs.secret, fileId)
}

6
go/weed/weed_server/filer_server_handlers.go
View File

@@ -170,7 +170,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
if ret.Name != "" {
path += ret.Name
} else {
operation.DeleteFile(fs.master, assignResult.Fid) //clean up
operation.DeleteFile(fs.master, assignResult.Fid, fs.jwt(assignResult.Fid)) //clean up
glog.V(0).Infoln("Can not to write to folder", path, "without a file name!")
writeJsonError(w, r, http.StatusInternalServerError,
errors.New("Can not to write to folder "+path+" without a file name"))
@@ -179,7 +179,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
}
glog.V(4).Infoln("saving", path, "=>", assignResult.Fid)
if db_err := fs.filer.CreateFile(path, assignResult.Fid); db_err != nil {
operation.DeleteFile(fs.master, assignResult.Fid) //clean up
operation.DeleteFile(fs.master, assignResult.Fid, fs.jwt(assignResult.Fid)) //clean up
glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err)
writeJsonError(w, r, http.StatusInternalServerError, db_err)
return
@@ -199,7 +199,7 @@ func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
} else {
fid, err = fs.filer.DeleteFile(r.URL.Path)
if err == nil {
err = operation.DeleteFile(fs.master, fid)
err = operation.DeleteFile(fs.master, fid, fs.jwt(fid))
}
}
if err == nil {

29
go/weed/weed_server/master_server.go
View File

@@ -23,6 +23,7 @@ type MasterServer struct {
pulseSeconds int
defaultReplicaPlacement string
garbageThreshold string
guard *security.Guard
Topo *topology.Topology
vg *topology.VolumeGrowth
@@ -57,22 +58,22 @@ func NewMasterServer(r *mux.Router, port int, metaFolder string,
ms.vg = topology.NewDefaultVolumeGrowth()
glog.V(0).Infoln("Volume Size Limit is", volumeSizeLimitMB, "MB")
guard := security.NewGuard(whiteList, secureKey)
ms.guard = security.NewGuard(whiteList, secureKey)
r.HandleFunc("/dir/assign", ms.proxyToLeader(guard.Secure(ms.dirAssignHandler)))
r.HandleFunc("/dir/lookup", ms.proxyToLeader(guard.Secure(ms.dirLookupHandler)))
r.HandleFunc("/dir/join", ms.proxyToLeader(guard.Secure(ms.dirJoinHandler)))
r.HandleFunc("/dir/status", ms.proxyToLeader(guard.Secure(ms.dirStatusHandler)))
r.HandleFunc("/col/delete", ms.proxyToLeader(guard.Secure(ms.collectionDeleteHandler)))
r.HandleFunc("/vol/lookup", ms.proxyToLeader(guard.Secure(ms.volumeLookupHandler)))
r.HandleFunc("/vol/grow", ms.proxyToLeader(guard.Secure(ms.volumeGrowHandler)))
r.HandleFunc("/vol/status", ms.proxyToLeader(guard.Secure(ms.volumeStatusHandler)))
r.HandleFunc("/vol/vacuum", ms.proxyToLeader(guard.Secure(ms.volumeVacuumHandler)))
r.HandleFunc("/submit", guard.Secure(ms.submitFromMasterServerHandler))
r.HandleFunc("/delete", guard.Secure(ms.deleteFromMasterServerHandler))
r.HandleFunc("/dir/assign", ms.proxyToLeader(ms.guard.WhiteList(ms.dirAssignHandler)))
r.HandleFunc("/dir/lookup", ms.proxyToLeader(ms.guard.WhiteList(ms.dirLookupHandler)))
r.HandleFunc("/dir/join", ms.proxyToLeader(ms.guard.WhiteList(ms.dirJoinHandler)))
r.HandleFunc("/dir/status", ms.proxyToLeader(ms.guard.WhiteList(ms.dirStatusHandler)))
r.HandleFunc("/col/delete", ms.proxyToLeader(ms.guard.WhiteList(ms.collectionDeleteHandler)))
r.HandleFunc("/vol/lookup", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeLookupHandler)))
r.HandleFunc("/vol/grow", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeGrowHandler)))
r.HandleFunc("/vol/status", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeStatusHandler)))
r.HandleFunc("/vol/vacuum", ms.proxyToLeader(ms.guard.WhiteList(ms.volumeVacuumHandler)))
r.HandleFunc("/submit", ms.guard.WhiteList(ms.submitFromMasterServerHandler))
r.HandleFunc("/delete", ms.guard.WhiteList(ms.deleteFromMasterServerHandler))
r.HandleFunc("/{fileId}", ms.redirectHandler)
r.HandleFunc("/stats/counter", guard.Secure(statsCounterHandler))
r.HandleFunc("/stats/memory", guard.Secure(statsMemoryHandler))
r.HandleFunc("/stats/counter", ms.guard.WhiteList(statsCounterHandler))
r.HandleFunc("/stats/memory", ms.guard.WhiteList(statsMemoryHandler))
ms.Topo.StartRefreshWritableVolumes(garbageThreshold)

5
go/weed/weed_server/master_server_handlers_admin.go
View File

@@ -58,7 +58,10 @@ func (ms *MasterServer) dirJoinHandler(w http.ResponseWriter, r *http.Request) {
}
ms.Topo.ProcessJoinMessage(joinMessage)
writeJsonQuiet(w, r, http.StatusOK, operation.JoinResult{VolumeSizeLimit: uint64(ms.volumeSizeLimitMB) * 1024 * 1024})
writeJsonQuiet(w, r, http.StatusOK, operation.JoinResult{
VolumeSizeLimit: uint64(ms.volumeSizeLimitMB) * 1024 * 1024,
SecretKey: string(ms.guard.SecretKey),
})
}
func (ms *MasterServer) dirStatusHandler(w http.ResponseWriter, r *http.Request) {

29
go/weed/weed_server/volume_server.go
View File

@@ -41,16 +41,16 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string,
vs.guard = security.NewGuard(whiteList, "")
adminMux.HandleFunc("/status", vs.guard.Secure(vs.statusHandler))
adminMux.HandleFunc("/admin/assign_volume", vs.guard.Secure(vs.assignVolumeHandler))
adminMux.HandleFunc("/admin/vacuum_volume_check", vs.guard.Secure(vs.vacuumVolumeCheckHandler))
adminMux.HandleFunc("/admin/vacuum_volume_compact", vs.guard.Secure(vs.vacuumVolumeCompactHandler))
adminMux.HandleFunc("/admin/vacuum_volume_commit", vs.guard.Secure(vs.vacuumVolumeCommitHandler))
adminMux.HandleFunc("/admin/freeze_volume", vs.guard.Secure(vs.freezeVolumeHandler))
adminMux.HandleFunc("/admin/delete_collection", vs.guard.Secure(vs.deleteCollectionHandler))
adminMux.HandleFunc("/stats/counter", vs.guard.Secure(statsCounterHandler))
adminMux.HandleFunc("/stats/memory", vs.guard.Secure(statsMemoryHandler))
adminMux.HandleFunc("/stats/disk", vs.guard.Secure(vs.statsDiskHandler))
adminMux.HandleFunc("/status", vs.guard.WhiteList(vs.statusHandler))
adminMux.HandleFunc("/admin/assign_volume", vs.guard.WhiteList(vs.assignVolumeHandler))
adminMux.HandleFunc("/admin/vacuum_volume_check", vs.guard.WhiteList(vs.vacuumVolumeCheckHandler))
adminMux.HandleFunc("/admin/vacuum_volume_compact", vs.guard.WhiteList(vs.vacuumVolumeCompactHandler))
adminMux.HandleFunc("/admin/vacuum_volume_commit", vs.guard.WhiteList(vs.vacuumVolumeCommitHandler))
adminMux.HandleFunc("/admin/freeze_volume", vs.guard.WhiteList(vs.freezeVolumeHandler))
adminMux.HandleFunc("/admin/delete_collection", vs.guard.WhiteList(vs.deleteCollectionHandler))
adminMux.HandleFunc("/stats/counter", vs.guard.WhiteList(statsCounterHandler))
adminMux.HandleFunc("/stats/memory", vs.guard.WhiteList(statsMemoryHandler))
adminMux.HandleFunc("/stats/disk", vs.guard.WhiteList(vs.statsDiskHandler))
publicMux.HandleFunc("/delete", vs.guard.Secure(vs.batchDeleteHandler))
publicMux.HandleFunc("/", vs.storeHandler)
@@ -61,12 +61,13 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string,
vs.store.SetDataCenter(vs.dataCenter)
vs.store.SetRack(vs.rack)
for {
master, err := vs.store.Join()
master, secretKey, err := vs.store.Join()
if err == nil {
if !connected {
connected = true
vs.SetMasterNode(master)
glog.V(0).Infoln("Volume Server Connected with master at", master, "and set it as masterNode")
vs.guard.SecretKey = secretKey
glog.V(0).Infoln("Volume Server Connected with master at", master)
}
} else {
glog.V(4).Infoln("Volume Server Failed to talk with master:", err.Error())
@@ -102,3 +103,7 @@ func (vs *VolumeServer) Shutdown() {
vs.store.Close()
glog.V(0).Infoln("Shut down successfully!")
}
func (vs *VolumeServer) jwt(fileId string) security.EncodedJwt {
return security.GenJwt(vs.guard.SecretKey, fileId)
}

3
go/weed/weed_server/volume_server_handlers.go
View File

@@ -253,7 +253,8 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) {
}
ret := operation.UploadResult{}
size, errorStatus := topology.ReplicatedWrite(vs.GetMasterNode(), vs.store, volumeId, needle, r)
size, errorStatus := topology.ReplicatedWrite(vs.GetMasterNode(),
vs.store, volumeId, needle, r)
httpStatus := http.StatusCreated
if errorStatus != "" {
httpStatus = http.StatusInternalServerError