JCLOUDS-1450: Use S3-style ETags for MPUs.

S3 uses a different ETag for multipart uploads (MPUs) than regular
objects. The ETag consists of the md5 hash of the concatenated ETags of
individual parts followed by the number of parts (separated by "-").

The patch changes the LocalBlobStore's implementation of
CompleteMultipartUpload to set the S3-style ETag before calling
putBlob() and return that ETag to the caller.

To simplify testing, a new protected method with a default NOOP
implementation is added to the BaseBlobIntegrationTest. It allows
providers to further verify MPUs (i.e. ensuring the correct ETag has
been stored alongside the object).

apis/filesystem/src/main/java/org/jclouds/filesystem/strategy/internal/FilesystemStorageStrategyImpl.java

@@ -16,6 +16,7 @@
  */
 package org.jclouds.filesystem.strategy.internal;
 
+import static com.google.common.base.Charsets.US_ASCII;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Strings.isNullOrEmpty;
 import static com.google.common.io.BaseEncoding.base16;
@@ -35,6 +36,7 @@ import static org.jclouds.util.Closeables2.closeQuietly;
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.AccessDeniedException;
@@ -48,6 +50,7 @@ import java.util.Date;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.regex.Pattern;
 
 import javax.annotation.Resource;
 import javax.inject.Inject;
@@ -115,6 +118,7 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
    private static final String XATTR_USER_METADATA_PREFIX = "user.user-metadata.";
    private static final byte[] DIRECTORY_MD5 =
            Hashing.md5().hashBytes(new byte[0]).asBytes();
+   private static final Pattern MPU_ETAG_FORMAT = Pattern.compile("\"[a-f0-9]{32}-\\d+\"");
 
    @Resource
    protected Logger logger = Logger.NULL;
@@ -353,6 +357,7 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
          String contentLanguage = null;
          String contentType = null;
          HashCode hashCode = null;
+         String eTag = null;
          Date expires = null;
          Tier tier = Tier.STANDARD;
          ImmutableMap.Builder<String, String> userMetadata = ImmutableMap.builder();
