store_ec.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392
  1. package storage
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "os"
  7. "sort"
  8. "sync"
  9. "time"
  10. "github.com/klauspost/reedsolomon"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/operation"
  13. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  14. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  15. "github.com/chrislusf/seaweedfs/weed/stats"
  16. "github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
  17. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  18. "github.com/chrislusf/seaweedfs/weed/storage/types"
  19. )
  20. func (s *Store) CollectErasureCodingHeartbeat() *master_pb.Heartbeat {
  21. var ecShardMessages []*master_pb.VolumeEcShardInformationMessage
  22. collectionEcShardSize := make(map[string]int64)
  23. for _, location := range s.Locations {
  24. location.ecVolumesLock.RLock()
  25. for _, ecShards := range location.ecVolumes {
  26. ecShardMessages = append(ecShardMessages, ecShards.ToVolumeEcShardInformationMessage()...)
  27. for _, ecShard := range ecShards.Shards {
  28. collectionEcShardSize[ecShards.Collection] += ecShard.Size()
  29. }
  30. }
  31. location.ecVolumesLock.RUnlock()
  32. }
  33. for col, size := range collectionEcShardSize {
  34. stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "ec").Set(float64(size))
  35. }
  36. return &master_pb.Heartbeat{
  37. EcShards: ecShardMessages,
  38. HasNoEcShards: len(ecShardMessages) == 0,
  39. }
  40. }
  41. func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId) error {
  42. for _, location := range s.Locations {
  43. if err := location.LoadEcShard(collection, vid, shardId); err == nil {
  44. glog.V(0).Infof("MountEcShards %d.%d", vid, shardId)
  45. var shardBits erasure_coding.ShardBits
  46. s.NewEcShardsChan <- master_pb.VolumeEcShardInformationMessage{
  47. Id: uint32(vid),
  48. Collection: collection,
  49. EcIndexBits: uint32(shardBits.AddShardId(shardId)),
  50. DiskType: string(location.DiskType),
  51. }
  52. return nil
  53. } else if err == os.ErrNotExist {
  54. continue
  55. } else {
  56. return fmt.Errorf("%s load ec shard %d.%d: %v", location.Directory, vid, shardId, err)
  57. }
  58. }
  59. return fmt.Errorf("MountEcShards %d.%d not found on disk", vid, shardId)
  60. }
  61. func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.ShardId) error {
  62. ecShard, found := s.findEcShard(vid, shardId)
  63. if !found {
  64. return nil
  65. }
  66. var shardBits erasure_coding.ShardBits
  67. message := master_pb.VolumeEcShardInformationMessage{
  68. Id: uint32(vid),
  69. Collection: ecShard.Collection,
  70. EcIndexBits: uint32(shardBits.AddShardId(shardId)),
  71. DiskType: string(ecShard.DiskType),
  72. }
  73. for _, location := range s.Locations {
  74. if deleted := location.UnloadEcShard(vid, shardId); deleted {
  75. glog.V(0).Infof("UnmountEcShards %d.%d", vid, shardId)
  76. s.DeletedEcShardsChan <- message
  77. return nil
  78. }
  79. }
  80. return fmt.Errorf("UnmountEcShards %d.%d not found on disk", vid, shardId)
  81. }
  82. func (s *Store) findEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolumeShard, bool) {
  83. for _, location := range s.Locations {
  84. if v, found := location.FindEcShard(vid, shardId); found {
  85. return v, found
  86. }
  87. }
  88. return nil, false
  89. }
  90. func (s *Store) FindEcVolume(vid needle.VolumeId) (*erasure_coding.EcVolume, bool) {
  91. for _, location := range s.Locations {
  92. if s, found := location.FindEcVolume(vid); found {
  93. return s, true
  94. }
  95. }
  96. return nil, false
  97. }
  98. func (s *Store) DestroyEcVolume(vid needle.VolumeId) {
  99. for _, location := range s.Locations {
  100. location.DestroyEcVolume(vid)
  101. }
  102. }
  103. func (s *Store) ReadEcShardNeedle(vid needle.VolumeId, n *needle.Needle) (int, error) {
  104. for _, location := range s.Locations {
  105. if localEcVolume, found := location.FindEcVolume(vid); found {
  106. offset, size, intervals, err := localEcVolume.LocateEcShardNeedle(n.Id, localEcVolume.Version)
  107. if err != nil {
  108. return 0, fmt.Errorf("locate in local ec volume: %v", err)
  109. }
  110. if size.IsDeleted() {
  111. return 0, ErrorDeleted
  112. }
  113. glog.V(3).Infof("read ec volume %d offset %d size %d intervals:%+v", vid, offset.ToActualOffset(), size, intervals)
  114. if len(intervals) > 1 {
  115. glog.V(3).Infof("ReadEcShardNeedle needle id %s intervals:%+v", n.String(), intervals)
  116. }
  117. bytes, isDeleted, err := s.readEcShardIntervals(vid, n.Id, localEcVolume, intervals)
  118. if err != nil {
  119. return 0, fmt.Errorf("ReadEcShardIntervals: %v", err)
  120. }
  121. if isDeleted {
  122. return 0, ErrorDeleted
  123. }
  124. err = n.ReadBytes(bytes, offset.ToActualOffset(), size, localEcVolume.Version)
  125. if err != nil {
  126. return 0, fmt.Errorf("readbytes: %v", err)
  127. }
  128. return len(bytes), nil
  129. }
  130. }
  131. return 0, fmt.Errorf("ec shard %d not found", vid)
  132. }
  133. func (s *Store) readEcShardIntervals(vid needle.VolumeId, needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, is_deleted bool, err error) {
  134. if err = s.cachedLookupEcShardLocations(ecVolume); err != nil {
  135. return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)
  136. }
  137. for i, interval := range intervals {
  138. if d, isDeleted, e := s.readOneEcShardInterval(needleId, ecVolume, interval); e != nil {
  139. return nil, isDeleted, e
  140. } else {
  141. if isDeleted {
  142. is_deleted = true
  143. }
  144. if i == 0 {
  145. data = d
  146. } else {
  147. data = append(data, d...)
  148. }
  149. }
  150. }
  151. return
  152. }
  153. func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, interval erasure_coding.Interval) (data []byte, is_deleted bool, err error) {
  154. shardId, actualOffset := interval.ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize)
  155. data = make([]byte, interval.Size)
  156. if shard, found := ecVolume.FindEcVolumeShard(shardId); found {
  157. if _, err = shard.ReadAt(data, actualOffset); err != nil {
  158. glog.V(0).Infof("read local ec shard %d.%d offset %d: %v", ecVolume.VolumeId, shardId, actualOffset, err)
  159. return
  160. }
  161. } else {
  162. ecVolume.ShardLocationsLock.RLock()
  163. sourceDataNodes, hasShardIdLocation := ecVolume.ShardLocations[shardId]
  164. ecVolume.ShardLocationsLock.RUnlock()
  165. // try reading directly
  166. if hasShardIdLocation {
  167. _, is_deleted, err = s.readRemoteEcShardInterval(sourceDataNodes, needleId, ecVolume.VolumeId, shardId, data, actualOffset)
  168. if err == nil {
  169. return
  170. }
  171. glog.V(0).Infof("clearing ec shard %d.%d locations: %v", ecVolume.VolumeId, shardId, err)
  172. forgetShardId(ecVolume, shardId)
  173. }
  174. // try reading by recovering from other shards
  175. _, is_deleted, err = s.recoverOneRemoteEcShardInterval(needleId, ecVolume, shardId, data, actualOffset)
  176. if err == nil {
  177. return
  178. }
  179. glog.V(0).Infof("recover ec shard %d.%d : %v", ecVolume.VolumeId, shardId, err)
  180. }
  181. return
  182. }
  183. func forgetShardId(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.ShardId) {
  184. // failed to access the source data nodes, clear it up
  185. ecVolume.ShardLocationsLock.Lock()
  186. delete(ecVolume.ShardLocations, shardId)
  187. ecVolume.ShardLocationsLock.Unlock()
  188. }
  189. func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) (err error) {
  190. shardCount := len(ecVolume.ShardLocations)
  191. if shardCount < erasure_coding.DataShardsCount &&
  192. ecVolume.ShardLocationsRefreshTime.Add(11*time.Second).After(time.Now()) ||
  193. shardCount == erasure_coding.TotalShardsCount &&
  194. ecVolume.ShardLocationsRefreshTime.Add(37*time.Minute).After(time.Now()) ||
  195. shardCount >= erasure_coding.DataShardsCount &&
  196. ecVolume.ShardLocationsRefreshTime.Add(7*time.Minute).After(time.Now()) {
  197. // still fresh
  198. return nil
  199. }
  200. glog.V(3).Infof("lookup and cache ec volume %d locations", ecVolume.VolumeId)
  201. err = operation.WithMasterServerClient(s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
  202. req := &master_pb.LookupEcVolumeRequest{
  203. VolumeId: uint32(ecVolume.VolumeId),
  204. }
  205. resp, err := masterClient.LookupEcVolume(context.Background(), req)
  206. if err != nil {
  207. return fmt.Errorf("lookup ec volume %d: %v", ecVolume.VolumeId, err)
  208. }
  209. if len(resp.ShardIdLocations) < erasure_coding.DataShardsCount {
  210. return fmt.Errorf("only %d shards found but %d required", len(resp.ShardIdLocations), erasure_coding.DataShardsCount)
  211. }
  212. ecVolume.ShardLocationsLock.Lock()
  213. for _, shardIdLocations := range resp.ShardIdLocations {
  214. shardId := erasure_coding.ShardId(shardIdLocations.ShardId)
  215. delete(ecVolume.ShardLocations, shardId)
  216. for _, loc := range shardIdLocations.Locations {
  217. ecVolume.ShardLocations[shardId] = append(ecVolume.ShardLocations[shardId], loc.Url)
  218. }
  219. }
  220. ecVolume.ShardLocationsRefreshTime = time.Now()
  221. ecVolume.ShardLocationsLock.Unlock()
  222. return nil
  223. })
  224. return
  225. }
  226. func (s *Store) readRemoteEcShardInterval(sourceDataNodes []string, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) {
  227. if len(sourceDataNodes) == 0 {
  228. return 0, false, fmt.Errorf("failed to find ec shard %d.%d", vid, shardId)
  229. }
  230. for _, sourceDataNode := range sourceDataNodes {
  231. glog.V(3).Infof("read remote ec shard %d.%d from %s", vid, shardId, sourceDataNode)
  232. n, is_deleted, err = s.doReadRemoteEcShardInterval(sourceDataNode, needleId, vid, shardId, buf, offset)
  233. if err == nil {
  234. return
  235. }
  236. glog.V(1).Infof("read remote ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, err)
  237. }
  238. return
  239. }
  240. func (s *Store) doReadRemoteEcShardInterval(sourceDataNode string, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) {
  241. err = operation.WithVolumeServerClient(sourceDataNode, s.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  242. // copy data slice
  243. shardReadClient, err := client.VolumeEcShardRead(context.Background(), &volume_server_pb.VolumeEcShardReadRequest{
  244. VolumeId: uint32(vid),
  245. ShardId: uint32(shardId),
  246. Offset: offset,
  247. Size: int64(len(buf)),
  248. FileKey: uint64(needleId),
  249. })
  250. if err != nil {
  251. return fmt.Errorf("failed to start reading ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, err)
  252. }
  253. for {
  254. resp, receiveErr := shardReadClient.Recv()
  255. if receiveErr == io.EOF {
  256. break
  257. }
  258. if receiveErr != nil {
  259. return fmt.Errorf("receiving ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, err)
  260. }
  261. if resp.IsDeleted {
  262. is_deleted = true
  263. }
  264. copy(buf[n:n+len(resp.Data)], resp.Data)
  265. n += len(resp.Data)
  266. }
  267. return nil
  268. })
  269. if err != nil {
  270. return 0, is_deleted, fmt.Errorf("read ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, err)
  271. }
  272. return
  273. }
  274. func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, shardIdToRecover erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) {
  275. glog.V(3).Infof("recover ec shard %d.%d from other locations", ecVolume.VolumeId, shardIdToRecover)
  276. enc, err := reedsolomon.New(erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount)
  277. if err != nil {
  278. return 0, false, fmt.Errorf("failed to create encoder: %v", err)
  279. }
  280. bufs := make([][]byte, erasure_coding.TotalShardsCount)
  281. var wg sync.WaitGroup
  282. ecVolume.ShardLocationsLock.RLock()
  283. for shardId, locations := range ecVolume.ShardLocations {
  284. // skip currnent shard or empty shard
  285. if shardId == shardIdToRecover {
  286. continue
  287. }
  288. if len(locations) == 0 {
  289. glog.V(3).Infof("readRemoteEcShardInterval missing %d.%d from %+v", ecVolume.VolumeId, shardId, locations)
  290. continue
  291. }
  292. // read from remote locations
  293. wg.Add(1)
  294. go func(shardId erasure_coding.ShardId, locations []string) {
  295. defer wg.Done()
  296. data := make([]byte, len(buf))
  297. nRead, isDeleted, readErr := s.readRemoteEcShardInterval(locations, needleId, ecVolume.VolumeId, shardId, data, offset)
  298. if readErr != nil {
  299. glog.V(3).Infof("recover: readRemoteEcShardInterval %d.%d %d bytes from %+v: %v", ecVolume.VolumeId, shardId, nRead, locations, readErr)
  300. forgetShardId(ecVolume, shardId)
  301. }
  302. if isDeleted {
  303. is_deleted = true
  304. }
  305. if nRead == len(buf) {
  306. bufs[shardId] = data
  307. }
  308. }(shardId, locations)
  309. }
  310. ecVolume.ShardLocationsLock.RUnlock()
  311. wg.Wait()
  312. if err = enc.ReconstructData(bufs); err != nil {
  313. glog.V(3).Infof("recovered ec shard %d.%d failed: %v", ecVolume.VolumeId, shardIdToRecover, err)
  314. return 0, false, err
  315. }
  316. glog.V(4).Infof("recovered ec shard %d.%d from other locations", ecVolume.VolumeId, shardIdToRecover)
  317. copy(buf, bufs[shardIdToRecover])
  318. return len(buf), is_deleted, nil
  319. }
  320. func (s *Store) EcVolumes() (ecVolumes []*erasure_coding.EcVolume) {
  321. for _, location := range s.Locations {
  322. location.ecVolumesLock.RLock()
  323. for _, v := range location.ecVolumes {
  324. ecVolumes = append(ecVolumes, v)
  325. }
  326. location.ecVolumesLock.RUnlock()
  327. }
  328. sort.Slice(ecVolumes, func(i, j int) bool {
  329. return ecVolumes[i].VolumeId > ecVolumes[j].VolumeId
  330. })
  331. return ecVolumes
  332. }