aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorchrislusf <chris.lu@gmail.com>2015-03-09 01:10:01 -0700
committerchrislusf <chris.lu@gmail.com>2015-03-09 01:10:04 -0700
commitf511b507a5230e931d0250b446f744073be879b3 (patch)
treee9728b47902fc4f31aa93d24a937cb2654becc6c
parentd48d76cb4f419a58539fb84c7a45c7eb863b854f (diff)
downloadseaweedfs-f511b507a5230e931d0250b446f744073be879b3.tar.xz
seaweedfs-f511b507a5230e931d0250b446f744073be879b3.zip
Add read only public port on volume server
Add read only public port on volume server
-rw-r--r--go/operation/assign_file_id.go3
-rw-r--r--go/operation/submit.go2
-rw-r--r--go/storage/store.go6
-rw-r--r--go/topology/allocate_volume.go2
-rw-r--r--go/topology/data_node.go6
-rw-r--r--go/topology/rack.go3
-rw-r--r--go/topology/topology.go6
-rw-r--r--go/topology/topology_vacuum.go12
-rw-r--r--go/weed/server.go35
-rw-r--r--go/weed/volume.go36
-rw-r--r--go/weed/weed_server/volume_server.go9
-rw-r--r--go/weed/weed_server/volume_server_handlers.go25
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)
}
}