store.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579
  1. package storage
  2. import (
  3. "fmt"
  4. "io"
  5. "path/filepath"
  6. "strings"
  7. "sync/atomic"
  8. "github.com/seaweedfs/seaweedfs/weed/pb"
  9. "github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
  10. "github.com/seaweedfs/seaweedfs/weed/util"
  11. "google.golang.org/grpc"
  12. "github.com/seaweedfs/seaweedfs/weed/glog"
  13. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  14. "github.com/seaweedfs/seaweedfs/weed/stats"
  15. "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
  16. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  17. "github.com/seaweedfs/seaweedfs/weed/storage/super_block"
  18. . "github.com/seaweedfs/seaweedfs/weed/storage/types"
  19. )
  20. const (
  21. MAX_TTL_VOLUME_REMOVAL_DELAY = 10 // 10 minutes
  22. )
  23. type ReadOption struct {
  24. // request
  25. ReadDeleted bool
  26. AttemptMetaOnly bool
  27. MustMetaOnly bool
  28. // response
  29. IsMetaOnly bool // read status
  30. VolumeRevision uint16
  31. IsOutOfRange bool // whether read over MaxPossibleVolumeSize
  32. // If HasSlowRead is set to true:
  33. // * read requests and write requests compete for the lock.
  34. // * large file read P99 latency on busy sites will go up, due to the need to get locks multiple times.
  35. // * write requests will see lower latency.
  36. // If HasSlowRead is set to false:
  37. // * read requests should complete asap, not blocking other requests.
  38. // * write requests may see high latency when downloading large files.
  39. HasSlowRead bool
  40. // increasing ReadBufferSize can reduce the number of get locks times and shorten read P99 latency.
  41. // but will increase memory usage a bit. Use with hasSlowRead normally.
  42. ReadBufferSize int
  43. }
  44. /*
  45. * A VolumeServer contains one Store
  46. */
  47. type Store struct {
  48. MasterAddress pb.ServerAddress
  49. grpcDialOption grpc.DialOption
  50. volumeSizeLimit uint64 // read from the master
  51. Ip string
  52. Port int
  53. GrpcPort int
  54. PublicUrl string
  55. Locations []*DiskLocation
  56. dataCenter string // optional information, overwriting master setting if exists
  57. rack string // optional information, overwriting master setting if exists
  58. connected bool
  59. NeedleMapKind NeedleMapKind
  60. NewVolumesChan chan master_pb.VolumeShortInformationMessage
  61. DeletedVolumesChan chan master_pb.VolumeShortInformationMessage
  62. NewEcShardsChan chan master_pb.VolumeEcShardInformationMessage
  63. DeletedEcShardsChan chan master_pb.VolumeEcShardInformationMessage
  64. isStopping bool
  65. }
  66. func (s *Store) String() (str string) {
  67. str = fmt.Sprintf("Ip:%s, Port:%d, GrpcPort:%d PublicUrl:%s, dataCenter:%s, rack:%s, connected:%v, volumeSizeLimit:%d", s.Ip, s.Port, s.GrpcPort, s.PublicUrl, s.dataCenter, s.rack, s.connected, s.GetVolumeSizeLimit())
  68. return
  69. }
  70. func NewStore(grpcDialOption grpc.DialOption, ip string, port int, grpcPort int, publicUrl string, dirnames []string, maxVolumeCounts []int32,
  71. minFreeSpaces []util.MinFreeSpace, idxFolder string, needleMapKind NeedleMapKind, diskTypes []DiskType) (s *Store) {
  72. s = &Store{grpcDialOption: grpcDialOption, Port: port, Ip: ip, GrpcPort: grpcPort, PublicUrl: publicUrl, NeedleMapKind: needleMapKind}
  73. s.Locations = make([]*DiskLocation, 0)
  74. for i := 0; i < len(dirnames); i++ {
  75. location := NewDiskLocation(dirnames[i], int32(maxVolumeCounts[i]), minFreeSpaces[i], idxFolder, diskTypes[i])
  76. location.loadExistingVolumes(needleMapKind)
  77. s.Locations = append(s.Locations, location)
  78. stats.VolumeServerMaxVolumeCounter.Add(float64(maxVolumeCounts[i]))
  79. }
  80. s.NewVolumesChan = make(chan master_pb.VolumeShortInformationMessage, 3)
  81. s.DeletedVolumesChan = make(chan master_pb.VolumeShortInformationMessage, 3)
  82. s.NewEcShardsChan = make(chan master_pb.VolumeEcShardInformationMessage, 3)
  83. s.DeletedEcShardsChan = make(chan master_pb.VolumeEcShardInformationMessage, 3)
  84. return
  85. }
  86. func (s *Store) AddVolume(volumeId needle.VolumeId, collection string, needleMapKind NeedleMapKind, replicaPlacement string, ttlString string, preallocate int64, MemoryMapMaxSizeMb uint32, diskType DiskType) error {
  87. rt, e := super_block.NewReplicaPlacementFromString(replicaPlacement)
  88. if e != nil {
  89. return e
  90. }
  91. ttl, e := needle.ReadTTL(ttlString)
  92. if e != nil {
  93. return e
  94. }
  95. e = s.addVolume(volumeId, collection, needleMapKind, rt, ttl, preallocate, MemoryMapMaxSizeMb, diskType)
  96. return e
  97. }
  98. func (s *Store) DeleteCollection(collection string) (e error) {
  99. for _, location := range s.Locations {
  100. e = location.DeleteCollectionFromDiskLocation(collection)
  101. if e != nil {
  102. return
  103. }
  104. stats.DeleteCollectionMetrics(collection)
  105. // let the heartbeat send the list of volumes, instead of sending the deleted volume ids to DeletedVolumesChan
  106. }
  107. return
  108. }
  109. func (s *Store) findVolume(vid needle.VolumeId) *Volume {
  110. for _, location := range s.Locations {
  111. if v, found := location.FindVolume(vid); found {
  112. return v
  113. }
  114. }
  115. return nil
  116. }
  117. func (s *Store) FindFreeLocation(diskType DiskType) (ret *DiskLocation) {
  118. max := int32(0)
  119. for _, location := range s.Locations {
  120. if diskType != location.DiskType {
  121. continue
  122. }
  123. if location.isDiskSpaceLow {
  124. continue
  125. }
  126. currentFreeCount := location.MaxVolumeCount - int32(location.VolumesLen())
  127. currentFreeCount *= erasure_coding.DataShardsCount
  128. currentFreeCount -= int32(location.EcVolumesLen())
  129. currentFreeCount /= erasure_coding.DataShardsCount
  130. if currentFreeCount > max {
  131. max = currentFreeCount
  132. ret = location
  133. }
  134. }
  135. return ret
  136. }
  137. func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32, diskType DiskType) error {
  138. if s.findVolume(vid) != nil {
  139. return fmt.Errorf("Volume Id %d already exists!", vid)
  140. }
  141. if location := s.FindFreeLocation(diskType); location != nil {
  142. glog.V(0).Infof("In dir %s adds volume:%v collection:%s replicaPlacement:%v ttl:%v",
  143. location.Directory, vid, collection, replicaPlacement, ttl)
  144. if volume, err := NewVolume(location.Directory, location.IdxDirectory, collection, vid, needleMapKind, replicaPlacement, ttl, preallocate, memoryMapMaxSizeMb); err == nil {
  145. location.SetVolume(vid, volume)
  146. glog.V(0).Infof("add volume %d", vid)
  147. s.NewVolumesChan <- master_pb.VolumeShortInformationMessage{
  148. Id: uint32(vid),
  149. Collection: collection,
  150. ReplicaPlacement: uint32(replicaPlacement.Byte()),
  151. Version: uint32(volume.Version()),
  152. Ttl: ttl.ToUint32(),
  153. DiskType: string(diskType),
  154. }
  155. return nil
  156. } else {
  157. return err
  158. }
  159. }
  160. return fmt.Errorf("No more free space left")
  161. }
  162. func (s *Store) VolumeInfos() (allStats []*VolumeInfo) {
  163. for _, location := range s.Locations {
  164. stats := collectStatsForOneLocation(location)
  165. allStats = append(allStats, stats...)
  166. }
  167. sortVolumeInfos(allStats)
  168. return allStats
  169. }
  170. func collectStatsForOneLocation(location *DiskLocation) (stats []*VolumeInfo) {
  171. location.volumesLock.RLock()
  172. defer location.volumesLock.RUnlock()
  173. for k, v := range location.volumes {
  174. s := collectStatForOneVolume(k, v)
  175. stats = append(stats, s)
  176. }
  177. return stats
  178. }
  179. func collectStatForOneVolume(vid needle.VolumeId, v *Volume) (s *VolumeInfo) {
  180. s = &VolumeInfo{
  181. Id: vid,
  182. Collection: v.Collection,
  183. ReplicaPlacement: v.ReplicaPlacement,
  184. Version: v.Version(),
  185. ReadOnly: v.IsReadOnly(),
  186. Ttl: v.Ttl,
  187. CompactRevision: uint32(v.CompactionRevision),
  188. DiskType: v.DiskType().String(),
  189. }
  190. s.RemoteStorageName, s.RemoteStorageKey = v.RemoteStorageNameKey()
  191. v.dataFileAccessLock.RLock()
  192. defer v.dataFileAccessLock.RUnlock()
  193. if v.nm == nil {
  194. return
  195. }
  196. s.FileCount = v.nm.FileCount()
  197. s.DeleteCount = v.nm.DeletedCount()
  198. s.DeletedByteCount = v.nm.DeletedSize()
  199. s.Size = v.nm.ContentSize()
  200. return
  201. }
  202. func (s *Store) SetDataCenter(dataCenter string) {
  203. s.dataCenter = dataCenter
  204. }
  205. func (s *Store) SetRack(rack string) {
  206. s.rack = rack
  207. }
  208. func (s *Store) GetDataCenter() string {
  209. return s.dataCenter
  210. }
  211. func (s *Store) GetRack() string {
  212. return s.rack
  213. }
  214. func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
  215. var volumeMessages []*master_pb.VolumeInformationMessage
  216. maxVolumeCounts := make(map[string]uint32)
  217. var maxFileKey NeedleId
  218. collectionVolumeSize := make(map[string]int64)
  219. collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
  220. for _, location := range s.Locations {
  221. var deleteVids []needle.VolumeId
  222. maxVolumeCounts[string(location.DiskType)] += uint32(location.MaxVolumeCount)
  223. location.volumesLock.RLock()
  224. for _, v := range location.volumes {
  225. curMaxFileKey, volumeMessage := v.ToVolumeInformationMessage()
  226. if volumeMessage == nil {
  227. continue
  228. }
  229. if maxFileKey < curMaxFileKey {
  230. maxFileKey = curMaxFileKey
  231. }
  232. shouldDeleteVolume := false
  233. if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
  234. volumeMessages = append(volumeMessages, volumeMessage)
  235. } else {
  236. if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
  237. deleteVids = append(deleteVids, v.Id)
  238. shouldDeleteVolume = true
  239. } else {
  240. glog.V(0).Infof("volume %d is expired", v.Id)
  241. }
  242. if v.lastIoError != nil {
  243. deleteVids = append(deleteVids, v.Id)
  244. shouldDeleteVolume = true
  245. glog.Warningf("volume %d has IO error: %v", v.Id, v.lastIoError)
  246. }
  247. }
  248. if _, exist := collectionVolumeSize[v.Collection]; !exist {
  249. collectionVolumeSize[v.Collection] = 0
  250. }
  251. if !shouldDeleteVolume {
  252. collectionVolumeSize[v.Collection] += int64(volumeMessage.Size)
  253. } else {
  254. collectionVolumeSize[v.Collection] -= int64(volumeMessage.Size)
  255. if collectionVolumeSize[v.Collection] <= 0 {
  256. delete(collectionVolumeSize, v.Collection)
  257. }
  258. }
  259. if _, exist := collectionVolumeReadOnlyCount[v.Collection]; !exist {
  260. collectionVolumeReadOnlyCount[v.Collection] = map[string]uint8{
  261. stats.IsReadOnly: 0,
  262. stats.NoWriteOrDelete: 0,
  263. stats.NoWriteCanDelete: 0,
  264. stats.IsDiskSpaceLow: 0,
  265. }
  266. }
  267. if !shouldDeleteVolume && v.IsReadOnly() {
  268. collectionVolumeReadOnlyCount[v.Collection][stats.IsReadOnly] += 1
  269. if v.noWriteOrDelete {
  270. collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteOrDelete] += 1
  271. }
  272. if v.noWriteCanDelete {
  273. collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteCanDelete] += 1
  274. }
  275. if v.location.isDiskSpaceLow {
  276. collectionVolumeReadOnlyCount[v.Collection][stats.IsDiskSpaceLow] += 1
  277. }
  278. }
  279. }
  280. location.volumesLock.RUnlock()
  281. if len(deleteVids) > 0 {
  282. // delete expired volumes.
  283. location.volumesLock.Lock()
  284. for _, vid := range deleteVids {
  285. found, err := location.deleteVolumeById(vid)
  286. if err == nil {
  287. if found {
  288. glog.V(0).Infof("volume %d is deleted", vid)
  289. }
  290. } else {
  291. glog.Warningf("delete volume %d: %v", vid, err)
  292. }
  293. }
  294. location.volumesLock.Unlock()
  295. }
  296. }
  297. var uuidList []string
  298. for _, loc := range s.Locations {
  299. uuidList = append(uuidList, loc.DirectoryUuid)
  300. }
  301. for col, size := range collectionVolumeSize {
  302. stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "normal").Set(float64(size))
  303. }
  304. for col, types := range collectionVolumeReadOnlyCount {
  305. for t, count := range types {
  306. stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues(col, t).Set(float64(count))
  307. }
  308. }
  309. return &master_pb.Heartbeat{
  310. Ip: s.Ip,
  311. Port: uint32(s.Port),
  312. GrpcPort: uint32(s.GrpcPort),
  313. PublicUrl: s.PublicUrl,
  314. MaxVolumeCounts: maxVolumeCounts,
  315. MaxFileKey: NeedleIdToUint64(maxFileKey),
  316. DataCenter: s.dataCenter,
  317. Rack: s.rack,
  318. Volumes: volumeMessages,
  319. HasNoVolumes: len(volumeMessages) == 0,
  320. LocationUuids: uuidList,
  321. }
  322. }
  323. func (s *Store) SetStopping() {
  324. s.isStopping = true
  325. for _, location := range s.Locations {
  326. location.SetStopping()
  327. }
  328. }
  329. func (s *Store) Close() {
  330. for _, location := range s.Locations {
  331. location.Close()
  332. }
  333. }
  334. func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, checkCookie bool, fsync bool) (isUnchanged bool, err error) {
  335. if v := s.findVolume(i); v != nil {
  336. if v.IsReadOnly() {
  337. err = fmt.Errorf("volume %d is read only", i)
  338. return
  339. }
  340. _, _, isUnchanged, err = v.writeNeedle2(n, checkCookie, fsync && s.isStopping)
  341. return
  342. }
  343. glog.V(0).Infoln("volume", i, "not found!")
  344. err = fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port)
  345. return
  346. }
  347. func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (Size, error) {
  348. if v := s.findVolume(i); v != nil {
  349. if v.noWriteOrDelete {
  350. return 0, fmt.Errorf("volume %d is read only", i)
  351. }
  352. return v.deleteNeedle2(n)
  353. }
  354. return 0, fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port)
  355. }
  356. func (s *Store) ReadVolumeNeedle(i needle.VolumeId, n *needle.Needle, readOption *ReadOption, onReadSizeFn func(size Size)) (int, error) {
  357. if v := s.findVolume(i); v != nil {
  358. return v.readNeedle(n, readOption, onReadSizeFn)
  359. }
  360. return 0, fmt.Errorf("volume %d not found", i)
  361. }
  362. func (s *Store) ReadVolumeNeedleMetaAt(i needle.VolumeId, n *needle.Needle, offset int64, size int32) error {
  363. if v := s.findVolume(i); v != nil {
  364. return v.readNeedleMetaAt(n, offset, size)
  365. }
  366. return fmt.Errorf("volume %d not found", i)
  367. }
  368. func (s *Store) ReadVolumeNeedleDataInto(i needle.VolumeId, n *needle.Needle, readOption *ReadOption, writer io.Writer, offset int64, size int64) error {
  369. if v := s.findVolume(i); v != nil {
  370. return v.readNeedleDataInto(n, readOption, writer, offset, size)
  371. }
  372. return fmt.Errorf("volume %d not found", i)
  373. }
  374. func (s *Store) GetVolume(i needle.VolumeId) *Volume {
  375. return s.findVolume(i)
  376. }
  377. func (s *Store) HasVolume(i needle.VolumeId) bool {
  378. v := s.findVolume(i)
  379. return v != nil
  380. }
  381. func (s *Store) MarkVolumeReadonly(i needle.VolumeId) error {
  382. v := s.findVolume(i)
  383. if v == nil {
  384. return fmt.Errorf("volume %d not found", i)
  385. }
  386. v.noWriteLock.Lock()
  387. v.noWriteOrDelete = true
  388. v.noWriteLock.Unlock()
  389. return nil
  390. }
  391. func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
  392. v := s.findVolume(i)
  393. if v == nil {
  394. return fmt.Errorf("volume %d not found", i)
  395. }
  396. v.noWriteLock.Lock()
  397. v.noWriteOrDelete = false
  398. v.noWriteLock.Unlock()
  399. return nil
  400. }
  401. func (s *Store) MountVolume(i needle.VolumeId) error {
  402. for _, location := range s.Locations {
  403. if found := location.LoadVolume(i, s.NeedleMapKind); found == true {
  404. glog.V(0).Infof("mount volume %d", i)
  405. v := s.findVolume(i)
  406. s.NewVolumesChan <- master_pb.VolumeShortInformationMessage{
  407. Id: uint32(v.Id),
  408. Collection: v.Collection,
  409. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  410. Version: uint32(v.Version()),
  411. Ttl: v.Ttl.ToUint32(),
  412. DiskType: string(v.location.DiskType),
  413. }
  414. return nil
  415. }
  416. }
  417. return fmt.Errorf("volume %d not found on disk", i)
  418. }
  419. func (s *Store) UnmountVolume(i needle.VolumeId) error {
  420. v := s.findVolume(i)
  421. if v == nil {
  422. return nil
  423. }
  424. message := master_pb.VolumeShortInformationMessage{
  425. Id: uint32(v.Id),
  426. Collection: v.Collection,
  427. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  428. Version: uint32(v.Version()),
  429. Ttl: v.Ttl.ToUint32(),
  430. DiskType: string(v.location.DiskType),
  431. }
  432. for _, location := range s.Locations {
  433. err := location.UnloadVolume(i)
  434. if err == nil {
  435. glog.V(0).Infof("UnmountVolume %d", i)
  436. stats.DeleteCollectionMetrics(v.Collection)
  437. s.DeletedVolumesChan <- message
  438. return nil
  439. } else if err == ErrVolumeNotFound {
  440. continue
  441. }
  442. }
  443. return fmt.Errorf("volume %d not found on disk", i)
  444. }
  445. func (s *Store) DeleteVolume(i needle.VolumeId) error {
  446. v := s.findVolume(i)
  447. if v == nil {
  448. return fmt.Errorf("delete volume %d not found on disk", i)
  449. }
  450. message := master_pb.VolumeShortInformationMessage{
  451. Id: uint32(v.Id),
  452. Collection: v.Collection,
  453. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  454. Version: uint32(v.Version()),
  455. Ttl: v.Ttl.ToUint32(),
  456. DiskType: string(v.location.DiskType),
  457. }
  458. for _, location := range s.Locations {
  459. err := location.DeleteVolume(i)
  460. if err == nil {
  461. glog.V(0).Infof("DeleteVolume %d", i)
  462. s.DeletedVolumesChan <- message
  463. return nil
  464. } else if err == ErrVolumeNotFound {
  465. continue
  466. } else {
  467. glog.Errorf("DeleteVolume %d: %v", i, err)
  468. }
  469. }
  470. return fmt.Errorf("volume %d not found on disk", i)
  471. }
  472. func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
  473. for _, location := range s.Locations {
  474. fileInfo, found := location.LocateVolume(i)
  475. if !found {
  476. continue
  477. }
  478. // load, modify, save
  479. baseFileName := strings.TrimSuffix(fileInfo.Name(), filepath.Ext(fileInfo.Name()))
  480. vifFile := filepath.Join(location.Directory, baseFileName+".vif")
  481. volumeInfo, _, _, err := volume_info.MaybeLoadVolumeInfo(vifFile)
  482. if err != nil {
  483. return fmt.Errorf("volume %d fail to load vif: %v", i, err)
  484. }
  485. volumeInfo.Replication = replication
  486. err = volume_info.SaveVolumeInfo(vifFile, volumeInfo)
  487. if err != nil {
  488. return fmt.Errorf("volume %d fail to save vif: %v", i, err)
  489. }
  490. return nil
  491. }
  492. return fmt.Errorf("volume %d not found on disk", i)
  493. }
  494. func (s *Store) SetVolumeSizeLimit(x uint64) {
  495. atomic.StoreUint64(&s.volumeSizeLimit, x)
  496. }
  497. func (s *Store) GetVolumeSizeLimit() uint64 {
  498. return atomic.LoadUint64(&s.volumeSizeLimit)
  499. }
  500. func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) {
  501. volumeSizeLimit := s.GetVolumeSizeLimit()
  502. if volumeSizeLimit == 0 {
  503. return
  504. }
  505. for _, diskLocation := range s.Locations {
  506. if diskLocation.OriginalMaxVolumeCount == 0 {
  507. currentMaxVolumeCount := atomic.LoadInt32(&diskLocation.MaxVolumeCount)
  508. diskStatus := stats.NewDiskStatus(diskLocation.Directory)
  509. unusedSpace := diskLocation.UnUsedSpace(volumeSizeLimit)
  510. unclaimedSpaces := int64(diskStatus.Free) - int64(unusedSpace)
  511. volCount := diskLocation.VolumesLen()
  512. maxVolumeCount := int32(volCount)
  513. if unclaimedSpaces > int64(volumeSizeLimit) {
  514. maxVolumeCount += int32(uint64(unclaimedSpaces)/volumeSizeLimit) - 1
  515. }
  516. atomic.StoreInt32(&diskLocation.MaxVolumeCount, maxVolumeCount)
  517. glog.V(4).Infof("disk %s max %d unclaimedSpace:%dMB, unused:%dMB volumeSizeLimit:%dMB",
  518. diskLocation.Directory, maxVolumeCount, unclaimedSpaces/1024/1024, unusedSpace/1024/1024, volumeSizeLimit/1024/1024)
  519. hasChanges = hasChanges || currentMaxVolumeCount != atomic.LoadInt32(&diskLocation.MaxVolumeCount)
  520. }
  521. }
  522. return
  523. }