HBASE-29368 [Feature] Key management for encryption at rest (MVP changes)

.rubocop.yml

@@ -9,3 +9,12 @@ Layout/LineLength:
 
 Metrics/MethodLength:
   Max: 75
+
+GlobalVars:
+  AllowedVariables:
+    - $CUST1_ENCODED
+    - $CUST1_ALIAS
+    - $CUST1_ENCODED
+    - $GLOB_CUST_ENCODED
+    - $TEST
+    - $TEST_CLUSTER

hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/balancer/StochasticLoadBalancer.java

@@ -581,7 +581,6 @@ protected List<RegionPlan> balanceTable(TableName tableName,
       rackManager, regionCacheRatioOnOldServerMap);
 
     long startTime = EnvironmentEdgeManager.currentTime();
-    cluster.setStopRequestedAt(startTime + maxRunningTime);
 
     initCosts(cluster);
     balancerConditionals.loadClusterState(cluster);
@@ -632,6 +631,10 @@ protected List<RegionPlan> balanceTable(TableName tableName,
       currentCost / sumMultiplier, functionCost(), computedMaxSteps);
 
     final String initFunctionTotalCosts = totalCostsPerFunc();
+    long searchStartTime = EnvironmentEdgeManager.currentTime();
+    // Budget maxRunningTime for the stochastic walk only; initialization (cluster costs, etc.)
+    // can be substantial on busy hosts and must not consume the search deadline.
+    cluster.setStopRequestedAt(searchStartTime + maxRunningTime);
     // Perform a stochastic walk to see if we can get a good fit.
     long step;
     boolean planImprovedConditionals = false;

hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java

@@ -2705,4 +2705,28 @@ List<LogEntry> getLogEntries(Set<ServerName> serverNames, String logType, Server
 
   @InterfaceAudience.Private
   void restoreBackupSystemTable(String snapshotName) throws IOException;
+
+  /**
+   * Refresh the system key cache on all specified region servers.
+   * @param regionServers the list of region servers to refresh the system key cache on
+   */
+  void refreshSystemKeyCacheOnServers(List<ServerName> regionServers) throws IOException;
+
+  /**
+   * Eject a specific managed key entry from the managed key data cache on all specified region
+   * servers.
+   * @param regionServers the list of region servers to eject the managed key entry from
+   * @param keyCustodian  the key custodian
+   * @param keyNamespace  the key namespace
+   * @param keyMetadata   the key metadata
+   */
+  void ejectManagedKeyDataCacheEntryOnServers(List<ServerName> regionServers, byte[] keyCustodian,
+    String keyNamespace, String keyMetadata) throws IOException;
+
+  /**
+   * Clear all entries in the managed key data cache on all specified region servers without having
+   * to restart the process.
+   * @param regionServers the list of region servers to clear the managed key data cache on
+   */
+  void clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) throws IOException;
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java

@@ -1157,4 +1157,21 @@ public List<String> getCachedFilesList(ServerName serverName) throws IOException
   public void restoreBackupSystemTable(String snapshotName) throws IOException {
     get(admin.restoreBackupSystemTable(snapshotName));
   }
