[release/2.8 backport] Ignore self reference object on empty prefix

registry/storage/driver/s3-aws/s3.go

@@ -1007,6 +1007,11 @@ func (d *driver) doWalk(parentCtx context.Context, objectCount *int64, path, pre
 		}
 
 		for _, file := range objects.Contents {
+			// empty prefixes are listed as objects inside its own prefix.
+			// https://docs.aws.amazon.com/AmazonS3/latest/user-guide/using-folders.html
+			if strings.HasSuffix(*file.Key, "/") {
+				continue
+			}
 			walkInfos = append(walkInfos, walkInfoContainer{
 				FileInfoFields: storagedriver.FileInfoFields{
 					IsDir:   false,

registry/storage/driver/s3-aws/s3_test.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"io/ioutil"
 	"os"
+	"reflect"
 	"strconv"
 	"testing"
 
@@ -164,6 +165,39 @@ func TestEmptyRootList(t *testing.T) {
 	}
 }
 
+// TestWalkEmptySubDirectory assures we list an empty sub directory only once when walking
+// through its parent directory.
+func TestWalkEmptySubDirectory(t *testing.T) {
+	if skipS3() != "" {
+		t.Skip(skipS3())
+	}
+
+	drv, err := s3DriverConstructor("", s3.StorageClassStandard)
+	if err != nil {
+		t.Fatalf("unexpected error creating rooted driver: %v", err)
+	}
+
+	// create an empty sub directory.
+	s3driver := drv.StorageDriver.(*driver)
+	if _, err := s3driver.S3.PutObject(&s3.PutObjectInput{
+		Bucket: aws.String(os.Getenv("S3_BUCKET")),
+		Key:    aws.String("/testdir/emptydir/"),
+	}); err != nil {
+		t.Fatalf("error creating empty directory: %s", err)
+	}
+
+	bucketFiles := []string{}
+	s3driver.Walk(context.Background(), "/testdir", func(fileInfo storagedriver.FileInfo) error {
+		bucketFiles = append(bucketFiles, fileInfo.Path())
+		return nil
+	})
+
+	expected := []string{"/testdir/emptydir"}
+	if !reflect.DeepEqual(bucketFiles, expected) {
+		t.Errorf("expecting files %+v, found %+v instead", expected, bucketFiles)
+	}
+}
+
 func TestStorageClass(t *testing.T) {
 	if skipS3() != "" {
 		t.Skip(skipS3())