Skip to content

Commit 72798c5

Browse files
alpass163gmailJackieTien97
alpass163gmail
authored and
JackieTien97
committed
Support high availability for the authority module (#18425)
1 parent 6349515 commit 72798c5

15 files changed

Lines changed: 766 additions & 135 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableAuthHAIT.java‎

Lines changed: 670 additions & 0 deletions
Large diffs are not rendered by default.

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

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,9 @@ public final class ProcedureMessages {
9393
public static final String AUTHENTICATION_FAILED = "Authentication failed.";
9494
public static final String AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY =
9595
"Auth procedure: clean datanode cache successfully";
96+
public static final String AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED =
97+
"Auth plan has been committed, but DataNode permission cache invalidation failed. "
98+
+ "Some DataNodes may have stale permissions; please clear their permission cache manually.";
9699
public static final String BEGIN_TO_CHANGE_DATANODE_STATUS_NODESTATUSMAP =
97100
"{}, Begin to change DataNode status, nodeStatusMap: {}";
98101
public static final String BEGIN_TO_STOP_DATANODES_AND_KILL_THE_DATANODE_PROCESS =
@@ -356,7 +359,7 @@ public final class ProcedureMessages {
356359
public static final String ERROR_IN_DESERIALIZE_PROCID_THIS_PROCEDURE_WILL_BE_IGNORED_IT =
357360
"Error in deserialize {} (procID {}). This procedure will be ignored. It may belong to old version and cannot be used now.";
358361
public static final String EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES =
359-
"Execute auth plan {} success. To invalidate datanodes: {}";
362+
"Execute auth plan {} success.";
360363
public static final String EXECUTING_ON_REGION_FOR_COLUMN_IN_WHEN_DROPPING_COLUMN =
361364
"Executing on region for column {} in {}.{} when dropping column";
362365
public static final String FAILED_TO_ACTIVE_CQ_BECAUSE_OF_NO_SUCH_CQ =
@@ -551,8 +554,6 @@ public final class ProcedureMessages {
551554
"Fail to drop trigger [%s] at STATE [%s]";
552555
public static final String FAIL_TO_DROP_TRIGGER_ON_DATA_NODES =
553556
"Fail to drop trigger [%s] on Data Nodes";
554-
public static final String FAIL_TO_EXECUTE_PLAN_AT_STATE =
555-
"Fail to execute plan [%s] at state[%s]";
556557
public static final String FAIL_TO_REMOVE_AINODE_AT_STATE =
557558
"Fail to remove AINode [%s] at STATE [%s], %s";
558559
public static final String FAIL_TO_REMOVE_AINODE_ON_CONFIG_NODES =
@@ -860,8 +861,6 @@ public final class ProcedureMessages {
860861
"Retrievable error trying to create pipe plugin [{}], state: {}";
861862
public static final String RETRIEVABLE_ERROR_TRYING_TO_DROP_PIPE_PLUGIN_STATE =
862863
"Retrievable error trying to drop pipe plugin [{}], state: {}";
863-
public static final String RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE =
864-
"Retrievable error trying to execute plan {}, state: {}";
865864
public static final String RETRIEVABLE_ERROR_TRYING_TO_REMOVE_AINODE_STATE =
866865
"Retrievable error trying to remove AINode [{}], state [{}]";
867866
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‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,8 @@ public final class ProcedureMessages {
9191
public static final String AUTHENTICATION_FAILED = "认证失败。";
9292
public static final String AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY =
9393
"Auth procedure：成功清理 datanode 缓存";
94+
public static final String AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED =
95+
"权限计划已提交，但 DataNode 权限缓存失效失败。部分 DataNode 可能仍持有过期权限，请手动清理权限缓存。";
9496
public static final String BEGIN_TO_CHANGE_DATANODE_STATUS_NODESTATUSMAP =
9597
"{}, 开始修改 DataNode 状态，nodeStatusMap：{}";
9698
public static final String BEGIN_TO_STOP_DATANODES_AND_KILL_THE_DATANODE_PROCESS =
@@ -344,7 +346,7 @@ public final class ProcedureMessages {
344346
public static final String ERROR_IN_DESERIALIZE_PROCID_THIS_PROCEDURE_WILL_BE_IGNORED_IT =
345347
"反序列化 {}（procID {}）出错。该 procedure 将被忽略。它可能属于旧版本，目前无法使用。";
346348
public static final String EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES =
347-
"执行 auth plan {} 成功。使 datanode 缓存失效：{}";
349+
"执行 auth plan {} 成功。";
348350
public static final String EXECUTING_ON_REGION_FOR_COLUMN_IN_WHEN_DROPPING_COLUMN =
349351
"删除列时在表 {}.{} 中列 {} 对应的 region 上执行";
350352
public static final String FAILED_TO_ACTIVE_CQ_BECAUSE_OF_NO_SUCH_CQ =
@@ -531,7 +533,6 @@ public final class ProcedureMessages {
531533
"重试 {} 次后删除 pipe plugin [{}] 仍失败";
532534
public static final String FAIL_TO_DROP_TRIGGER_AT_STATE = "在 STATE [%s] 处删除 trigger [%s] 失败";
533535
public static final String FAIL_TO_DROP_TRIGGER_ON_DATA_NODES = "在 DataNode 上删除 trigger [%s] 失败";
534-
public static final String FAIL_TO_EXECUTE_PLAN_AT_STATE = "在 state[%s] 处执行 plan [%s] 失败";
535536
public static final String FAIL_TO_REMOVE_AINODE_AT_STATE = "在 STATE [%s] 处移除 AINode [%s] 失败，%s";
536537
public static final String FAIL_TO_REMOVE_AINODE_ON_CONFIG_NODES =
537538
"在 ConfigNode [%s] 上移除 [%s] 个 AINode 失败";
@@ -818,8 +819,6 @@ public final class ProcedureMessages {
818819
"尝试创建 pipe plugin [{}] 时发生可重试错误，状态：{}";
819820
public static final String RETRIEVABLE_ERROR_TRYING_TO_DROP_PIPE_PLUGIN_STATE =
820821
"尝试删除 pipe plugin [{}] 时发生可重试错误，状态：{}";
821-
public static final String RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE =
822-
"尝试执行 plan {} 时发生可重试错误，状态：{}";
823822
public static final String RETRIEVABLE_ERROR_TRYING_TO_REMOVE_AINODE_STATE =
824823
"尝试移除 AINode [{}] 时发生可重试错误，状态 [{}]";
825824
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‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,4 +138,7 @@ public enum CnToDnAsyncRequestType {
138138
// audit log and event write-back
139139
INSERT_RECORD,
140140
ENABLE_SEPARATION_OF_ADMIN_POWERS,
141+
142+
// authority
143+
INVALIDATE_PERMISSION_CACHE,
141144
}

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,7 @@
8282
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
8383
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateColumnCacheReq;
8484
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq;
85+
import org.apache.iotdb.mpp.rpc.thrift.TInvalidatePermissionCacheReq;
8586
import org.apache.iotdb.mpp.rpc.thrift.TInvalidateTableCacheReq;
8687
import org.apache.iotdb.mpp.rpc.thrift.TKillQueryInstanceReq;
8788
import org.apache.iotdb.mpp.rpc.thrift.TNotifyRegionMigrationReq;
@@ -527,6 +528,11 @@ protected void initActionMapBuilder() {
527528
CnToDnAsyncRequestType.GET_BUILTIN_SERVICE,
528529
(req, client, handler) ->
529530
client.getBuiltInService((GetBuiltInExternalServiceRPCHandler) handler));
531+
actionMapBuilder.put(
532+
CnToDnAsyncRequestType.INVALIDATE_PERMISSION_CACHE,
533+
(req, client, handler) ->
534+
client.invalidatePermissionCache(
535+
(TInvalidatePermissionCacheReq) req, (DataNodeTSStatusRPCHandler) handler));
530536
}
531537

532538
@Override

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

Lines changed: 41 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -20,21 +20,25 @@
2020
package org.apache.iotdb.confignode.procedure.impl.sync;
2121

2222
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
23+
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
2324
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2425
import org.apache.iotdb.commons.conf.CommonConfig;
2526
import org.apache.iotdb.commons.conf.CommonDescriptor;
2627
import org.apache.iotdb.commons.exception.IoTDBException;
2728
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
28-
import org.apache.iotdb.confignode.client.sync.CnToDnSyncRequestType;
29-
import org.apache.iotdb.confignode.client.sync.SyncDataNodeClientPool;
29+
import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
30+
import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
31+
import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
3032
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
3133
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
3234
import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
3335
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
3436
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
37+
import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator;
3538
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
3639
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
3740
import org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure;
41+
import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils;
3842
import org.apache.iotdb.confignode.procedure.state.auth.AuthOperationProcedureState;
3943
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
4044
import org.apache.iotdb.consensus.exception.ConsensusException;
@@ -50,8 +54,8 @@
5054
import java.io.IOException;
5155
import java.nio.ByteBuffer;
5256
import java.util.ArrayList;
53-
import java.util.Iterator;
5457
import java.util.List;
58+
import java.util.Map;
5559
import java.util.Objects;
5660

5761
import static org.apache.iotdb.confignode.procedure.state.auth.AuthOperationProcedureState.DATANODE_AUTHCACHE_INVALIDING;
@@ -68,7 +72,6 @@ public class AuthOperationProcedure extends AbstractNodeProcedure<AuthOperationP
6872
private static final String CONSENSUS_WRITE_ERROR =
6973
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
7074

71-
private static final int RETRY_THRESHOLD = 2;
7275
private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
7376

7477
private final List<Pair<TDataNodeConfiguration, Long>> dataNodesToInvalid = new ArrayList<>();
@@ -97,54 +100,40 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, AuthOperationProcedu
97100
writePlan(env);
98101
return Flow.HAS_MORE_STATE;
99102
case DATANODE_AUTHCACHE_INVALIDING:
100-
TInvalidatePermissionCacheReq req = new TInvalidatePermissionCacheReq();
101-
TSStatus status;
102-
req.setUsername(user);
103-
req.setRoleName(role);
103+
TInvalidatePermissionCacheReq req = new TInvalidatePermissionCacheReq(user, role);
104104
if (plan.getAuthorType() == ConfigPhysicalPlanType.AccountUnlock
105105
|| plan.getAuthorType() == ConfigPhysicalPlanType.RAccountUnlock) {
106106
// For account unlock, role carries the optional login address.
107107
req.setNeedDisconnect(true);
108108
}
109-
Iterator<Pair<TDataNodeConfiguration, Long>> it = dataNodesToInvalid.iterator();
110-
while (it.hasNext()) {
111-
Pair<TDataNodeConfiguration, Long> pair = it.next();
112-
if (pair.getRight() + this.timeoutMS < System.currentTimeMillis()) {
113-
it.remove();
114-
continue;
115-
}
116-
status =
117-
(TSStatus)
118-
SyncDataNodeClientPool.getInstance()
119-
.sendSyncRequestToDataNodeWithRetry(
120-
pair.getLeft().getLocation().getInternalEndPoint(),
121-
req,
122-
CnToDnSyncRequestType.INVALIDATE_PERMISSION_CACHE);
123-
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
124-
it.remove();
125-
}
126-
}
127-
if (dataNodesToInvalid.isEmpty()) {
109+
final boolean proceeded =
110+
new ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager()))
111+
.propagate(targets -> broadcastAuthorityCache(req, targets));
112+
113+
if (proceeded) {
128114
LOGGER.info(ProcedureMessages.AUTH_PROCEDURE_CLEAN_DATANODE_CACHE_SUCCESSFULLY);
129115
return Flow.NO_MORE_STATE;
130-
} else {
131-
setNextState(AuthOperationProcedureState.DATANODE_AUTHCACHE_INVALIDING);
132116
}
133-
break;
117+
LOGGER.warn(ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED);
118+
setFailure(
119+
new ProcedureException(
120+
new IoTDBException(
121+
ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED,
122+
TSStatusCode.AUTH_OPERATE_EXCEPTION.getStatusCode())));
123+
return Flow.NO_MORE_STATE;
134124
}
135125
} catch (Exception e) {
136126
if (isRollbackSupported(state)) {
137127
LOGGER.error(ProcedureMessages.FAIL_WHEN_EXECUTE, plan);
138128
setFailure(new ProcedureException(e));
139129
} else {
140-
LOGGER.error(
141-
ProcedureMessages.RETRIEVABLE_ERROR_TRYING_TO_EXECUTE_PLAN_STATE, plan, state, e);
142-
if (getCycles() > RETRY_THRESHOLD) {
143-
setFailure(
144-
new ProcedureException(
145-
String.format(
146-
ProcedureMessages.FAIL_TO_EXECUTE_PLAN_AT_STATE, plan.toString(), state)));
147-
}
130+
LOGGER.warn(ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED, e);
131+
setFailure(
132+
new ProcedureException(
133+
new IoTDBException(
134+
ProcedureMessages.AUTH_PROCEDURE_CACHE_INVALIDATION_FAILED,
135+
e,
136+
TSStatusCode.AUTH_OPERATE_EXCEPTION.getStatusCode())));
148137
}
149138
}
150139
return Flow.HAS_MORE_STATE;
@@ -164,19 +153,26 @@ private void writePlan(ConfigNodeProcedureEnv env) {
164153
}
165154
if (res.code == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
166155
setNextState(DATANODE_AUTHCACHE_INVALIDING);
167-
for (TDataNodeConfiguration item : datanodes) {
168-
this.dataNodesToInvalid.add(new Pair<>(item, System.currentTimeMillis()));
169-
}
170-
LOGGER.info(
171-
ProcedureMessages.EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES,
172-
plan,
173-
dataNodesToInvalid);
156+
LOGGER.info(ProcedureMessages.EXECUTE_AUTH_PLAN_SUCCESS_TO_INVALIDATE_DATANODES, plan);
174157
} else {
175158
LOGGER.info(ProcedureMessages.FAILED_TO_EXECUTE_PLAN_BECAUSE, plan, res.message);
176159
setFailure(new ProcedureException(new IoTDBException(res)));
177160
}
178161
}
179162

163+
private static Map<Integer, TSStatus> broadcastAuthorityCache(
164+
final TInvalidatePermissionCacheReq req, final Map<Integer, TDataNodeLocation> targets) {
165+
final DataNodeAsyncRequestContext<TInvalidatePermissionCacheReq, TSStatus> clientHandler =
166+
new DataNodeAsyncRequestContext<>(
167+
CnToDnAsyncRequestType.INVALIDATE_PERMISSION_CACHE, req, targets);
168+
CnToDnInternalServiceAsyncRequestManager.getInstance()
169+
.sendAsyncRequest(
170+
clientHandler,
171+
ClusterCachePropagator.BROADCAST_RPC_RETRY,
172+
ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS);
173+
return clientHandler.getResponseMap();
174+
}
175+
180176
@Override
181177
protected boolean isRollbackSupported(AuthOperationProcedureState state) {
182178
return state == AuthOperationProcedureState.INIT;

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

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -947,12 +947,6 @@ private DataNodeMiscMessages() {}
947947
public static final String CACHE_ROLE_PATH_PRIVILEGES_ERROR =
948948
"cache role's path privileges error";
949949

950-
// ---------------------------------------------------------------------------
951-
// auth – BasicAuthorityCache
952-
// ---------------------------------------------------------------------------
953-
public static final String DATANODE_CACHE_INIT_FAILED =
954-
"datanode cache initialization failed";
955-
956950
// ---------------------------------------------------------------------------
957951
// trigger – TriggerExecutor
958952
// ---------------------------------------------------------------------------

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

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -945,11 +945,6 @@ private DataNodeMiscMessages() {}
945945
public static final String CACHE_ROLE_PATH_PRIVILEGES_ERROR =
946946
"缓存角色路径权限时发生错误";
947947

948-
// ---------------------------------------------------------------------------
949-
// auth – BasicAuthorityCache
950-
// ---------------------------------------------------------------------------
951-
public static final String DATANODE_CACHE_INIT_FAILED =
952-
"DataNode 缓存初始化失败";
953948

954949
// ---------------------------------------------------------------------------
955950
// trigger – TriggerExecutor

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,7 @@ public static boolean invalidateCache(String username, String roleName) {
135135
}
136136

137137
public static void invalidateAllCache() {
138+
PipeInsertionDataNodeListener.getInstance().invalidateAllCache();
138139
authorityFetcher.get().getAuthorCache().invalidAllCache();
139140
}
140141

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

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import org.apache.iotdb.commons.auth.entity.Role;
2323
import org.apache.iotdb.commons.auth.entity.User;
2424
import org.apache.iotdb.db.conf.IoTDBDescriptor;
25-
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
2625

2726
import com.github.benmanes.caffeine.cache.Cache;
2827
import com.github.benmanes.caffeine.cache.Caffeine;
@@ -90,19 +89,11 @@ public boolean invalidateCache(final String userName, final String roleName) {
9089
}
9190
userCache.invalidate(userName);
9291
}
93-
if (userCache.getIfPresent(userName) != null) {
94-
LOGGER.error(DataNodeMiscMessages.DATANODE_CACHE_INIT_FAILED);
95-
return false;
96-
}
9792
}
9893
if (roleName != null) {
9994
if (roleCache.getIfPresent(roleName) != null) {
10095
roleCache.invalidate(roleName);
10196
}
102-
if (roleCache.getIfPresent(roleName) != null) {
103-
LOGGER.error(DataNodeMiscMessages.DATANODE_CACHE_INIT_FAILED);
104-
return false;
105-
}
10697
}
10798
return true;
10899
}

0 commit comments

Comments
 (0)