feat(Ranger): Compatible with the old ACL and the new ACL (#1379)

#1054

This patch is compatible with old and new acl.

- Modify some method names and parameter names to make them more accurate.
- Defines the configuration parameters that the new ACL needs to use.
- Two new 'allowed()' methods are provided for meta_server and replica_server.
- Some incompatible methods will be removed, (allowed&pre_check)commented.

src/replica/replica_config.cpp

@@ -525,7 +525,7 @@ void replica::update_ac_allowed_users(const std::map<std::string, std::string> &
         allowed_users = iter->second;
     }
 
-    _access_controller->update(allowed_users);
+    _access_controller->update_allowed_users(allowed_users);
 }
 
 void replica::update_allow_ingest_behind(const std::map<std::string, std::string> &envs)

src/runtime/security/access_controller.cpp

@@ -26,11 +26,23 @@
 namespace dsn {
 namespace security {
 DSN_DEFINE_bool(security, enable_acl, false, "whether enable access controller or not");
-DSN_TAG_VARIABLE(enable_acl, FT_MUTABLE);
+DSN_DEFINE_bool(security,
+                enable_ranger_acl,
+                false,
+                "whether enable access controller integrate to Apache Ranger or not");
+DSN_DEFINE_string(security,
+                  super_users,
+                  "",
+                  "super users for access controller, comma-separated list of user names");
 
-DSN_DEFINE_string(security, super_users, "", "super user for access controller");
-
-access_controller::access_controller() { utils::split_args(FLAGS_super_users, _super_users, ','); }
+access_controller::access_controller()
+{
+    // when FLAGS_enable_ranger_acl is true, FLAGS_enable_acl must be true.
+    // TODO(wanghao): check with DSN_DEFINE_group_validator().
+    CHECK(!FLAGS_enable_ranger_acl || FLAGS_enable_acl,
+          "when FLAGS_enable_ranger_acl is true, FLAGS_enable_acl must be true too");
+    utils::split_args(FLAGS_super_users, _super_users, ',');
+}
 
 access_controller::~access_controller() {}
 
@@ -42,6 +54,13 @@ bool access_controller::pre_check(const std::string &user_name)
     return false;
 }
 
+bool access_controller::is_enable_ranger_acl() { return FLAGS_enable_ranger_acl; }
+
+bool access_controller::is_super_user(const std::string &user_name) const
+{
+    return _super_users.find(user_name) != _super_users.end();
+}
+
 std::unique_ptr<access_controller> create_meta_access_controller()
 {
     return make_unique<meta_access_controller>();

src/runtime/security/access_controller.h

@@ -21,6 +21,8 @@
 #include <string>
 #include <unordered_set>
 
+#include "runtime/ranger/ranger_resource_policy.h"
+
 namespace dsn {
 class message_ex;
 namespace security {
@@ -31,27 +33,41 @@ class access_controller
     access_controller();
     virtual ~access_controller() = 0;
 
-    /**
-     * update the access controller
-     *    acls - the new acls to update
-     **/
-    virtual void update(const std::string &acls){};
+    // Update the access controller.
+    // users - the new allowed users to update
+    virtual void update_allowed_users(const std::string &users) {}
+
+    // Check whether the Ranger ACL is enabled or not.
+    bool is_enable_ranger_acl();
+
+    // Check if the message received is allowd to access the system.
+    // msg - the message received
+    virtual bool allowed(message_ex *msg, dsn::ranger::access_type req_type) { return false; }
 
-    /**
-     * check if the message received is allowd to do something.
-     *   msg - the message received
-     **/
+    // Check if the message received is allowd to access the table.
+    // msg - the message received
+    // app_name - tables involved in ACL
+    virtual bool allowed(message_ex *msg, const std::string &app_name) { return false; }
+
+    // TODO(wanghao): this method will be deleted in the next patch.
+    // check if the message received is allowd to do something.
+    // msg - the message received
     virtual bool allowed(message_ex *msg) = 0;
 
 protected:
+    // TODO(wanghao): this method will be deleted in the next patch.
     bool pre_check(const std::string &user_name);
+
+    // Check if 'user_name' is the super user.
+    bool is_super_user(const std::string &user_name) const;
     friend class meta_access_controller_test;
 
     std::unordered_set<std::string> _super_users;
 };
 
 std::unique_ptr<access_controller> create_meta_access_controller();
 
-std::unique_ptr<access_controller> create_replica_access_controller(const std::string &name);
+std::unique_ptr<access_controller>
+create_replica_access_controller(const std::string &replica_name);
 } // namespace security
 } // namespace dsn

src/runtime/security/replica_access_controller.cpp

@@ -46,7 +46,7 @@ bool replica_access_controller::allowed(message_ex *msg)
     }
 }
 
-void replica_access_controller::update(const std::string &users)
+void replica_access_controller::update_allowed_users(const std::string &users)
 {
     {
         // check to see whether we should update it or not.

src/runtime/security/replica_access_controller.h

@@ -27,7 +27,7 @@ class replica_access_controller : public access_controller
 public:
     explicit replica_access_controller(const std::string &name);
     bool allowed(message_ex *msg) override;
-    void update(const std::string &users) override;
+    void update_allowed_users(const std::string &users) override;
 
 private:
     utils::rw_lock_nr _lock; // [