store.go 22 KB

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