{miniogw,testsuite/miniogw}: allow ListObjects(V2) to be fully compliant

This change adds --s3.fully-compatible-listing (default: false)
configuration attribute/flag configuring ListObjects/ListObjectsV2 to be
fully S3-compatible (but slow), e.g. maintain lexicographical ordering
of listing results.

Closes #49

Change-Id: I378b28417baa54045e05baf5002e41e18dc0a32c

miniogw/config.go

@@ -21,6 +21,7 @@ type S3CompatibilityConfig struct {
 	IncludeCustomMetadataListing bool `help:"include custom metadata in S3's ListObjects, ListObjectsV2 and ListMultipartUploads responses" default:"true"`
 	MaxKeysLimit                 int  `help:"MaxKeys parameter limit for S3's ListObjects and ListObjectsV2 responses" default:"1000"`
 	MaxKeysExhaustiveLimit       int  `help:"maximum number of items to list for gateway-side filtering using arbitrary delimiter/prefix" default:"100000"`
+	FullyCompatibleListing       bool `help:"make ListObjects(V2) fully S3-compatible (specifically: always return lexicographically ordered results) but slow" default:"false"`
 	DisableCopyObject            bool `help:"return 501 (Not Implemented) for CopyObject calls" devDefault:"false" releaseDefault:"true"`
 	MinPartSize                  int  `help:"minimum part size for multipart uploads" default:"5242880"` // 5 MiB
 }

miniogw/gateway.go

