[#2759] feat(server,core): Add service admin and metalake admin

core/src/main/java/com/datastrato/gravitino/Configs.java

@@ -9,6 +9,7 @@
 import com.datastrato.gravitino.config.ConfigEntry;
 import com.google.common.collect.Lists;
 import java.io.File;
+import java.util.List;
 import org.apache.commons.lang3.StringUtils;
 
 public interface Configs {
@@ -239,4 +240,19 @@ public interface Configs {
           .version(ConfigConstants.VERSION_0_4_0)
           .longConf()
           .createWithDefault(CLEAN_INTERVAL_IN_SECS);
+
+  ConfigEntry<Boolean> ENABLE_AUTHORIZATION =
+      new ConfigBuilder("gravitino.authorization.enable")
+          .doc("Enable the authorization")
+          .version(ConfigConstants.VERSION_0_5_0)
+          .booleanConf()
+          .createWithDefault(false);
+
+  ConfigEntry<List<String>> SERVICE_ADMINS =
+      new ConfigBuilder("gravitino.authorization.serviceAdmins")
+          .doc("The admins of Gravitino service")
+          .version(ConfigConstants.VERSION_0_5_0)
+          .stringConf()
+          .toSequence()
+          .create();
 }

core/src/main/java/com/datastrato/gravitino/GravitinoEnv.java

@@ -116,7 +116,7 @@ public void initialize(Config config) {
         new TopicOperationDispatcher(catalogManager, entityStore, idGenerator);
 
     // Create and initialize access control related modules
-    this.accessControlManager = new AccessControlManager(entityStore, idGenerator);
+    this.accessControlManager = new AccessControlManager(entityStore, idGenerator, config);
 
     this.auxServiceManager = new AuxiliaryServiceManager();
     this.auxServiceManager.serviceInit(

core/src/main/java/com/datastrato/gravitino/authorization/AccessControlManager.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.authorization;
 
+import com.datastrato.gravitino.Config;
 import com.datastrato.gravitino.EntityStore;
 import com.datastrato.gravitino.exceptions.GroupAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.NoSuchGroupException;
@@ -12,15 +13,17 @@
 import com.datastrato.gravitino.storage.IdGenerator;
 
 /**
- * AccessControlManager is used for manage users, roles, grant information, this class is an
+ * AccessControlManager is used for manage users, roles, admin, grant information, this class is an
  * entrance class for tenant management.
  */
 public class AccessControlManager {
 
   private final UserGroupManager userGroupManager;
+  private final AdminManager adminManager;
 
-  public AccessControlManager(EntityStore store, IdGenerator idGenerator) {
+  public AccessControlManager(EntityStore store, IdGenerator idGenerator, Config config) {
     this.userGroupManager = new UserGroupManager(store, idGenerator);
+    this.adminManager = new AdminManager(store, idGenerator, config);
   }
 
   /**
@@ -98,4 +101,47 @@ public boolean removeGroup(String metalake, String group) {
   public Group getGroup(String metalake, String group) throws NoSuchGroupException {
     return userGroupManager.getGroup(metalake, group);
   }
+
+  /**
+   * Adds a new metalake admin.
+   *
+   * @param user The name of the User.
+   * @return The added User instance.
+   * @throws UserAlreadyExistsException If a User with the same identifier already exists.
+   * @throws RuntimeException If adding the User encounters storage issues.
+   */
+  public User addMetalakeAdmin(String user) {
+    return adminManager.addMetalakeAdmin(user);
+  }
+
+  /**
+   * Removes a metalake admin.
+   *
+   * @param user The name of the User.
+   * @return `true` if the User was successfully removed, `false` otherwise.
+   * @throws RuntimeException If removing the User encounters storage issues.
+   */
+  public boolean removeMetalakeAdmin(String user) {
+    return adminManager.removeMetalakeAdmin(user);
+  }
+
+  /**
+   * Judges whether the user is the service admin.
+   *
+   * @param user the name of the user
+   * @return true, if the user is service admin, otherwise false.
+   */
+  public boolean isServiceAdmin(String user) {
+    return adminManager.isServiceAdmin(user);
+  }
+
+  /**
+   * Judges whether the user is the metalake admin.
+   *
+   * @param user the name of the user
+   * @return true, if the user is metalake admin, otherwise false.
+   */
+  public boolean isMetalakeAdmin(String user) {
+    return adminManager.isMetalakeAdmin(user);
+  }
 }

core/src/main/java/com/datastrato/gravitino/authorization/AdminManager.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.authorization;
+
+import com.datastrato.gravitino.Config;
+import com.datastrato.gravitino.Configs;
+import com.datastrato.gravitino.Entity;
+import com.datastrato.gravitino.EntityAlreadyExistsException;
+import com.datastrato.gravitino.EntityStore;
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.Namespace;
+import com.datastrato.gravitino.exceptions.UserAlreadyExistsException;
+import com.datastrato.gravitino.meta.AuditInfo;
+import com.datastrato.gravitino.meta.EntitySpecificConstants;
+import com.datastrato.gravitino.meta.UserEntity;
+import com.datastrato.gravitino.storage.IdGenerator;
+import com.datastrato.gravitino.utils.PrincipalUtils;
+import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * There are two kinds of admin roles in the system: service admin and metalake admin. The service
+ * admin is configured instead of managing by APIs. It is responsible for creating metalake admin.
+ * If Gravitino enables authorization, service admin is required. Metalake admin can create a
+ * metalake or drops its metalake. The metalake admin will be responsible for managing the access
+ * control. AdminManager operates underlying store using the lock because kv storage needs the lock.
+ */
+public class AdminManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(AdminManager.class);
+
+  private final EntityStore store;
+  private final IdGenerator idGenerator;
+  private final List<String> serviceAdmins;
+
+  public AdminManager(EntityStore store, IdGenerator idGenerator, Config config) {
+    this.store = store;
+    this.idGenerator = idGenerator;
+    this.serviceAdmins = config.get(Configs.SERVICE_ADMINS);
+  }
+
+  /**
+   * Adds a new metalake admin.
+   *
+   * @param user The name of the User.
+   * @return The added User instance.
+   * @throws UserAlreadyExistsException If a User with the same identifier already exists.
+   * @throws RuntimeException If adding the User encounters storage issues.
+   */
+  public synchronized User addMetalakeAdmin(String user) {
+
+    UserEntity userEntity =
+        UserEntity.builder()
+            .withId(idGenerator.nextId())
+            .withName(user)
+            .withNamespace(
+                Namespace.of(
+                    EntitySpecificConstants.SYSTEM_METALAKE_RESERVED_NAME,
+                    EntitySpecificConstants.AUTHORIZATION_CATALOG_NAME,
+                    EntitySpecificConstants.ADMIN_SCHEMA_NAME))
+            .withRoles(Lists.newArrayList())
+            .withAuditInfo(
+                AuditInfo.builder()
+                    .withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+                    .withCreateTime(Instant.now())
+                    .build())
+            .build();
+    try {
+      store.put(userEntity, false /* overwritten */);
+      return userEntity;
+    } catch (EntityAlreadyExistsException e) {
+      LOG.warn("User {} in the metalake admin already exists", user, e);
+      throw new UserAlreadyExistsException("User %s in the metalake admin already exists", user);
+    } catch (IOException ioe) {
+      LOG.error("Adding user {} failed to the metalake admin due to storage issues", user, ioe);
+      throw new RuntimeException(ioe);
+    }
+  }
+
+  /**
+   * Removes a metalake admin.
+   *
+   * @param user The name of the User.
+   * @return `true` if the User was successfully removed, `false` otherwise.
+   * @throws RuntimeException If removing the User encounters storage issues.
+   */
+  public synchronized boolean removeMetalakeAdmin(String user) {
+    try {
+      return store.delete(ofMetalakeAdmin(user), Entity.EntityType.USER);
+    } catch (IOException ioe) {
+      LOG.error(
+          "Removing user {} from the metalake admin {} failed due to storage issues", user, ioe);
+      throw new RuntimeException(ioe);
+    }
+  }
+
+  /**
+   * Judges whether the user is the service admin.
+   *
+   * @param user the name of the user
+   * @return true, if the user is service admin, otherwise false.
+   */
+  public boolean isServiceAdmin(String user) {
+    return serviceAdmins.contains(user);
+  }
+
+  /**
+   * Judges whether the user is the metalake admin.
+   *
+   * @param user the name of the user
+   * @return true, if the user is metalake admin, otherwise false.
+   */
+  public synchronized boolean isMetalakeAdmin(String user) {
+    try {
+      return store.exists(ofMetalakeAdmin(user), Entity.EntityType.USER);
+    } catch (IOException ioe) {
+      LOG.error(
+          "Fail to check whether {} is the metalake admin {} due to storage issues", user, ioe);
+      throw new RuntimeException(ioe);
+    }
+  }
+
+  private NameIdentifier ofMetalakeAdmin(String user) {
+    return NameIdentifier.of(
+        EntitySpecificConstants.SYSTEM_METALAKE_RESERVED_NAME,
+        EntitySpecificConstants.AUTHORIZATION_CATALOG_NAME,
+        EntitySpecificConstants.ADMIN_SCHEMA_NAME,
+        user);
+  }
+}

core/src/main/java/com/datastrato/gravitino/authorization/UserGroupManager.java

@@ -15,7 +15,7 @@
 import com.datastrato.gravitino.exceptions.NoSuchUserException;
 import com.datastrato.gravitino.exceptions.UserAlreadyExistsException;
 import com.datastrato.gravitino.meta.AuditInfo;
-import com.datastrato.gravitino.meta.CatalogEntity;
+import com.datastrato.gravitino.meta.EntitySpecificConstants;
 import com.datastrato.gravitino.meta.GroupEntity;
 import com.datastrato.gravitino.meta.UserEntity;
 import com.datastrato.gravitino.storage.IdGenerator;
@@ -66,8 +66,8 @@ public User addUser(String metalake, String name) throws UserAlreadyExistsExcept
             .withNamespace(
                 Namespace.of(
                     metalake,
-                    CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
-                    UserEntity.USER_SCHEMA_NAME))
+                    EntitySpecificConstants.SYSTEM_CATALOG_RESERVED_NAME,
+                    EntitySpecificConstants.USER_SCHEMA_NAME))
             .withRoles(Lists.newArrayList())
             .withAuditInfo(
                 AuditInfo.builder()
@@ -146,8 +146,8 @@ public Group addGroup(String metalake, String group) throws GroupAlreadyExistsEx
             .withNamespace(
                 Namespace.of(
                     metalake,
-                    CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
-                    GroupEntity.GROUP_SCHEMA_NAME))
+                    EntitySpecificConstants.SYSTEM_CATALOG_RESERVED_NAME,
+                    EntitySpecificConstants.GROUP_SCHEMA_NAME))
             .withRoles(Collections.emptyList())
             .withAuditInfo(
                 AuditInfo.builder()
@@ -213,11 +213,17 @@ public Group getGroup(String metalake, String group) {
 
   private NameIdentifier ofUser(String metalake, String user) {
     return NameIdentifier.of(
-        metalake, CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME, UserEntity.USER_SCHEMA_NAME, user);
+        metalake,
+        EntitySpecificConstants.SYSTEM_CATALOG_RESERVED_NAME,
+        EntitySpecificConstants.USER_SCHEMA_NAME,
+        user);
   }
 
   private NameIdentifier ofGroup(String metalake, String group) {
     return NameIdentifier.of(
-        metalake, CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME, GroupEntity.GROUP_SCHEMA_NAME, group);
+        metalake,
+        EntitySpecificConstants.SYSTEM_CATALOG_RESERVED_NAME,
+        EntitySpecificConstants.GROUP_SCHEMA_NAME,
+        group);
   }
 }

