This is an automated email from the ASF dual-hosted git repository.
abukor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new fcdffd4 [ranger] Add refreshing Ranger policies
fcdffd4 is described below
commit fcdffd4c0e5d980341653bf180b8a234244c2f48
Author: Attila Bukor <[email protected]>
AuthorDate: Fri Jul 10 15:59:45 2020 +0200
[ranger] Add refreshing Ranger policies
Authorization tests depend on frequent polling of policies from the
Ranger server to make sure newly created policies are cached in the
authorizer plugin. We've seen some flaky tests due to this recently, so
a better way to handle this would be to manually refresh the policies in
the authorizer plugin.
When the Ranger integration was initially implemented we decided to not
use the existing ResetAuthzCache methods in master as it behaves in a
different way than it did with Sentry. Sentry's cache was within Kudu
and could be fully reset. Unfortunately Ranger's cache lives in the
Ranger plugin within the Ranger subprocess which has a void
refreshPolicies() method that doesn't throw any checked exceptions. This
means there's no way to guarantee the cache is cleared and the only
failure we can know about is if the request times out or other failures
between the master and the subprocess, but not between the subprocess
and the Ranger server. This means that even if Ranger itself is down, we
can still get an OK response when calling this method and there's no way
to guarantee clearing the cache.
When Sentry integration was removed from Kudu, this ResetAuthzCache()
method was also removed along with the tooling, as we didn't have any
authz providers that would support resetting the cache.
Now we need a way to do this to deflake the tests, and as there is no
longer a discrepancy between the behavior of different authz providers
and the best effort basis of refreshing policies should be enough, this
method is readded by this commit and the tests are changed to use it
instead of the timer-based policy refreshes.
Change-Id: Ie90b1c23b92060e081ab514fbae417b4f31c2b66
Reviewed-on: http://gerrit.cloudera.org:8080/16160
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <[email protected]>
Reviewed-by: Andrew Wong <[email protected]>
---
.../subprocess/ranger/RangerProtocolHandler.java | 12 +-
.../ranger/authorization/RangerKuduAuthorizer.java | 21 ++-
.../subprocess/ranger/TestRangerSubprocess.java | 5 +-
src/kudu/integration-tests/master_authz-itest.cc | 140 ++++++++++++--------
src/kudu/integration-tests/ts_authz-itest.cc | 58 ++++++---
src/kudu/master/authz_provider.h | 3 +
src/kudu/master/default_authz_provider.h | 4 +
src/kudu/master/master.proto | 12 ++
src/kudu/master/master_service.cc | 12 ++
src/kudu/master/master_service.h | 6 +
src/kudu/master/ranger_authz_provider.cc | 4 +
src/kudu/master/ranger_authz_provider.h | 2 +
src/kudu/ranger/mini_ranger.h | 5 +-
src/kudu/ranger/ranger.proto | 14 ++
src/kudu/ranger/ranger_client.cc | 19 +++
src/kudu/ranger/ranger_client.h | 5 +
src/kudu/tools/kudu-admin-test.cc | 60 +++++++++
src/kudu/tools/kudu-tool-test.cc | 7 +
src/kudu/tools/tool_action_master.cc | 141 +++++++++++++++++++++
19 files changed, 446 insertions(+), 84 deletions(-)
diff --git
a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java
b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java
index 961ad52..6d6ef9c 100644
---
a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java
+++
b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java
@@ -17,12 +17,12 @@
package org.apache.kudu.subprocess.ranger;
-import com.google.common.base.Preconditions;
import org.apache.ranger.plugin.policyengine.RangerAccessResult;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.kudu.ranger.Ranger;
import org.apache.kudu.ranger.Ranger.RangerRequestListPB;
import org.apache.kudu.ranger.Ranger.RangerResponseListPB;
import org.apache.kudu.ranger.Ranger.RangerResponsePB;
@@ -47,7 +47,15 @@ class RangerProtocolHandler extends
ProtocolHandler<RangerRequestListPB,
@Override
protected RangerResponseListPB executeRequest(RangerRequestListPB requests) {
- return authz.authorize(requests);
+ RangerResponseListPB.Builder responses = authz.authorize(requests);
+ if (requests.hasControlRequest()) {
+ if (requests.getControlRequest().getRefreshPolicies()) {
+ authz.refreshPolicies();
+
responses.setControlResponse(Ranger.RangerControlResponsePB.newBuilder()
+ .setSuccess(true).build());
+ }
+ }
+ return responses.build();
}
@Override
diff --git
a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java
b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java
index 1e6235a..bd8c8f8 100644
---
a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java
+++
b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java
@@ -101,7 +101,7 @@ public class RangerKuduAuthorizer {
* @return a list of RangerAccessResult
*/
@VisibleForTesting
- public Ranger.RangerResponseListPB authorize(RangerRequestListPB requests) {
+ public Ranger.RangerResponseListPB.Builder authorize(RangerRequestListPB
requests) {
if (!requests.hasUser() || requests.getUser().isEmpty()) {
Ranger.RangerResponseListPB.Builder rangerResponseListPB =
Ranger.RangerResponseListPB
.newBuilder();
@@ -112,12 +112,25 @@ public class RangerKuduAuthorizer {
.build();
rangerResponseListPB.addResponses(response);
}
- return rangerResponseListPB.build();
+ return rangerResponseListPB;
}
return authorizeRequests(requests);
}
/**
+ * Refreshes the policies cached in the authorization provider on a
best-effort basis. It
+ * doesn't guarantee invalidating the cache or that the latest policies
could be pulled from
+ * the server and doesn't throw exceptions even if the server couldn't be
reached.
+ *
+ * TODO(abukor): Revisit if RANGER-2906 is fixed.
+ */
+ public void refreshPolicies() {
+ LOG.debug("Refreshing policies...");
+ plugin.refreshPoliciesAndTags();
+ LOG.debug("Refreshing policies... DONE");
+ }
+
+ /**
* Creates a Ranger access request for the specified user who performs
* the given action on the resource.
*
@@ -156,7 +169,7 @@ public class RangerKuduAuthorizer {
* @param requests the given RangerRequestListPB
* @return a list of RangerAccessRequest
*/
- private Ranger.RangerResponseListPB authorizeRequests(RangerRequestListPB
requests) {
+ private Ranger.RangerResponseListPB.Builder
authorizeRequests(RangerRequestListPB requests) {
Ranger.RangerResponseListPB.Builder rangerResponseList =
Ranger.RangerResponseListPB
.newBuilder();
Preconditions.checkArgument(requests.hasUser());
@@ -193,7 +206,7 @@ public class RangerKuduAuthorizer {
rangerResponseList.addResponses(rangerResponsePB);
}
- return rangerResponseList.build();
+ return rangerResponseList;
}
/**
diff --git
a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java
b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java
index aafb06c..3e64059 100644
---
a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java
+++
b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java
@@ -103,11 +103,10 @@ public class TestRangerSubprocess extends
SubprocessTestUtil {
final SubprocessRequestPB subprocessRequest =
createRangerSubprocessRequest(requestList);
// Mock the authorization results.
- RangerResponseListPB responseListPB = RangerResponseListPB.newBuilder()
+ RangerResponseListPB.Builder responseListPB =
RangerResponseListPB.newBuilder()
.addResponses(Ranger.RangerResponsePB.newBuilder().setAllowed(true).build())
.addResponses(Ranger.RangerResponsePB.newBuilder().setAllowed(false).build())
-
.addResponses(Ranger.RangerResponsePB.newBuilder().setAllowed(true).build())
- .build();
+
.addResponses(Ranger.RangerResponsePB.newBuilder().setAllowed(true).build());
Mockito.when(RangerProtocolHandler.authz.authorize(requestList))
.thenReturn(responseListPB);
diff --git a/src/kudu/integration-tests/master_authz-itest.cc
b/src/kudu/integration-tests/master_authz-itest.cc
index 4a4c975..3e8f645 100644
--- a/src/kudu/integration-tests/master_authz-itest.cc
+++ b/src/kudu/integration-tests/master_authz-itest.cc
@@ -32,6 +32,7 @@
#include "kudu/client/schema.h"
#include "kudu/client/shared_ptr.h" // IWYU pragma: keep
#include "kudu/common/table_util.h"
+#include "kudu/common/wire_protocol.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/hms/hms_client.h"
@@ -42,6 +43,7 @@
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/ranger/mini_ranger.h"
#include "kudu/ranger/ranger.pb.h"
+#include "kudu/rpc/rpc_controller.h"
#include "kudu/rpc/user_credentials.h"
#include "kudu/security/test/mini_kdc.h"
#include "kudu/util/decimal_util.h"
@@ -69,12 +71,15 @@ using kudu::cluster::ExternalMiniClusterOptions;
using kudu::hms::HmsClient;
using kudu::master::GetTableLocationsResponsePB;
using kudu::master::GetTabletLocationsResponsePB;
+using kudu::master::RefreshAuthzCacheRequestPB;
+using kudu::master::RefreshAuthzCacheResponsePB;
using kudu::master::MasterServiceProxy;
using kudu::master::VOTER_REPLICA;
using kudu::ranger::ActionPB;
using kudu::ranger::AuthorizationPolicy;
using kudu::ranger::MiniRanger;
using kudu::ranger::PolicyItem;
+using kudu::rpc::RpcController;
using kudu::rpc::UserCredentials;
using std::function;
using std::move;
@@ -151,6 +156,18 @@ class MasterAuthzITestHarness {
table_id, &table_locations);
}
+ static Status RefreshAuthzPolicies(const unique_ptr<ExternalMiniCluster>&
cluster) {
+ RefreshAuthzCacheRequestPB req;
+ RefreshAuthzCacheResponsePB resp;
+ RpcController rpc;
+ rpc.set_timeout(MonoDelta::FromSeconds(10));
+ RETURN_NOT_OK(cluster->master_proxy()->RefreshAuthzCache(req, &resp,
&rpc));
+ if (resp.has_error()) {
+ return StatusFromPB(resp.error().status());
+ }
+ return Status::OK();
+ }
+
static Status IsCreateTableDone(const OperationParams& p,
const shared_ptr<KuduClient>& client) {
bool in_progress = false;
@@ -296,20 +313,34 @@ class MasterAuthzITestHarness {
// authorized.
opts.extra_master_flags.emplace_back("--trusted_user_acl=impala");
opts.extra_master_flags.emplace_back("--user_acl=test-user,impala");
+ opts.extra_master_flags.emplace_back("--ranger_log_level=debug");
SetUpExternalMiniServiceOpts(&opts);
return opts;
}
- virtual Status GrantCreateTablePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantDropTablePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantAlterTablePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantRenameTablePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantGetMetadataTablePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantAllTablePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantAllDatabasePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantAllWithGrantTablePrivilege(const PrivilegeParams& p) = 0;
- virtual Status GrantAllWithGrantDatabasePrivilege(const PrivilegeParams& p)
= 0;
- virtual Status GrantGetMetadataDatabasePrivilege(const PrivilegeParams& p) =
0;
+ virtual Status GrantCreateTablePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster) = 0;
+ virtual Status GrantDropTablePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster) = 0;
+ virtual Status GrantAlterTablePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster) = 0;
+ virtual Status GrantRenameTablePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster) = 0;
+ virtual Status GrantGetMetadataTablePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster) = 0;
+ virtual Status GrantGetMetadataDatabasePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster)
+ = 0;
+ virtual Status GrantAllTablePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) = 0;
+ virtual Status GrantAllDatabasePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster) = 0;
+ virtual Status GrantAllWithGrantTablePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster)
+ = 0;
+ virtual Status GrantAllWithGrantDatabasePrivilege(
+ const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>& cluster) = 0;
virtual Status CreateTable(const OperationParams& p,
const shared_ptr<KuduClient>& client) = 0;
virtual void CheckTableDoesNotExist(const string& database_name, const
string& table_name,
@@ -338,27 +369,25 @@ class RangerITestHarness : public MasterAuthzITestHarness
{
public:
static constexpr int kSleepAfterNewPolicyMs = 1400;
- Status GrantCreateTablePrivilege(const PrivilegeParams& p) override {
+ Status GrantCreateTablePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) override {
AuthorizationPolicy policy;
policy.databases.emplace_back(p.db_name);
// IsCreateTableDone() requires METADATA on the table level.
policy.tables.emplace_back("*");
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::CREATE},
false));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
-
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantDropTablePrivilege(const PrivilegeParams& p) override {
+ Status GrantDropTablePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) override {
AuthorizationPolicy policy;
policy.databases.emplace_back(p.db_name);
policy.tables.emplace_back(p.table_name);
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::DROP},
false));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
-
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
void CheckTableDoesNotExist(const string& database_name, const string&
table_name,
@@ -368,17 +397,18 @@ class RangerITestHarness : public MasterAuthzITestHarness
{
ASSERT_TRUE(s.IsNotFound()) << s.ToString();
}
- Status GrantAlterTablePrivilege(const PrivilegeParams& p) override {
+ Status GrantAlterTablePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) override {
AuthorizationPolicy policy;
policy.databases.emplace_back(p.db_name);
policy.tables.emplace_back(p.table_name);
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALTER},
false));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantRenameTablePrivilege(const PrivilegeParams& p) override {
+ Status GrantRenameTablePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) override {
AuthorizationPolicy policy_new_table;
policy_new_table.databases.emplace_back(p.db_name);
// IsCreateTableDone() requires METADATA on the table level.
@@ -391,40 +421,41 @@ class RangerITestHarness : public MasterAuthzITestHarness
{
policy.tables.emplace_back(p.table_name);
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALL}, false));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantGetMetadataDatabasePrivilege(const PrivilegeParams& p) override {
+ Status GrantGetMetadataDatabasePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster)
+ override {
AuthorizationPolicy policy;
policy.databases.emplace_back(p.db_name);
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::METADATA},
false));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantGetMetadataTablePrivilege(const PrivilegeParams& p) override {
+ Status GrantGetMetadataTablePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) override {
AuthorizationPolicy policy;
policy.databases.emplace_back(p.db_name);
policy.tables.emplace_back(p.table_name);
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::METADATA},
false));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantAllTablePrivilege(const PrivilegeParams& p) override {
+ Status GrantAllTablePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) override {
AuthorizationPolicy policy;
policy.databases.emplace_back(p.db_name);
policy.tables.emplace_back(p.table_name);
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALL}, false));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantAllDatabasePrivilege(const PrivilegeParams& p) override {
+ Status GrantAllDatabasePrivilege(const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>&
cluster) override {
AuthorizationPolicy db_policy;
db_policy.databases.emplace_back(p.db_name);
db_policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALL},
false));
@@ -434,21 +465,22 @@ class RangerITestHarness : public MasterAuthzITestHarness
{
tbl_policy.tables.emplace_back("*");
tbl_policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALL},
false));
RETURN_NOT_OK(ranger_->AddPolicy(move(tbl_policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantAllWithGrantTablePrivilege(const PrivilegeParams& p) override {
+ Status GrantAllWithGrantTablePrivilege(const PrivilegeParams& p,
+ const
unique_ptr<ExternalMiniCluster>& cluster) override {
AuthorizationPolicy policy;
policy.databases.emplace_back(p.db_name);
policy.tables.emplace_back(p.table_name);
policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALL}, true));
RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
- Status GrantAllWithGrantDatabasePrivilege(const PrivilegeParams& p) override
{
+ Status GrantAllWithGrantDatabasePrivilege(
+ const PrivilegeParams& p,
+ const unique_ptr<ExternalMiniCluster>& cluster) override {
AuthorizationPolicy db_policy;
db_policy.databases.emplace_back(p.db_name);
db_policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALL},
true));
@@ -458,8 +490,7 @@ class RangerITestHarness : public MasterAuthzITestHarness {
tbl_policy.tables.emplace_back("*");
tbl_policy.items.emplace_back(PolicyItem({kTestUser}, {ActionPB::ALL},
true));
RETURN_NOT_OK(ranger_->AddPolicy(move(tbl_policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
Status CreateTable(const OperationParams& p,
@@ -496,9 +527,7 @@ class RangerITestHarness : public MasterAuthzITestHarness {
policy.databases.emplace_back(kDatabaseName);
policy.tables.emplace_back("*");
policy.items.emplace_back(PolicyItem({kAdminUser}, {ActionPB::ALL}, true));
- RETURN_NOT_OK(ranger_->AddPolicy(move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return ranger_->AddPolicy(move(policy));
}
private:
@@ -532,6 +561,7 @@ class MasterAuthzITestBase : public
ExternalMiniClusterITestBase {
ASSERT_OK(harness_->SetUpExternalServiceClients(cluster_));
ASSERT_OK(harness_->SetUpCredentials());
+ ASSERT_OK(harness_->RefreshAuthzPolicies(cluster_));
ASSERT_OK(harness_->SetUpTables(cluster_, client_));
// Log in as 'test-user' and reset the client to pick up the change in
user.
@@ -550,43 +580,43 @@ class MasterAuthzITestBase : public
ExternalMiniClusterITestBase {
}
Status GrantCreateTablePrivilege(const PrivilegeParams& p) {
- return harness_->GrantCreateTablePrivilege(p);
+ return harness_->GrantCreateTablePrivilege(p, cluster_);
}
Status GrantDropTablePrivilege(const PrivilegeParams& p) {
- return harness_->GrantDropTablePrivilege(p);
+ return harness_->GrantDropTablePrivilege(p, cluster_);
}
Status GrantAlterTablePrivilege(const PrivilegeParams& p) {
- return harness_->GrantAlterTablePrivilege(p);
+ return harness_->GrantAlterTablePrivilege(p, cluster_);
}
Status GrantRenameTablePrivilege(const PrivilegeParams& p) {
- return harness_->GrantRenameTablePrivilege(p);
+ return harness_->GrantRenameTablePrivilege(p, cluster_);
}
Status GrantGetMetadataTablePrivilege(const PrivilegeParams& p) {
- return harness_->GrantGetMetadataTablePrivilege(p);
+ return harness_->GrantGetMetadataTablePrivilege(p, cluster_);
}
Status GrantAllTablePrivilege(const PrivilegeParams& p) {
- return harness_->GrantAllTablePrivilege(p);
+ return harness_->GrantAllTablePrivilege(p, cluster_);
}
Status GrantAllDatabasePrivilege(const PrivilegeParams& p) {
- return harness_->GrantAllDatabasePrivilege(p);
+ return harness_->GrantAllDatabasePrivilege(p, cluster_);
}
Status GrantAllWithGrantTablePrivilege(const PrivilegeParams& p) {
- return harness_->GrantAllWithGrantTablePrivilege(p);
+ return harness_->GrantAllWithGrantTablePrivilege(p, cluster_);
}
Status GrantAllWithGrantDatabasePrivilege(const PrivilegeParams& p) {
- return harness_->GrantAllWithGrantDatabasePrivilege(p);
+ return harness_->GrantAllWithGrantDatabasePrivilege(p, cluster_);
}
Status GrantGetMetadataDatabasePrivilege(const PrivilegeParams& p) {
- return harness_->GrantGetMetadataDatabasePrivilege(p);
+ return harness_->GrantGetMetadataDatabasePrivilege(p, cluster_);
}
Status CreateTable(const OperationParams& p) {
diff --git a/src/kudu/integration-tests/ts_authz-itest.cc
b/src/kudu/integration-tests/ts_authz-itest.cc
index 45564ee..d3efa34 100644
--- a/src/kudu/integration-tests/ts_authz-itest.cc
+++ b/src/kudu/integration-tests/ts_authz-itest.cc
@@ -35,15 +35,19 @@
#include "kudu/client/shared_ptr.h" // IWYU pragma: keep
#include "kudu/client/write_op.h"
#include "kudu/common/partial_row.h"
+#include "kudu/common/wire_protocol.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/integration-tests/data_gen_util.h"
#include "kudu/integration-tests/external_mini_cluster-itest-base.h"
+#include "kudu/master/master.pb.h"
+#include "kudu/master/master.proxy.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/ranger/mini_ranger.h"
#include "kudu/ranger/ranger.pb.h"
+#include "kudu/rpc/rpc_controller.h"
#include "kudu/security/test/mini_kdc.h"
#include "kudu/tablet/ops/write_op.h"
#include "kudu/util/barrier.h"
@@ -73,10 +77,13 @@ using kudu::client::KuduTableAlterer;
using kudu::client::KuduTableCreator;
using kudu::cluster::ExternalMiniCluster;
using kudu::cluster::ExternalMiniClusterOptions;
+using kudu::master::RefreshAuthzCacheRequestPB;
+using kudu::master::RefreshAuthzCacheResponsePB;
using kudu::ranger::ActionPB;
using kudu::ranger::AuthorizationPolicy;
using kudu::ranger::PolicyItem;
using kudu::ranger::MiniRanger;
+using kudu::rpc::RpcController;
using kudu::tablet::WritePrivileges;
using kudu::tablet::WritePrivilegeType;
using std::pair;
@@ -286,9 +293,12 @@ class TSAuthzITestHarness {
// These abide the hierarchical definition of privileges, so if granting
privileges on a
// table, these grant privileges on the table and all columns in that table.
virtual Status GrantTablePrivilege(const string& db, const string& tbl,
const string& user,
- const string& action, bool admin) = 0;
+ const string& action, bool admin,
+ const unique_ptr<ExternalMiniCluster>&
cluster) = 0;
virtual Status GrantColumnPrivilege(const string& db, const string& tbl,
const string& col,
- const string& user, const string&
action, bool admin) = 0;
+ const string& user, const string&
action, bool admin,
+ const unique_ptr<ExternalMiniCluster>&
cluster) = 0;
+ virtual Status RefreshAuthzPolicies(const unique_ptr<ExternalMiniCluster>&
cluster) = 0;
// Creates a table named 'table_ident' with 'kNumColsPerTable' columns.
Status CreateTable(const string& table_ident, const shared_ptr<KuduClient>&
client) {
@@ -353,9 +363,7 @@ class RangerITestHarness : public TSAuthzITestHarness {
policy.tables = { "*" };
policy.columns = { "*" };
policy.items.emplace_back(PolicyItem({ kAdminUser }, { ActionPB::ALL },
true));
- RETURN_NOT_OK(ranger_->AddPolicy(std::move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return ranger_->AddPolicy(std::move(policy));
}
Status SetUpTables() override {
// Finally populate a set of column names to use for our tables.
@@ -364,27 +372,40 @@ class RangerITestHarness : public TSAuthzITestHarness {
}
return Status::OK();
}
+
+ Status RefreshAuthzPolicies(const unique_ptr<ExternalMiniCluster>& cluster)
override {
+ RefreshAuthzCacheRequestPB req;
+ RefreshAuthzCacheResponsePB resp;
+ RpcController rpc;
+ rpc.set_timeout(MonoDelta::FromSeconds(10));
+ RETURN_NOT_OK(cluster->master_proxy()->RefreshAuthzCache(req, &resp,
&rpc));
+ if (resp.has_error()) {
+ return StatusFromPB(resp.error().status());
+ }
+ return Status::OK();
+ }
+
Status GrantTablePrivilege(const string& db, const string& tbl, const
string& user,
- const string& action, bool admin) override {
+ const string& action, bool admin,
+ const unique_ptr<ExternalMiniCluster>& cluster)
override {
AuthorizationPolicy policy;
policy.databases = { db };
policy.tables = { tbl };
policy.columns = { "*" };
policy.items.emplace_back(PolicyItem({ user }, { StringToActionPB(action)
}, admin));
RETURN_NOT_OK(ranger_->AddPolicy(std::move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
Status GrantColumnPrivilege(const string& db, const string& tbl, const
string& col,
- const string& user, const string& action, bool
admin) override {
+ const string& user, const string& action, bool
admin,
+ const unique_ptr<ExternalMiniCluster>& cluster)
override {
AuthorizationPolicy policy;
policy.databases = { db };
policy.tables = { tbl };
policy.columns = { col };
policy.items.emplace_back(PolicyItem({ user }, { StringToActionPB(action)
}, admin));
RETURN_NOT_OK(ranger_->AddPolicy(std::move(policy)));
- SleepFor(MonoDelta::FromMilliseconds(kSleepAfterNewPolicyMs));
- return Status::OK();
+ return RefreshAuthzPolicies(cluster);
}
private:
MiniRanger* ranger_;
@@ -428,6 +449,7 @@ class TSAuthzITest : public ExternalMiniClusterITestBase,
ASSERT_OK(harness_->SetUpExternalServiceClients(cluster_));
ASSERT_OK(harness_->SetUpCredentials());
+ ASSERT_OK(harness_->RefreshAuthzPolicies(cluster_));
ASSERT_OK(harness_->SetUpTables());
// Create a client as the admin user, who now has admin privileges on the
@@ -496,11 +518,11 @@ TEST_P(TSAuthzITest, TestReadsAndWrites) {
RWPrivileges granted_privileges = GeneratePrivileges(cols, &prng);
for (const auto& wp : granted_privileges.table_write_privileges) {
ASSERT_OK(harness_->GrantTablePrivilege(kDb, table_name, user,
WritePrivilegeToString(wp),
- /*admin=*/false));
+ /*admin=*/false, cluster_));
}
for (const auto& col : granted_privileges.column_scan_privileges) {
ASSERT_OK(harness_->GrantColumnPrivilege(kDb, table_name, col, user,
"SELECT",
- /*admin=*/false));
+ /*admin=*/false, cluster_));
}
RWPrivileges not_granted_privileges = ComplementaryPrivileges(cols,
granted_privileges);
InsertOrDie(&table_to_privileges, table_name,
@@ -577,13 +599,14 @@ TEST_P(TSAuthzITest, TestAlters) {
// Note: we only need privileges on the metadata for OpenTable() calls.
// METADATA isn't a first-class privilege and won't get carried over
// on table rename, so we just grant INSERT privileges.
- ASSERT_OK(harness_->GrantTablePrivilege(kDb, kTableName, user, "INSERT",
/*admin=*/false));
+ ASSERT_OK(harness_->GrantTablePrivilege(kDb, kTableName, user, "INSERT",
/*admin=*/false,
+ cluster_));
// First, grant privileges on a new column that doesn't yet exist. Once that
// column is created, we should be able to scan it immediately.
const string new_column = Substitute("col$0", kNumColsPerTable);
ASSERT_OK(harness_->GrantColumnPrivilege(kDb, kTableName, new_column, user,
"SELECT",
- /*admin=*/false));
+ /*admin=*/false, cluster_));
{
unique_ptr<KuduTableAlterer>
table_alterer(client_->NewTableAlterer(table_ident));
table_alterer->AddColumn(new_column)->Type(KuduColumnSchema::INT32);
@@ -596,7 +619,7 @@ TEST_P(TSAuthzITest, TestAlters) {
// Now create another column and grant the user privileges for that column.
const string another_column = Substitute("col$0", kNumColsPerTable + 1);
ASSERT_OK(harness_->GrantColumnPrivilege(kDb, kTableName, another_column,
user, "SELECT",
- /*admin=*/false));
+ /*admin=*/false, cluster_));
{
unique_ptr<KuduTableAlterer>
table_alterer(client_->NewTableAlterer(table_ident));
table_alterer->AddColumn(another_column)->Type(KuduColumnSchema::INT32);
@@ -615,7 +638,8 @@ TEST_P(TSAuthzITest, TestAlters) {
// We need to explicitly grant privileges on the new table name.
const string kNewTableName = "newtable";
const string new_table_ident = Substitute("$0.$1", kDb, kNewTableName);
- ASSERT_OK(harness_->GrantTablePrivilege(kDb, kNewTableName, user, "SELECT",
/*admin=*/false));
+ ASSERT_OK(harness_->GrantTablePrivilege(kDb, kNewTableName, user, "SELECT",
/*admin=*/false,
+ cluster_));
{
unique_ptr<KuduTableAlterer>
table_alterer(client_->NewTableAlterer(table_ident));
table_alterer->RenameTo(new_table_ident);
diff --git a/src/kudu/master/authz_provider.h b/src/kudu/master/authz_provider.h
index 2a6981f..c8042ac 100644
--- a/src/kudu/master/authz_provider.h
+++ b/src/kudu/master/authz_provider.h
@@ -107,6 +107,9 @@ class AuthzProvider {
const SchemaPB& schema_pb,
security::TablePrivilegePB* pb)
WARN_UNUSED_RESULT = 0;
+ // Refreshes policies in the authorization provider plugin.
+ virtual Status RefreshPolicies() WARN_UNUSED_RESULT = 0;
+
// Checks if the given user is trusted and thus can be exempted from
// authorization validation.
bool IsTrustedUser(const std::string& user) const;
diff --git a/src/kudu/master/default_authz_provider.h
b/src/kudu/master/default_authz_provider.h
index 46c181c..3deb6d2 100644
--- a/src/kudu/master/default_authz_provider.h
+++ b/src/kudu/master/default_authz_provider.h
@@ -86,6 +86,10 @@ class DefaultAuthzProvider : public AuthzProvider {
pb->set_update_privilege(true);
return Status::OK();
}
+
+ Status RefreshPolicies() override WARN_UNUSED_RESULT {
+ return Status::OK();
+ }
};
} // namespace master
diff --git a/src/kudu/master/master.proto b/src/kudu/master/master.proto
index 906941f..0259db4 100644
--- a/src/kudu/master/master.proto
+++ b/src/kudu/master/master.proto
@@ -945,6 +945,15 @@ message ReplaceTabletResponsePB {
optional bytes replacement_tablet_id = 2;
}
+// RefreshAuthzCache{Request/Result}PB: refresh the authz privileges cache,
+// download new policies (does not guarantee the old cache is invalidated).
+message RefreshAuthzCacheRequestPB {
+}
+
+message RefreshAuthzCacheResponsePB {
+ optional MasterErrorPB error = 1;
+}
+
enum MasterFeatures {
UNKNOWN_FEATURE = 0;
// The master supports creating tables with non-covering range partitions.
@@ -1032,6 +1041,9 @@ service MasterService {
rpc ReplaceTablet(ReplaceTabletRequestPB) returns (ReplaceTabletResponsePB) {
option (kudu.rpc.authz_method) = "AuthorizeSuperUser";
}
+ rpc RefreshAuthzCache(RefreshAuthzCacheRequestPB) returns
(RefreshAuthzCacheResponsePB) {
+ option (kudu.rpc.authz_method) = "AuthorizeSuperUser";
+ }
rpc ChangeTServerState(ChangeTServerStateRequestPB) returns
(ChangeTServerStateResponsePB) {
option (kudu.rpc.authz_method) = "AuthorizeSuperUser";
diff --git a/src/kudu/master/master_service.cc
b/src/kudu/master/master_service.cc
index e622e6a..3b9dc5d 100644
--- a/src/kudu/master/master_service.cc
+++ b/src/kudu/master/master_service.cc
@@ -37,6 +37,7 @@
#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/hms/hms_catalog.h"
+#include "kudu/master/authz_provider.h"
#include "kudu/master/catalog_manager.h"
#include "kudu/master/location_cache.h"
#include "kudu/master/master.h"
@@ -728,6 +729,17 @@ void MasterServiceImpl::ReplaceTablet(const
ReplaceTabletRequestPB* req,
rpc->RespondSuccess();
}
+void MasterServiceImpl::RefreshAuthzCache(
+ const RefreshAuthzCacheRequestPB* /* req */,
+ RefreshAuthzCacheResponsePB* resp,
+ rpc::RpcContext* rpc) {
+ LOG(INFO) << Substitute("request to refresh authz privileges cache from $0",
+ rpc->requestor_string());
+ CheckRespErrorOrSetUnknown(
+ server_->catalog_manager()->authz_provider()->RefreshPolicies(), resp);
+ rpc->RespondSuccess();
+}
+
bool MasterServiceImpl::SupportsFeature(uint32_t feature) const {
switch (feature) {
case MasterFeatures::RANGE_PARTITION_BOUNDS: FALLTHROUGH_INTENDED;
diff --git a/src/kudu/master/master_service.h b/src/kudu/master/master_service.h
index 00fbdad..e5ba019 100644
--- a/src/kudu/master/master_service.h
+++ b/src/kudu/master/master_service.h
@@ -71,6 +71,8 @@ class PingRequestPB;
class PingResponsePB;
class ReplaceTabletRequestPB;
class ReplaceTabletResponsePB;
+class RefreshAuthzCacheRequestPB;
+class RefreshAuthzCacheResponsePB;
class TSHeartbeatRequestPB;
class TSHeartbeatResponsePB;
@@ -170,6 +172,10 @@ class MasterServiceImpl : public MasterServiceIf {
ReplaceTabletResponsePB* resp,
rpc::RpcContext* rpc) override;
+ void RefreshAuthzCache(const RefreshAuthzCacheRequestPB* req,
+ RefreshAuthzCacheResponsePB* resp,
+ rpc::RpcContext* rpc) override;
+
bool SupportsFeature(uint32_t feature) const override;
private:
diff --git a/src/kudu/master/ranger_authz_provider.cc
b/src/kudu/master/ranger_authz_provider.cc
index 8317098..42289cb 100644
--- a/src/kudu/master/ranger_authz_provider.cc
+++ b/src/kudu/master/ranger_authz_provider.cc
@@ -333,6 +333,10 @@ Status RangerAuthzProvider::FillTablePrivilegePB(const
string& table_name,
return Status::OK();
}
+Status RangerAuthzProvider::RefreshPolicies() {
+ return client_.RefreshPolicies();
+}
+
bool RangerAuthzProvider::IsEnabled() {
return !FLAGS_ranger_config_path.empty();
}
diff --git a/src/kudu/master/ranger_authz_provider.h
b/src/kudu/master/ranger_authz_provider.h
index e84e912..bfb8bdb 100644
--- a/src/kudu/master/ranger_authz_provider.h
+++ b/src/kudu/master/ranger_authz_provider.h
@@ -76,6 +76,8 @@ class RangerAuthzProvider : public AuthzProvider {
const SchemaPB& schema_pb,
security::TablePrivilegePB* pb) override
WARN_UNUSED_RESULT;
+ Status RefreshPolicies() override WARN_UNUSED_RESULT;
+
// Returns true if 'ranger_policy_server' flag is set indicating Ranger
// authorization is enabled.
static bool IsEnabled();
diff --git a/src/kudu/ranger/mini_ranger.h b/src/kudu/ranger/mini_ranger.h
index e4b7244..185472e 100644
--- a/src/kudu/ranger/mini_ranger.h
+++ b/src/kudu/ranger/mini_ranger.h
@@ -188,9 +188,8 @@ class MiniRanger {
uint16_t port_ = 0;
// Determines how frequently clients fetch policies from the server. The
- // default is 200ms so that tests don't have to wait too long until freshly
- // created policies can be used.
- uint32_t policy_poll_interval_ms_ = 200;
+ // default is 30s which aligns with Ranger's default.
+ uint32_t policy_poll_interval_ms_ = 30000;
// Stores existing policies since starting the MiniRanger instance. This is
// used for adding new policy items (list of users and privileges) to
existing
diff --git a/src/kudu/ranger/ranger.proto b/src/kudu/ranger/ranger.proto
index 3ca54bc..2bfcb13 100644
--- a/src/kudu/ranger/ranger.proto
+++ b/src/kudu/ranger/ranger.proto
@@ -78,6 +78,16 @@ message RangerResponsePB {
optional bool allowed = 1;
}
+message RangerControlRequestPB {
+ optional bool refresh_policies = 1;
+}
+
+message RangerControlResponsePB {
+ optional bool success = 1;
+
+ optional string error = 2;
+}
+
// Describes a list of authorization requests sent to the Ranger service
// for a single user.
message RangerRequestListPB {
@@ -85,6 +95,8 @@ message RangerRequestListPB {
// User performing the action.
optional string user = 2;
+
+ optional RangerControlRequestPB control_request = 3;
}
// Describes a list of authorization descison responded from the Ranger
@@ -92,4 +104,6 @@ message RangerRequestListPB {
// responses is determined by the order of requests in RangerRequestListPB.
message RangerResponseListPB {
repeated RangerResponsePB responses = 1;
+
+ optional RangerControlResponsePB control_response = 2;
}
diff --git a/src/kudu/ranger/ranger_client.cc b/src/kudu/ranger/ranger_client.cc
index be0f6f9..375f806 100644
--- a/src/kudu/ranger/ranger_client.cc
+++ b/src/kudu/ranger/ranger_client.cc
@@ -528,5 +528,24 @@ Status RangerClient::AuthorizeActions(const string&
user_name, const string& dat
return Status::OK();
}
+
+Status RangerClient::RefreshPolicies() {
+ DCHECK(subprocess_);
+
+ RangerRequestListPB req_list;
+ RangerResponseListPB resp_list;
+
+ req_list.mutable_control_request()->set_refresh_policies(true);
+
+ RETURN_NOT_OK(subprocess_->Execute(req_list, &resp_list));
+
+ if (PREDICT_TRUE(!resp_list.control_response().success())) {
+ string err = resp_list.control_response().error();
+ return Status::RemoteError(err);
+ }
+
+ return Status::OK();
+}
+
} // namespace ranger
} // namespace kudu
diff --git a/src/kudu/ranger/ranger_client.h b/src/kudu/ranger/ranger_client.h
index 804f7da..097793c 100644
--- a/src/kudu/ranger/ranger_client.h
+++ b/src/kudu/ranger/ranger_client.h
@@ -107,6 +107,11 @@ class RangerClient {
const std::string& table,
std::unordered_set<ActionPB, ActionHash>* actions)
WARN_UNUSED_RESULT;
+ // Refreshes policies in the Ranger subprocess. This does not invalidate the
+ // existing cache and doesn't fail if Ranger service is unavailable, it
simply
+ // tries to refresh the policies from the server on a best effort basis.
+ Status RefreshPolicies() WARN_UNUSED_RESULT;
+
// Replaces the subprocess server in the subprocess proxy.
void ReplaceServerForTests(std::unique_ptr<subprocess::SubprocessServer>
server) {
// Creates a dummy RangerSubprocess if it is not initialized.
diff --git a/src/kudu/tools/kudu-admin-test.cc
b/src/kudu/tools/kudu-admin-test.cc
index 44835ad..a91c0c7 100644
--- a/src/kudu/tools/kudu-admin-test.cc
+++ b/src/kudu/tools/kudu-admin-test.cc
@@ -52,6 +52,7 @@
#include "kudu/gutil/basictypes.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/stl_util.h"
+#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/split.h"
#include "kudu/gutil/strings/strip.h"
#include "kudu/gutil/strings/substitute.h"
@@ -2204,6 +2205,65 @@ TEST_F(AdminCliTest, TestDumpMemTrackers) {
ASSERT_STR_CONTAINS(stdout, Substitute("\"parent_id\":\"$0\"",
tablet_tracker_id));
}
+TEST_F(AdminCliTest, TestAuthzResetCacheIncorrectMasterAddressList) {
+ NO_FATALS(BuildAndStart());
+
+ const auto& master_addr =
cluster_->master()->bound_rpc_hostport().ToString();
+ const vector<string> dup_master_addresses = { master_addr, master_addr, };
+ const auto& dup_master_addresses_str = JoinStrings(dup_master_addresses,
",");
+ string out;
+ string err;
+ Status s;
+
+ s = RunKuduTool({
+ "master",
+ "authz_cache",
+ "refresh",
+ dup_master_addresses_str,
+ }, &out, &err);
+ ASSERT_TRUE(s.IsRuntimeError()) << ToolRunInfo(s, out, err);
+
+ const auto ref_err_msg = Substitute(
+ "Invalid argument: list of master addresses provided ($0) "
+ "does not match the actual cluster configuration ($1)",
+ dup_master_addresses_str, master_addr);
+ ASSERT_STR_CONTAINS(err, ref_err_msg);
+
+ // However, the '--force' option makes it possible to run the tool even
+ // if the specified list of master addresses does not match the actual
+ // list of master addresses in the cluster.
+ out.clear();
+ err.clear();
+ ASSERT_OK(RunKuduTool({
+ "master",
+ "authz_cache",
+ "refresh",
+ "--force",
+ dup_master_addresses_str,
+ }, &out, &err));
+}
+
+TEST_F(AdminCliTest, TestAuthzResetCacheNotAuthorized) {
+ vector<string> master_flags{ "--superuser_acl=no-such-user" };
+ NO_FATALS(BuildAndStart({}, master_flags));
+
+ // The tool should report an error: it's not possible to reset the cache
+ // since the OS user under which the tools is invoked is not a
superuser/admin
+ // (the --superuser_acl flag is set to contain a non-existent user only).
+ string out;
+ string err;
+ Status s = RunKuduTool({
+ "master",
+ "authz_cache",
+ "refresh",
+ cluster_->master()->bound_rpc_hostport().ToString(),
+ }, &out, &err);
+ ASSERT_TRUE(s.IsRuntimeError()) << ToolRunInfo(s, out, err);
+ ASSERT_STR_CONTAINS(err,
+ "Remote error: Not authorized: unauthorized access to method: "
+ "RefreshAuthzCache");
+}
+
TEST_F(AdminCliTest, TestExtraConfig) {
NO_FATALS(BuildAndStart());
diff --git a/src/kudu/tools/kudu-tool-test.cc b/src/kudu/tools/kudu-tool-test.cc
index c2b6a7d..9061209 100644
--- a/src/kudu/tools/kudu-tool-test.cc
+++ b/src/kudu/tools/kudu-tool-test.cc
@@ -1069,6 +1069,7 @@ TEST_F(ToolTest, TestModeHelp) {
}
{
const vector<string> kMasterModeRegexes = {
+ "authz_cache.*Operate on the authz caches of the Kudu Masters",
"dump_memtrackers.*Dump the memtrackers",
"get_flags.*Get the gflags",
"set_flag.*Change a gflag value",
@@ -1079,6 +1080,12 @@ TEST_F(ToolTest, TestModeHelp) {
NO_FATALS(RunTestHelp("master", kMasterModeRegexes));
}
{
+ const vector<string> kMasterAuthzCacheModeRegexes = {
+ "refresh.*Refresh the authorization policies",
+ };
+ NO_FATALS(RunTestHelp("master authz_cache", kMasterAuthzCacheModeRegexes));
+ }
+ {
const vector<string> kPbcModeRegexes = {
"dump.*Dump a PBC",
};
diff --git a/src/kudu/tools/tool_action_master.cc
b/src/kudu/tools/tool_action_master.cc
index 97a0bc5..f0464cc 100644
--- a/src/kudu/tools/tool_action_master.cc
+++ b/src/kudu/tools/tool_action_master.cc
@@ -19,12 +19,16 @@
#include <functional>
#include <iostream>
#include <iterator>
+#include <map>
#include <memory>
+#include <set>
#include <string>
#include <unordered_map>
+#include <utility>
#include <vector>
#include <boost/algorithm/string/predicate.hpp>
+#include <boost/optional/optional.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>
@@ -42,9 +46,11 @@
#include "kudu/master/master.proxy.h"
#include "kudu/master/master_runner.h"
#include "kudu/rpc/response_callback.h"
+#include "kudu/rpc/rpc_controller.h"
#include "kudu/tools/tool_action.h"
#include "kudu/tools/tool_action_common.h"
#include "kudu/util/init.h"
+#include "kudu/util/monotime.h"
#include "kudu/util/status.h"
DECLARE_bool(force);
@@ -57,8 +63,13 @@ using kudu::master::ListMastersRequestPB;
using kudu::master::ListMastersResponsePB;
using kudu::master::Master;
using kudu::master::MasterServiceProxy;
+using kudu::master::RefreshAuthzCacheRequestPB;
+using kudu::master::RefreshAuthzCacheResponsePB;
using kudu::consensus::RaftPeerPB;
+using kudu::rpc::RpcController;
using std::cout;
+using std::map;
+using std::set;
using std::string;
using std::unique_ptr;
using std::vector;
@@ -189,6 +200,117 @@ Status MasterDumpMemTrackers(const RunnerContext&
context) {
return DumpMemTrackers(address, Master::kDefaultPort);
}
+// Make sure the list of master addresses specified in 'master_addresses'
+// corresponds to the actual list of masters addresses in the cluster,
+// as reported in ConnectToMasterResponsePB::master_addrs.
+Status VerifyMasterAddressList(const vector<string>& master_addresses) {
+ map<string, set<string>> addresses_per_master;
+ for (const auto& address : master_addresses) {
+ unique_ptr<MasterServiceProxy> proxy;
+ RETURN_NOT_OK(BuildProxy(address, Master::kDefaultPort, &proxy));
+
+ RpcController ctl;
+ ctl.set_timeout(MonoDelta::FromMilliseconds(FLAGS_timeout_ms));
+ ConnectToMasterRequestPB req;
+ ConnectToMasterResponsePB resp;
+ RETURN_NOT_OK(proxy->ConnectToMaster(req, &resp, &ctl));
+ const auto& resp_master_addrs = resp.master_addrs();
+ if (resp_master_addrs.size() != master_addresses.size()) {
+ const auto addresses_provided = JoinStrings(master_addresses, ",");
+ const auto addresses_cluster_config = JoinMapped(
+ resp_master_addrs,
+ [](const HostPortPB& pb) {
+ return Substitute("$0:$1", pb.host(), pb.port());
+ }, ",");
+ return Status::InvalidArgument(Substitute(
+ "list of master addresses provided ($0) "
+ "does not match the actual cluster configuration ($1) ",
+ addresses_provided, addresses_cluster_config));
+ }
+ set<string> addr_set;
+ for (const auto& hp : resp_master_addrs) {
+ addr_set.emplace(Substitute("$0:$1", hp.host(), hp.port()));
+ }
+ addresses_per_master.emplace(address, std::move(addr_set));
+ }
+
+ bool mismatch = false;
+ if (addresses_per_master.size() > 1) {
+ const auto it_0 = addresses_per_master.cbegin();
+ auto it_1 = addresses_per_master.begin();
+ ++it_1;
+ for (auto it = it_1; it != addresses_per_master.end(); ++it) {
+ if (it->second != it_0->second) {
+ mismatch = true;
+ break;
+ }
+ }
+ }
+
+ if (mismatch) {
+ string err_msg = Substitute("specified: ($0);",
+ JoinStrings(master_addresses, ","));
+ for (const auto& e : addresses_per_master) {
+ err_msg += Substitute(" from master $0: ($1);",
+ e.first, JoinStrings(e.second, ","));
+ }
+ return Status::ConfigurationError(
+ Substitute("master address lists mismatch: $0", err_msg));
+ }
+
+ return Status::OK();
+}
+
+Status RefreshAuthzCacheAtMaster(const string& master_address) {
+ unique_ptr<MasterServiceProxy> proxy;
+ RETURN_NOT_OK(BuildProxy(master_address, Master::kDefaultPort, &proxy));
+
+ RpcController ctl;
+ ctl.set_timeout(MonoDelta::FromMilliseconds(FLAGS_timeout_ms));
+
+ RefreshAuthzCacheRequestPB req;
+ RefreshAuthzCacheResponsePB resp;
+ RETURN_NOT_OK(proxy->RefreshAuthzCache(req, &resp, &ctl));
+ if (resp.has_error()) {
+ return StatusFromPB(resp.error().status());
+ }
+ return Status::OK();
+}
+
+Status RefreshAuthzCache(const RunnerContext& context) {
+ vector<string> master_addresses;
+ RETURN_NOT_OK(ParseMasterAddresses(context, &master_addresses));
+
+ if (!FLAGS_force) {
+ // Make sure the list of master addresses specified for the command
+ // matches the actual list of masters in the cluster.
+ RETURN_NOT_OK(VerifyMasterAddressList(master_addresses));
+ }
+
+ // It makes sense to refresh privileges cache at every master in the cluster.
+ // Otherwise, the authorization provider might return inconsistent results
for
+ // authz requests upon master leadership change.
+ vector<Status> statuses;
+ statuses.reserve(master_addresses.size());
+ for (const auto& address : master_addresses) {
+ auto status = RefreshAuthzCacheAtMaster(address);
+ statuses.emplace_back(std::move(status));
+ }
+ DCHECK_EQ(master_addresses.size(), statuses.size());
+ string err_str;
+ for (auto i = 0; i < statuses.size(); ++i) {
+ const auto& s = statuses[i];
+ if (s.ok()) {
+ continue;
+ }
+ err_str += Substitute(" error from master at $0: $1",
+ master_addresses[i], s.ToString());
+ }
+ if (err_str.empty()) {
+ return Status::OK();
+ }
+ return Status::Incomplete(err_str);
+}
} // anonymous namespace
unique_ptr<Mode> BuildMasterMode() {
@@ -196,6 +318,25 @@ unique_ptr<Mode> BuildMasterMode() {
builder.Description("Operate on a Kudu Master");
{
+ unique_ptr<Action> action_refresh =
+ ActionBuilder("refresh", &RefreshAuthzCache)
+ .Description("Refresh the authorization policies")
+ .AddRequiredParameter(
+ {kMasterAddressesArg, kMasterAddressesArgDesc})
+ .AddOptionalParameter(
+ "force", boost::none,
+ string(
+ "Ignore mismatches of the specified and the actual lists "
+ "of master addresses in the cluster"))
+ .Build();
+
+ unique_ptr<Mode> mode_authz_cache = ModeBuilder("authz_cache")
+ .Description("Operate on the authz caches of the Kudu Masters")
+ .AddAction(std::move(action_refresh))
+ .Build();
+ builder.AddMode(std::move(mode_authz_cache));
+ }
+ {
unique_ptr<Action> dump_memtrackers =
ActionBuilder("dump_memtrackers", &MasterDumpMemTrackers)
.Description("Dump the memtrackers from a Kudu Master")
