HDDS-13919. S3 Conditional Writes (PutObject)

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -318,6 +318,8 @@ public final class OzoneConsts {
   public static final String TENANT = "tenant";
   public static final String USER_PREFIX = "userPrefix";
   public static final String REWRITE_GENERATION = "rewriteGeneration";
+  /** Sentinel generation used to request atomic create-if-not-exists(put if absent) semantics. */
+  public static final long EXPECTED_GEN_CREATE_IF_NOT_EXISTS = -1L;
   public static final String FROM_SNAPSHOT = "fromSnapshot";
   public static final String TO_SNAPSHOT = "toSnapshot";
   public static final String TOKEN = "token";

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java

@@ -54,6 +54,9 @@ public enum OzoneManagerVersion implements ComponentVersion {
 
   S3_LIST_MULTIPART_UPLOADS_PAGINATION(11,
       "OzoneManager version that supports S3 list multipart uploads API with pagination"),
+
+  ATOMIC_CREATE_IF_NOT_EXISTS(12,
+      "OzoneManager version that supports explicit create-if-not-exists key semantics"),
 
   FUTURE_VERSION(-1, "Used internally in the client when the server side is "
       + " newer and an unknown server version has arrived to the client.");

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java

@@ -508,7 +508,7 @@ public OzoneOutputStream createKey(String key, long size,
    *
    * @param keyName Existing key to rewrite. This must exist in the bucket.
    * @param size The size of the new key
-   * @param existingKeyGeneration The generation of the existing key which is checked for changes at key create
+   * @param existingKeyGeneration The positive generation of the existing key which is checked for changes at key create
    *                              and commit time.
    * @param replicationConfig The replication configuration for the key to be rewritten.
    * @param metadata custom key value metadata
@@ -520,6 +520,44 @@ public OzoneOutputStream rewriteKey(String keyName, long size, long existingKeyG
     return proxy.rewriteKey(volumeName, name, keyName, size, existingKeyGeneration, replicationConfig, metadata);
   }
 
+  /**
+   * Creates a key only if it does not exist (S3 If-None-Match: * semantics).
+   *
+   * @param keyName Name of the key
+   * @param size Size of the data
+   * @param replicationConfig Replication configuration
+   * @param metadata custom key value metadata
+   * @param tags Tags used for S3 object tags
+   * @return OzoneOutputStream to which the data has to be written.
+   * @throws IOException
+   */
+  public OzoneOutputStream createKeyIfNotExists(String keyName, long size,
+      ReplicationConfig replicationConfig, Map<String, String> metadata,
+      Map<String, String> tags) throws IOException {
+    return proxy.createKeyIfNotExists(volumeName, name, keyName, size,
+        replicationConfig, metadata, tags);
+  }
+
+  /**
+   * Rewrites a key only if its ETag matches (S3 If-Match semantics).
+   *
+   * @param keyName Name of the key
+   * @param size Size of the data
+   * @param expectedETag The ETag value the existing key must have
+   * @param replicationConfig Replication configuration
+   * @param metadata custom key value metadata
+   * @param tags Tags used for S3 object tags
+   * @return OzoneOutputStream to which the data has to be written.
+   * @throws IOException
+   */
+  public OzoneOutputStream rewriteKeyIfMatch(String keyName, long size,
+      String expectedETag, ReplicationConfig replicationConfig,
+      Map<String, String> metadata, Map<String, String> tags)
+      throws IOException {
+    return proxy.rewriteKeyIfMatch(volumeName, name, keyName, size,
+        expectedETag, replicationConfig, metadata, tags);
+  }
+
   /**
    * Creates a new key in the bucket, with default replication type RATIS and
    * with replication factor THREE.
@@ -1047,8 +1085,8 @@ public List<OzoneFileStatus> listStatus(String keyName, boolean recursive,
    *
    * @param prefix Optional string to filter for the selected keys.
    */
-  public OzoneMultipartUploadList listMultipartUploads(String prefix, 
-      String keyMarker, String uploadIdMarker, int maxUploads) 
+  public OzoneMultipartUploadList listMultipartUploads(String prefix,
+      String keyMarker, String uploadIdMarker, int maxUploads)
       throws IOException {
     return proxy.listMultipartUploads(volumeName, getName(), prefix, keyMarker, uploadIdMarker, maxUploads);
   }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -368,7 +368,7 @@ OzoneOutputStream createKey(String volumeName, String bucketName,
    * @param bucketName Name of the Bucket
    * @param keyName Existing key to rewrite. This must exist in the bucket.
    * @param size The size of the new key
-   * @param existingKeyGeneration The generation of the existing key which is checked for changes at key create
+   * @param existingKeyGeneration The positive generation of the existing key which is checked for changes at key create
    *                              and commit time.
    * @param replicationConfig The replication configuration for the key to be rewritten.
    * @param metadata custom key value metadata
@@ -379,6 +379,44 @@ OzoneOutputStream rewriteKey(String volumeName, String bucketName, String keyNam
       long size, long existingKeyGeneration, ReplicationConfig replicationConfig,
        Map<String, String> metadata) throws IOException;
 
+  /**
+   * Creates a key only if it does not exist (S3 If-None-Match: * semantics).
+   *
+   * @param volumeName Name of the Volume
+   * @param bucketName Name of the Bucket
+   * @param keyName Name of the Key
+   * @param size Size of the data
+   * @param replicationConfig The replication configuration
+   * @param metadata custom key value metadata
+   * @param tags Tags used for S3 object tags
+   * @return {@link OzoneOutputStream}
+   * @throws OMException with KEY_ALREADY_EXISTS if key exists
+   */
+  OzoneOutputStream createKeyIfNotExists(String volumeName, String bucketName,
+      String keyName, long size, ReplicationConfig replicationConfig,
+      Map<String, String> metadata, Map<String, String> tags)
+      throws IOException;
+
+  /**
+   * Rewrites a key only if its ETag matches (S3 If-Match semantics).
+   *
+   * @param volumeName Name of the Volume
+   * @param bucketName Name of the Bucket
+   * @param keyName Name of the Key
+   * @param size Size of the data
+   * @param expectedETag The ETag value the existing key must have
+   * @param replicationConfig The replication configuration
+   * @param metadata custom key value metadata
+   * @param tags Tags used for S3 object tags
+   * @return {@link OzoneOutputStream}
+   * @throws OMException with ETAG_MISMATCH, ETAG_NOT_AVAILABLE, or KEY_NOT_FOUND
+   */
+  @SuppressWarnings("checkstyle:parameternumber")
+  OzoneOutputStream rewriteKeyIfMatch(String volumeName, String bucketName,
+      String keyName, long size, String expectedETag,
+      ReplicationConfig replicationConfig, Map<String, String> metadata,
+      Map<String, String> tags) throws IOException;
+
   /**
    * Writes a key in an existing bucket.
    * @param volumeName Name of the Volume

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -677,7 +677,7 @@ public void createBucket(
       builder.setDefaultReplicationConfig(defaultReplicationConfig);
     }
 
-    String replicationType = defaultReplicationConfig == null 
+    String replicationType = defaultReplicationConfig == null
         ? "server-side default replication type"
         : defaultReplicationConfig.getType().toString();
 
@@ -1317,7 +1317,7 @@ public List<OzoneBucket> listBuckets(String volumeName, String bucketPrefix,
     List<OmBucketInfo> buckets = ozoneManagerClient.listBuckets(
         volumeName, prevBucket, bucketPrefix, maxListResult, hasSnapshot);
 
-    return buckets.stream().map(bucket -> 
+    return buckets.stream().map(bucket ->
             OzoneBucket.newBuilder(conf, this)
                 .setVolumeName(bucket.getVolumeName())
                 .setName(bucket.getBucketName())
@@ -1408,6 +1408,9 @@ public OzoneOutputStream rewriteKey(String volumeName, String bucketName, String
     if (omVersion.compareTo(OzoneManagerVersion.ATOMIC_REWRITE_KEY) < 0) {
       throw new IOException("OzoneManager does not support atomic key rewrite.");
     }
+    Preconditions.checkArgument(existingKeyGeneration > 0,
+        "existingKeyGeneration must be positive, but was %s",
+        existingKeyGeneration);
 
     createKeyPreChecks(volumeName, bucketName, keyName, replicationConfig);
 
@@ -1432,6 +1435,68 @@ public OzoneOutputStream rewriteKey(String volumeName, String bucketName, String
     return createOutputStream(openKey);
   }
 
+  @Override
+  public OzoneOutputStream createKeyIfNotExists(String volumeName,
+      String bucketName, String keyName, long size,
+      ReplicationConfig replicationConfig, Map<String, String> metadata,
+      Map<String, String> tags) throws IOException {
+    if (omVersion.compareTo(OzoneManagerVersion.ATOMIC_REWRITE_KEY) < 0) {
+      throw new IOException(
+          "OzoneManager does not support atomic key creation.");
+    }
+
+    createKeyPreChecks(volumeName, bucketName, keyName, replicationConfig);
+
+    OmKeyArgs.Builder builder = new OmKeyArgs.Builder()
+        .setVolumeName(volumeName)
+        .setBucketName(bucketName)
+        .setKeyName(keyName)
+        .setDataSize(size)
+        .setReplicationConfig(replicationConfig)
+        .addAllMetadataGdpr(metadata)
+        .addAllTags(tags)
+        .setLatestVersionLocation(getLatestVersionLocation)
+        .setExpectedDataGeneration(
+            OzoneConsts.EXPECTED_GEN_CREATE_IF_NOT_EXISTS);
+
+    OpenKeySession openKey = ozoneManagerClient.openKey(builder.build());
+    if (isS3GRequest.get() && size == 0) {
+      openKey.getKeyInfo().setDataSize(0);
+    }
+    return createOutputStream(openKey);
+  }
+
+  @Override
+  @SuppressWarnings("checkstyle:parameternumber")
+  public OzoneOutputStream rewriteKeyIfMatch(String volumeName,
+      String bucketName, String keyName, long size, String expectedETag,
+      ReplicationConfig replicationConfig, Map<String, String> metadata,
+      Map<String, String> tags) throws IOException {
+    if (omVersion.compareTo(OzoneManagerVersion.ATOMIC_REWRITE_KEY) < 0) {
+      throw new IOException(
+          "OzoneManager does not support conditional key rewrite.");
+    }
+
+    createKeyPreChecks(volumeName, bucketName, keyName, replicationConfig);
+
+    OmKeyArgs.Builder builder = new OmKeyArgs.Builder()
+        .setVolumeName(volumeName)
+        .setBucketName(bucketName)
+        .setKeyName(keyName)
+        .setDataSize(size)
+        .setReplicationConfig(replicationConfig)
+        .addAllMetadataGdpr(metadata)
+        .addAllTags(tags)
+        .setLatestVersionLocation(getLatestVersionLocation)
+        .setExpectedETag(expectedETag);
+
+    OpenKeySession openKey = ozoneManagerClient.openKey(builder.build());
+    if (isS3GRequest.get() && size == 0) {
+      openKey.getKeyInfo().setDataSize(0);
+    }
+    return createOutputStream(openKey);
+  }
+
   private void createKeyPreChecks(String volumeName, String bucketName, String keyName,
       ReplicationConfig replicationConfig) throws IOException {
     verifyVolumeName(volumeName);

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/exceptions/OMException.java

@@ -208,7 +208,7 @@ public enum ResultCodes {
     USER_MISMATCH, // Error code when requested user name passed is different
     // from remote user.
 
-    INVALID_PART, // When part name is not found or not matching with partname 
+    INVALID_PART, // When part name is not found or not matching with partname
     // in OM MPU partInfo.
 
     INVALID_PART_ORDER, // When list of parts mentioned to complete MPU are not
@@ -267,13 +267,17 @@ public enum ResultCodes {
     UNAUTHORIZED,
 
     S3_SECRET_ALREADY_EXISTS,
-    
+
     INVALID_PATH,
     TOO_MANY_BUCKETS,
     KEY_UNDER_LEASE_RECOVERY,
     KEY_ALREADY_CLOSED,
     KEY_UNDER_LEASE_SOFT_LIMIT_PERIOD,
 
     TOO_MANY_SNAPSHOTS,
+
+    ETAG_MISMATCH,
+
+    ETAG_NOT_AVAILABLE,
   }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyArgs.java

@@ -61,6 +61,7 @@ public final class OmKeyArgs extends WithMetadata implements Auditable {
   // This allows a key to be created an committed atomically if the original has not
   // been modified.
   private Long expectedDataGeneration = null;
+  private final String expectedETag;
 
   private OmKeyArgs(Builder b) {
     super(b);
@@ -82,6 +83,7 @@ private OmKeyArgs(Builder b) {
     this.ownerName = b.ownerName;
     this.tags = b.tags.build();
     this.expectedDataGeneration = b.expectedDataGeneration;
+    this.expectedETag = b.expectedETag;
   }
 
   public boolean getIsMultipartKey() {
@@ -164,6 +166,10 @@ public Long getExpectedDataGeneration() {
     return expectedDataGeneration;
   }
 
+  public String getExpectedETag() {
+    return expectedETag;
+  }
+
   @Override
   public Map<String, String> toAuditMap() {
     Map<String, String> auditMap = new LinkedHashMap<>();
@@ -209,6 +215,9 @@ public KeyArgs toProtobuf() {
     if (expectedDataGeneration != null) {
       builder.setExpectedDataGeneration(expectedDataGeneration);
     }
+    if (expectedETag != null) {
+      builder.setExpectedETag(expectedETag);
+    }
     return builder.build();
   }
 
@@ -234,6 +243,7 @@ public static class Builder extends WithMetadata.Builder {
     private boolean forceUpdateContainerCacheFromSCM;
     private final MapBuilder<String, String> tags;
     private Long expectedDataGeneration = null;
+    private String expectedETag;
 
     public Builder() {
       this(AclListBuilder.empty());
@@ -263,6 +273,7 @@ public Builder(OmKeyArgs obj) {
       this.forceUpdateContainerCacheFromSCM =
           obj.forceUpdateContainerCacheFromSCM;
       this.expectedDataGeneration = obj.expectedDataGeneration;
+      this.expectedETag = obj.expectedETag;
       this.tags = MapBuilder.of(obj.tags);
       this.acls = AclListBuilder.of(obj.acls);
     }
@@ -398,6 +409,11 @@ public Builder setExpectedDataGeneration(long generation) {
       return this;
     }
 
+    public Builder setExpectedETag(String eTag) {
+      this.expectedETag = eTag;
+      return this;
+    }
+
     public OmKeyArgs build() {
       return new OmKeyArgs(this);
     }