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.

346 lines
10 KiB

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