+
+  @Override
+  public void refreshSystemKeyCacheOnServers(List<ServerName> regionServers) throws IOException {
+    get(admin.refreshSystemKeyCacheOnServers(regionServers));
+  }
+
+  @Override
+  public void ejectManagedKeyDataCacheEntryOnServers(List<ServerName> regionServers,
+    byte[] keyCustodian, String keyNamespace, String keyMetadata) throws IOException {
+    get(admin.ejectManagedKeyDataCacheEntryOnServers(regionServers, keyCustodian, keyNamespace,
+      keyMetadata));
+  }
+
+  @Override
+  public void clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) throws IOException {
+    get(admin.clearManagedKeyDataCacheOnServers(regionServers));
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java

@@ -1892,4 +1892,27 @@ CompletableFuture<List<LogEntry>> getLogEntries(Set<ServerName> serverNames, Str
 
   @InterfaceAudience.Private
   CompletableFuture<Void> restoreBackupSystemTable(String snapshotName);
+
+  /**
+   * Refresh the system key cache on all specified region servers.
+   * @param regionServers the list of region servers to refresh the system key cache on
+   */
+  CompletableFuture<Void> refreshSystemKeyCacheOnServers(List<ServerName> regionServers);
+
+  /**
+   * Eject a specific managed key entry from the managed key data cache on all specified region
+   * servers.
+   * @param regionServers the list of region servers to eject the managed key entry from
+   * @param keyCustodian  the key custodian
+   * @param keyNamespace  the key namespace
+   * @param keyMetadata   the key metadata
+   */
+  CompletableFuture<Void> ejectManagedKeyDataCacheEntryOnServers(List<ServerName> regionServers,
+    byte[] keyCustodian, String keyNamespace, String keyMetadata);
+
+  /**
+   * Clear all entries in the managed key data cache on all specified region servers.
+   * @param regionServers the list of region servers to clear the managed key data cache on
+   */
+  CompletableFuture<Void> clearManagedKeyDataCacheOnServers(List<ServerName> regionServers);
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java

@@ -696,6 +696,23 @@ public CompletableFuture<Void> updateConfiguration(String groupName) {
     return wrap(rawAdmin.updateConfiguration(groupName));
   }
 
+  @Override
+  public CompletableFuture<Void> refreshSystemKeyCacheOnServers(List<ServerName> regionServers) {
+    return wrap(rawAdmin.refreshSystemKeyCacheOnServers(regionServers));
+  }
+
+  @Override
+  public CompletableFuture<Void> ejectManagedKeyDataCacheEntryOnServers(
+    List<ServerName> regionServers, byte[] keyCustodian, String keyNamespace, String keyMetadata) {
+    return wrap(rawAdmin.ejectManagedKeyDataCacheEntryOnServers(regionServers, keyCustodian,
+      keyNamespace, keyMetadata));
+  }
+
+  @Override
+  public CompletableFuture<Void> clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) {
+    return wrap(rawAdmin.clearManagedKeyDataCacheOnServers(regionServers));
+  }
+
   @Override
   public CompletableFuture<Void> rollWALWriter(ServerName serverName) {
     return wrap(rawAdmin.rollWALWriter(serverName));

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ColumnFamilyDescriptor.java

@@ -117,6 +117,9 @@ public interface ColumnFamilyDescriptor {
   /** Returns the encryption key namespace for this family */
   String getEncryptionKeyNamespace();
 
+  /** Returns the encryption key namespace for this family as a {@link Bytes} object */
+  Bytes getEncryptionKeyNamespaceBytes();
+
   /** Returns Return the encryption algorithm in use by this family */
   String getEncryptionType();
 

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ColumnFamilyDescriptorBuilder.java

@@ -1352,6 +1352,11 @@ public String getEncryptionKeyNamespace() {
       return getStringOrDefault(ENCRYPTION_KEY_NAMESPACE_BYTES, Function.identity(), null);
     }
 
+    @Override
+    public Bytes getEncryptionKeyNamespaceBytes() {
+      return getValue(ENCRYPTION_KEY_NAMESPACE_BYTES);
+    }
+
     /**
      * Set the encryption key namespace attribute for the family
      * @param keyNamespace the key namespace, or null to remove existing setting

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java

@@ -77,6 +77,7 @@
 import org.apache.hadoop.hbase.client.security.SecurityCapability;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
+import org.apache.hadoop.hbase.keymeta.ManagedKeyIdentityUtils;
 import org.apache.hadoop.hbase.net.Address;
 import org.apache.hadoop.hbase.quotas.QuotaFilter;
 import org.apache.hadoop.hbase.quotas.QuotaSettings;
@@ -150,7 +151,10 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.EmptyMsg;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.LastHighestWalFilenum;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ManagedKeyEntryRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ManagedKeyRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.NameStringPair;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ProcedureDescription;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
@@ -4697,4 +4701,86 @@ MasterProtos.RestoreBackupSystemTableResponse> procedureCall(request,
         MasterProtos.RestoreBackupSystemTableResponse::getProcId,
         new RestoreBackupSystemTableProcedureBiConsumer());
   }
+
+  @Override
+  public CompletableFuture<Void> refreshSystemKeyCacheOnServers(List<ServerName> regionServers) {
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    List<CompletableFuture<Void>> futures =
+      regionServers.stream().map(this::refreshSystemKeyCache).collect(Collectors.toList());
+    addListener(CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])),
+      (result, err) -> {
+        if (err != null) {
+          future.completeExceptionally(err);
+        } else {
+          future.complete(result);
+        }
+      });
+    return future;
+  }
+
+  private CompletableFuture<Void> refreshSystemKeyCache(ServerName serverName) {
+    return this.<Void> newAdminCaller()
+      .action((controller, stub) -> this.<EmptyMsg, EmptyMsg, Void> adminCall(controller, stub,
+        EmptyMsg.getDefaultInstance(),
+        (s, c, req, done) -> s.refreshSystemKeyCache(controller, req, done), resp -> null))
+      .serverName(serverName).call();
+  }
+
+  @Override
+  public CompletableFuture<Void> ejectManagedKeyDataCacheEntryOnServers(
+    List<ServerName> regionServers, byte[] keyCustodian, String keyNamespace, String keyMetadata) {
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    // Create the request once instead of repeatedly for each server
+    byte[] keyMetadataHash = ManagedKeyIdentityUtils.constructMetadataHash(keyMetadata);
+    ManagedKeyEntryRequest request = ManagedKeyEntryRequest.newBuilder()
+      .setKeyCustNs(ManagedKeyRequest.newBuilder().setKeyCust(ByteString.copyFrom(keyCustodian))
+        .setKeyNamespace(keyNamespace).build())
+      .setKeyMetadataHash(ByteString.copyFrom(keyMetadataHash)).build();
+    List<CompletableFuture<Void>> futures =
+      regionServers.stream().map(serverName -> ejectManagedKeyDataCacheEntry(serverName, request))
+        .collect(Collectors.toList());
+    addListener(CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])),
+      (result, err) -> {
+        if (err != null) {
+          future.completeExceptionally(err);
+        } else {
+          future.complete(result);
+        }
+      });
+    return future;
+  }
+
+  private CompletableFuture<Void> ejectManagedKeyDataCacheEntry(ServerName serverName,
+    ManagedKeyEntryRequest request) {
+    return this.<Void> newAdminCaller()
+      .action((controller, stub) -> this.<ManagedKeyEntryRequest, HBaseProtos.BooleanMsg,
+        Void> adminCall(controller, stub, request,
+          (s, c, req, done) -> s.ejectManagedKeyDataCacheEntry(controller, req, done),
+          resp -> null))
+      .serverName(serverName).call();
+  }
+
+  @Override
+  public CompletableFuture<Void> clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) {
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    List<CompletableFuture<Void>> futures =
+      regionServers.stream().map(this::clearManagedKeyDataCache).collect(Collectors.toList());
+    addListener(CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])),
+      (result, err) -> {
+        if (err != null) {
+          future.completeExceptionally(err);
+        } else {
+          future.complete(result);
+        }
+      });
+    return future;
+  }
+
+  private CompletableFuture<Void> clearManagedKeyDataCache(ServerName serverName) {
+    return this.<Void> newAdminCaller()
+      .action((controller, stub) -> this.<EmptyMsg, EmptyMsg, Void> adminCall(controller, stub,
+        EmptyMsg.getDefaultInstance(),
+        (s, c, req, done) -> s.clearManagedKeyDataCache(controller, req, done), resp -> null))
+      .serverName(serverName).call();
+  }
 }