@@ -439,14 +439,9 @@ func (layer *gatewayLayer) listObjectsExhaustive(
 ) {
 	defer mon.Task()(&ctx)(&err)
 
-	// From where listObjectsExhaustive is called, only one of the following can
-	// be true (or should be):
-	//
-	//  - prefix is empty or ends with "/";
-	//  - delimiter is empty or is "/".
-	//
-	// Filling Prefix and Recursive are a few optimizations that try to exploit
-	// this fact.
+	// Filling Prefix and Recursive are a few optimizations that try to make
+	// exhaustive listing less resource-intensive if it's possible. We still
+	// have to comply with (*uplink.Project).ListObjects' API.
 	var listPrefix string
 
 	if strings.HasSuffix(prefix, "/") {
@@ -455,7 +450,7 @@ func (layer *gatewayLayer) listObjectsExhaustive(
 
 	list := project.ListObjects(ctx, bucket, &uplink.ListObjectsOptions{
 		Prefix:    listPrefix,
-		Recursive: delimiter != "/" || strings.Contains(prefix, "/"),
+		Recursive: delimiter != "/" || (strings.Contains(prefix, "/") && !strings.HasSuffix(prefix, "/")),
 		System:    true,
 		Custom:    layer.compatibilityConfig.IncludeCustomMetadataListing,
 	})
@@ -556,6 +551,9 @@ func (layer *gatewayLayer) listObjectsExhaustive(
 // The only S3-incompatible thing that listObjectsGeneral represents is that for
 // fast listing, it will not return items lexicographically ordered, but it is a
 // burden we must all bear.
+//
+// If layer.compatibilityConfig.FullyCompatibleListing is true, it will always
+// list exhaustively to achieve full S3 compatibility. Use at your own risk.
 func (layer *gatewayLayer) listObjectsGeneral(
 	ctx context.Context,
 	project *uplink.Project,
@@ -573,7 +571,7 @@ func (layer *gatewayLayer) listObjectsGeneral(
 
 	supportedPrefix := prefix == "" || strings.HasSuffix(prefix, "/")
 
-	if supportedPrefix && (delimiter == "" || delimiter == "/") {
+	if !layer.compatibilityConfig.FullyCompatibleListing && supportedPrefix && (delimiter == "" || delimiter == "/") {
 		prefixes, objects, token, err = layer.listObjectsFast(
 			ctx,
 			project,

testsuite/miniogw/gateway_test.go

@@ -2022,6 +2022,271 @@ func testListObjectsLimits(t *testing.T, listObjects listObjectsFunc) {
 	})
 }
 
+// TestListObjectsFullyCompatible tests whether exhaustive listing's results
+// are, e.g., lexicographically ordered. It also tests cases when incorrectly
+// implemented obvious exhaustive listing optimizations (in the context of
+// libuplink API) might return wrong results. It cannot test whether such
+// optimizations exist (there are other ways to check this).
+func TestListObjectsFullyCompatible(t *testing.T) {
+	t.Parallel()
+
+	bucketName := testrand.BucketName()
+
+	const (
+		prefix  = "moHaq/" // "moHaq" is "prefix" in Klingon.
+		prefix2 = "ab/cd/"
+	)
+
+	var paths, prefixed []string
+
+	for i := 0; i < 100; i++ {
+		p := testrand.Path()
+
+		if i%2 == 0 {
+			p = prefix + prefix2 + p
+
+			if len(p) > 1024 {
+				p = p[:1024]
+			}
+
+			prefixed = append(prefixed, p)
+		}
+
+		paths = append(paths, p)
+	}
+
+	sort.Strings(paths)
+	sort.Strings(prefixed)
+
+	testplanet.Run(t, testplanet.Config{
+		SatelliteCount:   1,
+		StorageNodeCount: 4,
+		UplinkCount:      1,
+	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
+		access, err := setupAccess(ctx, t, planet, storj.EncNull, uplink.FullPermission())
+		require.NoError(t, err)
+
+		project, err := uplink.OpenProject(ctx, access)
+		require.NoError(t, err)
+
+		defer func() { require.NoError(t, project.Close()) }()
+
+		// Establish new context with *uplink.Project for the gateway to pick up.
+		ctxWithProject := miniogw.WithUplinkProject(ctx, project)
+
+		s3Compatibility := miniogw.S3CompatibilityConfig{
+			IncludeCustomMetadataListing: true,
+			MaxKeysLimit:                 51, // +1 to list 50 (see storj.io/gateway/miniogw.(*gatewayLayer).limitMaxKeys)
+			MaxKeysExhaustiveLimit:       100,
+			FullyCompatibleListing:       true, // this is what's important here
+		}
+
+		layer, err := miniogw.NewStorjGateway(s3Compatibility).NewGatewayLayer(auth.Credentials{})
+		require.NoError(t, err)
+
+		defer func() { require.NoError(t, layer.Shutdown(ctxWithProject)) }()
+
+		_, err = project.EnsureBucket(ctxWithProject, bucketName)
+		require.NoError(t, err)
+
+		for _, p := range paths {
+			_, err = createFile(ctxWithProject, project, bucketName, p, []byte{'t', 'e', 's', 't'}, nil)
+			require.NoError(t, err)
+		}
+
+		{
+			const name = "mustn't trigger optimizations: all"
+
+			var objects []minio.ObjectInfo
+
+			for isTruncated, nextMarker := true, ""; isTruncated; {
+				result, err := layer.ListObjects(ctxWithProject, bucketName, "", nextMarker, "", 100)
+				require.NoError(t, err, name)
+
+				isTruncated, nextMarker = result.IsTruncated, result.NextMarker
+
+				assert.Empty(t, result.Prefixes, name)
+
+				objects = append(objects, result.Objects...)
+			}
+
+			require.Len(t, objects, len(paths), name)
+
+			for i, p := range paths {
+				assert.Equal(t, p, objects[i].Name, name, i)
+			}
+		}
+		{
+			const name = "mustn't trigger optimizations: all (limit check)"
+
+			result, err := layer.ListObjects(ctxWithProject, bucketName, "", "", "", 100)
+			require.NoError(t, err, name)
+
+			assert.True(t, result.IsTruncated, name)
+			assert.Equal(t, result.NextMarker, paths[49], name)
+			assert.Empty(t, result.Prefixes, name)
+
+			require.Len(t, result.Objects, len(paths[:50]), name)
+
+			for i, p := range paths[:50] {
+				assert.Equal(t, p, result.Objects[i].Name, name, i)
+			}
+		}
+		{
+			const name = "mustn't trigger optimizations: prefix without trailing forward slash"
+
+			var objects []minio.ObjectInfo
+
+			for isTruncated, nextMarker := true, ""; isTruncated; {
+				result, err := layer.ListObjects(ctxWithProject, bucketName, strings.TrimSuffix(prefix, "/"), nextMarker, "", 10)
+				require.NoError(t, err, name)
+
+				isTruncated, nextMarker = result.IsTruncated, result.NextMarker
+
+				assert.Empty(t, result.Prefixes, name)
+
+				objects = append(objects, result.Objects...)
+			}
+
+			require.Len(t, objects, len(prefixed), name)
+
+			for i, p := range prefixed {
+				assert.Equal(t, p, objects[i].Name, name, i)
+			}
+		}
+		{
+			const name = "mustn't trigger optimizations: prefix without trailing forward slash contains forward slashes + forward slash delimiter"
+
+			// Set marker to something before prefix, so we don't trigger our
+			// optimization for non-forward-slash-terminated prefixes. If we
+			// triggered it, we are still fully compatible, but we specifically
+			// want to test the fallback to the exhaustive listing.
+			result, err := layer.ListObjects(ctxWithProject, bucketName, strings.TrimSuffix(prefix+prefix2, "/"), string(prefix[0]), "/", 100)
+			require.NoError(t, err, name)
+
+			assert.False(t, result.IsTruncated, name)
+			assert.Empty(t, result.NextMarker, name)
+			assert.Empty(t, result.Objects, name)
+
+			require.Len(t, result.Prefixes, 1, name)
+
+			assert.Equal(t, prefix+prefix2, result.Prefixes[0], name)
+		}
+		{
+			const name = "mustn't trigger optimizations: arbitrary delimiter"
+
+			var (
+				objects  []minio.ObjectInfo
+				prefixes []string
+			)
+
+			for isTruncated, nextMarker := true, ""; isTruncated; {
+				result, err := layer.ListObjects(ctxWithProject, bucketName, "", nextMarker, prefix2, 100)
+				require.NoError(t, err, name)
+
+				isTruncated, nextMarker = result.IsTruncated, result.NextMarker
+
+				objects = append(objects, result.Objects...)
+				prefixes = append(prefixes, result.Prefixes...)
+			}
+
+			require.Len(t, prefixes, 1, name)
+
+			assert.Equal(t, prefix+prefix2, prefixes[0], name)
+
+			var withoutPrefix []string
+
+			for _, p := range paths {
+				if !strings.HasPrefix(p, prefix+prefix2) {
+					withoutPrefix = append(withoutPrefix, p)
+				}
+			}
+
+			require.Len(t, objects, len(withoutPrefix), name)
+
+			for i, p := range withoutPrefix {
+				assert.Equal(t, p, objects[i].Name, name, i)
+			}
+		}
+		{
+			const name = "mustn't trigger optimizations: prefix without trailing forward slash + arbitrary delimiter"
+
+			result, err := layer.ListObjects(ctxWithProject, bucketName, strings.TrimSuffix(prefix, "/"), "", prefix2, 100)
+			require.NoError(t, err, name)
+
+			assert.False(t, result.IsTruncated, name)
+			assert.Empty(t, result.NextMarker, name)
+			assert.Empty(t, result.Objects, name)
+
+			require.Len(t, result.Prefixes, 1, name)
+
+			assert.Equal(t, prefix+prefix2, result.Prefixes[0], name)
+		}
+		{
+			const name = "partially optimized: prefix"
+
+			result, err := layer.ListObjects(ctxWithProject, bucketName, prefix, "", "", 100)
+			require.NoError(t, err, name)
+
+			assert.False(t, result.IsTruncated, name)
+			assert.Empty(t, result.NextMarker, name)
+			assert.Empty(t, result.Prefixes, name)
+
+			require.Len(t, result.Objects, len(prefixed), name)
+
+			for i, p := range prefixed {
+				assert.Equal(t, p, result.Objects[i].Name, name, i)
+			}
+		}
+		{
+			const name = "partially optimized: prefix without trailing forward slash + forward slash delimiter"
+
+			// Set marker to something before prefix, so we don't trigger our
+			// optimization for non-forward-slash-terminated prefixes. If we
+			// triggered it, we are still fully compatible, but we specifically
+			// want to test the fallback to the exhaustive listing.
+			result, err := layer.ListObjects(ctxWithProject, bucketName, strings.TrimSuffix(prefix, "/"), string(prefix[0]), "/", 100)
+			require.NoError(t, err, name)
+
+			assert.False(t, result.IsTruncated, name)
+			assert.Empty(t, result.NextMarker, name)
+			assert.Empty(t, result.Objects, name)
+
+			require.Len(t, result.Prefixes, 1, name)
+
+			assert.Equal(t, prefix, result.Prefixes[0], name)
+		}
+		{
+			const name = "partially optimized: prefix + arbitrary delimiter"
+
+			result, err := layer.ListObjects(ctxWithProject, bucketName, prefix, "", strings.SplitAfterN(prefix2, "/", 2)[1], 100)
+			require.NoError(t, err, name)
+
+			assert.False(t, result.IsTruncated, name)
+			assert.Empty(t, result.NextMarker, name)
+			assert.Empty(t, result.Objects, name)
+
+			require.Len(t, result.Prefixes, 1, name)
+
+			assert.Equal(t, prefix+prefix2, result.Prefixes[0], name)
+		}
+		{
+			const name = "optimized: prefix + forward slash delimiter"
+
+			result, err := layer.ListObjects(ctxWithProject, bucketName, prefix, "", "/", 100)
+			require.NoError(t, err, name)
+
+			assert.False(t, result.IsTruncated, name)
+			assert.Empty(t, result.NextMarker, name)
+			assert.Empty(t, result.Objects, name)
+
+			require.Len(t, result.Prefixes, 1, name)
+
+			assert.Equal(t, prefix+strings.SplitAfterN(prefix2, "/", 2)[0], result.Prefixes[0], name)
+		}
+	})
+}
+
 func TestListMultipartUploads(t *testing.T) {
 	t.Parallel()