You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

158 lines
4.6 KiB

10 years ago
10 years ago
10 years ago
  1. package topology
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "net/url"
  6. "time"
  7. "github.com/chrislusf/seaweedfs/go/glog"
  8. "github.com/chrislusf/seaweedfs/go/storage"
  9. "github.com/chrislusf/seaweedfs/go/util"
  10. )
  11. func batchVacuumVolumeCheck(vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList, garbageThreshold string) bool {
  12. ch := make(chan bool, locationlist.Length())
  13. for index, dn := range locationlist.list {
  14. go func(index int, url string, vid storage.VolumeId) {
  15. //glog.V(0).Infoln(index, "Check vacuuming", vid, "on", dn.Url())
  16. if e, ret := vacuumVolume_Check(url, vid, garbageThreshold); e != nil {
  17. //glog.V(0).Infoln(index, "Error when checking vacuuming", vid, "on", url, e)
  18. ch <- false
  19. } else {
  20. //glog.V(0).Infoln(index, "Checked vacuuming", vid, "on", url, "needVacuum", ret)
  21. ch <- ret
  22. }
  23. }(index, dn.Url(), vid)
  24. }
  25. isCheckSuccess := true
  26. for _ = range locationlist.list {
  27. select {
  28. case canVacuum := <-ch:
  29. isCheckSuccess = isCheckSuccess && canVacuum
  30. case <-time.After(30 * time.Minute):
  31. isCheckSuccess = false
  32. break
  33. }
  34. }
  35. return isCheckSuccess
  36. }
  37. func batchVacuumVolumeCompact(vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList) bool {
  38. vl.removeFromWritable(vid)
  39. ch := make(chan bool, locationlist.Length())
  40. for index, dn := range locationlist.list {
  41. go func(index int, url string, vid storage.VolumeId) {
  42. glog.V(0).Infoln(index, "Start vacuuming", vid, "on", url)
  43. if e := vacuumVolume_Compact(url, vid); e != nil {
  44. glog.V(0).Infoln(index, "Error when vacuuming", vid, "on", url, e)
  45. ch <- false
  46. } else {
  47. glog.V(0).Infoln(index, "Complete vacuuming", vid, "on", url)
  48. ch <- true
  49. }
  50. }(index, dn.Url(), vid)
  51. }
  52. isVacuumSuccess := true
  53. for _ = range locationlist.list {
  54. select {
  55. case _ = <-ch:
  56. case <-time.After(30 * time.Minute):
  57. isVacuumSuccess = false
  58. break
  59. }
  60. }
  61. return isVacuumSuccess
  62. }
  63. func batchVacuumVolumeCommit(vl *VolumeLayout, vid storage.VolumeId, locationlist *VolumeLocationList) bool {
  64. isCommitSuccess := true
  65. for _, dn := range locationlist.list {
  66. glog.V(0).Infoln("Start Commiting vacuum", vid, "on", dn.Url())
  67. if e := vacuumVolume_Commit(dn.Url(), vid); e != nil {
  68. glog.V(0).Infoln("Error when committing vacuum", vid, "on", dn.Url(), e)
  69. isCommitSuccess = false
  70. } else {
  71. glog.V(0).Infoln("Complete Commiting vacuum", vid, "on", dn.Url())
  72. }
  73. if isCommitSuccess {
  74. vl.SetVolumeAvailable(dn, vid)
  75. }
  76. }
  77. return isCommitSuccess
  78. }
  79. func (t *Topology) Vacuum(garbageThreshold string) int {
  80. glog.V(0).Infoln("Start vacuum on demand")
  81. for _, col := range t.collectionMap.Items {
  82. c := col.(*Collection)
  83. glog.V(0).Infoln("vacuum on collection:", c.Name)
  84. for _, vl := range c.storageType2VolumeLayout.Items {
  85. if vl != nil {
  86. volumeLayout := vl.(*VolumeLayout)
  87. for vid, locationlist := range volumeLayout.vid2location {
  88. glog.V(0).Infoln("vacuum on collection:", c.Name, "volume", vid)
  89. if batchVacuumVolumeCheck(volumeLayout, vid, locationlist, garbageThreshold) {
  90. if batchVacuumVolumeCompact(volumeLayout, vid, locationlist) {
  91. batchVacuumVolumeCommit(volumeLayout, vid, locationlist)
  92. }
  93. }
  94. }
  95. }
  96. }
  97. }
  98. return 0
  99. }
  100. type VacuumVolumeResult struct {
  101. Result bool
  102. Error string
  103. }
  104. func vacuumVolume_Check(urlLocation string, vid storage.VolumeId, garbageThreshold string) (error, bool) {
  105. values := make(url.Values)
  106. values.Add("volume", vid.String())
  107. values.Add("garbageThreshold", garbageThreshold)
  108. jsonBlob, err := util.Post("http://"+urlLocation+"/admin/vacuum/check", values)
  109. if err != nil {
  110. glog.V(0).Infoln("parameters:", values)
  111. return err, false
  112. }
  113. var ret VacuumVolumeResult
  114. if err := json.Unmarshal(jsonBlob, &ret); err != nil {
  115. return err, false
  116. }
  117. if ret.Error != "" {
  118. return errors.New(ret.Error), false
  119. }
  120. return nil, ret.Result
  121. }
  122. func vacuumVolume_Compact(urlLocation string, vid storage.VolumeId) error {
  123. values := make(url.Values)
  124. values.Add("volume", vid.String())
  125. jsonBlob, err := util.Post("http://"+urlLocation+"/admin/vacuum/compact", values)
  126. if err != nil {
  127. return err
  128. }
  129. var ret VacuumVolumeResult
  130. if err := json.Unmarshal(jsonBlob, &ret); err != nil {
  131. return err
  132. }
  133. if ret.Error != "" {
  134. return errors.New(ret.Error)
  135. }
  136. return nil
  137. }
  138. func vacuumVolume_Commit(urlLocation string, vid storage.VolumeId) error {
  139. values := make(url.Values)
  140. values.Add("volume", vid.String())
  141. jsonBlob, err := util.Post("http://"+urlLocation+"/admin/vacuum/commit", values)
  142. if err != nil {
  143. return err
  144. }
  145. var ret VacuumVolumeResult
  146. if err := json.Unmarshal(jsonBlob, &ret); err != nil {
  147. return err
  148. }
  149. if ret.Error != "" {
  150. return errors.New(ret.Error)
  151. }
  152. return nil
  153. }