filer_copy.go 20 KB

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