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.

436 lines
14 KiB

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