@@ -373,7 +378,15 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
                if (attributes.contains(XATTR_CONTENT_MD5)) {
                   ByteBuffer buf = ByteBuffer.allocate(view.size(XATTR_CONTENT_MD5));
                   view.read(XATTR_CONTENT_MD5, buf);
-                  hashCode = HashCode.fromBytes(buf.array());
+                  byte [] etagBytes = buf.array();
+                  if (etagBytes.length == 16) {
+                     // regular object
+                     hashCode = HashCode.fromBytes(buf.array());
+                     eTag = "\"" + hashCode + "\"";
+                  } else {
+                     // multi-part object
+                     eTag = new String(etagBytes, US_ASCII);
+                  }
                }
                if (attributes.contains(XATTR_EXPIRES)) {
                   ByteBuffer buf = ByteBuffer.allocate(view.size(XATTR_EXPIRES));
@@ -403,6 +416,7 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
                .contentLanguage(contentLanguage)
                .contentLength(byteSource.size())
                .contentMD5(hashCode)
+               .eTag(eTag)
                .contentType(contentType)
                .expires(expires)
                .tier(tier)
@@ -488,23 +502,36 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
       String tmpBlobName = blobKey + "-" + UUID.randomUUID();
       File tmpFile = getFileForBlobKey(containerName, tmpBlobName);
       Path tmpPath = tmpFile.toPath();
-      HashingInputStream his = null;
+      boolean isMpu = false;
+      if (blob.getMetadata() != null && blob.getMetadata().getETag() != null)
+         isMpu = MPU_ETAG_FORMAT.matcher(blob.getMetadata().getETag()).matches();
+      InputStream inputStream = null;
+      byte[] eTag = null;
       try {
          Files.createParentDirs(tmpFile);
-         his = new HashingInputStream(Hashing.md5(), payload.openStream());
+         if (isMpu) {
-         long actualSize = Files.asByteSink(tmpFile).writeFrom(his);
+            inputStream = payload.openStream();
+            eTag = blob.getMetadata().getETag().getBytes();
+         } else {
+            inputStream = new HashingInputStream(Hashing.md5(), payload.openStream());
+         }
+         long actualSize = Files.asByteSink(tmpFile).writeFrom(inputStream);
          Long expectedSize = blob.getMetadata().getContentMetadata().getContentLength();
          if (expectedSize != null && actualSize != expectedSize) {
             throw new IOException("Content-Length mismatch, actual: " + actualSize +
                   " expected: " + expectedSize);
          }
-         HashCode actualHashCode = his.hash();
+
-         HashCode expectedHashCode = payload.getContentMetadata().getContentMD5AsHashCode();
+         if (!isMpu) {
-         if (expectedHashCode != null && !actualHashCode.equals(expectedHashCode)) {
+            HashCode actualHashCode = ((HashingInputStream) inputStream).hash();
-            throw new IOException("MD5 hash code mismatch, actual: " + actualHashCode +
+            HashCode expectedHashCode = payload.getContentMetadata().getContentMD5AsHashCode();
-                  " expected: " + expectedHashCode);
+            if (expectedHashCode != null && !actualHashCode.equals(expectedHashCode)) {
+               throw new IOException("MD5 hash code mismatch, actual: " + actualHashCode +
+                       " expected: " + expectedHashCode);
+            }
+            payload.getContentMetadata().setContentMD5(actualHashCode);
+            eTag = actualHashCode.asBytes();
          }
-         payload.getContentMetadata().setContentMD5(actualHashCode);
 
          if (outputFile.exists()) {
             delete(outputFile);
@@ -513,7 +540,7 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
          UserDefinedFileAttributeView view = getUserDefinedFileAttributeView(tmpPath);
          if (view != null) {
             try {
-               view.write(XATTR_CONTENT_MD5, ByteBuffer.wrap(actualHashCode.asBytes()));
+               view.write(XATTR_CONTENT_MD5, ByteBuffer.wrap(eTag));
                writeCommonMetadataAttr(view, blob);
             } catch (IOException e) {
                logger.debug("xattrs not supported on %s", tmpPath);
@@ -527,7 +554,7 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
          }
          tmpFile = null;
 
-         return base16().lowerCase().encode(actualHashCode.asBytes());
+         return base16().lowerCase().encode(eTag);
       } finally {
          if (tmpFile != null) {
             try {
@@ -536,7 +563,7 @@ public class FilesystemStorageStrategyImpl implements LocalStorageStrategy {
                logger.debug("Could not delete %s: %s", tmpFile, e);
             }
          }
-         closeQuietly(his);
+         closeQuietly(inputStream);
          if (payload != null) {
             payload.release();
          }

apis/filesystem/src/test/java/org/jclouds/filesystem/integration/FilesystemBlobIntegrationTest.java

@@ -19,11 +19,17 @@ package org.jclouds.filesystem.integration;
 import static org.jclouds.filesystem.util.Utils.isMacOSX;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import static java.nio.charset.StandardCharsets.US_ASCII;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.common.hash.Hashing;
 import org.jclouds.blobstore.domain.Blob;
 import org.jclouds.blobstore.domain.BlobMetadata;
+import org.jclouds.blobstore.domain.MultipartPart;
 import org.jclouds.blobstore.domain.Tier;
 import org.jclouds.blobstore.integration.internal.BaseBlobIntegrationTest;
 import org.jclouds.blobstore.integration.internal.BaseBlobStoreIntegrationTest;
@@ -104,6 +110,22 @@ public class FilesystemBlobIntegrationTest extends BaseBlobIntegrationTest {
       throw new SkipException("filesystem does not support anonymous access");
    }
 
+   @Override
+   protected void checkMPUParts(Blob blob, List<MultipartPart> partsList) {
+      assertThat(blob.getMetadata().getETag()).endsWith(String.format("-%d\"", partsList.size()));
+      StringBuilder eTags = new StringBuilder();
+      for (MultipartPart part : partsList) {
+         eTags.append(part.partETag());
+      }
+      String expectedETag = new StringBuilder("\"")
+         .append(Hashing.md5().hashString(eTags.toString(), US_ASCII))
+         .append("-")
+         .append(partsList.size())
+         .append("\"")
+         .toString();
+      assertThat(blob.getMetadata().getETag()).isEqualTo(expectedETag);
+   }
+
    protected void checkExtendedAttributesSupport() {
       if (isMacOSX()) {
          throw new SkipException("filesystem does not support extended attributes in Mac OSX");

blobstore/src/main/java/org/jclouds/blobstore/config/LocalBlobStore.java

@@ -25,6 +25,7 @@ import static com.google.common.collect.Iterables.transform;
 import static com.google.common.collect.Iterables.tryFind;
 import static com.google.common.collect.Sets.filter;
 import static com.google.common.collect.Sets.newTreeSet;
+import static java.nio.charset.StandardCharsets.US_ASCII;
 import static org.jclouds.blobstore.options.ListContainerOptions.Builder.recursive;
 
 import java.io.File;
@@ -45,6 +46,7 @@ import javax.annotation.Resource;
 import javax.inject.Inject;
 import javax.inject.Singleton;
 
+import com.google.common.hash.Hashing;
 import org.jclouds.blobstore.BlobStore;
 import org.jclouds.blobstore.BlobStoreContext;
 import org.jclouds.blobstore.ContainerNotFoundException;
@@ -824,6 +826,8 @@ public final class LocalBlobStore implements BlobStore {
    public String completeMultipartUpload(MultipartUpload mpu, List<MultipartPart> parts) {
       ImmutableList.Builder<InputStream> streams = ImmutableList.builder();
       long contentLength = 0;
+      StringBuilder partHashes = new StringBuilder();
+
       for (MultipartPart part : parts) {
          Blob blobPart = getBlob(mpu.containerName(), MULTIPART_PREFIX + mpu.id() + "-" + mpu.blobName() + "-" + part.partNumber());
          contentLength += blobPart.getMetadata().getContentMetadata().getContentLength();
@@ -834,11 +838,19 @@ public final class LocalBlobStore implements BlobStore {
             throw propagate(ioe);
          }
          streams.add(is);
+         partHashes.append(blobPart.getMetadata().getETag());
       }
+      String mpuETag = new StringBuilder("\"")
+         .append(Hashing.md5().hashString(partHashes.toString(), US_ASCII).toString())
+         .append("-")
+         .append(parts.size())
+         .append("\"")
+         .toString();
       PayloadBlobBuilder blobBuilder = blobBuilder(mpu.blobName())
             .userMetadata(mpu.blobMetadata().getUserMetadata())
             .payload(new SequenceInputStream(Iterators.asEnumeration(streams.build().iterator())))
-            .contentLength(contentLength);
+            .contentLength(contentLength)
+            .eTag(mpuETag);
       String cacheControl = mpu.blobMetadata().getContentMetadata().getCacheControl();
       if (cacheControl != null) {
          blobBuilder.cacheControl(cacheControl);
@@ -869,7 +881,7 @@ public final class LocalBlobStore implements BlobStore {
           blobBuilder.tier(tier);
       }
 
-      String eTag = putBlob(mpu.containerName(), blobBuilder.build());
+      putBlob(mpu.containerName(), blobBuilder.build());
 
       for (MultipartPart part : parts) {
          removeBlob(mpu.containerName(), MULTIPART_PREFIX + mpu.id() + "-" + mpu.blobName() + "-" + part.partNumber());
@@ -878,7 +890,7 @@ public final class LocalBlobStore implements BlobStore {
 
       setBlobAccess(mpu.containerName(), mpu.blobName(), mpu.putOptions().getBlobAccess());
 
-      return eTag;
+      return mpuETag;
    }
 
    @Override

blobstore/src/main/java/org/jclouds/blobstore/domain/BlobBuilder.java

@@ -57,6 +57,12 @@ public interface BlobBuilder {
     */
    BlobBuilder userMetadata(Map<String, String> userMetadata);
 
+   /**
+    * @param eTag
+    *           Entity Tag associated with the Blob. Typically, content MD5 hash.
+    */
+   BlobBuilder eTag(String eTag);
+
    /**
     * 
     * @param payload
@@ -132,5 +138,7 @@ public interface BlobBuilder {
       PayloadBlobBuilder contentEncoding(String contentEncoding);
 
       PayloadBlobBuilder expires(Date expires);
+
+      PayloadBlobBuilder eTag(String string);
    }
 }

blobstore/src/main/java/org/jclouds/blobstore/domain/internal/BlobBuilderImpl.java

@@ -45,6 +45,7 @@ public class BlobBuilderImpl implements BlobBuilder {
    private Tier tier = Tier.STANDARD;
    private Map<String, String> userMetadata = Maps.newLinkedHashMap();
    private StorageType type = StorageType.BLOB;
+   private String eTag;
 
    @Override
    public BlobBuilder name(String name) {
@@ -66,6 +67,12 @@ public class BlobBuilderImpl implements BlobBuilder {
       return this;
    }
 
+   @Override
+   public BlobBuilder eTag(String eTag) {
+      this.eTag = eTag;
+      return this;
+   }
+
    @Override
    public BlobBuilder userMetadata(Map<String, String> userMetadata) {
       if (userMetadata != null)
@@ -126,6 +133,7 @@ public class BlobBuilderImpl implements BlobBuilder {
       blob.getMetadata().setUserMetadata(userMetadata);
       blob.getMetadata().setType(type);
       blob.getMetadata().setTier(tier);
+      blob.getMetadata().setETag(eTag);
       return blob;
    }
 
@@ -255,6 +263,12 @@ public class BlobBuilderImpl implements BlobBuilder {
          return this;
       }
 
+      @Override
+      public PayloadBlobBuilder eTag(String eTag) {
+         this.builder.eTag(eTag);
+         return this;
+      }
+
       @Override
       public PayloadBlobBuilder forSigning() {
          return builder.forSigning();

blobstore/src/test/java/org/jclouds/blobstore/integration/internal/BaseBlobIntegrationTest.java

@@ -899,6 +899,9 @@ public class BaseBlobIntegrationTest extends BaseBlobStoreIntegrationTest {
                .getMetadata().getContentMetadata().getContentLanguage();
    }
 
+   protected void checkMPUParts(Blob newBlob, List<MultipartPart> parts) {
+   }
+
    protected static volatile Crypto crypto;
    static {
       try {
@@ -1321,6 +1324,7 @@ public class BaseBlobIntegrationTest extends BaseBlobStoreIntegrationTest {
          assertThat(ByteStreams2.toByteArrayAndClose(newBlob.getPayload().openStream())).isEqualTo(byteSource.read());
          checkContentMetadata(newBlob);
          checkUserMetadata(newBlob.getMetadata().getUserMetadata(), blob.getMetadata().getUserMetadata());
+         checkMPUParts(newBlob, parts);
       } finally {
          returnContainer(container);
       }
@@ -1366,6 +1370,7 @@ public class BaseBlobIntegrationTest extends BaseBlobStoreIntegrationTest {
          assertThat(ByteStreams2.toByteArrayAndClose(newBlob.getPayload().openStream())).isEqualTo(byteSource.read());
          checkContentMetadata(newBlob);
          checkUserMetadata(newBlob.getMetadata().getUserMetadata(), blob.getMetadata().getUserMetadata());
+         checkMPUParts(newBlob, parts);
       } finally {
          returnContainer(container);
       }