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.

474 lines
14 KiB

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