[S3 API]Support list objectV2 S3 API (#2775)

* [S3 API]Support list objectV2 S3 API

* Address comments

* Address comment

---------

Co-authored-by: Sophie Guo <sopguo@sopguo-mn2.linkedin.biz>

ambry-api/src/main/java/com/github/ambry/frontend/NamedBlobPath.java

@@ -35,7 +35,11 @@ public class NamedBlobPath {
 
   static final String PREFIX_PARAM = "prefix";
   static final String PAGE_PARAM = "page";
+  static final String MARKER = "Marker";
+  static final String CONTINUATION_TOKEN = "ContinuationToken";
   static final int MAX_BLOB_NAME_LENGTH = 350;
+  static final String LIST_TYPE = "list-type";
+  static final String LIST_TYPE_VERSION_2 = "2";
 
   /**
    * Parse the input path if it's a named blob request.
@@ -90,7 +94,11 @@ public static NamedBlobPath parseS3(String path, Map<String, Object> args) throw
     String[] splitPath = path.split("/", 4);
     String blobNamePrefix = RestUtils.getHeader(args, PREFIX_PARAM, false);
     boolean isGetObjectLockRequest = args.containsKey(OBJECT_LOCK_PARAM);
-    boolean isListRequest = blobNamePrefix != null;
+    //There are two cases for S3 listing
+    //1.has prefix (Ex:GET /?prefix=prefixName&delimiter=&encoding-type=url)
+    //2.no prefix but listObjectV2 (Ex:GET /?list-type=2&prefix=&delimiter=&encoding-type=url)
+    boolean isListObjectV2Request = LIST_TYPE_VERSION_2.equals(RestUtils.getHeader(args, LIST_TYPE, false));
+    boolean isListRequest = blobNamePrefix != null || isListObjectV2Request;
     int expectedSegments = (isListRequest || isGetObjectLockRequest) ? 3 : 4;
     if (splitPath.length != expectedSegments || !Operations.NAMED_BLOB.equalsIgnoreCase(splitPath[0])) {
       throw new RestServiceException(String.format(
@@ -99,8 +107,15 @@ public static NamedBlobPath parseS3(String path, Map<String, Object> args) throw
     }
     String accountName = splitPath[1];
     String containerName = splitPath[2];
+    String pageToken;
+    //S3 listObject use Marker as page token.
+    //S3 listObjectV2 use ContinuationToken as page token.
     if (isListRequest) {
-      String pageToken = RestUtils.getHeader(args, PAGE_PARAM, false);
+      if (isListObjectV2Request) {
+        pageToken = RestUtils.getHeader(args, CONTINUATION_TOKEN, false);
+      } else {
+        pageToken = RestUtils.getHeader(args, MARKER, false);
+      }
       return new NamedBlobPath(accountName, containerName, null, blobNamePrefix, pageToken);
     }
     if (isGetObjectLockRequest) {
@@ -115,6 +130,7 @@ public static NamedBlobPath parseS3(String path, Map<String, Object> args) throw
       return new NamedBlobPath(accountName, containerName, blobName, null, null);
     }
   }
+
   /**
    * Parse the input path if it's a named blob request.
    * @param requestPath the {@link RequestPath} to be parsed.

ambry-api/src/main/java/com/github/ambry/frontend/s3/S3MessagePayload.java

@@ -222,22 +222,27 @@ public static class Contents {
     private String key;
     @JacksonXmlProperty(localName = "LastModified")
     private String lastModified;
+    @JacksonXmlProperty(localName = "Size")
+    private long size;
 
     private Contents() {
     }
 
-    public Contents(String key, String lastModified) {
+    public Contents(String key, String lastModified, long size) {
       this.key = key;
       this.lastModified = lastModified;
+      this.size = size;
     }
 
     public String getKey() {
       return key;
     }
 
+    public long getSize() { return size; }
+
     @Override
     public String toString() {
-      return "Key=" + key + ", " + "LastModified=" + lastModified;
+      return "Key=" + key + ", " + "LastModified=" + lastModified + ", " + "Size=" + size;
     }
   }
 

ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3ListHandler.java

@@ -54,6 +54,7 @@ public class S3ListHandler extends S3BaseHandler<ReadableStreamChannel> {
   private static final ObjectMapper xmlMapper = new XmlMapper();
   public static final String PREFIX_PARAM_NAME = "prefix";
   public static final String MAXKEYS_PARAM_NAME = "max-keys";
+  public static final int DEFAULT_MAX_KEY_VALUE = 1000;
   public static final String DELIMITER_PARAM_NAME = "delimiter";
   public static final String ENCODING_TYPE_PARAM_NAME = "encoding-type";
   private final NamedBlobListHandler namedBlobListHandler;
@@ -99,14 +100,16 @@ private ReadableStreamChannel serializeAsXml(RestRequest restRequest, Page<Named
     String delimiter = getHeader(restRequest.getArgs(), DELIMITER_PARAM_NAME, false);
     String encodingType = getHeader(restRequest.getArgs(), ENCODING_TYPE_PARAM_NAME, false);
     String maxKeys = getHeader(restRequest.getArgs(), MAXKEYS_PARAM_NAME, false);
-    int maxKeysValue = maxKeys == null ? Integer.MAX_VALUE : Integer.parseInt(maxKeys);
+    //By default S3 list returns up to 1000 key names.
+    int maxKeysValue = maxKeys == null ? DEFAULT_MAX_KEY_VALUE : Integer.parseInt(maxKeys);
     // Iterate through list of blob names.
     List<Contents> contentsList = new ArrayList<>();
     int keyCount = 0;
     for (NamedBlobListEntry namedBlobRecord : namedBlobRecordPage.getEntries()) {
       String blobName = namedBlobRecord.getBlobName();
       String todayDate = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ").format(Calendar.getInstance().getTime());
-      contentsList.add(new Contents(blobName, todayDate));
+      //Set size to -1 due to current use case don't need this value.
+      contentsList.add(new Contents(blobName, todayDate, -1));
       if (++keyCount == maxKeysValue) {
         break;
       }

ambry-frontend/src/test/java/com/github/ambry/frontend/S3ListHandlerTest.java

@@ -107,7 +107,7 @@ public void listS3BlobsTest() throws Exception {
     // 2. Get list of blobs by sending matching s3 request
     String s3_list_request_uri =
         S3_PREFIX + SLASH + account.getName() + SLASH + container.getName() + SLASH + "?prefix=" + PREFIX
-            + "&delimiter=/" + "&max-keys=1" + "&encoding-type=url";
+            + "&delimiter=/" + "&Marker=/" + "&max-keys=1" + "&encoding-type=url";
     request =
         FrontendRestRequestServiceTest.createRestRequest(RestMethod.GET, s3_list_request_uri, new JSONObject(), null);
     request.setArg(InternalKeys.REQUEST_PATH,
@@ -133,6 +133,51 @@ public void listS3BlobsTest() throws Exception {
     assertEquals("Mismatch in encoding type", "url", listBucketResult.getEncodingType());
   }
 
+  @Test
+  public void listObjectV2S3BlobsTest() throws Exception {
+    // 1. Put a named blob
+    String PREFIX = "directory-name";
+    String KEY_NAME = PREFIX + SLASH + "key_name";
+    String request_path =
+        NAMED_BLOB_PREFIX + SLASH + account.getName() + SLASH + container.getName() + SLASH + KEY_NAME;
+    JSONObject headers = new JSONObject();
+    FrontendRestRequestServiceTest.setAmbryHeadersForPut(headers, TestUtils.TTL_SECS, container.isCacheable(),
+        SERVICE_ID, CONTENT_TYPE, OWNER_ID, null, null, null);
+    byte[] content = TestUtils.getRandomBytes(1024);
+    RestRequest request = FrontendRestRequestServiceTest.createRestRequest(RestMethod.PUT, request_path, headers,
+        new LinkedList<>(Arrays.asList(ByteBuffer.wrap(content), null)));
+    request.setArg(InternalKeys.REQUEST_PATH,
+        RequestPath.parse(request, frontendConfig.pathPrefixesToRemove, CLUSTER_NAME));
+    RestResponseChannel restResponseChannel = new MockRestResponseChannel();
+    FutureResult<Void> putResult = new FutureResult<>();
+    namedBlobPutHandler.handle(request, restResponseChannel, putResult::done);
+    putResult.get();
+
+    // 2. Get list of blobs by sending matching s3 list object v2 request
+    String s3_list_request_uri =
+        S3_PREFIX + SLASH + account.getName() + SLASH + container.getName() + SLASH + "?list-type=2" + "&prefix="
+            + "&delimiter=/" + "&ContinuationToken=/" + "&encoding-type=url";
+    request =
+        FrontendRestRequestServiceTest.createRestRequest(RestMethod.GET, s3_list_request_uri, new JSONObject(), null);
+    request.setArg(InternalKeys.REQUEST_PATH,
+        RequestPath.parse(request, frontendConfig.pathPrefixesToRemove, CLUSTER_NAME));
+    restResponseChannel = new MockRestResponseChannel();
+    FutureResult<ReadableStreamChannel> futureResult = new FutureResult<>();
+    s3ListHandler.handle(request, restResponseChannel, futureResult::done);
+
+    // 3. Verify results
+    ReadableStreamChannel readableStreamChannel = futureResult.get();
+    ByteBuffer byteBuffer = ((ByteBufferReadableStreamChannel) readableStreamChannel).getContent();
+    ListBucketResult listBucketResult = xmlMapper.readValue(byteBuffer.array(), ListBucketResult.class);
+    assertEquals("Mismatch on status", ResponseStatus.Ok, restResponseChannel.getStatus());
+    assertEquals("Mismatch in content type", XML_CONTENT_TYPE, restResponseChannel.getHeader(Headers.CONTENT_TYPE));
+    Contents contents = listBucketResult.getContents().get(0);
+    assertEquals("Mismatch in key name", KEY_NAME, contents.getKey());
+    assertEquals("Mismatch in key count", 1, listBucketResult.getKeyCount());
+    assertEquals("Mismatch in delimiter", "/", listBucketResult.getDelimiter());
+    assertEquals("Mismatch in encoding type", "url", listBucketResult.getEncodingType());
+    assertEquals("Mismatch in size", -1, contents.getSize());
+  }
 
   /**
    * Initates a {@link NamedBlobPutHandler} and a {@link S3ListHandler}

ambry-named-mysql/src/integration-test/java/com/github/ambry/named/MySqlNamedBlobDbIntegrationTest.java

@@ -128,6 +128,20 @@ public void testPutGetListDeleteSequence() throws Exception {
       }
     }
 
+    // list all records in each container.
+    time.setCurrentMilliseconds(System.currentTimeMillis());
+    for (Account account : accountService.getAllAccounts()) {
+      for (Container container : account.getAllContainers()) {
+        //page with no token
+        Page<NamedBlobRecord> page = namedBlobDb.list(account.getName(), container.getName(), null, null).get();
+        assertNull("No continuation token expected", page.getNextPageToken());
+        assertEquals("Unexpected number of blobs in container", blobsPerContainer, page.getEntries().size());
+        //page with token
+        Page<NamedBlobRecord> pageWithToken = namedBlobDb.list(account.getName(), container.getName(), null, "name/4").get();
+        assertEquals("Unexpected number of blobs in container", blobsPerContainer / 5, pageWithToken.getEntries().size());
+      }
+    }
+
     // check that puts to the same keys fail.
     time.setCurrentMilliseconds(System.currentTimeMillis());
     for (Account account : accountService.getAllAccounts()) {