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.

553 lines
17 KiB

4 years ago
5 years ago
4 years ago
5 years ago
5 years ago
4 years ago
3 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
5 years ago
  1. package shell
  2. import (
  3. "bufio"
  4. "context"
  5. "flag"
  6. "fmt"
  7. "io"
  8. "math"
  9. "os"
  10. "path/filepath"
  11. "sync"
  12. "github.com/chrislusf/seaweedfs/weed/filer"
  13. "github.com/chrislusf/seaweedfs/weed/operation"
  14. "github.com/chrislusf/seaweedfs/weed/pb"
  15. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  16. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  17. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  18. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  19. "github.com/chrislusf/seaweedfs/weed/storage/needle_map"
  20. "github.com/chrislusf/seaweedfs/weed/storage/types"
  21. "github.com/chrislusf/seaweedfs/weed/util"
  22. )
  23. func init() {
  24. Commands = append(Commands, &commandVolumeFsck{})
  25. }
  26. type commandVolumeFsck struct {
  27. env *CommandEnv
  28. }
  29. func (c *commandVolumeFsck) Name() string {
  30. return "volume.fsck"
  31. }
  32. func (c *commandVolumeFsck) Help() string {
  33. return `check all volumes to find entries not used by the filer
  34. Important assumption!!!
  35. the system is all used by one filer.
  36. This command works this way:
  37. 1. collect all file ids from all volumes, as set A
  38. 2. collect all file ids from the filer, as set B
  39. 3. find out the set A subtract B
  40. If -findMissingChunksInFiler is enabled, this works
  41. in a reverse way:
  42. 1. collect all file ids from all volumes, as set A
  43. 2. collect all file ids from the filer, as set B
  44. 3. find out the set B subtract A
  45. `
  46. }
  47. func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  48. fsckCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
  49. verbose := fsckCommand.Bool("v", false, "verbose mode")
  50. findMissingChunksInFiler := fsckCommand.Bool("findMissingChunksInFiler", false, "see \"help volume.fsck\"")
  51. findMissingChunksInFilerPath := fsckCommand.String("findMissingChunksInFilerPath", "/", "used together with findMissingChunksInFiler")
  52. applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, "<expert only> delete data not referenced by the filer")
  53. if err = fsckCommand.Parse(args); err != nil {
  54. return nil
  55. }
  56. if err = commandEnv.confirmIsLocked(args); err != nil {
  57. return
  58. }
  59. c.env = commandEnv
  60. // create a temp folder
  61. tempFolder, err := os.MkdirTemp("", "sw_fsck")
  62. if err != nil {
  63. return fmt.Errorf("failed to create temp folder: %v", err)
  64. }
  65. if *verbose {
  66. fmt.Fprintf(writer, "working directory: %s\n", tempFolder)
  67. }
  68. defer os.RemoveAll(tempFolder)
  69. // collect all volume id locations
  70. volumeIdToVInfo, err := c.collectVolumeIds(commandEnv, *verbose, writer)
  71. if err != nil {
  72. return fmt.Errorf("failed to collect all volume locations: %v", err)
  73. }
  74. // collect each volume file ids
  75. for volumeId, vinfo := range volumeIdToVInfo {
  76. err = c.collectOneVolumeFileIds(tempFolder, volumeId, vinfo, *verbose, writer)
  77. if err != nil {
  78. return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, err)
  79. }
  80. }
  81. if *findMissingChunksInFiler {
  82. // collect all filer file ids and paths
  83. if err = c.collectFilerFileIdAndPaths(volumeIdToVInfo, tempFolder, writer, *findMissingChunksInFilerPath, *verbose, applyPurging); err != nil {
  84. return fmt.Errorf("collectFilerFileIdAndPaths: %v", err)
  85. }
  86. // for each volume, check filer file ids
  87. if err = c.findFilerChunksMissingInVolumeServers(volumeIdToVInfo, tempFolder, writer, *verbose, applyPurging); err != nil {
  88. return fmt.Errorf("findFilerChunksMissingInVolumeServers: %v", err)
  89. }
  90. } else {
  91. // collect all filer file ids
  92. if err = c.collectFilerFileIds(tempFolder, volumeIdToVInfo, *verbose, writer); err != nil {
  93. return fmt.Errorf("failed to collect file ids from filer: %v", err)
  94. }
  95. // volume file ids substract filer file ids
  96. if err = c.findExtraChunksInVolumeServers(volumeIdToVInfo, tempFolder, writer, *verbose, applyPurging); err != nil {
  97. return fmt.Errorf("findExtraChunksInVolumeServers: %v", err)
  98. }
  99. }
  100. return nil
  101. }
  102. func (c *commandVolumeFsck) collectFilerFileIdAndPaths(volumeIdToServer map[uint32]VInfo, tempFolder string, writer io.Writer, filerPath string, verbose bool, applyPurging *bool) error {
  103. if verbose {
  104. fmt.Fprintf(writer, "checking each file from filer ...\n")
  105. }
  106. files := make(map[uint32]*os.File)
  107. for vid := range volumeIdToServer {
  108. dst, openErr := os.OpenFile(getFilerFileIdFile(tempFolder, vid), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
  109. if openErr != nil {
  110. return fmt.Errorf("failed to create file %s: %v", getFilerFileIdFile(tempFolder, vid), openErr)
  111. }
  112. files[vid] = dst
  113. }
  114. defer func() {
  115. for _, f := range files {
  116. f.Close()
  117. }
  118. }()
  119. type Item struct {
  120. vid uint32
  121. fileKey uint64
  122. cookie uint32
  123. path util.FullPath
  124. }
  125. return doTraverseBfsAndSaving(c.env, nil, filerPath, false, func(outputChan chan interface{}) {
  126. buffer := make([]byte, 16)
  127. for item := range outputChan {
  128. i := item.(*Item)
  129. if f, ok := files[i.vid]; ok {
  130. util.Uint64toBytes(buffer, i.fileKey)
  131. util.Uint32toBytes(buffer[8:], i.cookie)
  132. util.Uint32toBytes(buffer[12:], uint32(len(i.path)))
  133. f.Write(buffer)
  134. f.Write([]byte(i.path))
  135. // fmt.Fprintf(writer, "%d,%x%08x %d %s\n", i.vid, i.fileKey, i.cookie, len(i.path), i.path)
  136. } else {
  137. fmt.Fprintf(writer, "%d,%x%08x %s volume not found\n", i.vid, i.fileKey, i.cookie, i.path)
  138. }
  139. }
  140. }, func(entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
  141. if verbose && entry.Entry.IsDirectory {
  142. fmt.Fprintf(writer, "checking directory %s\n", util.NewFullPath(entry.Dir, entry.Entry.Name))
  143. }
  144. dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
  145. if resolveErr != nil {
  146. return nil
  147. }
  148. dChunks = append(dChunks, mChunks...)
  149. for _, chunk := range dChunks {
  150. outputChan <- &Item{
  151. vid: chunk.Fid.VolumeId,
  152. fileKey: chunk.Fid.FileKey,
  153. cookie: chunk.Fid.Cookie,
  154. path: util.NewFullPath(entry.Dir, entry.Entry.Name),
  155. }
  156. }
  157. return nil
  158. })
  159. }
  160. func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInfo map[uint32]VInfo, tempFolder string, writer io.Writer, verbose bool, applyPurging *bool) error {
  161. for volumeId, vinfo := range volumeIdToVInfo {
  162. checkErr := c.oneVolumeFileIdsCheckOneVolume(tempFolder, volumeId, writer, verbose)
  163. if checkErr != nil {
  164. return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, checkErr)
  165. }
  166. }
  167. return nil
  168. }
  169. func (c *commandVolumeFsck) findExtraChunksInVolumeServers(volumeIdToVInfo map[uint32]VInfo, tempFolder string, writer io.Writer, verbose bool, applyPurging *bool) error {
  170. var totalInUseCount, totalOrphanChunkCount, totalOrphanDataSize uint64
  171. for volumeId, vinfo := range volumeIdToVInfo {
  172. inUseCount, orphanFileIds, orphanDataSize, checkErr := c.oneVolumeFileIdsSubtractFilerFileIds(tempFolder, volumeId, writer, verbose)
  173. if checkErr != nil {
  174. return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, checkErr)
  175. }
  176. totalInUseCount += inUseCount
  177. totalOrphanChunkCount += uint64(len(orphanFileIds))
  178. totalOrphanDataSize += orphanDataSize
  179. if verbose {
  180. for _, fid := range orphanFileIds {
  181. fmt.Fprintf(writer, "%s\n", fid)
  182. }
  183. }
  184. if *applyPurging && len(orphanFileIds) > 0 {
  185. if vinfo.isEcVolume {
  186. fmt.Fprintf(writer, "Skip purging for Erasure Coded volume %d.\n", volumeId)
  187. continue
  188. }
  189. if vinfo.isReadOnly {
  190. fmt.Fprintf(writer, "Skip purging for read only volume %d.\n", volumeId)
  191. continue
  192. }
  193. if inUseCount == 0 {
  194. if err := deleteVolume(c.env.option.GrpcDialOption, needle.VolumeId(volumeId), vinfo.server); err != nil {
  195. return fmt.Errorf("delete volume %d: %v", volumeId, err)
  196. }
  197. } else {
  198. if err := c.purgeFileIdsForOneVolume(volumeId, orphanFileIds, writer); err != nil {
  199. return fmt.Errorf("purge for volume %d: %v", volumeId, err)
  200. }
  201. }
  202. }
  203. }
  204. if totalOrphanChunkCount == 0 {
  205. fmt.Fprintf(writer, "no orphan data\n")
  206. return nil
  207. }
  208. if !*applyPurging {
  209. pct := float64(totalOrphanChunkCount*100) / (float64(totalOrphanChunkCount + totalInUseCount))
  210. fmt.Fprintf(writer, "\nTotal\t\tentries:%d\torphan:%d\t%.2f%%\t%dB\n",
  211. totalOrphanChunkCount+totalInUseCount, totalOrphanChunkCount, pct, totalOrphanDataSize)
  212. fmt.Fprintf(writer, "This could be normal if multiple filers or no filers are used.\n")
  213. }
  214. return nil
  215. }
  216. func (c *commandVolumeFsck) collectOneVolumeFileIds(tempFolder string, volumeId uint32, vinfo VInfo, verbose bool, writer io.Writer) error {
  217. if verbose {
  218. fmt.Fprintf(writer, "collecting volume %d file ids from %s ...\n", volumeId, vinfo.server)
  219. }
  220. return operation.WithVolumeServerClient(false, vinfo.server, c.env.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  221. ext := ".idx"
  222. if vinfo.isEcVolume {
  223. ext = ".ecx"
  224. }
  225. copyFileClient, err := volumeServerClient.CopyFile(context.Background(), &volume_server_pb.CopyFileRequest{
  226. VolumeId: volumeId,
  227. Ext: ext,
  228. CompactionRevision: math.MaxUint32,
  229. StopOffset: math.MaxInt64,
  230. Collection: vinfo.collection,
  231. IsEcVolume: vinfo.isEcVolume,
  232. IgnoreSourceFileNotFound: false,
  233. })
  234. if err != nil {
  235. return fmt.Errorf("failed to start copying volume %d%s: %v", volumeId, ext, err)
  236. }
  237. err = writeToFile(copyFileClient, getVolumeFileIdFile(tempFolder, volumeId))
  238. if err != nil {
  239. return fmt.Errorf("failed to copy %d%s from %s: %v", volumeId, ext, vinfo.server, err)
  240. }
  241. return nil
  242. })
  243. }
  244. func (c *commandVolumeFsck) collectFilerFileIds(tempFolder string, volumeIdToServer map[uint32]VInfo, verbose bool, writer io.Writer) error {
  245. if verbose {
  246. fmt.Fprintf(writer, "collecting file ids from filer ...\n")
  247. }
  248. files := make(map[uint32]*os.File)
  249. for vid := range volumeIdToServer {
  250. dst, openErr := os.OpenFile(getFilerFileIdFile(tempFolder, vid), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
  251. if openErr != nil {
  252. return fmt.Errorf("failed to create file %s: %v", getFilerFileIdFile(tempFolder, vid), openErr)
  253. }
  254. files[vid] = dst
  255. }
  256. defer func() {
  257. for _, f := range files {
  258. f.Close()
  259. }
  260. }()
  261. type Item struct {
  262. vid uint32
  263. fileKey uint64
  264. }
  265. return doTraverseBfsAndSaving(c.env, nil, "/", false, func(outputChan chan interface{}) {
  266. buffer := make([]byte, 8)
  267. for item := range outputChan {
  268. i := item.(*Item)
  269. util.Uint64toBytes(buffer, i.fileKey)
  270. files[i.vid].Write(buffer)
  271. }
  272. }, func(entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
  273. dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
  274. if resolveErr != nil {
  275. if verbose {
  276. fmt.Fprintf(writer, "resolving manifest chunks in %s: %v\n", util.NewFullPath(entry.Dir, entry.Entry.Name), resolveErr)
  277. }
  278. return nil
  279. }
  280. dChunks = append(dChunks, mChunks...)
  281. for _, chunk := range dChunks {
  282. outputChan <- &Item{
  283. vid: chunk.Fid.VolumeId,
  284. fileKey: chunk.Fid.FileKey,
  285. }
  286. }
  287. return nil
  288. })
  289. }
  290. func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(tempFolder string, volumeId uint32, writer io.Writer, verbose bool) (err error) {
  291. if verbose {
  292. fmt.Fprintf(writer, "find missing file chuns in volume %d ...\n", volumeId)
  293. }
  294. db := needle_map.NewMemDb()
  295. defer db.Close()
  296. if err = db.LoadFromIdx(getVolumeFileIdFile(tempFolder, volumeId)); err != nil {
  297. return
  298. }
  299. file := getFilerFileIdFile(tempFolder, volumeId)
  300. fp, err := os.Open(file)
  301. if err != nil {
  302. return
  303. }
  304. defer fp.Close()
  305. type Item struct {
  306. fileKey uint64
  307. cookie uint32
  308. path util.FullPath
  309. }
  310. br := bufio.NewReader(fp)
  311. buffer := make([]byte, 16)
  312. item := &Item{}
  313. var readSize int
  314. for {
  315. readSize, err = io.ReadFull(br, buffer)
  316. if err != nil || readSize != 16 {
  317. if err == io.EOF {
  318. return nil
  319. } else {
  320. break
  321. }
  322. }
  323. item.fileKey = util.BytesToUint64(buffer[:8])
  324. item.cookie = util.BytesToUint32(buffer[8:12])
  325. pathSize := util.BytesToUint32(buffer[12:16])
  326. pathBytes := make([]byte, int(pathSize))
  327. n, err := io.ReadFull(br, pathBytes)
  328. if err != nil {
  329. fmt.Fprintf(writer, "%d,%x%08x in unexpected error: %v\n", volumeId, item.fileKey, item.cookie, err)
  330. }
  331. if n != int(pathSize) {
  332. fmt.Fprintf(writer, "%d,%x%08x %d unexpected file name size %d\n", volumeId, item.fileKey, item.cookie, pathSize, n)
  333. }
  334. item.path = util.FullPath(string(pathBytes))
  335. if _, found := db.Get(types.NeedleId(item.fileKey)); !found {
  336. fmt.Fprintf(writer, "%d,%x%08x in %s %d not found\n", volumeId, item.fileKey, item.cookie, item.path, pathSize)
  337. }
  338. }
  339. return
  340. }
  341. func (c *commandVolumeFsck) oneVolumeFileIdsSubtractFilerFileIds(tempFolder string, volumeId uint32, writer io.Writer, verbose bool) (inUseCount uint64, orphanFileIds []string, orphanDataSize uint64, err error) {
  342. db := needle_map.NewMemDb()
  343. defer db.Close()
  344. if err = db.LoadFromIdx(getVolumeFileIdFile(tempFolder, volumeId)); err != nil {
  345. return
  346. }
  347. filerFileIdsData, err := os.ReadFile(getFilerFileIdFile(tempFolder, volumeId))
  348. if err != nil {
  349. return
  350. }
  351. dataLen := len(filerFileIdsData)
  352. if dataLen%8 != 0 {
  353. return 0, nil, 0, fmt.Errorf("filer data is corrupted")
  354. }
  355. for i := 0; i < len(filerFileIdsData); i += 8 {
  356. fileKey := util.BytesToUint64(filerFileIdsData[i : i+8])
  357. db.Delete(types.NeedleId(fileKey))
  358. inUseCount++
  359. }
  360. var orphanFileCount uint64
  361. db.AscendingVisit(func(n needle_map.NeedleValue) error {
  362. // fmt.Printf("%d,%x\n", volumeId, n.Key)
  363. orphanFileIds = append(orphanFileIds, fmt.Sprintf("%d,%s00000000", volumeId, n.Key.String()))
  364. orphanFileCount++
  365. orphanDataSize += uint64(n.Size)
  366. return nil
  367. })
  368. if orphanFileCount > 0 {
  369. pct := float64(orphanFileCount*100) / (float64(orphanFileCount + inUseCount))
  370. fmt.Fprintf(writer, "volume:%d\tentries:%d\torphan:%d\t%.2f%%\t%dB\n",
  371. volumeId, orphanFileCount+inUseCount, orphanFileCount, pct, orphanDataSize)
  372. }
  373. return
  374. }
  375. type VInfo struct {
  376. server pb.ServerAddress
  377. collection string
  378. isEcVolume bool
  379. isReadOnly bool
  380. }
  381. func (c *commandVolumeFsck) collectVolumeIds(commandEnv *CommandEnv, verbose bool, writer io.Writer) (volumeIdToServer map[uint32]VInfo, err error) {
  382. if verbose {
  383. fmt.Fprintf(writer, "collecting volume id and locations from master ...\n")
  384. }
  385. volumeIdToServer = make(map[uint32]VInfo)
  386. // collect topology information
  387. topologyInfo, _, err := collectTopologyInfo(commandEnv)
  388. if err != nil {
  389. return
  390. }
  391. eachDataNode(topologyInfo, func(dc string, rack RackId, t *master_pb.DataNodeInfo) {
  392. for _, diskInfo := range t.DiskInfos {
  393. for _, vi := range diskInfo.VolumeInfos {
  394. volumeIdToServer[vi.Id] = VInfo{
  395. server: pb.NewServerAddressFromDataNode(t),
  396. collection: vi.Collection,
  397. isEcVolume: false,
  398. isReadOnly: vi.ReadOnly,
  399. }
  400. }
  401. for _, ecShardInfo := range diskInfo.EcShardInfos {
  402. volumeIdToServer[ecShardInfo.Id] = VInfo{
  403. server: pb.NewServerAddressFromDataNode(t),
  404. collection: ecShardInfo.Collection,
  405. isEcVolume: true,
  406. isReadOnly: true,
  407. }
  408. }
  409. }
  410. })
  411. if verbose {
  412. fmt.Fprintf(writer, "collected %d volumes and locations.\n", len(volumeIdToServer))
  413. }
  414. return
  415. }
  416. func (c *commandVolumeFsck) purgeFileIdsForOneVolume(volumeId uint32, fileIds []string, writer io.Writer) (err error) {
  417. fmt.Fprintf(writer, "purging orphan data for volume %d...\n", volumeId)
  418. locations, found := c.env.MasterClient.GetLocations(volumeId)
  419. if !found {
  420. return fmt.Errorf("failed to find volume %d locations", volumeId)
  421. }
  422. resultChan := make(chan []*volume_server_pb.DeleteResult, len(locations))
  423. var wg sync.WaitGroup
  424. for _, location := range locations {
  425. wg.Add(1)
  426. go func(server pb.ServerAddress, fidList []string) {
  427. defer wg.Done()
  428. if deleteResults, deleteErr := operation.DeleteFilesAtOneVolumeServer(server, c.env.option.GrpcDialOption, fidList, false); deleteErr != nil {
  429. err = deleteErr
  430. } else if deleteResults != nil {
  431. resultChan <- deleteResults
  432. }
  433. }(location.ServerAddress(), fileIds)
  434. }
  435. wg.Wait()
  436. close(resultChan)
  437. for results := range resultChan {
  438. for _, result := range results {
  439. if result.Error != "" {
  440. fmt.Fprintf(writer, "purge error: %s\n", result.Error)
  441. }
  442. }
  443. }
  444. return
  445. }
  446. func getVolumeFileIdFile(tempFolder string, vid uint32) string {
  447. return filepath.Join(tempFolder, fmt.Sprintf("%d.idx", vid))
  448. }
  449. func getFilerFileIdFile(tempFolder string, vid uint32) string {
  450. return filepath.Join(tempFolder, fmt.Sprintf("%d.fid", vid))
  451. }
  452. func writeToFile(client volume_server_pb.VolumeServer_CopyFileClient, fileName string) error {
  453. flags := os.O_WRONLY | os.O_CREATE | os.O_TRUNC
  454. dst, err := os.OpenFile(fileName, flags, 0644)
  455. if err != nil {
  456. return nil
  457. }
  458. defer dst.Close()
  459. for {
  460. resp, receiveErr := client.Recv()
  461. if receiveErr == io.EOF {
  462. break
  463. }
  464. if receiveErr != nil {
  465. return fmt.Errorf("receiving %s: %v", fileName, receiveErr)
  466. }
  467. dst.Write(resp.FileContent)
  468. }
  469. return nil
  470. }