jerryshao commented on code in PR #10996:
URL: https://github.com/apache/gravitino/pull/10996#discussion_r3246229844
##########
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java:
##########
@@ -587,178 +881,302 @@ private List<GroupEntity>
resolveCurrentUserGroups(String metalake, EntityStore
return entityStore.batchGet(groupIdents, Entity.EntityType.GROUP,
GroupEntity.class);
}
- /**
- * Adds a role mapping for the given user in both enforcers and
asynchronously loads the role's
- * policies if they are not already cached. When a role needs loading, the
resulting {@link
- * CompletableFuture} is appended to {@code loadRoleFutures} so the caller
can join all futures
- * after processing both direct and group-inherited roles.
- */
- private void addRoleForUserAndLoadPolicies(
- Long userId,
- String metalake,
- Long roleId,
- String roleName,
- List<CompletableFuture<Void>> loadRoleFutures,
- EntityStore entityStore) {
- allowEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- denyEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- if (loadedRoles.getIfPresent(roleId) != null) {
- return;
- }
- CompletableFuture<Void> loadRoleFuture =
- CompletableFuture.supplyAsync(
- () -> {
- try {
- return entityStore.get(
- NameIdentifierUtil.ofRole(metalake, roleName),
- Entity.EntityType.ROLE,
- RoleEntity.class);
- } catch (Exception e) {
- throw new RuntimeException("Failed to load role: " +
roleName, e);
- }
- },
- executor)
- .thenAcceptAsync(
- roleEntity -> {
- loadPolicyByRoleEntity(roleEntity);
- loadedRoles.put(roleId, true);
- },
- executor);
- loadRoleFutures.add(loadRoleFuture);
- }
-
- private void loadOwnerPolicy(String metalake, MetadataObject metadataObject,
Long metadataId) {
- if (ownerRel.getIfPresent(metadataId) != null) {
- LOG.debug("Metadata {} OWNER has been loaded.", metadataId);
- return;
- }
- try {
- NameIdentifier entityIdent = MetadataObjectUtil.toEntityIdent(metalake,
metadataObject);
- EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- List<? extends Entity> owners =
- entityStore
- .relationOperations()
- .listEntitiesByRelation(
- SupportsRelationOperations.Type.OWNER_REL,
- entityIdent,
- Entity.EntityType.valueOf(metadataObject.type().name()));
- if (owners.isEmpty()) {
- ownerRel.put(metadataId, Optional.empty());
- } else {
- for (Entity ownerEntity : owners) {
- if (ownerEntity instanceof UserEntity) {
- UserEntity user = (UserEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(user.id(), Entity.EntityType.USER,
user.name())));
- } else if (ownerEntity instanceof GroupEntity) {
- GroupEntity group = (GroupEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(group.id(), Entity.EntityType.GROUP,
group.name())));
- }
- }
+ private void versionCheckAndLoadRoles(
+ String metalake, List<Long> roleIds, AuthorizationRequestContext
requestContext) {
+ // Step 3: batch fetch (roleId, roleName, updated_at) for all role IDs — 1
query
+ List<Long> uniqueRoleIds =
roleIds.stream().distinct().collect(Collectors.toList());
+ List<RoleUpdatedAt> roleVersions =
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class, m -> m.batchGetRoleUpdatedAt(uniqueRoleIds));
+
+ for (RoleUpdatedAt rv : roleVersions) {
+ long roleId = rv.getRoleId();
+ long dbUpdatedAt = rv.getUpdatedAt();
+ Optional<Long> cachedUpdatedAt = loadedRoles.getIfPresent(roleId);
+
+ if (cachedUpdatedAt.isPresent() && cachedUpdatedAt.get() >= dbUpdatedAt)
{
+ // Role policies are still current
+ continue;
}
- } catch (IOException e) {
- LOG.warn("Can not load metadata owner", e);
+
+ // Stale or missing — evict old policies and reload
+ if (cachedUpdatedAt.isPresent()) {
+ allowEnforcer.deleteRole(String.valueOf(roleId));
+ denyEnforcer.deleteRole(String.valueOf(roleId));
+ }
+
+ // Load full role entity using roleName from the batch query (no extra
DB scan)
+ try {
+ EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
+ RoleEntity roleEntity =
+ entityStore.get(
+ NameIdentifierUtil.ofRole(metalake, rv.getRoleName()),
+ Entity.EntityType.ROLE,
+ RoleEntity.class);
+ loadPolicyByRoleEntity(roleEntity, requestContext);
+ } catch (Exception e) {
+ LOG.warn("Failed to load role policies for roleId {}", roleId, e);
+ continue;
+ }
+
+ loadedRoles.put(roleId, dbUpdatedAt);
}
}
- private void loadPolicyByRoleEntity(RoleEntity roleEntity) {
+ private void bindUserRoles(long userId, List<Long> roleIds) {
+ for (Long roleId : roleIds) {
+ allowEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
+ denyEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
+ }
+ }
+
+ //
---------------------------------------------------------------------------
+ // Policy loading from role entity
+ //
---------------------------------------------------------------------------
+
+ private void loadPolicyByRoleEntity(
+ RoleEntity roleEntity, AuthorizationRequestContext requestContext) {
String metalake =
NameIdentifierUtil.getMetalake(roleEntity.nameIdentifier());
List<SecurableObject> securableObjects = roleEntity.securableObjects();
for (SecurableObject securableObject : securableObjects) {
+ Long securableId = resolveMetadataId(securableObject, metalake,
requestContext);
for (Privilege privilege : securableObject.privileges()) {
Privilege.Condition condition = privilege.condition();
if (AuthConstants.DENY.equalsIgnoreCase(condition.name())) {
denyEnforcer.addPolicy(
String.valueOf(roleEntity.id()),
securableObject.type().name(),
- String.valueOf(MetadataIdConverter.getID(securableObject,
metalake)),
+ String.valueOf(securableId),
AuthorizationUtils.replaceLegacyPrivilegeName(privilege.name())
.name()
- .toUpperCase(java.util.Locale.ROOT),
+ .toUpperCase(Locale.ROOT),
AuthConstants.ALLOW);
}
- // Since different roles of a user may simultaneously hold both
"allow" and "deny"
- // permissions
- // for the same privilege on a given MetadataObject, the allowEnforcer
must also incorporate
- // the "deny" privilege to ensure that the authorize method correctly
returns false in such
- // cases. For example, if role1 has an "allow" privilege for
SELECT_TABLE on table1, while
- // role2 has a "deny" privilege for the same action on table1, then a
user assigned both
- // roles should receive a false result when calling the authorize
method.
allowEnforcer.addPolicy(
String.valueOf(roleEntity.id()),
securableObject.type().name(),
- String.valueOf(MetadataIdConverter.getID(securableObject,
metalake)),
+ String.valueOf(securableId),
AuthorizationUtils.replaceLegacyPrivilegeName(privilege.name())
.name()
- .toUpperCase(java.util.Locale.ROOT),
- condition.name().toLowerCase(java.util.Locale.ROOT));
+ .toUpperCase(Locale.ROOT),
+ condition.name().toLowerCase(Locale.ROOT));
}
}
}
+ //
---------------------------------------------------------------------------
+ // Change poller (eventual consistency for HA)
+ //
---------------------------------------------------------------------------
+
+ @VisibleForTesting
+ void pollChanges() {
+ try {
+ LOG.debug("Polling for owner changes after id {}", ownerPollHighWaterId);
+ pollOwnerChanges();
+ } catch (Exception e) {
+ LOG.warn("Owner change poll failed", e);
+ }
+
+ try {
+ LOG.debug("Polling for entity changes after id {}",
entityPollHighWaterId);
+ pollEntityChanges();
+ } catch (Exception e) {
+ LOG.warn("Entity change poll failed", e);
+ }
+ }
+
/**
- * Checks whether the given principal is the owner of the metadata object
identified by
- * metadataId. Supports both user and group ownership.
+ * Drains owner-change rows past {@link #ownerPollHighWaterId} and
invalidates the affected {@link
+ * #ownerRelCache} entries. Each row carries {@code metadataObjectId}, so
invalidation is a direct
+ * key removal — no name resolution needed.
*/
- private boolean checkOwnership(Principal principal, String metalake, Long
metadataId) {
- Optional<OwnerInfo> ownerOpt = ownerRel.getIfPresent(metadataId);
- if (ownerOpt == null || !ownerOpt.isPresent()) {
- return false;
+ private void pollOwnerChanges() {
+ List<ChangedOwnerInfo> changes =
+ SessionUtils.getWithoutCommit(
+ OwnerMetaMapper.class, m ->
m.selectChangedOwners(ownerPollHighWaterId));
+
+ long maxSeenId = ownerPollHighWaterId;
+ for (ChangedOwnerInfo change : changes) {
+ ownerRelCache.invalidate(change.getMetadataObjectId());
+ if (change.getId() > maxSeenId) {
+ maxSeenId = change.getId();
+ }
}
- OwnerInfo owner = ownerOpt.get();
- // We compare by entity ID rather than name to guard against stale cache
entries.
- // If a user/group is deleted and recreated with the same name, the cached
OwnerInfo
- // still holds the old ID. A name-only comparison would incorrectly grant
ownership
- // to the new entity. The extra IO to fetch the current entity ensures
correctness.
- if (owner.type == Entity.EntityType.USER) {
+ ownerPollHighWaterId = maxSeenId;
+ }
+
+ /**
+ * Drains entity-change rows past {@link #entityPollHighWaterId} and
invalidates the affected
+ * {@link #metadataIdCache} keys.
+ *
+ * <p><b>Contract with the writer side:</b> {@code
entity_change_log.full_name} must be the
+ * <i>pre-mutation</i> name (the name that consumers currently have cached).
The writers in {@code
+ * SchemaMetaService} / {@code TableMetaService} / etc. emit {@code
oldFullName} on rename and the
+ * current name on drop, so the cacheKey we build here resolves to the entry
a peer node would
+ * have populated under that name. If a future change starts emitting the
new post-rename name,
+ * this invalidation will silently miss and stale entries will only clear
via LRU eviction.
+ */
+ private void pollEntityChanges() {
+ List<EntityChangeRecord> changes =
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
+ m -> m.selectEntityChanges(entityPollHighWaterId,
POLLER_MAX_ROWS));
+
+ long maxSeenId = entityPollHighWaterId;
+ for (EntityChangeRecord change : changes) {
+ String metalake = change.getMetalakeName();
+ String entityType = change.getEntityType();
+ String fullName = change.getFullName();
+
+ MetadataObject.Type mdType;
try {
- UserEntity userEntity = getUserEntity(principal.getName(), metalake);
- return Objects.equals(userEntity.id(), owner.id);
- } catch (Exception e) {
- LOG.debug("Can not get user entity for ownership check", e);
- return false;
- }
- } else if (owner.type == Entity.EntityType.GROUP) {
- if (principal instanceof UserPrincipal) {
- List<UserGroup> groups = ((UserPrincipal) principal).getGroups();
- if (groups.isEmpty()) {
- return false;
- }
- try {
- List<NameIdentifier> groupIdents =
- groups.stream()
- .map(g -> NameIdentifierUtil.ofGroup(metalake,
g.getGroupname()))
- .collect(Collectors.toList());
- List<GroupEntity> groupEntities =
- GravitinoEnv.getInstance()
- .entityStore()
- .batchGet(groupIdents, Entity.EntityType.GROUP,
GroupEntity.class);
- return groupEntities.stream().anyMatch(ge -> Objects.equals(ge.id(),
owner.id));
- } catch (Exception e) {
- LOG.debug("Can not get group entities for ownership check", e);
- return false;
+ mdType =
MetadataObject.Type.valueOf(entityType.toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Unknown entity type in change log: {}", entityType);
+ if (change.getId() > maxSeenId) {
+ maxSeenId = change.getId();
}
+ continue;
+ }
+
+ MetadataObject mdObj = metadataObjectFromChangeLog(metalake, fullName,
mdType);
+ String cacheKey = buildCacheKey(metalake, mdObj);
+
+ if (isNonLeaf(mdType)) {
+ metadataIdCache.invalidateByPrefix(cacheKey);
+ } else {
+ metadataIdCache.invalidate(cacheKey);
+ }
+
+ if (change.getId() > maxSeenId) {
+ maxSeenId = change.getId();
}
- return false;
}
- return false;
+ entityPollHighWaterId = maxSeenId;
+ }
+
+ //
---------------------------------------------------------------------------
+ // Helpers
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Builds a hierarchical cache key for the metadataIdCache. Non-leaf objects
end with "::" to
+ * enable prefix-based cascade invalidation.
+ *
+ * <p>Examples: metalake:: , metalake::catalog:: ,
metalake::catalog::schema:: ,
+ * metalake::catalog::schema::table::TABLE
+ */
+ @VisibleForTesting
+ static String buildCacheKey(String metalake, MetadataObject metadataObject) {
+ if (metadataObject.type() == MetadataObject.Type.METALAKE) {
+ return metalake + KEY_SEP;
+ }
+ StringBuilder sb = new StringBuilder(metalake);
+ sb.append(KEY_SEP);
+ // fullName uses '.' as separator, e.g. "catalog1.schema1.table1"
+ String[] parts = metadataObject.fullName().split("\\.");
+ sb.append(String.join(KEY_SEP, parts));
+ if (isNonLeaf(metadataObject.type())) {
+ // Trailing separator enables prefix-based cascade invalidation
+ sb.append(KEY_SEP);
+ } else {
+ // Leaf nodes get the type suffix to avoid collisions
+ sb.append(KEY_SEP);
+ sb.append(metadataObject.type().name());
+ }
+ return sb.toString();
+ }
+
+ @VisibleForTesting
+ static MetadataObject metadataObjectFromChangeLog(
+ String metalake, String fullName, MetadataObject.Type type) {
+ List<String> names = new ArrayList<>(Arrays.asList(fullName.split("\\.")));
+ if (type != MetadataObject.Type.METALAKE
+ && !names.isEmpty()
+ && Objects.equals(names.get(0), metalake)) {
+ names.remove(0);
+ }
+ return MetadataObjects.of(names, type);
+ }
+
+ /** Returns true for entity types that can contain children (metalake,
catalog, schema). */
+ @VisibleForTesting
+ static boolean isNonLeaf(MetadataObject.Type type) {
+ return type == MetadataObject.Type.METALAKE
+ || type == MetadataObject.Type.CATALOG
+ || type == MetadataObject.Type.SCHEMA;
}
- /** Holds the owner identity for a metadata object in the owner cache. */
- static class OwnerInfo {
- final Long id;
- final Entity.EntityType type;
- final String name;
+ private static long nullToZero(Long value) {
+ return value == null ? 0L : value;
+ }
+
+ //
---------------------------------------------------------------------------
+ // LoadedRoles cache — wraps CaffeineGravitinoCache with eviction
side-effects
+ //
---------------------------------------------------------------------------
+
+ /**
+ * A specialized GravitinoCache for loaded roles that cleans up JCasbin
policies on eviction. This
+ * uses a raw Caffeine cache internally so that we can attach a removal
listener.
+ */
+ private static class LoadedRolesCache implements GravitinoCache<Long, Long> {
+
+ private final Cache<Long, Long> cache;
+
+ LoadedRolesCache(long ttlMs, long maxSize, Enforcer allowEnforcer,
Enforcer denyEnforcer) {
+ this.cache =
+ com.github.benmanes.caffeine.cache.Caffeine.newBuilder()
+ .expireAfterAccess(ttlMs, TimeUnit.MILLISECONDS)
+ .maximumSize(maxSize)
+ .executor(Runnable::run)
+ .removalListener(
+ (roleId, value, cause) -> {
+ if (roleId != null) {
+ allowEnforcer.deleteRole(String.valueOf(roleId));
+ denyEnforcer.deleteRole(String.valueOf(roleId));
+ }
+ })
+ .build();
+ }
+
+ @Override
+ public Optional<Long> getIfPresent(Long key) {
+ Long v = cache.getIfPresent(key);
+ return Optional.ofNullable(v);
+ }
+
+ @Override
+ public void put(Long key, Long value) {
+ cache.put(key, value);
+ }
+
+ @Override
+ public void invalidate(Long key) {
+ cache.invalidate(key);
+ }
+
+ @Override
+ public void invalidateAll() {
+ cache.invalidateAll();
+ }
+
+ @Override
+ public void invalidateByPrefix(String prefix) {
+ cache.asMap().keySet().removeIf(k -> k.toString().startsWith(prefix));
+ }
+
+ @Override
+ public long size() {
+ cache.cleanUp();
+ return cache.estimatedSize();
+ }
- OwnerInfo(Long id, Entity.EntityType type, String name) {
- this.id = id;
- this.type = type;
- this.name = name;
+ @Override
+ public void close() {
+ cache.invalidateAll();
+ cache.cleanUp();
}
}
}
Review Comment:
This class is so big, I suggest that we can split into several small classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
