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.

497 lines
15 KiB

7 years ago
7 years ago
7 years ago
7 years ago
4 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
2 years ago
2 years ago
4 years ago
4 years ago
2 years ago
  1. package s3api
  2. import (
  3. "context"
  4. "encoding/xml"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/glog"
  7. "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
  8. "io"
  9. "net/http"
  10. "net/url"
  11. "strconv"
  12. "strings"
  13. "time"
  14. "github.com/seaweedfs/seaweedfs/weed/filer"
  15. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  16. "github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
  17. )
  18. const cutoffTimeNewEmptyDir = 3
  19. type ListBucketResultV2 struct {
  20. XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListBucketResult"`
  21. Name string `xml:"Name"`
  22. Prefix string `xml:"Prefix"`
  23. MaxKeys int `xml:"MaxKeys"`
  24. Delimiter string `xml:"Delimiter,omitempty"`
  25. IsTruncated bool `xml:"IsTruncated"`
  26. Contents []ListEntry `xml:"Contents,omitempty"`
  27. CommonPrefixes []PrefixEntry `xml:"CommonPrefixes,omitempty"`
  28. ContinuationToken string `xml:"ContinuationToken,omitempty"`
  29. NextContinuationToken string `xml:"NextContinuationToken,omitempty"`
  30. KeyCount int `xml:"KeyCount"`
  31. StartAfter string `xml:"StartAfter,omitempty"`
  32. }
  33. func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Request) {
  34. // https://docs.aws.amazon.com/AmazonS3/latest/API/v2-RESTBucketGET.html
  35. // collect parameters
  36. bucket, _ := s3_constants.GetBucketAndObject(r)
  37. glog.V(3).Infof("ListObjectsV2Handler %s", bucket)
  38. originalPrefix, continuationToken, startAfter, delimiter, _, maxKeys := getListObjectsV2Args(r.URL.Query())
  39. if maxKeys < 0 {
  40. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxKeys)
  41. return
  42. }
  43. if delimiter != "" && delimiter != "/" {
  44. s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
  45. return
  46. }
  47. marker := continuationToken
  48. if continuationToken == "" {
  49. marker = startAfter
  50. }
  51. response, err := s3a.listFilerEntries(bucket, originalPrefix, maxKeys, marker, delimiter)
  52. if err != nil {
  53. s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
  54. return
  55. }
  56. if len(response.Contents) == 0 {
  57. if exists, existErr := s3a.exists(s3a.option.BucketsPath, bucket, true); existErr == nil && !exists {
  58. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
  59. return
  60. }
  61. }
  62. responseV2 := &ListBucketResultV2{
  63. XMLName: response.XMLName,
  64. Name: response.Name,
  65. CommonPrefixes: response.CommonPrefixes,
  66. Contents: response.Contents,
  67. ContinuationToken: continuationToken,
  68. Delimiter: response.Delimiter,
  69. IsTruncated: response.IsTruncated,
  70. KeyCount: len(response.Contents) + len(response.CommonPrefixes),
  71. MaxKeys: response.MaxKeys,
  72. NextContinuationToken: response.NextMarker,
  73. Prefix: response.Prefix,
  74. StartAfter: startAfter,
  75. }
  76. writeSuccessResponseXML(w, r, responseV2)
  77. }
  78. func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Request) {
  79. // https://docs.aws.amazon.com/AmazonS3/latest/API/RESTBucketGET.html
  80. // collect parameters
  81. bucket, _ := s3_constants.GetBucketAndObject(r)
  82. glog.V(3).Infof("ListObjectsV1Handler %s", bucket)
  83. originalPrefix, marker, delimiter, maxKeys := getListObjectsV1Args(r.URL.Query())
  84. if maxKeys < 0 {
  85. s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxKeys)
  86. return
  87. }
  88. if delimiter != "" && delimiter != "/" {
  89. s3err.WriteErrorResponse(w, r, s3err.ErrNotImplemented)
  90. return
  91. }
  92. response, err := s3a.listFilerEntries(bucket, originalPrefix, maxKeys, marker, delimiter)
  93. if err != nil {
  94. s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
  95. return
  96. }
  97. if len(response.Contents) == 0 {
  98. if exists, existErr := s3a.exists(s3a.option.BucketsPath, bucket, true); existErr == nil && !exists {
  99. s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
  100. return
  101. }
  102. }
  103. writeSuccessResponseXML(w, r, response)
  104. }
  105. func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, maxKeys int, originalMarker string, delimiter string) (response ListBucketResult, err error) {
  106. // convert full path prefix into directory name and prefix for entry name
  107. requestDir, prefix, marker := normalizePrefixMarker(originalPrefix, originalMarker)
  108. bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, bucket)
  109. reqDir := bucketPrefix[:len(bucketPrefix)-1]
  110. if requestDir != "" {
  111. reqDir = fmt.Sprintf("%s%s", bucketPrefix, requestDir)
  112. }
  113. var contents []ListEntry
  114. var commonPrefixes []PrefixEntry
  115. var doErr error
  116. var nextMarker string
  117. cursor := &ListingCursor{
  118. maxKeys: maxKeys,
  119. prefixEndsOnDelimiter: strings.HasSuffix(originalPrefix, "/") && len(originalMarker) == 0,
  120. }
  121. // check filer
  122. err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
  123. for {
  124. empty := true
  125. nextMarker, doErr = s3a.doListFilerEntries(client, reqDir, prefix, cursor, marker, delimiter, false, func(dir string, entry *filer_pb.Entry) {
  126. empty = false
  127. if entry.IsDirectory {
  128. if entry.IsDirectoryKeyObject() {
  129. contents = append(contents, ListEntry{
  130. Key: fmt.Sprintf("%s/%s/", dir, entry.Name)[len(bucketPrefix):],
  131. LastModified: time.Unix(entry.Attributes.Mtime, 0).UTC(),
  132. ETag: "\"" + filer.ETag(entry) + "\"",
  133. Owner: CanonicalUser{
  134. ID: fmt.Sprintf("%x", entry.Attributes.Uid),
  135. DisplayName: entry.Attributes.UserName,
  136. },
  137. StorageClass: "STANDARD",
  138. })
  139. cursor.maxKeys--
  140. // https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html
  141. } else if delimiter == "/" { // A response can contain CommonPrefixes only if you specify a delimiter.
  142. commonPrefixes = append(commonPrefixes, PrefixEntry{
  143. Prefix: fmt.Sprintf("%s/%s/", dir, entry.Name)[len(bucketPrefix):],
  144. })
  145. //All of the keys (up to 1,000) rolled up into a common prefix count as a single return when calculating the number of returns.
  146. cursor.maxKeys--
  147. }
  148. } else {
  149. storageClass := "STANDARD"
  150. if v, ok := entry.Extended[s3_constants.AmzStorageClass]; ok {
  151. storageClass = string(v)
  152. }
  153. contents = append(contents, ListEntry{
  154. Key: fmt.Sprintf("%s/%s", dir, entry.Name)[len(bucketPrefix):],
  155. LastModified: time.Unix(entry.Attributes.Mtime, 0).UTC(),
  156. ETag: "\"" + filer.ETag(entry) + "\"",
  157. Size: int64(filer.FileSize(entry)),
  158. Owner: CanonicalUser{
  159. ID: fmt.Sprintf("%x", entry.Attributes.Uid),
  160. DisplayName: entry.Attributes.UserName,
  161. },
  162. StorageClass: StorageClass(storageClass),
  163. })
  164. cursor.maxKeys--
  165. }
  166. })
  167. if doErr != nil {
  168. return doErr
  169. }
  170. if cursor.isTruncated {
  171. if requestDir != "" {
  172. nextMarker = requestDir + "/" + nextMarker
  173. }
  174. break
  175. } else if empty || strings.HasSuffix(originalPrefix, "/") {
  176. nextMarker = ""
  177. break
  178. } else {
  179. // start next loop
  180. marker = nextMarker
  181. }
  182. }
  183. response = ListBucketResult{
  184. Name: bucket,
  185. Prefix: originalPrefix,
  186. Marker: originalMarker,
  187. NextMarker: nextMarker,
  188. MaxKeys: maxKeys,
  189. Delimiter: delimiter,
  190. IsTruncated: cursor.isTruncated,
  191. Contents: contents,
  192. CommonPrefixes: commonPrefixes,
  193. }
  194. return nil
  195. })
  196. return
  197. }
  198. type ListingCursor struct {
  199. maxKeys int
  200. isTruncated bool
  201. prefixEndsOnDelimiter bool
  202. }
  203. // the prefix and marker may be in different directories
  204. // normalizePrefixMarker ensures the prefix and marker both starts from the same directory
  205. func normalizePrefixMarker(prefix, marker string) (alignedDir, alignedPrefix, alignedMarker string) {
  206. // alignedDir should not end with "/"
  207. // alignedDir, alignedPrefix, alignedMarker should only have "/" in middle
  208. if len(marker) == 0 {
  209. prefix = strings.Trim(prefix, "/")
  210. } else {
  211. prefix = strings.TrimLeft(prefix, "/")
  212. }
  213. marker = strings.TrimLeft(marker, "/")
  214. if prefix == "" {
  215. return "", "", marker
  216. }
  217. if marker == "" {
  218. alignedDir, alignedPrefix = toDirAndName(prefix)
  219. return
  220. }
  221. if !strings.HasPrefix(marker, prefix) {
  222. // something wrong
  223. return "", prefix, marker
  224. }
  225. if strings.HasPrefix(marker, prefix+"/") {
  226. alignedDir = prefix
  227. alignedPrefix = ""
  228. alignedMarker = marker[len(alignedDir)+1:]
  229. return
  230. }
  231. alignedDir, alignedPrefix = toDirAndName(prefix)
  232. if alignedDir != "" {
  233. alignedMarker = marker[len(alignedDir)+1:]
  234. } else {
  235. alignedMarker = marker
  236. }
  237. return
  238. }
  239. func toDirAndName(dirAndName string) (dir, name string) {
  240. sepIndex := strings.LastIndex(dirAndName, "/")
  241. if sepIndex >= 0 {
  242. dir, name = dirAndName[0:sepIndex], dirAndName[sepIndex+1:]
  243. } else {
  244. name = dirAndName
  245. }
  246. return
  247. }
  248. func toParentAndDescendants(dirAndName string) (dir, name string) {
  249. sepIndex := strings.Index(dirAndName, "/")
  250. if sepIndex >= 0 {
  251. dir, name = dirAndName[0:sepIndex], dirAndName[sepIndex+1:]
  252. } else {
  253. name = dirAndName
  254. }
  255. return
  256. }
  257. func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, dir, prefix string, cursor *ListingCursor, marker, delimiter string, inclusiveStartFrom bool, eachEntryFn func(dir string, entry *filer_pb.Entry)) (nextMarker string, err error) {
  258. // invariants
  259. // prefix and marker should be under dir, marker may contain "/"
  260. // maxKeys should be updated for each recursion
  261. // glog.V(4).Infof("doListFilerEntries dir: %s, prefix: %s, marker %s, maxKeys: %d, prefixEndsOnDelimiter: %+v", dir, prefix, marker, cursor.maxKeys, cursor.prefixEndsOnDelimiter)
  262. if prefix == "/" && delimiter == "/" {
  263. return
  264. }
  265. if cursor.maxKeys <= 0 {
  266. return
  267. }
  268. if strings.Contains(marker, "/") {
  269. subDir, subMarker := toParentAndDescendants(marker)
  270. // println("doListFilerEntries dir", dir+"/"+subDir, "subMarker", subMarker)
  271. subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+subDir, "", cursor, subMarker, delimiter, false, eachEntryFn)
  272. if subErr != nil {
  273. err = subErr
  274. return
  275. }
  276. nextMarker = subDir + "/" + subNextMarker
  277. // finished processing this sub directory
  278. marker = subDir
  279. }
  280. if cursor.isTruncated {
  281. return
  282. }
  283. // now marker is also a direct child of dir
  284. request := &filer_pb.ListEntriesRequest{
  285. Directory: dir,
  286. Prefix: prefix,
  287. Limit: uint32(cursor.maxKeys + 2), // bucket root directory needs to skip additional s3_constants.MultipartUploadsFolder folder
  288. StartFromFileName: marker,
  289. InclusiveStartFrom: inclusiveStartFrom,
  290. }
  291. if cursor.prefixEndsOnDelimiter {
  292. request.Limit = uint32(1)
  293. }
  294. ctx, cancel := context.WithCancel(context.Background())
  295. defer cancel()
  296. stream, listErr := client.ListEntries(ctx, request)
  297. if listErr != nil {
  298. err = fmt.Errorf("list entires %+v: %v", request, listErr)
  299. return
  300. }
  301. for {
  302. resp, recvErr := stream.Recv()
  303. if recvErr != nil {
  304. if recvErr == io.EOF {
  305. break
  306. } else {
  307. err = fmt.Errorf("iterating entires %+v: %v", request, recvErr)
  308. return
  309. }
  310. }
  311. if cursor.maxKeys <= 0 {
  312. cursor.isTruncated = true
  313. return
  314. }
  315. entry := resp.Entry
  316. nextMarker = entry.Name
  317. if cursor.prefixEndsOnDelimiter {
  318. if entry.Name == prefix && entry.IsDirectory {
  319. if delimiter != "/" {
  320. cursor.prefixEndsOnDelimiter = false
  321. }
  322. } else {
  323. continue
  324. }
  325. }
  326. if entry.IsDirectory {
  327. // glog.V(4).Infof("List Dir Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
  328. if entry.Name == s3_constants.MultipartUploadsFolder { // FIXME no need to apply to all directories. this extra also affects maxKeys
  329. continue
  330. }
  331. if delimiter != "/" || cursor.prefixEndsOnDelimiter {
  332. if cursor.prefixEndsOnDelimiter {
  333. cursor.prefixEndsOnDelimiter = false
  334. if entry.IsDirectoryKeyObject() {
  335. eachEntryFn(dir, entry)
  336. }
  337. } else {
  338. eachEntryFn(dir, entry)
  339. }
  340. subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", cursor, "", delimiter, false, eachEntryFn)
  341. if subErr != nil {
  342. err = fmt.Errorf("doListFilerEntries2: %v", subErr)
  343. return
  344. }
  345. // println("doListFilerEntries2 dir", dir+"/"+entry.Name, "subNextMarker", subNextMarker)
  346. nextMarker = entry.Name + "/" + subNextMarker
  347. if cursor.isTruncated {
  348. return
  349. }
  350. // println("doListFilerEntries2 nextMarker", nextMarker)
  351. } else {
  352. var isEmpty bool
  353. if !s3a.option.AllowEmptyFolder && !entry.IsDirectoryKeyObject() {
  354. if isEmpty, err = s3a.ensureDirectoryAllEmpty(client, dir, entry.Name); err != nil {
  355. glog.Errorf("check empty folder %s: %v", dir, err)
  356. }
  357. }
  358. if !isEmpty {
  359. eachEntryFn(dir, entry)
  360. }
  361. }
  362. } else {
  363. eachEntryFn(dir, entry)
  364. // glog.V(4).Infof("List File Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
  365. }
  366. if cursor.prefixEndsOnDelimiter {
  367. cursor.prefixEndsOnDelimiter = false
  368. }
  369. }
  370. return
  371. }
  372. func getListObjectsV2Args(values url.Values) (prefix, token, startAfter, delimiter string, fetchOwner bool, maxkeys int) {
  373. prefix = values.Get("prefix")
  374. token = values.Get("continuation-token")
  375. startAfter = values.Get("start-after")
  376. delimiter = values.Get("delimiter")
  377. if values.Get("max-keys") != "" {
  378. maxkeys, _ = strconv.Atoi(values.Get("max-keys"))
  379. } else {
  380. maxkeys = maxObjectListSizeLimit
  381. }
  382. fetchOwner = values.Get("fetch-owner") == "true"
  383. return
  384. }
  385. func getListObjectsV1Args(values url.Values) (prefix, marker, delimiter string, maxkeys int) {
  386. prefix = values.Get("prefix")
  387. marker = values.Get("marker")
  388. delimiter = values.Get("delimiter")
  389. if values.Get("max-keys") != "" {
  390. maxkeys, _ = strconv.Atoi(values.Get("max-keys"))
  391. } else {
  392. maxkeys = maxObjectListSizeLimit
  393. }
  394. return
  395. }
  396. func (s3a *S3ApiServer) ensureDirectoryAllEmpty(filerClient filer_pb.SeaweedFilerClient, parentDir, name string) (isEmpty bool, err error) {
  397. // println("+ ensureDirectoryAllEmpty", dir, name)
  398. glog.V(4).Infof("+ isEmpty %s/%s", parentDir, name)
  399. defer glog.V(4).Infof("- isEmpty %s/%s %v", parentDir, name, isEmpty)
  400. var fileCounter int
  401. var subDirs []string
  402. currentDir := parentDir + "/" + name
  403. var startFrom string
  404. var isExhausted bool
  405. var foundEntry bool
  406. cutOffTimeAtSec := time.Now().Unix() + cutoffTimeNewEmptyDir
  407. for fileCounter == 0 && !isExhausted && err == nil {
  408. err = filer_pb.SeaweedList(filerClient, currentDir, "", func(entry *filer_pb.Entry, isLast bool) error {
  409. foundEntry = true
  410. if entry.IsDirectory {
  411. if entry.Attributes != nil && cutOffTimeAtSec >= entry.Attributes.GetCrtime() {
  412. fileCounter++
  413. } else {
  414. subDirs = append(subDirs, entry.Name)
  415. }
  416. } else {
  417. fileCounter++
  418. }
  419. startFrom = entry.Name
  420. isExhausted = isExhausted || isLast
  421. glog.V(4).Infof(" * %s/%s isLast: %t", currentDir, startFrom, isLast)
  422. return nil
  423. }, startFrom, false, 8)
  424. if !foundEntry {
  425. break
  426. }
  427. }
  428. if err != nil {
  429. return false, err
  430. }
  431. if fileCounter > 0 {
  432. return false, nil
  433. }
  434. for _, subDir := range subDirs {
  435. isSubEmpty, subErr := s3a.ensureDirectoryAllEmpty(filerClient, currentDir, subDir)
  436. if subErr != nil {
  437. return false, subErr
  438. }
  439. if !isSubEmpty {
  440. return false, nil
  441. }
  442. }
  443. glog.V(1).Infof("deleting empty folder %s", currentDir)
  444. if err = doDeleteEntry(filerClient, parentDir, name, true, true); err != nil {
  445. return
  446. }
  447. return true, nil
  448. }