Support high availability for the authority module

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -93,6 +93,9 @@ public final class ProcedureMessages {
   public static final String AUTHENTICATION_FAILED = "Authentication failed.";
   public static final String AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY =
       "Auth procedure: clean datanode cache successfully";
+  public static final String AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED =
+      "Auth plan has been committed, but DataNode permission cache invalidation failed. "
+          + "Some DataNodes may have stale permissions; please clear their permission cache manually.";
   public static final String BEGIN_TO_CHANGE_DATANODE_STATUS_NODESTATUSMAP =
       "{}, Begin to change DataNode status, nodeStatusMap: {}";
   public static final String BEGIN_TO_STOP_DATANODES_AND_KILL_THE_DATANODE_PROCESS =
@@ -356,7 +359,7 @@ public final class ProcedureMessages {
   public static final String ERROR_IN_DESERIALIZE_PROCID_THIS_PROCEDURE_WILL_BE_IGNORED_IT =
       "Error in deserialize {} (procID {}). This procedure will be ignored. It may belong to old version and cannot be used now.";
   public static final String EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES =
-      "Execute auth plan {} success. To invalidate datanodes: {}";
+      "Execute auth plan {} success.";
   public static final String EXECUTING_ON_REGION_FOR_COLUMN_IN_WHEN_DROPPING_COLUMN =
       "Executing on region for column {} in {}.{} when dropping column";
   public static final String FAILED_TO_ACTIVE_CQ_BECAUSE_OF_NO_SUCH_CQ =
@@ -551,8 +554,6 @@ public final class ProcedureMessages {
       "Fail to drop trigger [%s] at STATE [%s]";
   public static final String FAIL_TO_DROP_TRIGGER_ON_DATA_NODES =
       "Fail to drop trigger [%s] on Data Nodes";
-  public static final String FAIL_TO_EXECUTE_PLAN_AT_STATE =
-      "Fail to execute plan [%s] at state[%s]";
   public static final String FAIL_TO_REMOVE_AINODE_AT_STATE =
       "Fail to remove AINode [%s] at STATE [%s], %s";
   public static final String FAIL_TO_REMOVE_AINODE_ON_CONFIG_NODES =
@@ -870,8 +871,6 @@ public final class ProcedureMessages {
       "Retrievable error trying to create pipe plugin [{}], state: {}";
   public static final String RETRIEVABLE_ERROR_TRYING_TO_DROP_PIPE_PLUGIN_STATE =
       "Retrievable error trying to drop pipe plugin [{}], state: {}";
-  public static final String RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE =
-      "Retrievable error trying to execute plan {}, state: {}";
   public static final String RETRIEVABLE_ERROR_TRYING_TO_REMOVE_AINODE_STATE =
       "Retrievable error trying to remove AINode [{}], state [{}]";
   public static final String ROLLBACK_CREATETABLE_COSTS_MS = "Rollback CreateTable-{} costs {}ms.";

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -91,6 +91,8 @@ public final class ProcedureMessages {
   public static final String AUTHENTICATION_FAILED = "认证失败。";
   public static final String AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY =
       "Auth procedure：成功清理 datanode 缓存";
+  public static final String AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED =
+      "权限计划已提交，但 DataNode 权限缓存失效失败。部分 DataNode 可能仍持有过期权限，请手动清理权限缓存。";
   public static final String BEGIN_TO_CHANGE_DATANODE_STATUS_NODESTATUSMAP =
       "{}, 开始修改 DataNode 状态，nodeStatusMap：{}";
   public static final String BEGIN_TO_STOP_DATANODES_AND_KILL_THE_DATANODE_PROCESS =
@@ -344,7 +346,7 @@ public final class ProcedureMessages {
   public static final String ERROR_IN_DESERIALIZE_PROCID_THIS_PROCEDURE_WILL_BE_IGNORED_IT =
       "反序列化 {}（procID {}）出错。该 procedure 将被忽略。它可能属于旧版本，目前无法使用。";
   public static final String EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES =
-      "执行 auth plan {} 成功。使 datanode 缓存失效：{}";
+      "执行 auth plan {} 成功。";
   public static final String EXECUTING_ON_REGION_FOR_COLUMN_IN_WHEN_DROPPING_COLUMN =
       "删除列时在表 {}.{} 中列 {} 对应的 region 上执行";
   public static final String FAILED_TO_ACTIVE_CQ_BECAUSE_OF_NO_SUCH_CQ =
@@ -531,7 +533,6 @@ public final class ProcedureMessages {
       "重试 {} 次后删除 pipe plugin [{}] 仍失败";
   public static final String FAIL_TO_DROP_TRIGGER_AT_STATE = "在 STATE [%s] 处删除 trigger [%s] 失败";
   public static final String FAIL_TO_DROP_TRIGGER_ON_DATA_NODES = "在 DataNode 上删除 trigger [%s] 失败";
-  public static final String FAIL_TO_EXECUTE_PLAN_AT_STATE = "在 state[%s] 处执行 plan [%s] 失败";
   public static final String FAIL_TO_REMOVE_AINODE_AT_STATE = "在 STATE [%s] 处移除 AINode [%s] 失败，%s";
   public static final String FAIL_TO_REMOVE_AINODE_ON_CONFIG_NODES =
       "在 ConfigNode [%s] 上移除 [%s] 个 AINode 失败";
@@ -828,8 +829,6 @@ public final class ProcedureMessages {
       "尝试创建 pipe plugin [{}] 时发生可重试错误，状态：{}";
   public static final String RETRIEVABLE_ERROR_TRYING_TO_DROP_PIPE_PLUGIN_STATE =
       "尝试删除 pipe plugin [{}] 时发生可重试错误，状态：{}";
-  public static final String RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE =
-      "尝试执行 plan {} 时发生可重试错误，状态：{}";
   public static final String RETRIEVABLE_ERROR_TRYING_TO_REMOVE_AINODE_STATE =
       "尝试移除 AINode [{}] 时发生可重试错误，状态 [{}]";
   public static final String ROLLBACK_CREATETABLE_COSTS_MS = "Rollback CreateTable-{} costs {}ms.";

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java

@@ -137,4 +137,7 @@ public enum CnToDnAsyncRequestType {
   // audit log and event write-back
   INSERT_RECORD,
   ENABLE_SEPARATION_OF_ADMIN_POWERS,
+
+  // authority
+  INVALIDATE_PERMISSION_CACHE,
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java

@@ -81,6 +81,7 @@
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidateColumnCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq;
+import org.apache.iotdb.mpp.rpc.thrift.TInvalidatePermissionCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidateTableCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TKillQueryInstanceReq;
 import org.apache.iotdb.mpp.rpc.thrift.TNotifyRegionMigrationReq;
@@ -522,6 +523,11 @@ protected void initActionMapBuilder() {
         CnToDnAsyncRequestType.GET_BUILTIN_SERVICE,
         (req, client, handler) ->
             client.getBuiltInService((GetBuiltInExternalServiceRPCHandler) handler));
+    actionMapBuilder.put(
+        CnToDnAsyncRequestType.INVALIDATE_PERMISSION_CACHE,
+        (req, client, handler) ->
+            client.invalidatePermissionCache(
+                (TInvalidatePermissionCacheReq) req, (DataNodeTSStatusRPCHandler) handler));
   }
 
   @Override

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java

@@ -20,21 +20,25 @@
 package org.apache.iotdb.confignode.procedure.impl.sync;
 
 import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.conf.CommonConfig;
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.exception.IoTDBException;
 import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
-import org.apache.iotdb.confignode.client.sync.CnToDnSyncRequestType;
-import org.apache.iotdb.confignode.client.sync.SyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
+import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
+import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
 import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
 import org.apache.iotdb.confignode.i18n.ProcedureMessages;
+import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator;
 import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
 import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
 import org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils;
 import org.apache.iotdb.confignode.procedure.state.auth.AuthOperationProcedureState;
 import org.apache.iotdb.confignode.procedure.store.ProcedureType;
 import org.apache.iotdb.consensus.exception.ConsensusException;
@@ -50,8 +54,8 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
-import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 import static org.apache.iotdb.confignode.procedure.state.auth.AuthOperationProcedureState.DATANODE_AUTHCACHE_INVALIDING;
@@ -68,7 +72,6 @@ public class AuthOperationProcedure extends AbstractNodeProcedure<AuthOperationP
   private static final String CONSENSUS_WRITE_ERROR =
       ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
 
-  private static final int RETRY_THRESHOLD = 2;
   private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
 
   private final List<Pair<TDataNodeConfiguration, Long>> dataNodesToInvalid = new ArrayList<>();
@@ -97,54 +100,40 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, AuthOperationProcedu
           writePlan(env);
           return Flow.HAS_MORE_STATE;
         case DATANODE_AUTHCACHE_INVALIDING:
-          TInvalidatePermissionCacheReq req = new TInvalidatePermissionCacheReq();
-          TSStatus status;
-          req.setUsername(user);
-          req.setRoleName(role);
+          TInvalidatePermissionCacheReq req = new TInvalidatePermissionCacheReq(user, role);
           if (plan.getAuthorType() == ConfigPhysicalPlanType.AccountUnlock
               || plan.getAuthorType() == ConfigPhysicalPlanType.RAccountUnlock) {
             // For account unlock, role carries the optional login address.
             req.setNeedDisconnect(true);
           }
-          Iterator<Pair<TDataNodeConfiguration, Long>> it = dataNodesToInvalid.iterator();
-          while (it.hasNext()) {
-            Pair<TDataNodeConfiguration, Long> pair = it.next();
-            if (pair.getRight() + this.timeoutMS < System.currentTimeMillis()) {
-              it.remove();
-              continue;
-            }
-            status =
-                (TSStatus)
-                    SyncDataNodeClientPool.getInstance()
-                        .sendSyncRequestToDataNodeWithRetry(
-                            pair.getLeft().getLocation().getInternalEndPoint(),
-                            req,
-                            CnToDnSyncRequestType.INVALIDATE_PERMISSION_CACHE);
-            if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-              it.remove();
-            }
-          }
-          if (dataNodesToInvalid.isEmpty()) {
+          final boolean proceeded =
+              new ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager()))
+                  .propagate(targets -> broadcastAuthorityCache(req, targets));
+
+          if (proceeded) {
             LOGGER.info(ProcedureMessages.AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY);
             return Flow.NO_MORE_STATE;
-          } else {
-            setNextState(AuthOperationProcedureState.DATANODE_AUTHCACHE_INVALIDING);
           }
-          break;
+          LOGGER.warn(ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED);
+          setFailure(
+              new ProcedureException(
+                  new IoTDBException(
+                      ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED,
+                      TSStatusCode.AUTH_OPERATE_EXCEPTION.getStatusCode())));
+          return Flow.NO_MORE_STATE;
       }
     } catch (Exception e) {
       if (isRollbackSupported(state)) {
         LOGGER.error(ProcedureMessages.FAIL_WHEN_EXECUTE, plan);
         setFailure(new ProcedureException(e));
       } else {
-        LOGGER.error(
-            ProcedureMessages.RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE, plan, state, e);
-        if (getCycles() > RETRY_THRESHOLD) {
-          setFailure(
-              new ProcedureException(
-                  String.format(
-                      ProcedureMessages.FAIL_TO_EXECUTE_PLAN_AT_STATE, plan.toString(), state)));
-        }
+        LOGGER.warn(ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED, e);
+        setFailure(
+            new ProcedureException(
+                new IoTDBException(
+                    ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED,
+                    e,
+                    TSStatusCode.AUTH_OPERATE_EXCEPTION.getStatusCode())));
       }
     }
     return Flow.HAS_MORE_STATE;
@@ -164,19 +153,26 @@ private void writePlan(ConfigNodeProcedureEnv env) {
     }
     if (res.code == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
       setNextState(DATANODE_AUTHCACHE_INVALIDING);
-      for (TDataNodeConfiguration item : datanodes) {
-        this.dataNodesToInvalid.add(new Pair<>(item, System.currentTimeMillis()));
-      }
-      LOGGER.info(
-          ProcedureMessages.EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES,
-          plan,
-          dataNodesToInvalid);
+      LOGGER.info(ProcedureMessages.EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES, plan);
     } else {
       LOGGER.info(ProcedureMessages.FAILED_TO_EXECUTE_PLAN_BECAUSE, plan, res.message);
       setFailure(new ProcedureException(new IoTDBException(res)));
     }
   }
 
+  private static Map<Integer, TSStatus> broadcastAuthorityCache(
+      final TInvalidatePermissionCacheReq req, final Map<Integer, TDataNodeLocation> targets) {
+    final DataNodeAsyncRequestContext<TInvalidatePermissionCacheReq, TSStatus> clientHandler =
+        new DataNodeAsyncRequestContext<>(
+            CnToDnAsyncRequestType.INVALIDATE_PERMISSION_CACHE, req, targets);
+    CnToDnInternalServiceAsyncRequestManager.getInstance()
+        .sendAsyncRequest(
+            clientHandler,
+            ClusterCachePropagator.BROADCAST_RPC_RETRY,
+            ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
+    return clientHandler.getResponseMap();
+  }
+
   @Override
   protected boolean isRollbackSupported(AuthOperationProcedureState state) {
     return state == AuthOperationProcedureState.INIT;

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java

@@ -949,12 +949,6 @@ private DataNodeMiscMessages() {}
   public static final String CACHE_ROLE_PATH_PRIVILEGES_ERROR =
       "cache role's path privileges error";
 
-  // ---------------------------------------------------------------------------
-  // auth – BasicAuthorityCache
-  // ---------------------------------------------------------------------------
-  public static final String DATANODE_CACHE_INIT_FAILED =
-      "datanode cache initialization failed";
-
   // ---------------------------------------------------------------------------
   // trigger – TriggerExecutor
   // ---------------------------------------------------------------------------

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java

@@ -947,11 +947,6 @@ private DataNodeMiscMessages() {}
   public static final String CACHE_ROLE_PATH_PRIVILEGES_ERROR =
       "缓存角色路径权限时发生错误";
 
-  // ---------------------------------------------------------------------------
-  // auth – BasicAuthorityCache
-  // ---------------------------------------------------------------------------
-  public static final String DATANODE_CACHE_INIT_FAILED =
-      "DataNode 缓存初始化失败";
 
   // ---------------------------------------------------------------------------
   // trigger – TriggerExecutor

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/AuthorityChecker.java

@@ -136,6 +136,7 @@ public static boolean invalidateCache(String username, String roleName) {
   }
 
   public static void invalidateAllCache() {
+    PipeInsertionDataNodeListener.getInstance().invalidateAllCache();
     authorityFetcher.get().getAuthorCache().invalidAllCache();
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/BasicAuthorityCache.java

@@ -22,7 +22,6 @@
 import org.apache.iotdb.commons.auth.entity.Role;
 import org.apache.iotdb.commons.auth.entity.User;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
@@ -90,19 +89,11 @@ public boolean invalidateCache(final String userName, final String roleName) {
         }
         userCache.invalidate(userName);
       }
-      if (userCache.getIfPresent(userName) != null) {
-        LOGGER.error(DataNodeMiscMessages.DATANODE_CACHE_INIT_FAILED);
-        return false;
-      }
     }
     if (roleName != null) {
       if (roleCache.getIfPresent(roleName) != null) {
         roleCache.invalidate(roleName);
       }
-      if (roleCache.getIfPresent(roleName) != null) {
-        LOGGER.error(DataNodeMiscMessages.DATANODE_CACHE_INIT_FAILED);
-        return false;
-      }
     }
     return true;
   }