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.

498 lines
15 KiB

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