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.

747 lines
24 KiB

3 years ago
3 years ago
5 years ago
5 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
more solid weed mount (#4089) * compare chunks by timestamp * fix slab clearing error * fix test compilation * move oldest chunk to sealed, instead of by fullness * lock on fh.entryViewCache * remove verbose logs * revert slat clearing * less logs * less logs * track write and read by timestamp * remove useless logic * add entry lock on file handle release * use mem chunk only, swap file chunk has problems * comment out code that maybe used later * add debug mode to compare data read and write * more efficient readResolvedChunks with linked list * small optimization * fix test compilation * minor fix on writer * add SeparateGarbageChunks * group chunks into sections * turn off debug mode * fix tests * fix tests * tmp enable swap file chunk * Revert "tmp enable swap file chunk" This reverts commit 985137ec472924e4815f258189f6ca9f2168a0a7. * simple refactoring * simple refactoring * do not re-use swap file chunk. Sealed chunks should not be re-used. * comment out debugging facilities * either mem chunk or swap file chunk is fine now * remove orderedMutex as *semaphore.Weighted not found impactful * optimize size calculation for changing large files * optimize performance to avoid going through the long list of chunks * still problems with swap file chunk * rename * tiny optimization * swap file chunk save only successfully read data * fix * enable both mem and swap file chunk * resolve chunks with range * rename * fix chunk interval list * also change file handle chunk group when adding chunks * pick in-active chunk with time-decayed counter * fix compilation * avoid nil with empty fh.entry * refactoring * rename * rename * refactor visible intervals to *list.List * refactor chunkViews to *list.List * add IntervalList for generic interval list * change visible interval to use IntervalList in generics * cahnge chunkViews to *IntervalList[*ChunkView] * use NewFileChunkSection to create * rename variables * refactor * fix renaming leftover * renaming * renaming * add insert interval * interval list adds lock * incrementally add chunks to readers Fixes: 1. set start and stop offset for the value object 2. clone the value object 3. use pointer instead of copy-by-value when passing to interval.Value 4. use insert interval since adding chunk could be out of order * fix tests compilation * fix tests compilation
2 years ago
4 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
5 years ago
5 years ago
2 years ago
2 years ago
4 years ago
4 years ago
4 years ago
3 years ago
5 years ago
5 years ago
3 years ago
4 years ago
4 years ago
4 years ago
3 years ago
3 years ago
3 years ago
  1. package shell
  2. import (
  3. "bufio"
  4. "bytes"
  5. "context"
  6. "errors"
  7. "flag"
  8. "fmt"
  9. "github.com/seaweedfs/seaweedfs/weed/filer"
  10. "github.com/seaweedfs/seaweedfs/weed/operation"
  11. "github.com/seaweedfs/seaweedfs/weed/pb"
  12. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  13. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  14. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  15. "github.com/seaweedfs/seaweedfs/weed/storage"
  16. "github.com/seaweedfs/seaweedfs/weed/storage/idx"
  17. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  18. "github.com/seaweedfs/seaweedfs/weed/storage/needle_map"
  19. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  20. "github.com/seaweedfs/seaweedfs/weed/util"
  21. "golang.org/x/sync/errgroup"
  22. "io"
  23. "math"
  24. "net/http"
  25. "net/url"
  26. "os"
  27. "path"
  28. "path/filepath"
  29. "strconv"
  30. "strings"
  31. "sync"
  32. "time"
  33. )
  34. func init() {
  35. Commands = append(Commands, &commandVolumeFsck{})
  36. }
  37. const (
  38. readbufferSize = 16
  39. )
  40. type commandVolumeFsck struct {
  41. env *CommandEnv
  42. writer io.Writer
  43. bucketsPath string
  44. collection *string
  45. volumeIds map[uint32]bool
  46. tempFolder string
  47. verbose *bool
  48. forcePurging *bool
  49. findMissingChunksInFiler *bool
  50. verifyNeedle *bool
  51. }
  52. func (c *commandVolumeFsck) Name() string {
  53. return "volume.fsck"
  54. }
  55. func (c *commandVolumeFsck) Help() string {
  56. return `check all volumes to find entries not used by the filer
  57. Important assumption!!!
  58. the system is all used by one filer.
  59. This command works this way:
  60. 1. collect all file ids from all volumes, as set A
  61. 2. collect all file ids from the filer, as set B
  62. 3. find out the set A subtract B
  63. If -findMissingChunksInFiler is enabled, this works
  64. in a reverse way:
  65. 1. collect all file ids from all volumes, as set A
  66. 2. collect all file ids from the filer, as set B
  67. 3. find out the set B subtract A
  68. `
  69. }
  70. func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  71. fsckCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  72. c.verbose = fsckCommand.Bool("v", false, "verbose mode")
  73. c.findMissingChunksInFiler = fsckCommand.Bool("findMissingChunksInFiler", false, "see \"help volume.fsck\"")
  74. c.collection = fsckCommand.String("collection", "", "the collection name")
  75. volumeIds := fsckCommand.String("volumeId", "", "comma separated the volume id")
  76. applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, "<expert only!> after detection, delete missing data from volumes / delete missing file entries from filer. Currently this only works with default filerGroup.")
  77. c.forcePurging = fsckCommand.Bool("forcePurging", false, "delete missing data from volumes in one replica used together with applyPurging")
  78. purgeAbsent := fsckCommand.Bool("reallyDeleteFilerEntries", false, "<expert only!> delete missing file entries from filer if the corresponding volume is missing for any reason, please ensure all still existing/expected volumes are connected! used together with findMissingChunksInFiler")
  79. tempPath := fsckCommand.String("tempPath", path.Join(os.TempDir()), "path for temporary idx files")
  80. cutoffTimeAgo := fsckCommand.Duration("cutoffTimeAgo", 5*time.Minute, "only include entries on volume servers before this cutoff time to check orphan chunks")
  81. modifyTimeAgo := fsckCommand.Duration("modifyTimeAgo", 0, "only include entries after this modify time to check orphan chunks")
  82. c.verifyNeedle = fsckCommand.Bool("verifyNeedles", false, "check needles status from volume server")
  83. if err = fsckCommand.Parse(args); err != nil {
  84. return nil
  85. }
  86. if err = commandEnv.confirmIsLocked(args); err != nil {
  87. return
  88. }
  89. c.volumeIds = make(map[uint32]bool)
  90. if *volumeIds != "" {
  91. for _, volumeIdStr := range strings.Split(*volumeIds, ",") {
  92. if volumeIdInt, err := strconv.ParseUint(volumeIdStr, 10, 32); err == nil {
  93. c.volumeIds[uint32(volumeIdInt)] = true
  94. } else {
  95. return fmt.Errorf("parse volumeId string %s to int: %v", volumeIdStr, err)
  96. }
  97. }
  98. }
  99. c.env = commandEnv
  100. c.writer = writer
  101. c.bucketsPath, err = readFilerBucketsPath(commandEnv)
  102. if err != nil {
  103. return fmt.Errorf("read filer buckets path: %v", err)
  104. }
  105. // create a temp folder
  106. c.tempFolder, err = os.MkdirTemp(*tempPath, "sw_fsck")
  107. if err != nil {
  108. return fmt.Errorf("failed to create temp folder: %v", err)
  109. }
  110. if *c.verbose {
  111. fmt.Fprintf(c.writer, "working directory: %s\n", c.tempFolder)
  112. }
  113. defer os.RemoveAll(c.tempFolder)
  114. // collect all volume id locations
  115. dataNodeVolumeIdToVInfo, err := c.collectVolumeIds()
  116. if err != nil {
  117. return fmt.Errorf("failed to collect all volume locations: %v", err)
  118. }
  119. if err != nil {
  120. return fmt.Errorf("read filer buckets path: %v", err)
  121. }
  122. var collectCutoffFromAtNs int64 = 0
  123. if cutoffTimeAgo.Seconds() != 0 {
  124. collectCutoffFromAtNs = time.Now().Add(-*cutoffTimeAgo).UnixNano()
  125. }
  126. var collectModifyFromAtNs int64 = 0
  127. if modifyTimeAgo.Seconds() != 0 {
  128. collectModifyFromAtNs = time.Now().Add(-*modifyTimeAgo).UnixNano()
  129. }
  130. // collect each volume file ids
  131. eg, gCtx := errgroup.WithContext(context.Background())
  132. _ = gCtx
  133. for _dataNodeId, _volumeIdToVInfo := range dataNodeVolumeIdToVInfo {
  134. dataNodeId, volumeIdToVInfo := _dataNodeId, _volumeIdToVInfo
  135. eg.Go(func() error {
  136. for volumeId, vinfo := range volumeIdToVInfo {
  137. if len(c.volumeIds) > 0 {
  138. if _, ok := c.volumeIds[volumeId]; !ok {
  139. delete(volumeIdToVInfo, volumeId)
  140. continue
  141. }
  142. }
  143. if *c.collection != "" && vinfo.collection != *c.collection {
  144. delete(volumeIdToVInfo, volumeId)
  145. continue
  146. }
  147. err = c.collectOneVolumeFileIds(dataNodeId, volumeId, vinfo, uint64(collectModifyFromAtNs), uint64(collectCutoffFromAtNs))
  148. if err != nil {
  149. return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, err)
  150. }
  151. }
  152. if *c.verbose {
  153. fmt.Fprintf(c.writer, "dn %+v filtred %d volumes and locations.\n", dataNodeId, len(dataNodeVolumeIdToVInfo[dataNodeId]))
  154. }
  155. return nil
  156. })
  157. }
  158. err = eg.Wait()
  159. if err != nil {
  160. fmt.Fprintf(c.writer, "got error: %v", err)
  161. return err
  162. }
  163. if *c.findMissingChunksInFiler {
  164. // collect all filer file ids and paths
  165. if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, *purgeAbsent, collectModifyFromAtNs, collectCutoffFromAtNs); err != nil {
  166. return fmt.Errorf("collectFilerFileIdAndPaths: %v", err)
  167. }
  168. for dataNodeId, volumeIdToVInfo := range dataNodeVolumeIdToVInfo {
  169. // for each volume, check filer file ids
  170. if err = c.findFilerChunksMissingInVolumeServers(volumeIdToVInfo, dataNodeId, *applyPurging); err != nil {
  171. return fmt.Errorf("findFilerChunksMissingInVolumeServers: %v", err)
  172. }
  173. }
  174. } else {
  175. // collect all filer file ids
  176. if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, false, 0, 0); err != nil {
  177. return fmt.Errorf("failed to collect file ids from filer: %v", err)
  178. }
  179. // volume file ids subtract filer file ids
  180. if err = c.findExtraChunksInVolumeServers(dataNodeVolumeIdToVInfo, *applyPurging); err != nil {
  181. return fmt.Errorf("findExtraChunksInVolumeServers: %v", err)
  182. }
  183. }
  184. return nil
  185. }
  186. func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, collectModifyFromAtNs int64, cutoffFromAtNs int64) error {
  187. if *c.verbose {
  188. fmt.Fprintf(c.writer, "checking each file from filer path %s...\n", c.getCollectFilerFilePath())
  189. }
  190. files := make(map[uint32]*os.File)
  191. for _, volumeIdToServer := range dataNodeVolumeIdToVInfo {
  192. for vid := range volumeIdToServer {
  193. if _, ok := files[vid]; ok {
  194. continue
  195. }
  196. dst, openErr := os.OpenFile(getFilerFileIdFile(c.tempFolder, vid), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
  197. if openErr != nil {
  198. return fmt.Errorf("failed to create file %s: %v", getFilerFileIdFile(c.tempFolder, vid), openErr)
  199. }
  200. files[vid] = dst
  201. }
  202. }
  203. defer func() {
  204. for _, f := range files {
  205. f.Close()
  206. }
  207. }()
  208. return doTraverseBfsAndSaving(c.env, c.writer, c.getCollectFilerFilePath(), false,
  209. func(entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
  210. if *c.verbose && entry.Entry.IsDirectory {
  211. fmt.Fprintf(c.writer, "checking directory %s\n", util.NewFullPath(entry.Dir, entry.Entry.Name))
  212. }
  213. dataChunks, manifestChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.GetChunks(), 0, math.MaxInt64)
  214. if resolveErr != nil {
  215. return fmt.Errorf("failed to ResolveChunkManifest: %+v", resolveErr)
  216. }
  217. dataChunks = append(dataChunks, manifestChunks...)
  218. for _, chunk := range dataChunks {
  219. if cutoffFromAtNs != 0 && chunk.ModifiedTsNs > cutoffFromAtNs {
  220. continue
  221. }
  222. if collectModifyFromAtNs != 0 && chunk.ModifiedTsNs < collectModifyFromAtNs {
  223. continue
  224. }
  225. outputChan <- &Item{
  226. vid: chunk.Fid.VolumeId,
  227. fileKey: chunk.Fid.FileKey,
  228. cookie: chunk.Fid.Cookie,
  229. path: util.NewFullPath(entry.Dir, entry.Entry.Name),
  230. }
  231. }
  232. return nil
  233. },
  234. func(outputChan chan interface{}) {
  235. buffer := make([]byte, readbufferSize)
  236. for item := range outputChan {
  237. i := item.(*Item)
  238. if f, ok := files[i.vid]; ok {
  239. util.Uint64toBytes(buffer, i.fileKey)
  240. util.Uint32toBytes(buffer[8:], i.cookie)
  241. util.Uint32toBytes(buffer[12:], uint32(len(i.path)))
  242. f.Write(buffer)
  243. f.Write([]byte(i.path))
  244. } else if *c.findMissingChunksInFiler && len(c.volumeIds) == 0 {
  245. fmt.Fprintf(c.writer, "%d,%x%08x %s volume not found\n", i.vid, i.fileKey, i.cookie, i.path)
  246. if purgeAbsent {
  247. fmt.Printf("deleting path %s after volume not found", i.path)
  248. c.httpDelete(i.path)
  249. }
  250. }
  251. }
  252. })
  253. }
  254. func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInfo map[uint32]VInfo, dataNodeId string, applyPurging bool) error {
  255. for volumeId, vinfo := range volumeIdToVInfo {
  256. checkErr := c.oneVolumeFileIdsCheckOneVolume(dataNodeId, volumeId, applyPurging)
  257. if checkErr != nil {
  258. return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, checkErr)
  259. }
  260. }
  261. return nil
  262. }
  263. func (c *commandVolumeFsck) findExtraChunksInVolumeServers(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, applyPurging bool) error {
  264. var totalInUseCount, totalOrphanChunkCount, totalOrphanDataSize uint64
  265. volumeIdOrphanFileIds := make(map[uint32]map[string]bool)
  266. isSeveralReplicas := make(map[uint32]bool)
  267. isEcVolumeReplicas := make(map[uint32]bool)
  268. isReadOnlyReplicas := make(map[uint32]bool)
  269. serverReplicas := make(map[uint32][]pb.ServerAddress)
  270. for dataNodeId, volumeIdToVInfo := range dataNodeVolumeIdToVInfo {
  271. for volumeId, vinfo := range volumeIdToVInfo {
  272. inUseCount, orphanFileIds, orphanDataSize, checkErr := c.oneVolumeFileIdsSubtractFilerFileIds(dataNodeId, volumeId, &vinfo)
  273. if checkErr != nil {
  274. return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, checkErr)
  275. }
  276. isSeveralReplicas[volumeId] = false
  277. if _, found := volumeIdOrphanFileIds[volumeId]; !found {
  278. volumeIdOrphanFileIds[volumeId] = make(map[string]bool)
  279. } else {
  280. isSeveralReplicas[volumeId] = true
  281. }
  282. for _, fid := range orphanFileIds {
  283. if isSeveralReplicas[volumeId] {
  284. if _, found := volumeIdOrphanFileIds[volumeId][fid]; !found {
  285. continue
  286. }
  287. }
  288. volumeIdOrphanFileIds[volumeId][fid] = isSeveralReplicas[volumeId]
  289. }
  290. totalInUseCount += inUseCount
  291. totalOrphanChunkCount += uint64(len(orphanFileIds))
  292. totalOrphanDataSize += orphanDataSize
  293. if *c.verbose {
  294. for _, fid := range orphanFileIds {
  295. fmt.Fprintf(c.writer, "%s:%s\n", vinfo.collection, fid)
  296. }
  297. }
  298. isEcVolumeReplicas[volumeId] = vinfo.isEcVolume
  299. if isReadOnly, found := isReadOnlyReplicas[volumeId]; !(found && isReadOnly) {
  300. isReadOnlyReplicas[volumeId] = vinfo.isReadOnly
  301. }
  302. serverReplicas[volumeId] = append(serverReplicas[volumeId], vinfo.server)
  303. }
  304. for volumeId, orphanReplicaFileIds := range volumeIdOrphanFileIds {
  305. if !(applyPurging && len(orphanReplicaFileIds) > 0) {
  306. continue
  307. }
  308. orphanFileIds := []string{}
  309. for fid, foundInAllReplicas := range orphanReplicaFileIds {
  310. if !isSeveralReplicas[volumeId] || *c.forcePurging || (isSeveralReplicas[volumeId] && foundInAllReplicas) {
  311. orphanFileIds = append(orphanFileIds, fid)
  312. }
  313. }
  314. if !(len(orphanFileIds) > 0) {
  315. continue
  316. }
  317. if *c.verbose {
  318. fmt.Fprintf(c.writer, "purging process for volume %d.\n", volumeId)
  319. }
  320. if isEcVolumeReplicas[volumeId] {
  321. fmt.Fprintf(c.writer, "skip purging for Erasure Coded volume %d.\n", volumeId)
  322. continue
  323. }
  324. for _, server := range serverReplicas[volumeId] {
  325. needleVID := needle.VolumeId(volumeId)
  326. if isReadOnlyReplicas[volumeId] {
  327. err := markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, true)
  328. if err != nil {
  329. return fmt.Errorf("mark volume %d read/write: %v", volumeId, err)
  330. }
  331. fmt.Fprintf(c.writer, "temporarily marked %d on server %v writable for forced purge\n", volumeId, server)
  332. defer markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, false)
  333. fmt.Fprintf(c.writer, "marked %d on server %v writable for forced purge\n", volumeId, server)
  334. }
  335. if *c.verbose {
  336. fmt.Fprintf(c.writer, "purging files from volume %d\n", volumeId)
  337. }
  338. if err := c.purgeFileIdsForOneVolume(volumeId, orphanFileIds); err != nil {
  339. return fmt.Errorf("purging volume %d: %v", volumeId, err)
  340. }
  341. }
  342. }
  343. }
  344. if !applyPurging {
  345. pct := float64(totalOrphanChunkCount*100) / (float64(totalOrphanChunkCount + totalInUseCount))
  346. fmt.Fprintf(c.writer, "\nTotal\t\tentries:%d\torphan:%d\t%.2f%%\t%dB\n",
  347. totalOrphanChunkCount+totalInUseCount, totalOrphanChunkCount, pct, totalOrphanDataSize)
  348. fmt.Fprintf(c.writer, "This could be normal if multiple filers or no filers are used.\n")
  349. }
  350. if totalOrphanChunkCount == 0 {
  351. fmt.Fprintf(c.writer, "no orphan data\n")
  352. }
  353. return nil
  354. }
  355. func (c *commandVolumeFsck) collectOneVolumeFileIds(dataNodeId string, volumeId uint32, vinfo VInfo, modifyFrom uint64, cutoffFrom uint64) error {
  356. if *c.verbose {
  357. fmt.Fprintf(c.writer, "collecting volume %d file ids from %s ...\n", volumeId, vinfo.server)
  358. }
  359. return operation.WithVolumeServerClient(false, vinfo.server, c.env.option.GrpcDialOption,
  360. func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  361. ext := ".idx"
  362. if vinfo.isEcVolume {
  363. ext = ".ecx"
  364. }
  365. copyFileClient, err := volumeServerClient.CopyFile(context.Background(), &volume_server_pb.CopyFileRequest{
  366. VolumeId: volumeId,
  367. Ext: ext,
  368. CompactionRevision: math.MaxUint32,
  369. StopOffset: math.MaxInt64,
  370. Collection: vinfo.collection,
  371. IsEcVolume: vinfo.isEcVolume,
  372. IgnoreSourceFileNotFound: false,
  373. })
  374. if err != nil {
  375. return fmt.Errorf("failed to start copying volume %d%s: %v", volumeId, ext, err)
  376. }
  377. var buf bytes.Buffer
  378. for {
  379. resp, err := copyFileClient.Recv()
  380. if errors.Is(err, io.EOF) {
  381. break
  382. }
  383. if err != nil {
  384. return err
  385. }
  386. buf.Write(resp.FileContent)
  387. }
  388. if !vinfo.isReadOnly && (modifyFrom != 0 || cutoffFrom != 0) {
  389. index, err := idx.FirstInvalidIndex(buf.Bytes(),
  390. func(key types.NeedleId, offset types.Offset, size types.Size) (bool, error) {
  391. resp, err := volumeServerClient.ReadNeedleMeta(context.Background(), &volume_server_pb.ReadNeedleMetaRequest{
  392. VolumeId: volumeId,
  393. NeedleId: uint64(key),
  394. Offset: offset.ToActualOffset(),
  395. Size: int32(size),
  396. })
  397. if err != nil {
  398. return false, fmt.Errorf("read needle meta with id %d from volume %d: %v", key, volumeId, err)
  399. }
  400. if (modifyFrom == 0 || modifyFrom <= resp.AppendAtNs) && (cutoffFrom == 0 || resp.AppendAtNs <= cutoffFrom) {
  401. return true, nil
  402. }
  403. return false, nil
  404. })
  405. if err != nil {
  406. fmt.Fprintf(c.writer, "Failed to search for last valid index on volume %d with error %v\n", volumeId, err)
  407. } else {
  408. buf.Truncate(index * types.NeedleMapEntrySize)
  409. }
  410. }
  411. idxFilename := getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)
  412. err = writeToFile(buf.Bytes(), idxFilename)
  413. if err != nil {
  414. return fmt.Errorf("failed to copy %d%s from %s: %v", volumeId, ext, vinfo.server, err)
  415. }
  416. return nil
  417. })
  418. }
  419. type Item struct {
  420. vid uint32
  421. fileKey uint64
  422. cookie uint32
  423. path util.FullPath
  424. }
  425. func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleId types.NeedleId, itemPath util.FullPath)) error {
  426. fp, err := os.Open(getFilerFileIdFile(c.tempFolder, volumeId))
  427. if err != nil {
  428. return err
  429. }
  430. defer fp.Close()
  431. br := bufio.NewReader(fp)
  432. buffer := make([]byte, readbufferSize)
  433. var readSize int
  434. var readErr error
  435. item := &Item{vid: volumeId}
  436. for {
  437. readSize, readErr = io.ReadFull(br, buffer)
  438. if errors.Is(readErr, io.EOF) {
  439. break
  440. }
  441. if readErr != nil {
  442. return readErr
  443. }
  444. if readSize != readbufferSize {
  445. return fmt.Errorf("readSize mismatch")
  446. }
  447. item.fileKey = util.BytesToUint64(buffer[:8])
  448. item.cookie = util.BytesToUint32(buffer[8:12])
  449. pathSize := util.BytesToUint32(buffer[12:16])
  450. pathBytes := make([]byte, int(pathSize))
  451. n, err := io.ReadFull(br, pathBytes)
  452. if err != nil {
  453. fmt.Fprintf(c.writer, "%d,%x%08x in unexpected error: %v\n", volumeId, item.fileKey, item.cookie, err)
  454. }
  455. if n != int(pathSize) {
  456. fmt.Fprintf(c.writer, "%d,%x%08x %d unexpected file name size %d\n", volumeId, item.fileKey, item.cookie, pathSize, n)
  457. }
  458. item.path = util.FullPath(pathBytes)
  459. needleId := types.NeedleId(item.fileKey)
  460. fn(needleId, item.path)
  461. }
  462. return nil
  463. }
  464. func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, volumeId uint32, applyPurging bool) (err error) {
  465. if *c.verbose {
  466. fmt.Fprintf(c.writer, "find missing file chunks in dataNodeId %s volume %d ...\n", dataNodeId, volumeId)
  467. }
  468. db := needle_map.NewMemDb()
  469. defer db.Close()
  470. if err = db.LoadFromIdx(getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)); err != nil {
  471. return
  472. }
  473. if err = c.readFilerFileIdFile(volumeId, func(needleId types.NeedleId, itemPath util.FullPath) {
  474. if _, found := db.Get(needleId); !found {
  475. fmt.Fprintf(c.writer, "%s\n", itemPath)
  476. if applyPurging {
  477. c.httpDelete(itemPath)
  478. }
  479. }
  480. }); err != nil {
  481. return
  482. }
  483. return nil
  484. }
  485. func (c *commandVolumeFsck) httpDelete(path util.FullPath) {
  486. req, err := http.NewRequest(http.MethodDelete, "", nil)
  487. req.URL = &url.URL{
  488. Scheme: "http",
  489. Host: c.env.option.FilerAddress.ToHttpAddress(),
  490. Path: string(path),
  491. }
  492. if *c.verbose {
  493. fmt.Fprintf(c.writer, "full HTTP delete request to be sent: %v\n", req)
  494. }
  495. if err != nil {
  496. fmt.Fprintf(c.writer, "HTTP delete request error: %v\n", err)
  497. }
  498. client := &http.Client{}
  499. resp, err := client.Do(req)
  500. if err != nil {
  501. fmt.Fprintf(c.writer, "DELETE fetch error: %v\n", err)
  502. }
  503. defer resp.Body.Close()
  504. _, err = io.ReadAll(resp.Body)
  505. if err != nil {
  506. fmt.Fprintf(c.writer, "DELETE response error: %v\n", err)
  507. }
  508. if *c.verbose {
  509. fmt.Fprintln(c.writer, "delete response Status : ", resp.Status)
  510. fmt.Fprintln(c.writer, "delete response Headers : ", resp.Header)
  511. }
  512. }
  513. func (c *commandVolumeFsck) oneVolumeFileIdsSubtractFilerFileIds(dataNodeId string, volumeId uint32, vinfo *VInfo) (inUseCount uint64, orphanFileIds []string, orphanDataSize uint64, err error) {
  514. volumeFileIdDb := needle_map.NewMemDb()
  515. defer volumeFileIdDb.Close()
  516. if err = volumeFileIdDb.LoadFromIdx(getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)); err != nil {
  517. err = fmt.Errorf("failed to LoadFromIdx %+v", err)
  518. return
  519. }
  520. if err = c.readFilerFileIdFile(volumeId, func(filerNeedleId types.NeedleId, itemPath util.FullPath) {
  521. inUseCount++
  522. if *c.verifyNeedle {
  523. if needleValue, ok := volumeFileIdDb.Get(filerNeedleId); ok && !needleValue.Size.IsDeleted() {
  524. if _, err := readNeedleStatus(c.env.option.GrpcDialOption, vinfo.server, volumeId, *needleValue); err != nil {
  525. // files may be deleted during copying filesIds
  526. if !strings.Contains(err.Error(), storage.ErrorDeleted.Error()) {
  527. fmt.Fprintf(c.writer, "failed to read %d:%s needle status of file %s: %+v\n",
  528. volumeId, filerNeedleId.String(), itemPath, err)
  529. if *c.forcePurging {
  530. return
  531. }
  532. }
  533. }
  534. }
  535. }
  536. if err = volumeFileIdDb.Delete(filerNeedleId); err != nil && *c.verbose {
  537. fmt.Fprintf(c.writer, "failed to nm.delete %s(%+v): %+v", itemPath, filerNeedleId, err)
  538. }
  539. }); err != nil {
  540. err = fmt.Errorf("failed to readFilerFileIdFile %+v", err)
  541. return
  542. }
  543. var orphanFileCount uint64
  544. if err = volumeFileIdDb.AscendingVisit(func(n needle_map.NeedleValue) error {
  545. if n.Size.IsDeleted() {
  546. return nil
  547. }
  548. orphanFileIds = append(orphanFileIds, n.Key.FileId(volumeId))
  549. orphanFileCount++
  550. orphanDataSize += uint64(n.Size)
  551. return nil
  552. }); err != nil {
  553. err = fmt.Errorf("failed to AscendingVisit %+v", err)
  554. return
  555. }
  556. if orphanFileCount > 0 {
  557. pct := float64(orphanFileCount*100) / (float64(orphanFileCount + inUseCount))
  558. fmt.Fprintf(c.writer, "dataNode:%s\tvolume:%d\tentries:%d\torphan:%d\t%.2f%%\t%dB\n",
  559. dataNodeId, volumeId, orphanFileCount+inUseCount, orphanFileCount, pct, orphanDataSize)
  560. }
  561. return
  562. }
  563. type VInfo struct {
  564. server pb.ServerAddress
  565. collection string
  566. isEcVolume bool
  567. isReadOnly bool
  568. }
  569. func (c *commandVolumeFsck) collectVolumeIds() (volumeIdToServer map[string]map[uint32]VInfo, err error) {
  570. if *c.verbose {
  571. fmt.Fprintf(c.writer, "collecting volume id and locations from master ...\n")
  572. }
  573. volumeIdToServer = make(map[string]map[uint32]VInfo)
  574. // collect topology information
  575. topologyInfo, _, err := collectTopologyInfo(c.env, 0)
  576. if err != nil {
  577. return
  578. }
  579. eachDataNode(topologyInfo, func(dc string, rack RackId, t *master_pb.DataNodeInfo) {
  580. var volumeCount, ecShardCount int
  581. dataNodeId := t.GetId()
  582. for _, diskInfo := range t.DiskInfos {
  583. if _, ok := volumeIdToServer[dataNodeId]; !ok {
  584. volumeIdToServer[dataNodeId] = make(map[uint32]VInfo)
  585. }
  586. for _, vi := range diskInfo.VolumeInfos {
  587. volumeIdToServer[dataNodeId][vi.Id] = VInfo{
  588. server: pb.NewServerAddressFromDataNode(t),
  589. collection: vi.Collection,
  590. isEcVolume: false,
  591. isReadOnly: vi.ReadOnly,
  592. }
  593. volumeCount += 1
  594. }
  595. for _, ecShardInfo := range diskInfo.EcShardInfos {
  596. volumeIdToServer[dataNodeId][ecShardInfo.Id] = VInfo{
  597. server: pb.NewServerAddressFromDataNode(t),
  598. collection: ecShardInfo.Collection,
  599. isEcVolume: true,
  600. isReadOnly: true,
  601. }
  602. ecShardCount += 1
  603. }
  604. }
  605. if *c.verbose {
  606. fmt.Fprintf(c.writer, "dn %+v collected %d volumes and %d ec shards.\n", dataNodeId, volumeCount, ecShardCount)
  607. }
  608. })
  609. return
  610. }
  611. func (c *commandVolumeFsck) purgeFileIdsForOneVolume(volumeId uint32, fileIds []string) (err error) {
  612. fmt.Fprintf(c.writer, "purging orphan data for volume %d...\n", volumeId)
  613. locations, found := c.env.MasterClient.GetLocations(volumeId)
  614. if !found {
  615. return fmt.Errorf("failed to find volume %d locations", volumeId)
  616. }
  617. resultChan := make(chan []*volume_server_pb.DeleteResult, len(locations))
  618. var wg sync.WaitGroup
  619. for _, location := range locations {
  620. wg.Add(1)
  621. go func(server pb.ServerAddress, fidList []string) {
  622. defer wg.Done()
  623. if deleteResults, deleteErr := operation.DeleteFilesAtOneVolumeServer(server, c.env.option.GrpcDialOption, fidList, false); deleteErr != nil {
  624. err = deleteErr
  625. } else if deleteResults != nil {
  626. resultChan <- deleteResults
  627. }
  628. }(location.ServerAddress(), fileIds)
  629. }
  630. wg.Wait()
  631. close(resultChan)
  632. for results := range resultChan {
  633. for _, result := range results {
  634. if result.Error != "" {
  635. fmt.Fprintf(c.writer, "purge error: %s\n", result.Error)
  636. }
  637. }
  638. }
  639. return
  640. }
  641. func (c *commandVolumeFsck) getCollectFilerFilePath() string {
  642. if *c.collection != "" {
  643. return fmt.Sprintf("%s/%s", c.bucketsPath, *c.collection)
  644. }
  645. return "/"
  646. }
  647. func getVolumeFileIdFile(tempFolder string, dataNodeid string, vid uint32) string {
  648. return filepath.Join(tempFolder, fmt.Sprintf("%s_%d.idx", dataNodeid, vid))
  649. }
  650. func getFilerFileIdFile(tempFolder string, vid uint32) string {
  651. return filepath.Join(tempFolder, fmt.Sprintf("%d.fid", vid))
  652. }
  653. func writeToFile(bytes []byte, fileName string) error {
  654. flags := os.O_WRONLY | os.O_CREATE | os.O_TRUNC
  655. dst, err := os.OpenFile(fileName, flags, 0644)
  656. if err != nil {
  657. return nil
  658. }
  659. defer dst.Close()
  660. dst.Write(bytes)
  661. return nil
  662. }