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.

517 lines
15 KiB

5 years ago
7 years ago
4 years ago
5 years ago
4 years ago
5 years ago
4 years ago
5 years ago
5 years ago
4 years ago
5 years ago
5 years ago
5 years ago
  1. package command
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "io/ioutil"
  7. "net/http"
  8. "net/url"
  9. "os"
  10. "path/filepath"
  11. "strconv"
  12. "strings"
  13. "sync"
  14. "time"
  15. "google.golang.org/grpc"
  16. "github.com/chrislusf/seaweedfs/weed/util/grace"
  17. "github.com/chrislusf/seaweedfs/weed/operation"
  18. "github.com/chrislusf/seaweedfs/weed/pb"
  19. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  20. "github.com/chrislusf/seaweedfs/weed/security"
  21. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  22. "github.com/chrislusf/seaweedfs/weed/util"
  23. "github.com/chrislusf/seaweedfs/weed/wdclient"
  24. )
  25. var (
  26. copy CopyOptions
  27. waitGroup sync.WaitGroup
  28. )
  29. type CopyOptions struct {
  30. include *string
  31. replication *string
  32. collection *string
  33. ttl *string
  34. maxMB *int
  35. masterClient *wdclient.MasterClient
  36. concurrenctFiles *int
  37. concurrenctChunks *int
  38. grpcDialOption grpc.DialOption
  39. masters []string
  40. cipher bool
  41. ttlSec int32
  42. }
  43. func init() {
  44. cmdCopy.Run = runCopy // break init cycle
  45. cmdCopy.IsDebug = cmdCopy.Flag.Bool("debug", false, "verbose debug information")
  46. copy.include = cmdCopy.Flag.String("include", "", "pattens of files to copy, e.g., *.pdf, *.html, ab?d.txt, works together with -dir")
  47. copy.replication = cmdCopy.Flag.String("replication", "", "replication type")
  48. copy.collection = cmdCopy.Flag.String("collection", "", "optional collection name")
  49. copy.ttl = cmdCopy.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
  50. copy.maxMB = cmdCopy.Flag.Int("maxMB", 32, "split files larger than the limit")
  51. copy.concurrenctFiles = cmdCopy.Flag.Int("c", 8, "concurrent file copy goroutines")
  52. copy.concurrenctChunks = cmdCopy.Flag.Int("concurrentChunks", 8, "concurrent chunk copy goroutines for each file")
  53. }
  54. var cmdCopy = &Command{
  55. UsageLine: "filer.copy file_or_dir1 [file_or_dir2 file_or_dir3] http://localhost:8888/path/to/a/folder/",
  56. Short: "copy one or a list of files to a filer folder",
  57. Long: `copy one or a list of files, or batch copy one whole folder recursively, to a filer folder
  58. It can copy one or a list of files or folders.
  59. If copying a whole folder recursively:
  60. All files under the folder and subfolders will be copyed.
  61. Optional parameter "-include" allows you to specify the file name patterns.
  62. If "maxMB" is set to a positive number, files larger than it would be split into chunks.
  63. `,
  64. }
  65. func runCopy(cmd *Command, args []string) bool {
  66. util.LoadConfiguration("security", false)
  67. if len(args) <= 1 {
  68. return false
  69. }
  70. filerDestination := args[len(args)-1]
  71. fileOrDirs := args[0 : len(args)-1]
  72. filerUrl, err := url.Parse(filerDestination)
  73. if err != nil {
  74. fmt.Printf("The last argument should be a URL on filer: %v\n", err)
  75. return false
  76. }
  77. urlPath := filerUrl.Path
  78. if !strings.HasSuffix(urlPath, "/") {
  79. fmt.Printf("The last argument should be a folder and end with \"/\": %v\n", err)
  80. return false
  81. }
  82. if filerUrl.Port() == "" {
  83. fmt.Printf("The filer port should be specified.\n")
  84. return false
  85. }
  86. filerPort, parseErr := strconv.ParseUint(filerUrl.Port(), 10, 64)
  87. if parseErr != nil {
  88. fmt.Printf("The filer port parse error: %v\n", parseErr)
  89. return false
  90. }
  91. filerGrpcPort := filerPort + 10000
  92. filerGrpcAddress := fmt.Sprintf("%s:%d", filerUrl.Hostname(), filerGrpcPort)
  93. copy.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
  94. masters, collection, replication, dirBuckets, maxMB, cipher, err := readFilerConfiguration(copy.grpcDialOption, filerGrpcAddress)
  95. if err != nil {
  96. fmt.Printf("read from filer %s: %v\n", filerGrpcAddress, err)
  97. return false
  98. }
  99. if strings.HasPrefix(urlPath, dirBuckets+"/") {
  100. restPath := urlPath[len(dirBuckets)+1:]
  101. if strings.Index(restPath, "/") > 0 {
  102. expectedBucket := restPath[:strings.Index(restPath, "/")]
  103. if *copy.collection == "" {
  104. *copy.collection = expectedBucket
  105. } else if *copy.collection != expectedBucket {
  106. fmt.Printf("destination %s uses collection \"%s\": unexpected collection \"%v\"\n", urlPath, expectedBucket, *copy.collection)
  107. return true
  108. }
  109. }
  110. }
  111. if *copy.collection == "" {
  112. *copy.collection = collection
  113. }
  114. if *copy.replication == "" {
  115. *copy.replication = replication
  116. }
  117. if *copy.maxMB == 0 {
  118. *copy.maxMB = int(maxMB)
  119. }
  120. copy.masters = masters
  121. copy.cipher = cipher
  122. ttl, err := needle.ReadTTL(*copy.ttl)
  123. if err != nil {
  124. fmt.Printf("parsing ttl %s: %v\n", *copy.ttl, err)
  125. return false
  126. }
  127. copy.ttlSec = int32(ttl.Minutes()) * 60
  128. if *cmdCopy.IsDebug {
  129. grace.SetupProfiling("filer.copy.cpu.pprof", "filer.copy.mem.pprof")
  130. }
  131. fileCopyTaskChan := make(chan FileCopyTask, *copy.concurrenctFiles)
  132. go func() {
  133. defer close(fileCopyTaskChan)
  134. for _, fileOrDir := range fileOrDirs {
  135. if err := genFileCopyTask(fileOrDir, urlPath, fileCopyTaskChan); err != nil {
  136. fmt.Fprintf(os.Stderr, "gen file list error: %v\n", err)
  137. break
  138. }
  139. }
  140. }()
  141. for i := 0; i < *copy.concurrenctFiles; i++ {
  142. waitGroup.Add(1)
  143. go func() {
  144. defer waitGroup.Done()
  145. worker := FileCopyWorker{
  146. options: &copy,
  147. filerHost: filerUrl.Host,
  148. filerGrpcAddress: filerGrpcAddress,
  149. }
  150. if err := worker.copyFiles(fileCopyTaskChan); err != nil {
  151. fmt.Fprintf(os.Stderr, "copy file error: %v\n", err)
  152. return
  153. }
  154. }()
  155. }
  156. waitGroup.Wait()
  157. return true
  158. }
  159. func readFilerConfiguration(grpcDialOption grpc.DialOption, filerGrpcAddress string) (masters []string, collection, replication string, dirBuckets string, maxMB uint32, cipher bool, err error) {
  160. err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  161. resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
  162. if err != nil {
  163. return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err)
  164. }
  165. masters, collection, replication, maxMB = resp.Masters, resp.Collection, resp.Replication, resp.MaxMb
  166. dirBuckets = resp.DirBuckets
  167. cipher = resp.Cipher
  168. return nil
  169. })
  170. return
  171. }
  172. func genFileCopyTask(fileOrDir string, destPath string, fileCopyTaskChan chan FileCopyTask) error {
  173. fi, err := os.Stat(fileOrDir)
  174. if err != nil {
  175. fmt.Fprintf(os.Stderr, "Failed to get stat for file %s: %v\n", fileOrDir, err)
  176. return nil
  177. }
  178. mode := fi.Mode()
  179. if mode.IsDir() {
  180. files, _ := ioutil.ReadDir(fileOrDir)
  181. for _, subFileOrDir := range files {
  182. if err = genFileCopyTask(fileOrDir+"/"+subFileOrDir.Name(), destPath+fi.Name()+"/", fileCopyTaskChan); err != nil {
  183. return err
  184. }
  185. }
  186. return nil
  187. }
  188. uid, gid := util.GetFileUidGid(fi)
  189. fileCopyTaskChan <- FileCopyTask{
  190. sourceLocation: fileOrDir,
  191. destinationUrlPath: destPath,
  192. fileSize: fi.Size(),
  193. fileMode: fi.Mode(),
  194. uid: uid,
  195. gid: gid,
  196. }
  197. return nil
  198. }
  199. type FileCopyWorker struct {
  200. options *CopyOptions
  201. filerHost string
  202. filerGrpcAddress string
  203. }
  204. func (worker *FileCopyWorker) copyFiles(fileCopyTaskChan chan FileCopyTask) error {
  205. for task := range fileCopyTaskChan {
  206. if err := worker.doEachCopy(task); err != nil {
  207. return err
  208. }
  209. }
  210. return nil
  211. }
  212. type FileCopyTask struct {
  213. sourceLocation string
  214. destinationUrlPath string
  215. fileSize int64
  216. fileMode os.FileMode
  217. uid uint32
  218. gid uint32
  219. }
  220. func (worker *FileCopyWorker) doEachCopy(task FileCopyTask) error {
  221. f, err := os.Open(task.sourceLocation)
  222. if err != nil {
  223. fmt.Printf("Failed to open file %s: %v\n", task.sourceLocation, err)
  224. if _, ok := err.(*os.PathError); ok {
  225. fmt.Printf("skipping %s\n", task.sourceLocation)
  226. return nil
  227. }
  228. return err
  229. }
  230. defer f.Close()
  231. // this is a regular file
  232. if *worker.options.include != "" {
  233. if ok, _ := filepath.Match(*worker.options.include, filepath.Base(task.sourceLocation)); !ok {
  234. return nil
  235. }
  236. }
  237. // find the chunk count
  238. chunkSize := int64(*worker.options.maxMB * 1024 * 1024)
  239. chunkCount := 1
  240. if chunkSize > 0 && task.fileSize > chunkSize {
  241. chunkCount = int(task.fileSize/chunkSize) + 1
  242. }
  243. if chunkCount == 1 {
  244. return worker.uploadFileAsOne(task, f)
  245. }
  246. return worker.uploadFileInChunks(task, f, chunkCount, chunkSize)
  247. }
  248. func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) error {
  249. // upload the file content
  250. fileName := filepath.Base(f.Name())
  251. mimeType := detectMimeType(f)
  252. data, err := ioutil.ReadAll(f)
  253. if err != nil {
  254. return err
  255. }
  256. var chunks []*filer_pb.FileChunk
  257. var assignResult *filer_pb.AssignVolumeResponse
  258. var assignError error
  259. if task.fileSize > 0 {
  260. // assign a volume
  261. err := pb.WithGrpcFilerClient(worker.filerGrpcAddress, worker.options.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  262. request := &filer_pb.AssignVolumeRequest{
  263. Count: 1,
  264. Replication: *worker.options.replication,
  265. Collection: *worker.options.collection,
  266. TtlSec: worker.options.ttlSec,
  267. Path: task.destinationUrlPath,
  268. }
  269. assignResult, assignError = client.AssignVolume(context.Background(), request)
  270. if assignError != nil {
  271. return fmt.Errorf("assign volume failure %v: %v", request, assignError)
  272. }
  273. if assignResult.Error != "" {
  274. return fmt.Errorf("assign volume failure %v: %v", request, assignResult.Error)
  275. }
  276. return nil
  277. })
  278. if err != nil {
  279. return fmt.Errorf("Failed to assign from %v: %v\n", worker.options.masters, err)
  280. }
  281. targetUrl := "http://" + assignResult.Url + "/" + assignResult.FileId
  282. uploadResult, err := operation.UploadData(targetUrl, fileName, worker.options.cipher, data, false, mimeType, nil, security.EncodedJwt(assignResult.Auth))
  283. if err != nil {
  284. return fmt.Errorf("upload data %v to %s: %v\n", fileName, targetUrl, err)
  285. }
  286. if uploadResult.Error != "" {
  287. return fmt.Errorf("upload %v to %s result: %v\n", fileName, targetUrl, uploadResult.Error)
  288. }
  289. fmt.Printf("uploaded %s to %s\n", fileName, targetUrl)
  290. chunks = append(chunks, uploadResult.ToPbFileChunk(assignResult.FileId, 0))
  291. fmt.Printf("copied %s => http://%s%s%s\n", fileName, worker.filerHost, task.destinationUrlPath, fileName)
  292. }
  293. if err := pb.WithGrpcFilerClient(worker.filerGrpcAddress, worker.options.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  294. request := &filer_pb.CreateEntryRequest{
  295. Directory: task.destinationUrlPath,
  296. Entry: &filer_pb.Entry{
  297. Name: fileName,
  298. Attributes: &filer_pb.FuseAttributes{
  299. Crtime: time.Now().Unix(),
  300. Mtime: time.Now().Unix(),
  301. Gid: task.gid,
  302. Uid: task.uid,
  303. FileSize: uint64(task.fileSize),
  304. FileMode: uint32(task.fileMode),
  305. Mime: mimeType,
  306. Replication: *worker.options.replication,
  307. Collection: *worker.options.collection,
  308. TtlSec: worker.options.ttlSec,
  309. },
  310. Chunks: chunks,
  311. },
  312. }
  313. if err := filer_pb.CreateEntry(client, request); err != nil {
  314. return fmt.Errorf("update fh: %v", err)
  315. }
  316. return nil
  317. }); err != nil {
  318. return fmt.Errorf("upload data %v to http://%s%s%s: %v\n", fileName, worker.filerHost, task.destinationUrlPath, fileName, err)
  319. }
  320. return nil
  321. }
  322. func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File, chunkCount int, chunkSize int64) error {
  323. fileName := filepath.Base(f.Name())
  324. mimeType := detectMimeType(f)
  325. chunksChan := make(chan *filer_pb.FileChunk, chunkCount)
  326. concurrentChunks := make(chan struct{}, *worker.options.concurrenctChunks)
  327. var wg sync.WaitGroup
  328. var uploadError error
  329. var collection, replication string
  330. fmt.Printf("uploading %s in %d chunks ...\n", fileName, chunkCount)
  331. for i := int64(0); i < int64(chunkCount) && uploadError == nil; i++ {
  332. wg.Add(1)
  333. concurrentChunks <- struct{}{}
  334. go func(i int64) {
  335. defer func() {
  336. wg.Done()
  337. <-concurrentChunks
  338. }()
  339. // assign a volume
  340. var assignResult *filer_pb.AssignVolumeResponse
  341. var assignError error
  342. err := pb.WithGrpcFilerClient(worker.filerGrpcAddress, worker.options.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  343. request := &filer_pb.AssignVolumeRequest{
  344. Count: 1,
  345. Replication: *worker.options.replication,
  346. Collection: *worker.options.collection,
  347. TtlSec: worker.options.ttlSec,
  348. Path: task.destinationUrlPath + fileName,
  349. }
  350. assignResult, assignError = client.AssignVolume(context.Background(), request)
  351. if assignError != nil {
  352. return fmt.Errorf("assign volume failure %v: %v", request, assignError)
  353. }
  354. if assignResult.Error != "" {
  355. return fmt.Errorf("assign volume failure %v: %v", request, assignResult.Error)
  356. }
  357. return nil
  358. })
  359. if err != nil {
  360. fmt.Printf("Failed to assign from %v: %v\n", worker.options.masters, err)
  361. }
  362. if err != nil {
  363. fmt.Printf("Failed to assign from %v: %v\n", worker.options.masters, err)
  364. }
  365. targetUrl := "http://" + assignResult.Url + "/" + assignResult.FileId
  366. if collection == "" {
  367. collection = assignResult.Collection
  368. }
  369. if replication == "" {
  370. replication = assignResult.Replication
  371. }
  372. uploadResult, err, _ := operation.Upload(targetUrl, fileName+"-"+strconv.FormatInt(i+1, 10), worker.options.cipher, io.NewSectionReader(f, i*chunkSize, chunkSize), false, "", nil, security.EncodedJwt(assignResult.Auth))
  373. if err != nil {
  374. uploadError = fmt.Errorf("upload data %v to %s: %v\n", fileName, targetUrl, err)
  375. return
  376. }
  377. if uploadResult.Error != "" {
  378. uploadError = fmt.Errorf("upload %v to %s result: %v\n", fileName, targetUrl, uploadResult.Error)
  379. return
  380. }
  381. chunksChan <- uploadResult.ToPbFileChunk(assignResult.FileId, i*chunkSize)
  382. fmt.Printf("uploaded %s-%d to %s [%d,%d)\n", fileName, i+1, targetUrl, i*chunkSize, i*chunkSize+int64(uploadResult.Size))
  383. }(i)
  384. }
  385. wg.Wait()
  386. close(chunksChan)
  387. var chunks []*filer_pb.FileChunk
  388. for chunk := range chunksChan {
  389. chunks = append(chunks, chunk)
  390. }
  391. if uploadError != nil {
  392. var fileIds []string
  393. for _, chunk := range chunks {
  394. fileIds = append(fileIds, chunk.FileId)
  395. }
  396. operation.DeleteFiles(copy.masters[0], false, worker.options.grpcDialOption, fileIds)
  397. return uploadError
  398. }
  399. if err := pb.WithGrpcFilerClient(worker.filerGrpcAddress, worker.options.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  400. request := &filer_pb.CreateEntryRequest{
  401. Directory: task.destinationUrlPath,
  402. Entry: &filer_pb.Entry{
  403. Name: fileName,
  404. Attributes: &filer_pb.FuseAttributes{
  405. Crtime: time.Now().Unix(),
  406. Mtime: time.Now().Unix(),
  407. Gid: task.gid,
  408. Uid: task.uid,
  409. FileSize: uint64(task.fileSize),
  410. FileMode: uint32(task.fileMode),
  411. Mime: mimeType,
  412. Replication: replication,
  413. Collection: collection,
  414. TtlSec: worker.options.ttlSec,
  415. },
  416. Chunks: chunks,
  417. },
  418. }
  419. if err := filer_pb.CreateEntry(client, request); err != nil {
  420. return fmt.Errorf("update fh: %v", err)
  421. }
  422. return nil
  423. }); err != nil {
  424. return fmt.Errorf("upload data %v to http://%s%s%s: %v\n", fileName, worker.filerHost, task.destinationUrlPath, fileName, err)
  425. }
  426. fmt.Printf("copied %s => http://%s%s%s\n", fileName, worker.filerHost, task.destinationUrlPath, fileName)
  427. return nil
  428. }
  429. func detectMimeType(f *os.File) string {
  430. head := make([]byte, 512)
  431. f.Seek(0, io.SeekStart)
  432. n, err := f.Read(head)
  433. if err == io.EOF {
  434. return ""
  435. }
  436. if err != nil {
  437. fmt.Printf("read head of %v: %v\n", f.Name(), err)
  438. return ""
  439. }
  440. f.Seek(0, io.SeekStart)
  441. mimeType := http.DetectContentType(head[:n])
  442. if mimeType == "application/octet-stream" {
  443. return ""
  444. }
  445. return mimeType
  446. }