lots of fix

1. sending 404 if not found
2. handle node-up/node-down/changing-max/volume-become-full

7 files changed, 23 insertions, 20 deletions

diff --git a/weed-fs/src/pkg/replication/volume_growth.go b/weed-fs/src/pkg/replication/volume_growth.go
index 41d7bd18c..0176033d8 100644
--- a/weed-fs/src/pkg/replication/volume_growth.go
+++ b/weed-fs/src/pkg/replication/volume_growth.go
@@ -126,7 +126,7 @@ func (vg *VolumeGrowth) GrowByCountAndType(count int, repType storage.Replicatio
 func (vg *VolumeGrowth) grow(topo *topology.Topology, vid storage.VolumeId, repType storage.ReplicationType, servers ...*topology.DataNode) error {
 	for _, server := range servers {
 		if err := AllocateVolume(server, vid, repType); err == nil {
-			vi := storage.VolumeInfo{Id: vid, Size: 0}
+			vi := storage.VolumeInfo{Id: vid, Size: 0, RepType:repType}
 			server.AddOrUpdateVolume(vi)
 			topo.RegisterVolumeLayout(&vi, server)
 			fmt.Println("Created Volume", vid, "on", server)
diff --git a/weed-fs/src/pkg/topology/data_center.go b/weed-fs/src/pkg/topology/data_center.go
index c661090e8..a5d8cc749 100644
--- a/weed-fs/src/pkg/topology/data_center.go
+++ b/weed-fs/src/pkg/topology/data_center.go
@@ -38,6 +38,7 @@ func (dc *DataCenter) GetOrCreateRack(ip string) *Rack {
 
 func (dc *DataCenter) ToMap() interface{}{
   m := make(map[string]interface{})
+  m["Max"] = dc.GetMaxVolumeCount()
   m["Free"] = dc.FreeSpace()
   var racks []interface{}
   for _, c := range dc.Children() {
diff --git a/weed-fs/src/pkg/topology/data_node.go b/weed-fs/src/pkg/topology/data_node.go
index cb625a41b..cfdae45c3 100644
--- a/weed-fs/src/pkg/topology/data_node.go
+++ b/weed-fs/src/pkg/topology/data_node.go
@@ -49,8 +49,8 @@ func (dn *DataNode) ToMap() interface{} {
 	ret["Ip"] = dn.Ip
 	ret["Port"] = dn.Port
 	ret["Volumes"] = dn.GetActiveVolumeCount()
-	ret["MaxVolumeCount"] = dn.GetMaxVolumeCount()
-	ret["FreeVolumeCount"] = dn.FreeSpace()
+	ret["Max"] = dn.GetMaxVolumeCount()
+	ret["Free"] = dn.FreeSpace()
 	ret["PublicUrl"] = dn.PublicUrl
 	return ret
 }
diff --git a/weed-fs/src/pkg/topology/rack.go b/weed-fs/src/pkg/topology/rack.go
index 16520d14a..bea7d61d7 100644
--- a/weed-fs/src/pkg/topology/rack.go
+++ b/weed-fs/src/pkg/topology/rack.go
@@ -32,10 +32,10 @@ func (r *Rack) GetOrCreateDataNode(ip string, port int, publicUrl string, maxVol
 		if dn.MatchLocation(ip, port) {
 			dn.LastSeen = time.Now().Unix()
 			if dn.Dead {
-			  dn.Dead = false
-			  r.GetTopology().chanRecoveredDataNodes <- dn
+				dn.Dead = false
+				r.GetTopology().chanRecoveredDataNodes <- dn
+				dn.UpAdjustMaxVolumeCountDelta(maxVolumeCount - dn.maxVolumeCount)
 			}
-			dn.UpAdjustMaxVolumeCountDelta(maxVolumeCount - dn.maxVolumeCount)
 			return dn
 		}
 	}
@@ -51,6 +51,7 @@ func (r *Rack) GetOrCreateDataNode(ip string, port int, publicUrl string, maxVol
 
 func (rack *Rack) ToMap() interface{} {
 	m := make(map[string]interface{})
+	m["Max"] = rack.GetMaxVolumeCount()
 	m["Free"] = rack.FreeSpace()
 	var dns []interface{}
 	for _, c := range rack.Children() {
diff --git a/weed-fs/src/pkg/topology/topology.go b/weed-fs/src/pkg/topology/topology.go
index 8dcd64dca..90f402e28 100644
--- a/weed-fs/src/pkg/topology/topology.go
+++ b/weed-fs/src/pkg/topology/topology.go
@@ -23,8 +23,6 @@ type Topology struct {
 	chanDeadDataNodes      chan *DataNode
 	chanRecoveredDataNodes chan *DataNode
 	chanFullVolumes        chan *storage.VolumeInfo
-	chanIncomplemteVolumes chan *storage.VolumeInfo
-	chanRecoveredVolumes   chan *storage.VolumeInfo
 }
 
 func NewTopology(id string, dirname string, filename string, volumeSizeLimit uint64, pulse int) *Topology {
@@ -42,8 +40,6 @@ func NewTopology(id string, dirname string, filename string, volumeSizeLimit uin
 	t.chanDeadDataNodes = make(chan *DataNode)
 	t.chanRecoveredDataNodes = make(chan *DataNode)
 	t.chanFullVolumes = make(chan *storage.VolumeInfo)
-	t.chanIncomplemteVolumes = make(chan *storage.VolumeInfo)
-	t.chanRecoveredVolumes = make(chan *storage.VolumeInfo)
 
 	return t
 }
@@ -124,6 +120,7 @@ func (t *Topology) GetOrCreateDataCenter(ip string) *DataCenter {
 
 func (t *Topology) ToMap() interface{} {
 	m := make(map[string]interface{})
+  m["Max"] = t.GetMaxVolumeCount()
 	m["Free"] = t.FreeSpace()
 	var dcs []interface{}
 	for _, c := range t.Children() {
diff --git a/weed-fs/src/pkg/topology/topology_event_handling.go b/weed-fs/src/pkg/topology/topology_event_handling.go
index 813826a61..0ede2ba1e 100644
--- a/weed-fs/src/pkg/topology/topology_event_handling.go
+++ b/weed-fs/src/pkg/topology/topology_event_handling.go
@@ -18,10 +18,6 @@ func (t *Topology) StartRefreshWritableVolumes() {
 	go func() {
 		for {
 			select {
-			case v := <-t.chanIncomplemteVolumes:
-				fmt.Println("Volume", v, "is incomplete!")
-			case v := <-t.chanRecoveredVolumes:
-				fmt.Println("Volume", v, "is recovered!")
 			case v := <-t.chanFullVolumes:
 				t.SetVolumeCapacityFull(v)
 				fmt.Println("Volume", v, "is full!")
@@ -38,6 +34,9 @@ func (t *Topology) StartRefreshWritableVolumes() {
 func (t *Topology) SetVolumeCapacityFull(volumeInfo *storage.VolumeInfo) {
 	vl := t.GetVolumeLayout(volumeInfo.RepType)
 	vl.SetVolumeCapacityFull(volumeInfo.Id)
+	for _, dn := range vl.vid2location[volumeInfo.Id].list {
+		dn.UpAdjustActiveVolumeCountDelta(-1)
+	}
 }
 func (t *Topology) UnRegisterDataNode(dn *DataNode) {
 	for _, v := range dn.volumes {
@@ -45,6 +44,9 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) {
 		vl := t.GetVolumeLayout(v.RepType)
 		vl.SetVolumeUnavailable(dn, v.Id)
 	}
+	dn.UpAdjustActiveVolumeCountDelta(-dn.GetActiveVolumeCount())
+	dn.UpAdjustMaxVolumeCountDelta(-dn.GetMaxVolumeCount())
+	dn.Parent().UnlinkChildNode(dn.Id())
 }
 func (t *Topology) RegisterRecoveredDataNode(dn *DataNode) {
 	for _, v := range dn.volumes {
diff --git a/weed-fs/src/pkg/topology/volume_layout.go b/weed-fs/src/pkg/topology/volume_layout.go
index ba091b9bb..5159e8e75 100644
--- a/weed-fs/src/pkg/topology/volume_layout.go
+++ b/weed-fs/src/pkg/topology/volume_layout.go
@@ -76,16 +76,18 @@ func (vl *VolumeLayout) setVolumeWritable(vid storage.VolumeId) bool {
 }
 
 func (vl *VolumeLayout) SetVolumeUnavailable(dn *DataNode, vid storage.VolumeId) bool {
-  if vl.vid2location[vid].Remove(dn) {
-    if vl.vid2location[vid].Length() < vl.repType.GetCopyCount() {
-      return vl.removeFromWritable(vid)
-    }
-  }
-  return false
+	if vl.vid2location[vid].Remove(dn) {
+		if vl.vid2location[vid].Length() < vl.repType.GetCopyCount() {
+			fmt.Println("Volume", vid, "has", vl.vid2location[vid].Length(), "replica, less than required", vl.repType.GetCopyCount())
+			return vl.removeFromWritable(vid)
+		}
+	}
+	return false
 }
 func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid storage.VolumeId) bool {
 	if vl.vid2location[vid].Add(dn) {
 		if vl.vid2location[vid].Length() >= vl.repType.GetCopyCount() {
+			fmt.Println("Volume", vid, "becomes writable")
 			return vl.setVolumeWritable(vid)
 		}
 	}