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.

376 lines
9.2 KiB

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