store_ec.go 13 KB

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