core/src/main/java/com/datastrato/gravitino/catalog/CatalogManager.java

@@ -32,6 +32,7 @@
 import com.datastrato.gravitino.messaging.TopicCatalog;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.CatalogEntity;
+import com.datastrato.gravitino.meta.EntitySpecificConstants;
 import com.datastrato.gravitino.rel.SupportsSchemas;
 import com.datastrato.gravitino.rel.TableCatalog;
 import com.datastrato.gravitino.storage.IdGenerator;
@@ -287,7 +288,7 @@ public Catalog createCatalog(
       Map<String, String> properties)
       throws NoSuchMetalakeException, CatalogAlreadyExistsException {
 
-    if (CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME.equals(ident.name())) {
+    if (EntitySpecificConstants.SYSTEM_CATALOG_RESERVED_NAME.equals(ident.name())) {
       throw new IllegalArgumentException("Can't create a catalog with with reserved name `system`");
     }
 
@@ -678,7 +679,7 @@ private CatalogEntity.Builder updateEntity(
       if (change instanceof CatalogChange.RenameCatalog) {
         CatalogChange.RenameCatalog rename = (CatalogChange.RenameCatalog) change;
 
-        if (CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME.equals(
+        if (EntitySpecificConstants.SYSTEM_CATALOG_RESERVED_NAME.equals(
             ((CatalogChange.RenameCatalog) change).getNewName())) {
           throw new IllegalArgumentException(
               "Can't rename a catalog with with reserved name `system`");

core/src/main/java/com/datastrato/gravitino/meta/CatalogEntity.java

@@ -24,8 +24,6 @@
 @ToString
 public class CatalogEntity implements Entity, Auditable, HasIdentifier {
 
-  public static final String SYSTEM_CATALOG_RESERVED_NAME = "system";
-
   public static final Field ID =
       Field.required("id", Long.class, "The catalog's unique identifier");
   public static final Field NAME = Field.required("name", String.class, "The catalog's name");

core/src/main/java/com/datastrato/gravitino/meta/EntitySpecificConstants.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.meta;
+
+public class EntitySpecificConstants {
+
+  private EntitySpecificConstants() {}
+
+  public static final String SYSTEM_METALAKE_RESERVED_NAME = "system";
+  public static final String SYSTEM_CATALOG_RESERVED_NAME = "system";
+  public static final String AUTHORIZATION_CATALOG_NAME = "authorization";
+  public static final String USER_SCHEMA_NAME = "user";
+  public static final String GROUP_SCHEMA_NAME = "group";
+  public static final String ADMIN_SCHEMA_NAME = "admin";
+}

core/src/main/java/com/datastrato/gravitino/meta/GroupEntity.java

@@ -18,8 +18,6 @@
 
 public class GroupEntity implements Group, Entity, Auditable, HasIdentifier {
 
-  public static final String GROUP_SCHEMA_NAME = "group";
-
   public static final Field ID =
       Field.required("id", Long.class, " The unique id of the group entity.");
 

core/src/main/java/com/datastrato/gravitino/meta/UserEntity.java

@@ -21,8 +21,6 @@
 @ToString
 public class UserEntity implements User, Entity, Auditable, HasIdentifier {
 
-  public static final String USER_SCHEMA_NAME = "user";
-
   public static final Field ID =
       Field.required("id", Long.class, " The unique id of the user entity.");
 

core/src/main/java/com/datastrato/gravitino/metalake/MetalakeManager.java

@@ -18,6 +18,7 @@
 import com.datastrato.gravitino.exceptions.NoSuchMetalakeException;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.BaseMetalake;
+import com.datastrato.gravitino.meta.EntitySpecificConstants;
 import com.datastrato.gravitino.meta.SchemaVersion;
 import com.datastrato.gravitino.storage.IdGenerator;
 import com.datastrato.gravitino.utils.PrincipalUtils;
@@ -105,6 +106,11 @@ public BaseMetalake createMetalake(
     long uid = idGenerator.nextId();
     StringIdentifier stringId = StringIdentifier.fromId(uid);
 
+    if (EntitySpecificConstants.SYSTEM_METALAKE_RESERVED_NAME.equals(ident.name())) {
+      throw new IllegalArgumentException(
+          "Can't create a metalake with with reserved name `system`");
+    }
+
     BaseMetalake metalake =
         BaseMetalake.builder()
             .withId(uid)