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.

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