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.

333 lines
8.2 KiB

5 years ago
5 years ago
5 years ago
6 years ago
5 years ago
5 years ago
6 years ago
6 years ago
6 years ago
5 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
5 years ago
6 years ago
6 years ago
6 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package storage
  2. import (
  3. "fmt"
  4. "io/ioutil"
  5. "os"
  6. "path/filepath"
  7. "strings"
  8. "sync"
  9. "time"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/stats"
  12. "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
  13. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  14. "github.com/chrislusf/seaweedfs/weed/util"
  15. )
  16. type DiskLocation struct {
  17. Directory string
  18. MaxVolumeCount int
  19. OriginalMaxVolumeCount int
  20. MinFreeSpacePercent float32
  21. volumes map[needle.VolumeId]*Volume
  22. volumesLock sync.RWMutex
  23. // erasure coding
  24. ecVolumes map[needle.VolumeId]*erasure_coding.EcVolume
  25. ecVolumesLock sync.RWMutex
  26. isDiskSpaceLow bool
  27. }
  28. func NewDiskLocation(dir string, maxVolumeCount int, minFreeSpacePercent float32) *DiskLocation {
  29. location := &DiskLocation{Directory: dir, MaxVolumeCount: maxVolumeCount, OriginalMaxVolumeCount: maxVolumeCount, MinFreeSpacePercent: minFreeSpacePercent}
  30. location.volumes = make(map[needle.VolumeId]*Volume)
  31. location.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume)
  32. go location.CheckDiskSpace()
  33. return location
  34. }
  35. func (l *DiskLocation) volumeIdFromPath(dir os.FileInfo) (needle.VolumeId, string, error) {
  36. name := dir.Name()
  37. if !dir.IsDir() && strings.HasSuffix(name, ".idx") {
  38. base := name[:len(name)-len(".idx")]
  39. collection, volumeId, err := parseCollectionVolumeId(base)
  40. return volumeId, collection, err
  41. }
  42. return 0, "", fmt.Errorf("Path is not a volume: %s", name)
  43. }
  44. func parseCollectionVolumeId(base string) (collection string, vid needle.VolumeId, err error) {
  45. i := strings.LastIndex(base, "_")
  46. if i > 0 {
  47. collection, base = base[0:i], base[i+1:]
  48. }
  49. vol, err := needle.NewVolumeId(base)
  50. return collection, vol, err
  51. }
  52. func (l *DiskLocation) loadExistingVolume(fileInfo os.FileInfo, needleMapKind NeedleMapType) bool {
  53. name := fileInfo.Name()
  54. if !fileInfo.IsDir() && strings.HasSuffix(name, ".idx") {
  55. name := name[:len(name)-len(".idx")]
  56. noteFile := l.Directory + "/" + name + ".note"
  57. if util.FileExists(noteFile) {
  58. note, _ := ioutil.ReadFile(noteFile)
  59. glog.Warningf("volume %s was not completed: %s", name, string(note))
  60. removeVolumeFiles(l.Directory + "/" + name)
  61. return false
  62. }
  63. vid, collection, err := l.volumeIdFromPath(fileInfo)
  64. if err != nil {
  65. glog.Warningf("get volume id failed, %s, err : %s", name, err)
  66. return false
  67. }
  68. // void loading one volume more than once
  69. l.volumesLock.RLock()
  70. _, found := l.volumes[vid]
  71. l.volumesLock.RUnlock()
  72. if found {
  73. glog.V(1).Infof("loaded volume, %v", vid)
  74. return true
  75. }
  76. v, e := NewVolume(l.Directory, collection, vid, needleMapKind, nil, nil, 0, 0)
  77. if e != nil {
  78. glog.V(0).Infof("new volume %s error %s", name, e)
  79. return false
  80. }
  81. l.SetVolume(vid, v)
  82. size, _, _ := v.FileStat()
  83. glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s",
  84. l.Directory+"/"+name+".dat", v.ReplicaPlacement, v.Version(), size, v.Ttl.String())
  85. return true
  86. }
  87. return false
  88. }
  89. func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapType, concurrency int) {
  90. task_queue := make(chan os.FileInfo, 10*concurrency)
  91. go func() {
  92. if dirs, err := ioutil.ReadDir(l.Directory); err == nil {
  93. for _, dir := range dirs {
  94. task_queue <- dir
  95. }
  96. }
  97. close(task_queue)
  98. }()
  99. var wg sync.WaitGroup
  100. for workerNum := 0; workerNum < concurrency; workerNum++ {
  101. wg.Add(1)
  102. go func() {
  103. defer wg.Done()
  104. for dir := range task_queue {
  105. _ = l.loadExistingVolume(dir, needleMapKind)
  106. }
  107. }()
  108. }
  109. wg.Wait()
  110. }
  111. func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapType) {
  112. l.concurrentLoadingVolumes(needleMapKind, 10)
  113. glog.V(0).Infof("Store started on dir: %s with %d volumes max %d", l.Directory, len(l.volumes), l.MaxVolumeCount)
  114. l.loadAllEcShards()
  115. glog.V(0).Infof("Store started on dir: %s with %d ec shards", l.Directory, len(l.ecVolumes))
  116. }
  117. func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (e error) {
  118. l.volumesLock.Lock()
  119. delVolsMap := l.unmountVolumeByCollection(collection)
  120. l.volumesLock.Unlock()
  121. l.ecVolumesLock.Lock()
  122. delEcVolsMap := l.unmountEcVolumeByCollection(collection)
  123. l.ecVolumesLock.Unlock()
  124. errChain := make(chan error, 2)
  125. var wg sync.WaitGroup
  126. wg.Add(2)
  127. go func() {
  128. for _, v := range delVolsMap {
  129. if err := v.Destroy(); err != nil {
  130. errChain <- err
  131. }
  132. }
  133. wg.Done()
  134. }()
  135. go func() {
  136. for _, v := range delEcVolsMap {
  137. v.Destroy()
  138. }
  139. wg.Done()
  140. }()
  141. go func() {
  142. wg.Wait()
  143. close(errChain)
  144. }()
  145. errBuilder := strings.Builder{}
  146. for err := range errChain {
  147. errBuilder.WriteString(err.Error())
  148. errBuilder.WriteString("; ")
  149. }
  150. if errBuilder.Len() > 0 {
  151. e = fmt.Errorf(errBuilder.String())
  152. }
  153. return
  154. }
  155. func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId) (found bool, e error) {
  156. v, ok := l.volumes[vid]
  157. if !ok {
  158. return
  159. }
  160. e = v.Destroy()
  161. if e != nil {
  162. return
  163. }
  164. found = true
  165. delete(l.volumes, vid)
  166. return
  167. }
  168. func (l *DiskLocation) LoadVolume(vid needle.VolumeId, needleMapKind NeedleMapType) bool {
  169. if fileInfo, found := l.LocateVolume(vid); found {
  170. return l.loadExistingVolume(fileInfo, needleMapKind)
  171. }
  172. return false
  173. }
  174. func (l *DiskLocation) DeleteVolume(vid needle.VolumeId) error {
  175. l.volumesLock.Lock()
  176. defer l.volumesLock.Unlock()
  177. _, ok := l.volumes[vid]
  178. if !ok {
  179. return fmt.Errorf("Volume not found, VolumeId: %d", vid)
  180. }
  181. _, err := l.deleteVolumeById(vid)
  182. return err
  183. }
  184. func (l *DiskLocation) UnloadVolume(vid needle.VolumeId) error {
  185. l.volumesLock.Lock()
  186. defer l.volumesLock.Unlock()
  187. v, ok := l.volumes[vid]
  188. if !ok {
  189. return fmt.Errorf("Volume not loaded, VolumeId: %d", vid)
  190. }
  191. v.Close()
  192. delete(l.volumes, vid)
  193. return nil
  194. }
  195. func (l *DiskLocation) unmountVolumeByCollection(collectionName string) map[needle.VolumeId]*Volume {
  196. deltaVols := make(map[needle.VolumeId]*Volume, 0)
  197. for k, v := range l.volumes {
  198. if v.Collection == collectionName && !v.isCompacting {
  199. deltaVols[k] = v
  200. }
  201. }
  202. for k := range deltaVols {
  203. delete(l.volumes, k)
  204. }
  205. return deltaVols
  206. }
  207. func (l *DiskLocation) SetVolume(vid needle.VolumeId, volume *Volume) {
  208. l.volumesLock.Lock()
  209. defer l.volumesLock.Unlock()
  210. l.volumes[vid] = volume
  211. volume.location = l
  212. }
  213. func (l *DiskLocation) FindVolume(vid needle.VolumeId) (*Volume, bool) {
  214. l.volumesLock.RLock()
  215. defer l.volumesLock.RUnlock()
  216. v, ok := l.volumes[vid]
  217. return v, ok
  218. }
  219. func (l *DiskLocation) VolumesLen() int {
  220. l.volumesLock.RLock()
  221. defer l.volumesLock.RUnlock()
  222. return len(l.volumes)
  223. }
  224. func (l *DiskLocation) Close() {
  225. l.volumesLock.Lock()
  226. for _, v := range l.volumes {
  227. v.Close()
  228. }
  229. l.volumesLock.Unlock()
  230. l.ecVolumesLock.Lock()
  231. for _, ecVolume := range l.ecVolumes {
  232. ecVolume.Close()
  233. }
  234. l.ecVolumesLock.Unlock()
  235. return
  236. }
  237. func (l *DiskLocation) LocateVolume(vid needle.VolumeId) (os.FileInfo, bool) {
  238. if fileInfos, err := ioutil.ReadDir(l.Directory); err == nil {
  239. for _, fileInfo := range fileInfos {
  240. volId, _, err := l.volumeIdFromPath(fileInfo)
  241. if vid == volId && err == nil {
  242. return fileInfo, true
  243. }
  244. }
  245. }
  246. return nil, false
  247. }
  248. func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64) {
  249. l.volumesLock.RLock()
  250. defer l.volumesLock.RUnlock()
  251. for _, vol := range l.volumes {
  252. if vol.IsReadOnly() {
  253. continue
  254. }
  255. datSize, idxSize, _ := vol.FileStat()
  256. unUsedSpace += volumeSizeLimit - (datSize + idxSize)
  257. }
  258. return
  259. }
  260. func (l *DiskLocation) CheckDiskSpace() {
  261. for {
  262. if dir, e := filepath.Abs(l.Directory); e == nil {
  263. s := stats.NewDiskStatus(dir)
  264. stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "all").Set(float64(s.All))
  265. stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "used").Set(float64(s.Used))
  266. stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "free").Set(float64(s.Free))
  267. if (s.PercentFree < l.MinFreeSpacePercent) != l.isDiskSpaceLow {
  268. l.isDiskSpaceLow = !l.isDiskSpaceLow
  269. }
  270. if l.isDiskSpaceLow {
  271. glog.V(0).Infof("dir %s freePercent %.2f%% < min %.2f%%, isLowDiskSpace: %v", dir, s.PercentFree, l.MinFreeSpacePercent, l.isDiskSpaceLow)
  272. } else {
  273. glog.V(4).Infof("dir %s freePercent %.2f%% < min %.2f%%, isLowDiskSpace: %v", dir, s.PercentFree, l.MinFreeSpacePercent, l.isDiskSpaceLow)
  274. }
  275. }
  276. time.Sleep(time.Minute)
  277. }
  278. }