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.

343 lines
9.7 KiB

7 years ago
7 years ago
7 years ago
7 years ago
  1. package command
  2. import (
  3. "fmt"
  4. "io/ioutil"
  5. "net/url"
  6. "os"
  7. "path/filepath"
  8. "strings"
  9. "context"
  10. "github.com/chrislusf/seaweedfs/weed/operation"
  11. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  12. "github.com/chrislusf/seaweedfs/weed/security"
  13. "github.com/chrislusf/seaweedfs/weed/util"
  14. "io"
  15. "net/http"
  16. "strconv"
  17. "time"
  18. )
  19. var (
  20. copy CopyOptions
  21. )
  22. type CopyOptions struct {
  23. filerGrpcPort *int
  24. master *string
  25. include *string
  26. replication *string
  27. collection *string
  28. ttl *string
  29. maxMB *int
  30. secretKey *string
  31. secret security.Secret
  32. }
  33. func init() {
  34. cmdCopy.Run = runCopy // break init cycle
  35. cmdCopy.IsDebug = cmdCopy.Flag.Bool("debug", false, "verbose debug information")
  36. copy.master = cmdCopy.Flag.String("master", "localhost:9333", "SeaweedFS master location")
  37. copy.include = cmdCopy.Flag.String("include", "", "pattens of files to copy, e.g., *.pdf, *.html, ab?d.txt, works together with -dir")
  38. copy.replication = cmdCopy.Flag.String("replication", "", "replication type")
  39. copy.collection = cmdCopy.Flag.String("collection", "", "optional collection name")
  40. copy.ttl = cmdCopy.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y")
  41. copy.maxMB = cmdCopy.Flag.Int("maxMB", 0, "split files larger than the limit")
  42. copy.filerGrpcPort = cmdCopy.Flag.Int("filer.port.grpc", 0, "filer grpc server listen port, default to filer port + 10000")
  43. copy.secretKey = cmdCopy.Flag.String("secure.secret", "", "secret to encrypt Json Web Token(JWT)")
  44. }
  45. var cmdCopy = &Command{
  46. UsageLine: "filer.copy file_or_dir1 [file_or_dir2 file_or_dir3] http://localhost:8888/path/to/a/folder/",
  47. Short: "copy one or a list of files to a filer folder",
  48. Long: `copy one or a list of files, or batch copy one whole folder recursively, to a filer folder
  49. It can copy one or a list of files or folders.
  50. If copying a whole folder recursively:
  51. All files under the folder and subfolders will be copyed.
  52. Optional parameter "-include" allows you to specify the file name patterns.
  53. If "maxMB" is set to a positive number, files larger than it would be split into chunks.
  54. `,
  55. }
  56. func runCopy(cmd *Command, args []string) bool {
  57. copy.secret = security.Secret(*copy.secretKey)
  58. if len(args) <= 1 {
  59. return false
  60. }
  61. filerDestination := args[len(args)-1]
  62. fileOrDirs := args[0 : len(args)-1]
  63. filerUrl, err := url.Parse(filerDestination)
  64. if err != nil {
  65. fmt.Printf("The last argument should be a URL on filer: %v\n", err)
  66. return false
  67. }
  68. urlPath := filerUrl.Path
  69. if !strings.HasSuffix(urlPath, "/") {
  70. fmt.Printf("The last argument should be a folder and end with \"/\": %v\n", err)
  71. return false
  72. }
  73. if filerUrl.Port() == "" {
  74. fmt.Printf("The filer port should be specified.\n")
  75. return false
  76. }
  77. filerPort, parseErr := strconv.ParseUint(filerUrl.Port(), 10, 64)
  78. if parseErr != nil {
  79. fmt.Printf("The filer port parse error: %v\n", parseErr)
  80. return false
  81. }
  82. filerGrpcPort := filerPort + 10000
  83. if *copy.filerGrpcPort != 0 {
  84. filerGrpcPort = uint64(*copy.filerGrpcPort)
  85. }
  86. filerGrpcAddress := fmt.Sprintf("%s:%d", filerUrl.Hostname(), filerGrpcPort)
  87. for _, fileOrDir := range fileOrDirs {
  88. if !doEachCopy(fileOrDir, filerUrl.Host, filerGrpcAddress, urlPath) {
  89. return false
  90. }
  91. }
  92. return true
  93. }
  94. func doEachCopy(fileOrDir string, filerAddress, filerGrpcAddress string, path string) bool {
  95. f, err := os.Open(fileOrDir)
  96. if err != nil {
  97. fmt.Printf("Failed to open file %s: %v\n", fileOrDir, err)
  98. return false
  99. }
  100. defer f.Close()
  101. fi, err := f.Stat()
  102. if err != nil {
  103. fmt.Printf("Failed to get stat for file %s: %v\n", fileOrDir, err)
  104. return false
  105. }
  106. mode := fi.Mode()
  107. if mode.IsDir() {
  108. files, _ := ioutil.ReadDir(fileOrDir)
  109. for _, subFileOrDir := range files {
  110. if !doEachCopy(fileOrDir+"/"+subFileOrDir.Name(), filerAddress, filerGrpcAddress, path+fi.Name()+"/") {
  111. return false
  112. }
  113. }
  114. return true
  115. }
  116. // this is a regular file
  117. if *copy.include != "" {
  118. if ok, _ := filepath.Match(*copy.include, filepath.Base(fileOrDir)); !ok {
  119. return true
  120. }
  121. }
  122. // find the chunk count
  123. chunkSize := int64(*copy.maxMB * 1024 * 1024)
  124. chunkCount := 1
  125. if chunkSize > 0 && fi.Size() > chunkSize {
  126. chunkCount = int(fi.Size()/chunkSize) + 1
  127. }
  128. if chunkCount == 1 {
  129. return uploadFileAsOne(filerAddress, filerGrpcAddress, path, f, fi)
  130. }
  131. return uploadFileInChunks(filerAddress, filerGrpcAddress, path, f, fi, chunkCount, chunkSize)
  132. }
  133. func uploadFileAsOne(filerAddress, filerGrpcAddress string, urlFolder string, f *os.File, fi os.FileInfo) bool {
  134. // upload the file content
  135. fileName := filepath.Base(f.Name())
  136. mimeType := detectMimeType(f)
  137. var chunks []*filer_pb.FileChunk
  138. if fi.Size() > 0 {
  139. // assign a volume
  140. assignResult, err := operation.Assign(*copy.master, &operation.VolumeAssignRequest{
  141. Count: 1,
  142. Replication: *copy.replication,
  143. Collection: *copy.collection,
  144. Ttl: *copy.ttl,
  145. })
  146. if err != nil {
  147. fmt.Printf("Failed to assign from %s: %v\n", *copy.master, err)
  148. }
  149. targetUrl := "http://" + assignResult.Url + "/" + assignResult.Fid
  150. uploadResult, err := operation.Upload(targetUrl, fileName, f, false, mimeType, nil, "")
  151. if err != nil {
  152. fmt.Printf("upload data %v to %s: %v\n", fileName, targetUrl, err)
  153. return false
  154. }
  155. if uploadResult.Error != "" {
  156. fmt.Printf("upload %v to %s result: %v\n", fileName, targetUrl, uploadResult.Error)
  157. return false
  158. }
  159. fmt.Printf("uploaded %s to %s\n", fileName, targetUrl)
  160. chunks = append(chunks, &filer_pb.FileChunk{
  161. FileId: assignResult.Fid,
  162. Offset: 0,
  163. Size: uint64(uploadResult.Size),
  164. Mtime: time.Now().UnixNano(),
  165. ETag: uploadResult.ETag,
  166. })
  167. fmt.Printf("copied %s => http://%s%s%s\n", fileName, filerAddress, urlFolder, fileName)
  168. }
  169. if err := withFilerClient(filerGrpcAddress, func(client filer_pb.SeaweedFilerClient) error {
  170. request := &filer_pb.CreateEntryRequest{
  171. Directory: urlFolder,
  172. Entry: &filer_pb.Entry{
  173. Name: fileName,
  174. Attributes: &filer_pb.FuseAttributes{
  175. Crtime: time.Now().Unix(),
  176. Mtime: time.Now().Unix(),
  177. Gid: uint32(os.Getgid()),
  178. Uid: uint32(os.Getuid()),
  179. FileSize: uint64(fi.Size()),
  180. FileMode: uint32(fi.Mode()),
  181. Mime: mimeType,
  182. Replication: *copy.replication,
  183. Collection: *copy.collection,
  184. TtlSec: int32(util.ParseInt(*copy.ttl, 0)),
  185. },
  186. Chunks: chunks,
  187. },
  188. }
  189. if _, err := client.CreateEntry(context.Background(), request); err != nil {
  190. return fmt.Errorf("update fh: %v", err)
  191. }
  192. return nil
  193. }); err != nil {
  194. fmt.Printf("upload data %v to http://%s%s%s: %v\n", fileName, filerAddress, urlFolder, fileName, err)
  195. return false
  196. }
  197. return true
  198. }
  199. func uploadFileInChunks(filerAddress, filerGrpcAddress string, urlFolder string, f *os.File, fi os.FileInfo, chunkCount int, chunkSize int64) bool {
  200. fileName := filepath.Base(f.Name())
  201. mimeType := detectMimeType(f)
  202. var chunks []*filer_pb.FileChunk
  203. for i := int64(0); i < int64(chunkCount); i++ {
  204. // assign a volume
  205. assignResult, err := operation.Assign(*copy.master, &operation.VolumeAssignRequest{
  206. Count: 1,
  207. Replication: *copy.replication,
  208. Collection: *copy.collection,
  209. Ttl: *copy.ttl,
  210. })
  211. if err != nil {
  212. fmt.Printf("Failed to assign from %s: %v\n", *copy.master, err)
  213. }
  214. targetUrl := "http://" + assignResult.Url + "/" + assignResult.Fid
  215. uploadResult, err := operation.Upload(targetUrl,
  216. fileName+"-"+strconv.FormatInt(i+1, 10),
  217. io.LimitReader(f, chunkSize),
  218. false, "application/octet-stream", nil, "")
  219. if err != nil {
  220. fmt.Printf("upload data %v to %s: %v\n", fileName, targetUrl, err)
  221. return false
  222. }
  223. if uploadResult.Error != "" {
  224. fmt.Printf("upload %v to %s result: %v\n", fileName, targetUrl, uploadResult.Error)
  225. return false
  226. }
  227. chunks = append(chunks, &filer_pb.FileChunk{
  228. FileId: assignResult.Fid,
  229. Offset: i * chunkSize,
  230. Size: uint64(uploadResult.Size),
  231. Mtime: time.Now().UnixNano(),
  232. ETag: uploadResult.ETag,
  233. })
  234. fmt.Printf("uploaded %s-%d to %s [%d,%d)\n", fileName, i+1, targetUrl, i*chunkSize, i*chunkSize+int64(uploadResult.Size))
  235. }
  236. if err := withFilerClient(filerGrpcAddress, func(client filer_pb.SeaweedFilerClient) error {
  237. request := &filer_pb.CreateEntryRequest{
  238. Directory: urlFolder,
  239. Entry: &filer_pb.Entry{
  240. Name: fileName,
  241. Attributes: &filer_pb.FuseAttributes{
  242. Crtime: time.Now().Unix(),
  243. Mtime: time.Now().Unix(),
  244. Gid: uint32(os.Getgid()),
  245. Uid: uint32(os.Getuid()),
  246. FileSize: uint64(fi.Size()),
  247. FileMode: uint32(fi.Mode()),
  248. Mime: mimeType,
  249. Replication: *copy.replication,
  250. Collection: *copy.collection,
  251. TtlSec: int32(util.ParseInt(*copy.ttl, 0)),
  252. },
  253. Chunks: chunks,
  254. },
  255. }
  256. if _, err := client.CreateEntry(context.Background(), request); err != nil {
  257. return fmt.Errorf("update fh: %v", err)
  258. }
  259. return nil
  260. }); err != nil {
  261. fmt.Printf("upload data %v to http://%s%s%s: %v\n", fileName, filerAddress, urlFolder, fileName, err)
  262. return false
  263. }
  264. fmt.Printf("copied %s => http://%s%s%s\n", fileName, filerAddress, urlFolder, fileName)
  265. return true
  266. }
  267. func detectMimeType(f *os.File) string {
  268. head := make([]byte, 512)
  269. f.Seek(0, io.SeekStart)
  270. n, err := f.Read(head)
  271. if err == io.EOF {
  272. return ""
  273. }
  274. if err != nil {
  275. fmt.Printf("read head of %v: %v\n", f.Name(), err)
  276. return "application/octet-stream"
  277. }
  278. f.Seek(0, io.SeekStart)
  279. mimeType := http.DetectContentType(head[:n])
  280. return mimeType
  281. }
  282. func withFilerClient(filerAddress string, fn func(filer_pb.SeaweedFilerClient) error) error {
  283. grpcConnection, err := util.GrpcDial(filerAddress)
  284. if err != nil {
  285. return fmt.Errorf("fail to dial %s: %v", filerAddress, err)
  286. }
  287. defer grpcConnection.Close()
  288. client := filer_pb.NewSeaweedFilerClient(grpcConnection)
  289. return fn(client)
  290. }