filer_copy.go 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608
  1. package command
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "net/http"
  7. "os"
  8. "path/filepath"
  9. "strconv"
  10. "strings"
  11. "sync"
  12. "time"
  13. "google.golang.org/grpc"
  14. "github.com/seaweedfs/seaweedfs/weed/filer"
  15. "github.com/seaweedfs/seaweedfs/weed/operation"
  16. "github.com/seaweedfs/seaweedfs/weed/pb"
  17. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  18. "github.com/seaweedfs/seaweedfs/weed/security"
  19. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  20. "github.com/seaweedfs/seaweedfs/weed/util"
  21. "github.com/seaweedfs/seaweedfs/weed/util/grace"
  22. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  23. )
  24. var (
  25. copy CopyOptions
  26. waitGroup sync.WaitGroup
  27. )
  28. type CopyOptions struct {
  29. include *string
  30. replication *string
  31. collection *string
  32. ttl *string
  33. diskType *string
  34. maxMB *int
  35. masterClient *wdclient.MasterClient
  36. concurrentFiles *int
  37. concurrentChunks *int
  38. grpcDialOption grpc.DialOption
  39. masters []string
  40. cipher bool
  41. ttlSec int32
  42. checkSize *bool
  43. verbose *bool
  44. volumeServerAccess *string
  45. }
  46. func init() {
  47. cmdFilerCopy.Run = runCopy // break init cycle
  48. cmdFilerCopy.IsDebug = cmdFilerCopy.Flag.Bool("debug", false, "verbose debug information")
  49. copy.include = cmdFilerCopy.Flag.String("include", "", "pattens of files to copy, e.g., *.pdf, *.html, ab?d.txt, works together with -dir")
  50. copy.replication = cmdFilerCopy.Flag.String("replication", "", "replication type")
  51. copy.collection = cmdFilerCopy.Flag.String("collection", "", "optional collection name")
  52. copy.ttl = cmdFilerCopy.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
  53. copy.diskType = cmdFilerCopy.Flag.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
  54. copy.maxMB = cmdFilerCopy.Flag.Int("maxMB", 4, "split files larger than the limit")
  55. copy.concurrentFiles = cmdFilerCopy.Flag.Int("c", 8, "concurrent file copy goroutines")
  56. copy.concurrentChunks = cmdFilerCopy.Flag.Int("concurrentChunks", 8, "concurrent chunk copy goroutines for each file")
  57. copy.checkSize = cmdFilerCopy.Flag.Bool("check.size", false, "copy when the target file size is different from the source file")
  58. copy.verbose = cmdFilerCopy.Flag.Bool("verbose", false, "print out details during copying")
  59. copy.volumeServerAccess = cmdFilerCopy.Flag.String("volumeServerAccess", "direct", "access volume servers by [direct|publicUrl]")
  60. }
  61. var cmdFilerCopy = &Command{
  62. UsageLine: "filer.copy file_or_dir1 [file_or_dir2 file_or_dir3] http://localhost:8888/path/to/a/folder/",
  63. Short: "copy one or a list of files to a filer folder",
  64. Long: `copy one or a list of files, or batch copy one whole folder recursively, to a filer folder
  65. It can copy one or a list of files or folders.
  66. If copying a whole folder recursively:
  67. All files under the folder and sub folders will be copied.
  68. Optional parameter "-include" allows you to specify the file name patterns.
  69. If "maxMB" is set to a positive number, files larger than it would be split into chunks.
  70. `,
  71. }
  72. func runCopy(cmd *Command, args []string) bool {
  73. util.LoadSecurityConfiguration()
  74. if len(args) <= 1 {
  75. return false
  76. }
  77. filerDestination := args[len(args)-1]
  78. fileOrDirs := args[0 : len(args)-1]
  79. filerAddress, urlPath, err := pb.ParseUrl(filerDestination)
  80. if err != nil {
  81. fmt.Printf("The last argument should be a URL on filer: %v\n", err)
  82. return false
  83. }
  84. if !strings.HasSuffix(urlPath, "/") {
  85. fmt.Printf("The last argument should be a folder and end with \"/\"\n")
  86. return false
  87. }
  88. copy.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
  89. masters, collection, replication, dirBuckets, maxMB, cipher, err := readFilerConfiguration(copy.grpcDialOption, filerAddress)
  90. if err != nil {
  91. fmt.Printf("read from filer %s: %v\n", filerAddress, err)
  92. return false
  93. }
  94. if strings.HasPrefix(urlPath, dirBuckets+"/") {
  95. restPath := urlPath[len(dirBuckets)+1:]
  96. if strings.Index(restPath, "/") > 0 {
  97. expectedBucket := restPath[:strings.Index(restPath, "/")]
  98. if *copy.collection == "" {
  99. *copy.collection = expectedBucket
  100. } else if *copy.collection != expectedBucket {
  101. fmt.Printf("destination %s uses collection \"%s\": unexpected collection \"%v\"\n", urlPath, expectedBucket, *copy.collection)
  102. return true
  103. }
  104. }
  105. }
  106. if *copy.collection == "" {
  107. *copy.collection = collection
  108. }
  109. if *copy.replication == "" {
  110. *copy.replication = replication
  111. }
  112. if *copy.maxMB == 0 {
  113. *copy.maxMB = int(maxMB)
  114. }
  115. copy.masters = masters
  116. copy.cipher = cipher
  117. ttl, err := needle.ReadTTL(*copy.ttl)
  118. if err != nil {
  119. fmt.Printf("parsing ttl %s: %v\n", *copy.ttl, err)
  120. return false
  121. }
  122. copy.ttlSec = int32(ttl.Minutes()) * 60
  123. if *cmdFilerCopy.IsDebug {
  124. grace.SetupProfiling("filer.copy.cpu.pprof", "filer.copy.mem.pprof")
  125. }
  126. fileCopyTaskChan := make(chan FileCopyTask, *copy.concurrentFiles)
  127. go func() {
  128. defer close(fileCopyTaskChan)
  129. for _, fileOrDir := range fileOrDirs {
  130. if err := genFileCopyTask(fileOrDir, urlPath, fileCopyTaskChan); err != nil {
  131. fmt.Fprintf(os.Stderr, "genFileCopyTask : %v\n", err)
  132. break
  133. }
  134. }
  135. }()
  136. for i := 0; i < *copy.concurrentFiles; i++ {
  137. waitGroup.Add(1)
  138. go func() {
  139. defer waitGroup.Done()
  140. worker := FileCopyWorker{
  141. options: &copy,
  142. filerAddress: filerAddress,
  143. signature: util.RandomInt32(),
  144. }
  145. if err := worker.copyFiles(fileCopyTaskChan); err != nil {
  146. fmt.Fprintf(os.Stderr, "copy file error: %v\n", err)
  147. return
  148. }
  149. }()
  150. }
  151. waitGroup.Wait()
  152. return true
  153. }
  154. func readFilerConfiguration(grpcDialOption grpc.DialOption, filerGrpcAddress pb.ServerAddress) (masters []string, collection, replication string, dirBuckets string, maxMB uint32, cipher bool, err error) {
  155. err = pb.WithGrpcFilerClient(false, 0, filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  156. resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
  157. if err != nil {
  158. return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err)
  159. }
  160. masters, collection, replication, maxMB = resp.Masters, resp.Collection, resp.Replication, resp.MaxMb
  161. dirBuckets = resp.DirBuckets
  162. cipher = resp.Cipher
  163. return nil
  164. })
  165. return
  166. }
  167. func genFileCopyTask(fileOrDir string, destPath string, fileCopyTaskChan chan FileCopyTask) error {
  168. fi, err := os.Stat(fileOrDir)
  169. if err != nil {
  170. fmt.Fprintf(os.Stderr, "Error: read file %s: %v\n", fileOrDir, err)
  171. return nil
  172. }
  173. mode := fi.Mode()
  174. uid, gid := util.GetFileUidGid(fi)
  175. fileSize := fi.Size()
  176. if mode.IsDir() {
  177. fileSize = 0
  178. }
  179. fileCopyTaskChan <- FileCopyTask{
  180. sourceLocation: fileOrDir,
  181. destinationUrlPath: destPath,
  182. fileSize: fileSize,
  183. fileMode: fi.Mode(),
  184. uid: uid,
  185. gid: gid,
  186. }
  187. if mode.IsDir() {
  188. files, _ := os.ReadDir(fileOrDir)
  189. for _, subFileOrDir := range files {
  190. cleanedDestDirectory := destPath + fi.Name()
  191. if err = genFileCopyTask(fileOrDir+"/"+subFileOrDir.Name(), cleanedDestDirectory+"/", fileCopyTaskChan); err != nil {
  192. return err
  193. }
  194. }
  195. }
  196. return nil
  197. }
  198. type FileCopyWorker struct {
  199. options *CopyOptions
  200. filerAddress pb.ServerAddress
  201. signature int32
  202. }
  203. func (worker *FileCopyWorker) copyFiles(fileCopyTaskChan chan FileCopyTask) error {
  204. for task := range fileCopyTaskChan {
  205. if err := worker.doEachCopy(task); err != nil {
  206. return err
  207. }
  208. }
  209. return nil
  210. }
  211. type FileCopyTask struct {
  212. sourceLocation string
  213. destinationUrlPath string
  214. fileSize int64
  215. fileMode os.FileMode
  216. uid uint32
  217. gid uint32
  218. }
  219. func (worker *FileCopyWorker) doEachCopy(task FileCopyTask) error {
  220. f, err := os.Open(task.sourceLocation)
  221. if err != nil {
  222. fmt.Printf("Failed to open file %s: %v\n", task.sourceLocation, err)
  223. if _, ok := err.(*os.PathError); ok {
  224. fmt.Printf("skipping %s\n", task.sourceLocation)
  225. return nil
  226. }
  227. return err
  228. }
  229. defer f.Close()
  230. // this is a regular file
  231. if *worker.options.include != "" {
  232. if ok, _ := filepath.Match(*worker.options.include, filepath.Base(task.sourceLocation)); !ok {
  233. return nil
  234. }
  235. }
  236. if shouldCopy, err := worker.checkExistingFileFirst(task, f); err != nil {
  237. return fmt.Errorf("check existing file: %v", err)
  238. } else if !shouldCopy {
  239. if *worker.options.verbose {
  240. fmt.Printf("skipping copied file: %v\n", f.Name())
  241. }
  242. return nil
  243. }
  244. // find the chunk count
  245. chunkSize := int64(*worker.options.maxMB * 1024 * 1024)
  246. chunkCount := 1
  247. if chunkSize > 0 && task.fileSize > chunkSize {
  248. chunkCount = int(task.fileSize/chunkSize) + 1
  249. }
  250. if chunkCount == 1 {
  251. return worker.uploadFileAsOne(task, f)
  252. }
  253. return worker.uploadFileInChunks(task, f, chunkCount, chunkSize)
  254. }
  255. func (worker *FileCopyWorker) checkExistingFileFirst(task FileCopyTask, f *os.File) (shouldCopy bool, err error) {
  256. shouldCopy = true
  257. if !*worker.options.checkSize {
  258. return
  259. }
  260. fileStat, err := f.Stat()
  261. if err != nil {
  262. shouldCopy = false
  263. return
  264. }
  265. err = pb.WithGrpcFilerClient(false, worker.signature, worker.filerAddress, worker.options.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  266. request := &filer_pb.LookupDirectoryEntryRequest{
  267. Directory: task.destinationUrlPath,
  268. Name: filepath.Base(f.Name()),
  269. }
  270. resp, lookupErr := client.LookupDirectoryEntry(context.Background(), request)
  271. if lookupErr != nil {
  272. // mostly not found error
  273. return nil
  274. }
  275. if fileStat.Size() == int64(filer.FileSize(resp.Entry)) {
  276. shouldCopy = false
  277. }
  278. return nil
  279. })
  280. return
  281. }
  282. func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) error {
  283. // upload the file content
  284. fileName := filepath.Base(f.Name())
  285. var mimeType string
  286. var chunks []*filer_pb.FileChunk
  287. if task.fileMode&os.ModeDir == 0 && task.fileSize > 0 {
  288. mimeType = detectMimeType(f)
  289. data, err := io.ReadAll(f)
  290. if err != nil {
  291. return err
  292. }
  293. uploader, uploaderErr := operation.NewUploader()
  294. if uploaderErr != nil {
  295. return uploaderErr
  296. }
  297. finalFileId, uploadResult, flushErr, _ := uploader.UploadWithRetry(
  298. worker,
  299. &filer_pb.AssignVolumeRequest{
  300. Count: 1,
  301. Replication: *worker.options.replication,
  302. Collection: *worker.options.collection,
  303. TtlSec: worker.options.ttlSec,
  304. DiskType: *worker.options.diskType,
  305. Path: task.destinationUrlPath,
  306. },
  307. &operation.UploadOption{
  308. Filename: fileName,
  309. Cipher: worker.options.cipher,
  310. IsInputCompressed: false,
  311. MimeType: mimeType,
  312. PairMap: nil,
  313. },
  314. func(host, fileId string) string {
  315. return fmt.Sprintf("http://%s/%s", host, fileId)
  316. },
  317. util.NewBytesReader(data),
  318. )
  319. if flushErr != nil {
  320. return flushErr
  321. }
  322. chunks = append(chunks, uploadResult.ToPbFileChunk(finalFileId, 0, time.Now().UnixNano()))
  323. }
  324. if err := pb.WithGrpcFilerClient(false, worker.signature, worker.filerAddress, worker.options.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  325. request := &filer_pb.CreateEntryRequest{
  326. Directory: task.destinationUrlPath,
  327. Entry: &filer_pb.Entry{
  328. Name: fileName,
  329. Attributes: &filer_pb.FuseAttributes{
  330. Crtime: time.Now().Unix(),
  331. Mtime: time.Now().Unix(),
  332. Gid: task.gid,
  333. Uid: task.uid,
  334. FileSize: uint64(task.fileSize),
  335. FileMode: uint32(task.fileMode),
  336. Mime: mimeType,
  337. TtlSec: worker.options.ttlSec,
  338. },
  339. Chunks: chunks,
  340. },
  341. }
  342. if err := filer_pb.CreateEntry(client, request); err != nil {
  343. return fmt.Errorf("update fh: %v", err)
  344. }
  345. return nil
  346. }); err != nil {
  347. return fmt.Errorf("upload data %v to http://%s%s%s: %v\n", fileName, worker.filerAddress.ToHttpAddress(), task.destinationUrlPath, fileName, err)
  348. }
  349. return nil
  350. }
  351. func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File, chunkCount int, chunkSize int64) error {
  352. fileName := filepath.Base(f.Name())
  353. mimeType := detectMimeType(f)
  354. chunksChan := make(chan *filer_pb.FileChunk, chunkCount)
  355. concurrentChunks := make(chan struct{}, *worker.options.concurrentChunks)
  356. var wg sync.WaitGroup
  357. var uploadError error
  358. fmt.Printf("uploading %s in %d chunks ...\n", fileName, chunkCount)
  359. for i := int64(0); i < int64(chunkCount) && uploadError == nil; i++ {
  360. wg.Add(1)
  361. concurrentChunks <- struct{}{}
  362. go func(i int64) {
  363. defer func() {
  364. wg.Done()
  365. <-concurrentChunks
  366. }()
  367. uploader, err := operation.NewUploader()
  368. if err != nil {
  369. uploadError = fmt.Errorf("upload data %v: %v\n", fileName, err)
  370. return
  371. }
  372. fileId, uploadResult, err, _ := uploader.UploadWithRetry(
  373. worker,
  374. &filer_pb.AssignVolumeRequest{
  375. Count: 1,
  376. Replication: *worker.options.replication,
  377. Collection: *worker.options.collection,
  378. TtlSec: worker.options.ttlSec,
  379. DiskType: *worker.options.diskType,
  380. Path: task.destinationUrlPath + fileName,
  381. },
  382. &operation.UploadOption{
  383. Filename: fileName + "-" + strconv.FormatInt(i+1, 10),
  384. Cipher: worker.options.cipher,
  385. IsInputCompressed: false,
  386. MimeType: "",
  387. PairMap: nil,
  388. },
  389. func(host, fileId string) string {
  390. return fmt.Sprintf("http://%s/%s", host, fileId)
  391. },
  392. io.NewSectionReader(f, i*chunkSize, chunkSize),
  393. )
  394. if err != nil {
  395. uploadError = fmt.Errorf("upload data %v: %v\n", fileName, err)
  396. return
  397. }
  398. if uploadResult.Error != "" {
  399. uploadError = fmt.Errorf("upload %v result: %v\n", fileName, uploadResult.Error)
  400. return
  401. }
  402. chunksChan <- uploadResult.ToPbFileChunk(fileId, i*chunkSize, time.Now().UnixNano())
  403. fmt.Printf("uploaded %s-%d [%d,%d)\n", fileName, i+1, i*chunkSize, i*chunkSize+int64(uploadResult.Size))
  404. }(i)
  405. }
  406. wg.Wait()
  407. close(chunksChan)
  408. var chunks []*filer_pb.FileChunk
  409. for chunk := range chunksChan {
  410. chunks = append(chunks, chunk)
  411. }
  412. if uploadError != nil {
  413. var fileIds []string
  414. for _, chunk := range chunks {
  415. fileIds = append(fileIds, chunk.FileId)
  416. }
  417. operation.DeleteFileIds(func(_ context.Context) pb.ServerAddress {
  418. return pb.ServerAddress(copy.masters[0])
  419. }, false, worker.options.grpcDialOption, fileIds)
  420. return uploadError
  421. }
  422. manifestedChunks, manifestErr := filer.MaybeManifestize(worker.saveDataAsChunk, chunks)
  423. if manifestErr != nil {
  424. return fmt.Errorf("create manifest: %v", manifestErr)
  425. }
  426. if err := pb.WithGrpcFilerClient(false, worker.signature, worker.filerAddress, worker.options.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  427. request := &filer_pb.CreateEntryRequest{
  428. Directory: task.destinationUrlPath,
  429. Entry: &filer_pb.Entry{
  430. Name: fileName,
  431. Attributes: &filer_pb.FuseAttributes{
  432. Crtime: time.Now().Unix(),
  433. Mtime: time.Now().Unix(),
  434. Gid: task.gid,
  435. Uid: task.uid,
  436. FileSize: uint64(task.fileSize),
  437. FileMode: uint32(task.fileMode),
  438. Mime: mimeType,
  439. TtlSec: worker.options.ttlSec,
  440. },
  441. Chunks: manifestedChunks,
  442. },
  443. }
  444. if err := filer_pb.CreateEntry(client, request); err != nil {
  445. return fmt.Errorf("update fh: %v", err)
  446. }
  447. return nil
  448. }); err != nil {
  449. return fmt.Errorf("upload data %v to http://%s%s%s: %v\n", fileName, worker.filerAddress.ToHttpAddress(), task.destinationUrlPath, fileName, err)
  450. }
  451. fmt.Printf("copied %s => http://%s%s%s\n", f.Name(), worker.filerAddress.ToHttpAddress(), task.destinationUrlPath, fileName)
  452. return nil
  453. }
  454. func detectMimeType(f *os.File) string {
  455. head := make([]byte, 512)
  456. f.Seek(0, io.SeekStart)
  457. n, err := f.Read(head)
  458. if err == io.EOF {
  459. return ""
  460. }
  461. if err != nil {
  462. fmt.Printf("read head of %v: %v\n", f.Name(), err)
  463. return ""
  464. }
  465. f.Seek(0, io.SeekStart)
  466. mimeType := http.DetectContentType(head[:n])
  467. if mimeType == "application/octet-stream" {
  468. return ""
  469. }
  470. return mimeType
  471. }
  472. func (worker *FileCopyWorker) saveDataAsChunk(reader io.Reader, name string, offset int64, tsNs int64) (chunk *filer_pb.FileChunk, err error) {
  473. uploader, uploaderErr := operation.NewUploader()
  474. if uploaderErr != nil {
  475. return nil, fmt.Errorf("upload data: %v", uploaderErr)
  476. }
  477. finalFileId, uploadResult, flushErr, _ := uploader.UploadWithRetry(
  478. worker,
  479. &filer_pb.AssignVolumeRequest{
  480. Count: 1,
  481. Replication: *worker.options.replication,
  482. Collection: *worker.options.collection,
  483. TtlSec: worker.options.ttlSec,
  484. DiskType: *worker.options.diskType,
  485. Path: name,
  486. },
  487. &operation.UploadOption{
  488. Filename: name,
  489. Cipher: worker.options.cipher,
  490. IsInputCompressed: false,
  491. MimeType: "",
  492. PairMap: nil,
  493. },
  494. func(host, fileId string) string {
  495. return fmt.Sprintf("http://%s/%s", host, fileId)
  496. },
  497. reader,
  498. )
  499. if flushErr != nil {
  500. return nil, fmt.Errorf("upload data: %v", flushErr)
  501. }
  502. if uploadResult.Error != "" {
  503. return nil, fmt.Errorf("upload result: %v", uploadResult.Error)
  504. }
  505. return uploadResult.ToPbFileChunk(finalFileId, offset, tsNs), nil
  506. }
  507. var _ = filer_pb.FilerClient(&FileCopyWorker{})
  508. func (worker *FileCopyWorker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) (err error) {
  509. filerGrpcAddress := worker.filerAddress.ToGrpcAddress()
  510. err = pb.WithGrpcClient(streamingMode, worker.signature, func(grpcConnection *grpc.ClientConn) error {
  511. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  512. return fn(client)
  513. }, filerGrpcAddress, false, worker.options.grpcDialOption)
  514. return
  515. }
  516. func (worker *FileCopyWorker) AdjustedUrl(location *filer_pb.Location) string {
  517. if *worker.options.volumeServerAccess == "publicUrl" {
  518. return location.PublicUrl
  519. }
  520. return location.Url
  521. }
  522. func (worker *FileCopyWorker) GetDataCenter() string {
  523. return ""
  524. }