HDDS-1895. Support Key ACL operations for OM HA.

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

@@ -19,12 +19,14 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.BitSet;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+import com.google.protobuf.ByteString;
 import org.apache.hadoop.fs.FileEncryptionInfo;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyInfo;
@@ -34,6 +36,8 @@
 
 import com.google.common.base.Preconditions;
 
+import static org.apache.hadoop.ozone.OzoneAcl.ZERO_BITSET;
+
 /**
  * Args for key block. The block instance for the key requested in putKey.
  * This is returned from OM to client, and client use class to talk to
@@ -235,6 +239,119 @@ public List<OzoneAclInfo> getAcls() {
     return acls;
   }
 
+  /**
+   * Add an ozoneAcl to list of existing Acl set.
+   * @param ozoneAcl
+   * @return true - if successfully added, false if not added or acl is
+   * already existing in the acl list.
+   */
+  public boolean addAcl(OzoneAclInfo ozoneAcl) {
+    // Case 1: When we are adding more rights to existing user/group.
+    boolean addToExistingAcl = false;
+    for(OzoneAclInfo existingAcl: getAcls()) {
+      if(existingAcl.getName().equals(ozoneAcl.getName()) &&
+          existingAcl.getType().equals(ozoneAcl.getType())) {
+
+        // We need to do "or" before comparision because think of a case like
+        // existing acl is 777 and newly added acl is 444, we have already
+        // that acl set. In this case if we do direct check they will not
+        // be equal, but if we do or and then check, we shall know it
+        // has acl's already set or not.
+        BitSet newAclBits = BitSet.valueOf(
+            existingAcl.getRights().toByteArray());
+
+        newAclBits.or(BitSet.valueOf(ozoneAcl.getRights().toByteArray()));
+
+        if (newAclBits.equals(BitSet.valueOf(
+            existingAcl.getRights().toByteArray()))) {
+          return false;
+        } else {
+          OzoneAclInfo newAcl = OzoneAclInfo.newBuilder()
+              .setType(ozoneAcl.getType())
+              .setName(ozoneAcl.getName())
+              .setAclScope(ozoneAcl.getAclScope())
+              .setRights(ByteString.copyFrom(newAclBits.toByteArray()))
+              .build();
+          getAcls().remove(existingAcl);
+          getAcls().add(newAcl);
+          addToExistingAcl = true;
+          break;
+        }
+      }
+    }
+
+    // Case 2: When a completely new acl is added.
+    if(!addToExistingAcl) {
+      getAcls().add(ozoneAcl);
+    }
+    return true;
+  }
+
+  /**
+   * Remove acl from existing acl list.
+   * @param ozoneAcl
+   * @return true - if successfully removed, false if not able to remove due
+   * to that acl is not in the existing acl list.
+   */
+  public boolean removeAcl(OzoneAclInfo ozoneAcl) {
+    boolean removed = false;
+
+    // When we are removing subset of rights from existing acl.
+    for(OzoneAclInfo existingAcl: getAcls()) {
+      if (existingAcl.getName().equals(ozoneAcl.getName()) &&
+          existingAcl.getType().equals(ozoneAcl.getType())) {
+
+        BitSet bits = BitSet.valueOf(ozoneAcl.getRights().toByteArray());
+        BitSet existingAclBits =
+            BitSet.valueOf(existingAcl.getRights().toByteArray());
+        bits.and(existingAclBits);
+
+        // This happens when the acl bitset asked to remove is not set for
+        // matched name and type.
+        // Like a case we have 444 permission, 333 is asked to removed.
+        if (bits.equals(ZERO_BITSET)) {
+          return false;
+        }
+
+        // We have some matching. Remove them.
+        bits.xor(existingAclBits);
+
+        // If existing acl has same bitset as passed acl bitset, remove that
+        // acl from the list
+        if (bits.equals(ZERO_BITSET)) {
+          getAcls().remove(existingAcl);
+        } else {
+          // Remove old acl and add new acl.
+          OzoneAclInfo newAcl = OzoneAclInfo.newBuilder()
+              .setType(ozoneAcl.getType())
+              .setName(ozoneAcl.getName())
+              .setAclScope(ozoneAcl.getAclScope())
+              .setRights(ByteString.copyFrom(bits.toByteArray()))
+              .build();
+          getAcls().remove(existingAcl);
+          getAcls().add(newAcl);
+        }
+        removed = true;
+        break;
+      }
+    }
+
+    return removed;
+  }
+
+  /**
+   * Reset the existing acl list.
+   * @param ozoneAcls
+   * @return true - if successfully able to reset.
+   */
+  public boolean setAcls(List<OzoneAclInfo> ozoneAcls) {
+    this.acls.clear();
+    this.acls = ozoneAcls;
+    return true;
+  }
+
+
+
   /**
    * Builder of OmKeyInfo.
    */
@@ -320,7 +437,8 @@ public Builder setFileEncryptionInfo(FileEncryptionInfo feInfo) {
     }
 
     public Builder setAcls(List<OzoneAclInfo> listOfAcls) {
-      this.acls = listOfAcls;
+      this.acls = new ArrayList<>();
+      this.acls.addAll(listOfAcls);
       return this;
     }
 
@@ -359,22 +477,22 @@ public KeyInfo getProtobuf() {
   }
 
   public static OmKeyInfo getFromProtobuf(KeyInfo keyInfo) {
-    return new OmKeyInfo(
-        keyInfo.getVolumeName(),
-        keyInfo.getBucketName(),
-        keyInfo.getKeyName(),
-        keyInfo.getKeyLocationListList().stream()
+    return new OmKeyInfo.Builder()
+        .setVolumeName(keyInfo.getVolumeName())
+        .setBucketName(keyInfo.getBucketName())
+        .setKeyName(keyInfo.getKeyName())
+        .setOmKeyLocationInfos(keyInfo.getKeyLocationListList().stream()
             .map(OmKeyLocationInfoGroup::getFromProtobuf)
-            .collect(Collectors.toList()),
-        keyInfo.getDataSize(),
-        keyInfo.getCreationTime(),
-        keyInfo.getModificationTime(),
-        keyInfo.getType(),
-        keyInfo.getFactor(),
-        KeyValueUtil.getFromProtobuf(keyInfo.getMetadataList()),
-        keyInfo.hasFileEncryptionInfo() ? OMPBHelper.convert(keyInfo
-            .getFileEncryptionInfo()): null,
-        keyInfo.getAclsList());
+            .collect(Collectors.toList()))
+        .setDataSize(keyInfo.getDataSize())
+        .setCreationTime(keyInfo.getCreationTime())
+        .setModificationTime(keyInfo.getModificationTime())
+        .setReplicationType(keyInfo.getType())
+        .setReplicationFactor(keyInfo.getFactor())
+        .addAllMetadata(KeyValueUtil.getFromProtobuf(keyInfo.getMetadataList()))
+        .setFileEncryptionInfo(keyInfo.hasFileEncryptionInfo() ?
+            OMPBHelper.convert(keyInfo.getFileEncryptionInfo()): null)
+        .setAcls(keyInfo.getAclsList()).build();
   }
 
   @Override