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

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

@@ -239,4 +239,18 @@ 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<String> SERVICE_ADMIN =
+      new ConfigBuilder("gravitino.authorization.serviceAdmin")
+          .doc("The admin of Gravitino service")
+          .version(ConfigConstants.VERSION_0_5_0)
+          .stringConf()
+          .create();
 }

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

@@ -12,15 +12,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) {
     this.userGroupManager = new UserGroupManager(store, idGenerator);
+    this.adminManager = new AdminManager(store, idGenerator);
   }
 
   /**
@@ -98,4 +100,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.Configs;
+import com.datastrato.gravitino.Entity;
+import com.datastrato.gravitino.EntityAlreadyExistsException;
+import com.datastrato.gravitino.EntityStore;
+import com.datastrato.gravitino.GravitinoEnv;
+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.BaseMetalake;
+import com.datastrato.gravitino.meta.CatalogEntity;
+import com.datastrato.gravitino.meta.SchemaEntity;
+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 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.
+ */
+public class AdminManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(AdminManager.class);
+
+  private final EntityStore store;
+  private final IdGenerator idGenerator;
+
+  public AdminManager(EntityStore store, IdGenerator idGenerator) {
+    this.store = store;
+    this.idGenerator = idGenerator;
+  }
+
+  /**
+   * 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) {
+
+    UserEntity userEntity =
+        UserEntity.builder()
+            .withId(idGenerator.nextId())
+            .withName(user)
+            .withNamespace(
+                Namespace.of(
+                    BaseMetalake.SYSTEM_METALAKE_RESERVED_NAME,
+                    CatalogEntity.AUTHORIZATION_CATALOG_NAME,
+                    SchemaEntity.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 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) {
+    String admin = GravitinoEnv.getInstance().config().get(Configs.SERVICE_ADMIN);
+    return admin.equals(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) {
+    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(
+        BaseMetalake.SYSTEM_METALAKE_RESERVED_NAME,
+        CatalogEntity.AUTHORIZATION_CATALOG_NAME,
+        SchemaEntity.ADMIN_SCHEMA_NAME,
+        user);
+  }
+}

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

@@ -17,6 +17,7 @@
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.CatalogEntity;
 import com.datastrato.gravitino.meta.GroupEntity;
+import com.datastrato.gravitino.meta.SchemaEntity;
 import com.datastrato.gravitino.meta.UserEntity;
 import com.datastrato.gravitino.storage.IdGenerator;
 import com.datastrato.gravitino.utils.PrincipalUtils;
@@ -67,7 +68,7 @@ public User addUser(String metalake, String name) throws UserAlreadyExistsExcept
                 Namespace.of(
                     metalake,
                     CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
-                    UserEntity.USER_SCHEMA_NAME))
+                    SchemaEntity.USER_SCHEMA_NAME))
             .withRoles(Lists.newArrayList())
             .withAuditInfo(
                 AuditInfo.builder()
@@ -147,7 +148,7 @@ public Group addGroup(String metalake, String group) throws GroupAlreadyExistsEx
                 Namespace.of(
                     metalake,
                     CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
-                    GroupEntity.GROUP_SCHEMA_NAME))
+                    SchemaEntity.GROUP_SCHEMA_NAME))
             .withRoles(Collections.emptyList())
             .withAuditInfo(
                 AuditInfo.builder()
@@ -213,11 +214,14 @@ 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, CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME, SchemaEntity.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,
+        CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
+        SchemaEntity.GROUP_SCHEMA_NAME,
+        group);
   }
 }

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

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

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

@@ -25,6 +25,7 @@
 public class CatalogEntity implements Entity, Auditable, HasIdentifier {
 
   public static final String SYSTEM_CATALOG_RESERVED_NAME = "system";
+  public static final String AUTHORIZATION_CATALOG_NAME = "authorization";
 
   public static final Field ID =
       Field.required("id", Long.class, "The catalog's unique identifier");

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/SchemaEntity.java

@@ -27,6 +27,9 @@ public class SchemaEntity implements Entity, Auditable, HasIdentifier {
       Field.optional("comment", String.class, "The comment or description of the schema");
   public static final Field PROPERTIES =
       Field.optional("properties", Map.class, "The properties of the schema");
+  public static final String USER_SCHEMA_NAME = "user";
+  public static final String GROUP_SCHEMA_NAME = "group";
+  public static final String ADMIN_SCHEMA_NAME = "admin";
 
   private Long id;
 

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

@@ -104,6 +104,11 @@ public BaseMetalake createMetalake(
     long uid = idGenerator.nextId();
     StringIdentifier stringId = StringIdentifier.fromId(uid);
 
+    if (BaseMetalake.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)

core/src/test/java/com/datastrato/gravitino/authorization/TestAccessControlManager.java

@@ -150,4 +150,25 @@ public void testRemoveGroup() {
     boolean removed1 = accessControlManager.removeUser("metalake", "no-exist");
     Assertions.assertFalse(removed1);
   }
+
+  @Test
+  public void testMetalakeAdmin() {
+    User user = accessControlManager.addMetalakeAdmin("test");
+    Assertions.assertEquals("test", user.name());
+    Assertions.assertTrue(user.roles().isEmpty());
+    Assertions.assertTrue(accessControlManager.isMetalakeAdmin("test"));
+
+    // Test with UserAlreadyExistsException
+    Assertions.assertThrows(
+        UserAlreadyExistsException.class, () -> accessControlManager.addMetalakeAdmin("test"));
+
+    // Test to remove admin
+    boolean removed = accessControlManager.removeMetalakeAdmin("test");
+    Assertions.assertTrue(removed);
+    Assertions.assertFalse(accessControlManager.isMetalakeAdmin("test"));
+
+    // Test to remove non-existed admin
+    boolean removed1 = accessControlManager.removeMetalakeAdmin("no-exist");
+    Assertions.assertFalse(removed1);
+  }
 }

core/src/test/java/com/datastrato/gravitino/storage/TestEntityStorage.java

@@ -874,7 +874,9 @@ private static UserEntity createUser(String metalake, String name, AuditInfo aud
         .withId(1L)
         .withNamespace(
             Namespace.of(
-                metalake, CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME, UserEntity.USER_SCHEMA_NAME))
+                metalake,
+                CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
+                SchemaEntity.USER_SCHEMA_NAME))
         .withName(name)
         .withAuditInfo(auditInfo)
         .withRoles(Lists.newArrayList())
@@ -888,7 +890,7 @@ private static GroupEntity createGroup(String metalake, String name, AuditInfo a
             Namespace.of(
                 metalake,
                 CatalogEntity.SYSTEM_CATALOG_RESERVED_NAME,
-                GroupEntity.GROUP_SCHEMA_NAME))
+                SchemaEntity.GROUP_SCHEMA_NAME))
         .withName(name)
         .withAuditInfo(auditInfo)
         .withRoles(Lists.newArrayList())

server/src/main/java/com/datastrato/gravitino/server/GravitinoServer.java

@@ -4,6 +4,7 @@
  */
 package com.datastrato.gravitino.server;
 
+import com.datastrato.gravitino.Configs;
 import com.datastrato.gravitino.GravitinoEnv;
 import com.datastrato.gravitino.authorization.AccessControlManager;
 import com.datastrato.gravitino.catalog.CatalogManager;
@@ -58,6 +59,11 @@ public GravitinoServer(ServerConfig config) {
   public void initialize() {
     gravitinoEnv.initialize(serverConfig);
 
+    boolean enableAuthorization = serverConfig.get(Configs.ENABLE_AUTHORIZATION);
+    if (enableAuthorization && serverConfig.get(Configs.SERVICE_ADMIN) == null) {
+      throw new IllegalArgumentException("The service admin can't be null");
+    }
+
     JettyServerConfig jettyServerConfig =
         JettyServerConfig.fromConfig(serverConfig, WEBSERVER_CONF_PREFIX);
     server.initialize(jettyServerConfig, SERVER_NAME, true /* shouldEnableUI */);