From f511b507a5230e931d0250b446f744073be879b3 Mon Sep 17 00:00:00 2001 From: chrislusf Date: Mon, 9 Mar 2015 01:10:01 -0700 Subject: [PATCH] Add read only public port on volume server Add read only public port on volume server --- go/operation/assign_file_id.go | 3 +- go/operation/submit.go | 2 +- go/storage/store.go | 6 ++-- go/topology/allocate_volume.go | 2 +- go/topology/data_node.go | 6 ---- go/topology/rack.go | 3 +- go/topology/topology.go | 6 +--- go/topology/topology_vacuum.go | 12 +++---- go/weed/server.go | 35 ++++++++++++++---- go/weed/volume.go | 36 +++++++++---------- go/weed/weed_server/volume_server.go | 9 +++-- go/weed/weed_server/volume_server_handlers.go | 25 ++++--------- 12 files changed, 71 insertions(+), 74 deletions(-) diff --git a/go/operation/assign_file_id.go b/go/operation/assign_file_id.go index 672bfa99c..8b6dfe5ba 100644 --- a/go/operation/assign_file_id.go +++ b/go/operation/assign_file_id.go @@ -3,6 +3,7 @@ package operation import ( "encoding/json" "errors" + "fmt" "net/url" "strconv" @@ -38,7 +39,7 @@ func Assign(server string, count int, replication string, collection string, ttl var ret AssignResult err = json.Unmarshal(jsonBlob, &ret) if err != nil { - return nil, err + return nil, fmt.Errorf("JSON unmarshal error:%v, json:%s", err, string(jsonBlob)) } if ret.Count <= 0 { return nil, errors.New(ret.Error) diff --git a/go/operation/submit.go b/go/operation/submit.go index 03551b1e8..16304d2bd 100644 --- a/go/operation/submit.go +++ b/go/operation/submit.go @@ -55,7 +55,7 @@ func SubmitFiles(master string, files []FilePart, if index > 0 { file.Fid = file.Fid + "_" + strconv.Itoa(index) } - file.Server = ret.PublicUrl + file.Server = ret.Url file.Replication = replication file.Collection = collection results[index].Size, err = file.Upload(maxMB, master, secret) diff --git a/go/storage/store.go b/go/storage/store.go index d280175f2..fa618b44b 100644 --- a/go/storage/store.go +++ b/go/storage/store.go @@ -76,7 +76,6 @@ func (mn *MasterNodes) findMaster() (string, error) { type Store struct { Ip string Port int - AdminPort int PublicUrl string Locations []*DiskLocation dataCenter string //optional informaton, overwriting master setting if exists @@ -91,8 +90,8 @@ func (s *Store) String() (str string) { return } -func NewStore(port, adminPort int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int) (s *Store) { - s = &Store{Port: port, AdminPort: adminPort, Ip: ip, PublicUrl: publicUrl} +func NewStore(port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int) (s *Store) { + s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl} s.Locations = make([]*DiskLocation, 0) for i := 0; i < len(dirnames); i++ { location := &DiskLocation{Directory: dirnames[i], MaxVolumeCount: maxVolumeCounts[i]} @@ -310,7 +309,6 @@ func (s *Store) Join() (masterNode string, secretKey security.Secret, e error) { DataCenter: proto.String(s.dataCenter), Rack: proto.String(s.rack), Volumes: volumeMessages, - AdminPort: proto.Uint32(uint32(s.AdminPort)), } data, err := proto.Marshal(joinMessage) diff --git a/go/topology/allocate_volume.go b/go/topology/allocate_volume.go index 22730211d..be478d0cf 100644 --- a/go/topology/allocate_volume.go +++ b/go/topology/allocate_volume.go @@ -20,7 +20,7 @@ func AllocateVolume(dn *DataNode, vid storage.VolumeId, option *VolumeGrowOption values.Add("collection", option.Collection) values.Add("replication", option.ReplicaPlacement.String()) values.Add("ttl", option.Ttl.String()) - jsonBlob, err := util.Post("http://"+dn.AdminUrl()+"/admin/assign_volume", values) + jsonBlob, err := util.Post("http://"+dn.Url()+"/admin/assign_volume", values) if err != nil { return err } diff --git a/go/topology/data_node.go b/go/topology/data_node.go index e0d12cbcc..27461945b 100644 --- a/go/topology/data_node.go +++ b/go/topology/data_node.go @@ -13,7 +13,6 @@ type DataNode struct { volumes map[storage.VolumeId]storage.VolumeInfo Ip string Port int - AdminPort int PublicUrl string LastSeen int64 // unix time in seconds Dead bool @@ -90,10 +89,6 @@ func (dn *DataNode) Url() string { return dn.Ip + ":" + strconv.Itoa(dn.Port) } -func (dn *DataNode) AdminUrl() string { - return dn.Ip + ":" + strconv.Itoa(dn.AdminPort) -} - func (dn *DataNode) ToMap() interface{} { ret := make(map[string]interface{}) ret["Url"] = dn.Url() @@ -101,6 +96,5 @@ func (dn *DataNode) ToMap() interface{} { ret["Max"] = dn.GetMaxVolumeCount() ret["Free"] = dn.FreeSpace() ret["PublicUrl"] = dn.PublicUrl - ret["AdminUrl"] = dn.AdminUrl() return ret } diff --git a/go/topology/rack.go b/go/topology/rack.go index 50ad5f009..40e19dd0d 100644 --- a/go/topology/rack.go +++ b/go/topology/rack.go @@ -27,7 +27,7 @@ func (r *Rack) FindDataNode(ip string, port int) *DataNode { } return nil } -func (r *Rack) GetOrCreateDataNode(ip string, port, adminPort int, publicUrl string, maxVolumeCount int) *DataNode { +func (r *Rack) GetOrCreateDataNode(ip string, port int, publicUrl string, maxVolumeCount int) *DataNode { for _, c := range r.Children() { dn := c.(*DataNode) if dn.MatchLocation(ip, port) { @@ -43,7 +43,6 @@ func (r *Rack) GetOrCreateDataNode(ip string, port, adminPort int, publicUrl str dn := NewDataNode(ip + ":" + strconv.Itoa(port)) dn.Ip = ip dn.Port = port - dn.AdminPort = adminPort dn.PublicUrl = publicUrl dn.maxVolumeCount = maxVolumeCount dn.LastSeen = time.Now().Unix() diff --git a/go/topology/topology.go b/go/topology/topology.go index e06df6227..b64ff420a 100644 --- a/go/topology/topology.go +++ b/go/topology/topology.go @@ -157,12 +157,8 @@ func (t *Topology) ProcessJoinMessage(joinMessage *operation.JoinMessage) { if *joinMessage.IsInit && dn != nil { t.UnRegisterDataNode(dn) } - adminPort := *joinMessage.Port - if joinMessage.AdminPort != nil { - adminPort = *joinMessage.AdminPort - } dn = rack.GetOrCreateDataNode(*joinMessage.Ip, - int(*joinMessage.Port), int(adminPort), *joinMessage.PublicUrl, + int(*joinMessage.Port), *joinMessage.PublicUrl, int(*joinMessage.MaxVolumeCount)) var volumeInfos []storage.VolumeInfo for _, v := range joinMessage.Volumes { diff --git a/go/topology/topology_vacuum.go b/go/topology/topology_vacuum.go index 0aeb9f132..d6fa2213e 100644 --- a/go/topology/topology_vacuum.go +++ b/go/topology/topology_vacuum.go @@ -23,7 +23,7 @@ func batchVacuumVolumeCheck(vl *VolumeLayout, vid storage.VolumeId, locationlist //glog.V(0).Infoln(index, "Checked vacuuming", vid, "on", url, "needVacuum", ret) ch <- ret } - }(index, dn.AdminUrl(), vid) + }(index, dn.Url(), vid) } isCheckSuccess := true for _ = range locationlist.list { @@ -50,7 +50,7 @@ func batchVacuumVolumeCompact(vl *VolumeLayout, vid storage.VolumeId, locationli glog.V(0).Infoln(index, "Complete vacuuming", vid, "on", url) ch <- true } - }(index, dn.AdminUrl(), vid) + }(index, dn.Url(), vid) } isVacuumSuccess := true for _ = range locationlist.list { @@ -66,12 +66,12 @@ func batchVacuumVolumeCompact(vl *VolumeLayout, vid storage.VolumeId, locationli func batchVacuumVolumeCommit(vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList) bool { isCommitSuccess := true for _, dn := range locationlist.list { - glog.V(0).Infoln("Start Commiting vacuum", vid, "on", dn.AdminUrl()) - if e := vacuumVolume_Commit(dn.AdminUrl(), vid); e != nil { - glog.V(0).Infoln("Error when committing vacuum", vid, "on", dn.AdminUrl(), e) + glog.V(0).Infoln("Start Commiting vacuum", vid, "on", dn.Url()) + if e := vacuumVolume_Commit(dn.Url(), vid); e != nil { + glog.V(0).Infoln("Error when committing vacuum", vid, "on", dn.Url(), e) isCommitSuccess = false } else { - glog.V(0).Infoln("Complete Commiting vacuum", vid, "on", dn.AdminUrl()) + glog.V(0).Infoln("Complete Commiting vacuum", vid, "on", dn.Url()) } if isCommitSuccess { vl.SetVolumeAvailable(dn, vid) diff --git a/go/weed/server.go b/go/weed/server.go index dd05d035a..dd0355d55 100644 --- a/go/weed/server.go +++ b/go/weed/server.go @@ -64,7 +64,7 @@ var ( masterConfFile = cmdServer.Flag.String("master.conf", "/etc/weedfs/weedfs.conf", "xml configuration file") masterDefaultReplicaPlacement = cmdServer.Flag.String("master.defaultReplicaPlacement", "000", "Default replication type if not specified.") volumePort = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port") - volumeAdminPort = cmdServer.Flag.Int("volume.port.admin", 0, "volume server admin port to talk with master and other volume servers") + volumePublicPort = cmdServer.Flag.Int("volume.port.public", 0, "volume server public port") volumeDataFolders = cmdServer.Flag.String("dir", os.TempDir(), "directories to store data files. dir[,dir]...") volumeMaxDataVolumeCounts = cmdServer.Flag.String("volume.max", "7", "maximum numbers of volumes, count[,count]...") volumePulse = cmdServer.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats") @@ -109,8 +109,8 @@ func runServer(cmd *Command, args []string) bool { *filerOptions.defaultReplicaPlacement = *masterDefaultReplicaPlacement } - if *volumeAdminPort == 0 { - *volumeAdminPort = *volumePort + if *volumePublicPort == 0 { + *volumePublicPort = *volumePort } if *serverMaxCpu < 1 { @@ -223,9 +223,17 @@ func runServer(cmd *Command, args []string) bool { volumeWait.Wait() time.Sleep(100 * time.Millisecond) - r := http.NewServeMux() - volumeServer := weed_server.NewVolumeServer(r, r, - *serverIp, *volumePort, *volumeAdminPort, *serverPublicUrl, + if *volumePublicPort == 0 { + *volumePublicPort = *volumePort + } + isSeperatedPublicPort := *volumePublicPort != *volumePort + volumeMux := http.NewServeMux() + publicVolumeMux := volumeMux + if isSeperatedPublicPort { + publicVolumeMux = http.NewServeMux() + } + volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux, + *serverIp, *volumePort, *serverPublicUrl, folders, maxCounts, *serverIp+":"+strconv.Itoa(*masterPort), *volumePulse, *serverDataCenter, *serverRack, serverWhiteList, *volumeFixJpgOrientation, @@ -239,13 +247,26 @@ func runServer(cmd *Command, args []string) bool { if eListen != nil { glog.Fatalf("Volume server listener error: %v", eListen) } + if isSeperatedPublicPort { + publicListeningAddress := *serverIp + ":" + strconv.Itoa(*volumePublicPort) + glog.V(0).Infoln("Start Seaweed volume server", util.VERSION, "public at", publicListeningAddress) + publicListener, e := util.NewListener(publicListeningAddress, time.Duration(*serverTimeout)*time.Second) + if e != nil { + glog.Fatalf("Volume server listener error:%v", e) + } + go func() { + if e := http.Serve(publicListener, publicVolumeMux); e != nil { + glog.Fatalf("Volume server fail to serve public: %v", e) + } + }() + } OnInterrupt(func() { volumeServer.Shutdown() pprof.StopCPUProfile() }) - if e := http.Serve(volumeListener, r); e != nil { + if e := http.Serve(volumeListener, volumeMux); e != nil { glog.Fatalf("Volume server fail to serve:%v", e) } diff --git a/go/weed/volume.go b/go/weed/volume.go index f346d4b1c..aa2643d20 100644 --- a/go/weed/volume.go +++ b/go/weed/volume.go @@ -19,7 +19,7 @@ var ( type VolumeServerOptions struct { port *int - adminPort *int + publicPort *int folders []string folderMaxLimits []int ip *string @@ -38,7 +38,7 @@ type VolumeServerOptions struct { func init() { cmdVolume.Run = runVolume // break init cycle v.port = cmdVolume.Flag.Int("port", 8080, "http listen port") - v.adminPort = cmdVolume.Flag.Int("port.admin", 0, "admin port to talk with master and other volume servers") + v.publicPort = cmdVolume.Flag.Int("port.public", 0, "port opened to public") v.ip = cmdVolume.Flag.String("ip", "", "ip or server name") v.publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible address") v.bindIp = cmdVolume.Flag.String("ip.bind", "0.0.0.0", "ip address to bind to") @@ -102,19 +102,19 @@ func runVolume(cmd *Command, args []string) bool { *v.publicUrl = *v.ip + ":" + strconv.Itoa(*v.port) } - if *v.adminPort == 0 { - *v.adminPort = *v.port + if *v.publicPort == 0 { + *v.publicPort = *v.port } - isSeperatedAdminPort := *v.adminPort != *v.port + isSeperatedPublicPort := *v.publicPort != *v.port - publicMux := http.NewServeMux() - adminMux := publicMux - if isSeperatedAdminPort { - adminMux = http.NewServeMux() + volumeMux := http.NewServeMux() + publicVolumeMux := volumeMux + if isSeperatedPublicPort { + publicVolumeMux = http.NewServeMux() } - volumeServer := weed_server.NewVolumeServer(publicMux, adminMux, - *v.ip, *v.port, *v.adminPort, *v.publicUrl, + volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux, + *v.ip, *v.port, *v.publicUrl, v.folders, v.folderMaxLimits, *v.master, *v.pulseSeconds, *v.dataCenter, *v.rack, v.whiteList, @@ -127,16 +127,16 @@ func runVolume(cmd *Command, args []string) bool { if e != nil { glog.Fatalf("Volume server listener error:%v", e) } - if isSeperatedAdminPort { - adminListeningAddress := *v.bindIp + ":" + strconv.Itoa(*v.adminPort) - glog.V(0).Infoln("Start Seaweed volume server", util.VERSION, "admin at", adminListeningAddress) - adminListener, e := util.NewListener(adminListeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second) + if isSeperatedPublicPort { + publicListeningAddress := *v.bindIp + ":" + strconv.Itoa(*v.publicPort) + glog.V(0).Infoln("Start Seaweed volume server", util.VERSION, "public at", publicListeningAddress) + publicListener, e := util.NewListener(publicListeningAddress, time.Duration(*v.idleConnectionTimeout)*time.Second) if e != nil { glog.Fatalf("Volume server listener error:%v", e) } go func() { - if e := http.Serve(adminListener, adminMux); e != nil { - glog.Fatalf("Volume server fail to serve admin: %v", e) + if e := http.Serve(publicListener, publicVolumeMux); e != nil { + glog.Fatalf("Volume server fail to serve public: %v", e) } }() } @@ -145,7 +145,7 @@ func runVolume(cmd *Command, args []string) bool { volumeServer.Shutdown() }) - if e := http.Serve(listener, publicMux); e != nil { + if e := http.Serve(listener, volumeMux); e != nil { glog.Fatalf("Volume server fail to serve: %v", e) } return true diff --git a/go/weed/weed_server/volume_server.go b/go/weed/weed_server/volume_server.go index 9fb7fac99..0bcbc693c 100644 --- a/go/weed/weed_server/volume_server.go +++ b/go/weed/weed_server/volume_server.go @@ -23,8 +23,8 @@ type VolumeServer struct { FixJpgOrientation bool } -func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string, - port, adminPort int, publicUrl string, +func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, + port int, publicUrl string, folders []string, maxCounts []int, masterNode string, pulseSeconds int, dataCenter string, rack string, @@ -37,7 +37,7 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string, FixJpgOrientation: fixJpgOrientation, } vs.SetMasterNode(masterNode) - vs.store = storage.NewStore(port, adminPort, ip, publicUrl, folders, maxCounts) + vs.store = storage.NewStore(port, ip, publicUrl, folders, maxCounts) vs.guard = security.NewGuard(whiteList, "") @@ -56,8 +56,7 @@ func NewVolumeServer(publicMux, adminMux *http.ServeMux, ip string, adminMux.HandleFunc("/delete", vs.guard.WhiteList(vs.batchDeleteHandler)) adminMux.HandleFunc("/", vs.privateStoreHandler) } - publicMux.HandleFunc("/delete", vs.guard.Secure(vs.batchDeleteHandler)) - publicMux.HandleFunc("/", vs.publicStoreHandler) + publicMux.HandleFunc("/", vs.publicReadOnlyHandler) go func() { connected := true diff --git a/go/weed/weed_server/volume_server_handlers.go b/go/weed/weed_server/volume_server_handlers.go index a9179b284..8bd5934ae 100644 --- a/go/weed/weed_server/volume_server_handlers.go +++ b/go/weed/weed_server/volume_server_handlers.go @@ -8,19 +8,17 @@ import ( /* -Public port supports reads. Writes on public port can have one of the 3 +If volume server is started with a separated public port, the public port will +be more "secure". + +Public port currently only supports reads. + +Later writes on public port can have one of the 3 security settings: 1. not secured 2. secured by white list 3. secured by JWT(Json Web Token) -If volume server is started with a separated admin port, the admin port will -have less "security" for easier implementation. -Admin port always supports reads. Writes on admin port can have one of -the 2 security settings: -1. not secured -2. secured by white list - */ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Request) { @@ -43,7 +41,7 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque } } -func (vs *VolumeServer) publicStoreHandler(w http.ResponseWriter, r *http.Request) { +func (vs *VolumeServer) publicReadOnlyHandler(w http.ResponseWriter, r *http.Request) { switch r.Method { case "GET": stats.ReadRequest() @@ -51,14 +49,5 @@ func (vs *VolumeServer) publicStoreHandler(w http.ResponseWriter, r *http.Reques case "HEAD": stats.ReadRequest() vs.GetOrHeadHandler(w, r) - case "DELETE": - stats.DeleteRequest() - vs.guard.Secure(vs.DeleteHandler)(w, r) - case "PUT": - stats.WriteRequest() - vs.guard.Secure(vs.PostHandler)(w, r) - case "POST": - stats.WriteRequest() - vs.guard.Secure(vs.PostHandler)(w, r) } }