You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2019/04/26 22:02:24 UTC
[kudu] 02/02: [master] add RPC to reset AuthzProvider's cache
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit fdf2ae03216146208fff08a9b191e761fe0fe8fc
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Tue Apr 16 22:21:40 2019 -0700
[master] add RPC to reset AuthzProvider's cache
This changelist adds an RPC end-point to control SentryAuthzProvider's
privileges cache. As of now, it's only possible to reset the cache
of the aggregated SentryPrivilegesFetcher via the newly added
ResetAuthzCache() method.
The method has been added into the master's RPC interface. It's
necessary to have admin/superuser credentials to successfully invoke
the method via Kudu RPC.
Also, added few tests to cover the new functionality.
Change-Id: Ib9a5c1f84172acf5751f68edfb8a9bb25df6b3f6
Reviewed-on: http://gerrit.cloudera.org:8080/13063
Tested-by: Kudu Jenkins
Reviewed-by: Hao Hao <ha...@cloudera.com>
Reviewed-by: Andrew Wong <aw...@cloudera.com>
---
src/kudu/integration-tests/master_sentry-itest.cc | 240 +++++++++++++++++++++-
src/kudu/master/authz_provider.h | 5 +
src/kudu/master/catalog_manager.h | 4 +
src/kudu/master/default_authz_provider.h | 4 +
src/kudu/master/master.cc | 13 +-
src/kudu/master/master.proto | 13 ++
src/kudu/master/master_service.cc | 13 ++
src/kudu/master/master_service.h | 7 +-
src/kudu/master/sentry_authz_provider-test.cc | 4 +-
src/kudu/master/sentry_authz_provider.cc | 4 +
src/kudu/master/sentry_authz_provider.h | 2 +
src/kudu/master/sentry_privileges_fetcher.cc | 82 +++++---
src/kudu/master/sentry_privileges_fetcher.h | 24 ++-
src/kudu/mini-cluster/external_mini_cluster.h | 2 -
14 files changed, 354 insertions(+), 63 deletions(-)
diff --git a/src/kudu/integration-tests/master_sentry-itest.cc b/src/kudu/integration-tests/master_sentry-itest.cc
index a110162..80678cc 100644
--- a/src/kudu/integration-tests/master_sentry-itest.cc
+++ b/src/kudu/integration-tests/master_sentry-itest.cc
@@ -15,15 +15,19 @@
// specific language governing permissions and limitations
// under the License.
+#include <atomic>
+#include <cstdlib>
#include <functional>
#include <memory>
#include <ostream>
#include <string>
+#include <thread>
#include <unordered_set>
#include <utility>
#include <vector>
#include <boost/optional/optional.hpp>
+#include <glog/logging.h>
#include <gtest/gtest.h>
#include "kudu/client/client.h"
@@ -31,16 +35,20 @@
#include "kudu/client/shared_ptr.h"
#include "kudu/common/common.pb.h"
#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"
#include "kudu/hms/mini_hms.h"
#include "kudu/integration-tests/cluster_itest_util.h"
+#include "kudu/integration-tests/external_mini_cluster-itest-base.h"
#include "kudu/integration-tests/hms_itest-base.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/master.proxy.h"
#include "kudu/master/sentry_authz_provider-test-base.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
+#include "kudu/rpc/rpc_controller.h"
+#include "kudu/rpc/rpc_header.pb.h"
#include "kudu/rpc/user_credentials.h"
#include "kudu/security/test/mini_kdc.h"
#include "kudu/sentry/mini_sentry.h"
@@ -48,17 +56,17 @@
#include "kudu/sentry/sentry_policy_service_types.h"
#include "kudu/thrift/client.h"
#include "kudu/util/monotime.h"
+#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/slice.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
-using ::sentry::TSentryGrantOption;
-using ::sentry::TSentryPrivilege;
using boost::make_optional;
using boost::none;
using boost::optional;
using kudu::client::KuduClient;
+using kudu::client::KuduColumnSchema;
using kudu::client::KuduScanToken;
using kudu::client::KuduScanTokenBuilder;
using kudu::client::KuduSchema;
@@ -68,12 +76,19 @@ using kudu::client::sp::shared_ptr;
using kudu::hms::HmsClient;
using kudu::master::GetTableLocationsResponsePB;
using kudu::master::MasterServiceProxy;
+using kudu::master::ResetAuthzCacheRequestPB;
+using kudu::master::ResetAuthzCacheResponsePB;
using kudu::master::TabletLocationsPB;
+using kudu::rpc::RpcController;
using kudu::rpc::UserCredentials;
using kudu::sentry::SentryClient;
+using sentry::TSentryGrantOption;
+using sentry::TSentryPrivilege;
+using std::atomic;
using std::function;
using std::ostream;
using std::string;
+using std::thread;
using std::unique_ptr;
using std::unordered_set;
using std::vector;
@@ -180,8 +195,8 @@ class SentryITestBase : public HmsITestBase {
return client_->IsCreateTableInProgress(table, &in_progress);
}
- Status DeleteTable(const string& table,
- const string& /*new_table*/) {
+ Status DropTable(const string& table,
+ const string& /*new_table*/) {
return client_->DeleteTable(table);
}
@@ -509,9 +524,9 @@ static const AuthzDescriptor kAuthzCombinations[] = {
},
{
{
- &SentryITestBase::DeleteTable,
+ &SentryITestBase::DropTable,
&SentryITestBase::GrantDropTablePrivilege,
- "DeleteTable",
+ "DropTable",
},
SentryITestBase::kDatabaseName,
SentryITestBase::kTableName,
@@ -663,9 +678,9 @@ TEST_P(AuthzErrorHandlingTest, TestNonExistentTable) {
static const AuthzFuncs kAuthzFuncCombinations[] = {
{
- &SentryITestBase::DeleteTable,
+ &SentryITestBase::DropTable,
&SentryITestBase::GrantDropTablePrivilege,
- "DeleteTable"
+ "DropTable"
},
{
&SentryITestBase::AlterTable,
@@ -702,4 +717,213 @@ static const AuthzFuncs kAuthzFuncCombinations[] = {
INSTANTIATE_TEST_CASE_P(AuthzFuncCombinations,
AuthzErrorHandlingTest,
::testing::ValuesIn(kAuthzFuncCombinations));
+
+// Class for test scenarios verifying functionality of managing AuthzProvider's
+// privileges cache via Kudu RPC.
+class SentryAuthzProviderCacheITest : public SentryITestBase {
+ public:
+ bool IsAuthzPrivilegeCacheEnabled() const override {
+ return true;
+ }
+
+ Status ResetCache() {
+ // ResetAuthzCache() RPC requires admin/superuser credentials, so this
+ // method calls Kinit(kAdminUser) to authenticate appropriately. However,
+ // it's necessary to return back the credentials of the regular user after
+ // resetting the cache since the rest of the scenario is supposed to run
+ // without superuser credentials.
+ SCOPED_CLEANUP({
+ WARN_NOT_OK(cluster_->kdc()->Kinit(kTestUser),
+ "could not restore Kerberos credentials");
+ });
+ RETURN_NOT_OK(cluster_->kdc()->Kinit(kAdminUser));
+ std::shared_ptr<MasterServiceProxy> proxy = cluster_->master_proxy();
+ UserCredentials user_credentials;
+ user_credentials.set_real_user(kAdminUser);
+ proxy->set_user_credentials(user_credentials);
+
+ RpcController ctl;
+ ResetAuthzCacheResponsePB res;
+ RETURN_NOT_OK(proxy->ResetAuthzCache(
+ ResetAuthzCacheRequestPB(), &res, &ctl));
+ return res.has_error() ? StatusFromPB(res.error().status()) : Status::OK();
+ }
+};
+
+// This test scenario uses AlterTable() to make sure AuthzProvider's cache
+// empties upon successful ResetAuthzCache() RPC.
+TEST_F(SentryAuthzProviderCacheITest, AlterTable) {
+ const auto table_name = Substitute("$0.$1", kDatabaseName, kTableName);
+ ASSERT_OK(GrantAlterTablePrivilege(kDatabaseName, kTableName));
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(
+ client_->NewTableAlterer(table_name)->DropColumn("int8_val"));
+ auto s = table_alterer->Alter();
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ }
+ ASSERT_OK(StopSentry());
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(
+ client_->NewTableAlterer(table_name)->DropColumn("int16_val"));
+ auto s = table_alterer->Alter();
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ }
+ ASSERT_OK(ResetCache());
+ {
+ // After resetting the cache, it should not be possible to perform another
+ // ALTER TABLE operation: no entries are in the cache, so authz provider
+ // needs to fetch information from Sentry directly.
+ unique_ptr<KuduTableAlterer> table_alterer(
+ client_->NewTableAlterer(table_name)->DropColumn("int32_val"));
+ auto s = table_alterer->Alter();
+ ASSERT_TRUE(s.IsNetworkError()) << s.ToString();
+ }
+
+ // Try to do the same after starting Sentry back. It should be a success.
+ ASSERT_OK(StartSentry());
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(
+ client_->NewTableAlterer(table_name)->DropColumn("int32_val"));
+ auto s = table_alterer->Alter();
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ }
+}
+
+// This test scenario calls a couple of authz methods of SentryAuthzProvider
+// while resetting its fetcher's cache in parallel using master's
+// ResetAuthzCache() RPC.
+TEST_F(SentryAuthzProviderCacheITest, ResetAuthzCacheConcurrentAlterTable) {
+ constexpr const auto num_threads = 16;
+ const auto run_interval = AllowSlowTests() ? MonoDelta::FromSeconds(8)
+ : MonoDelta::FromSeconds(1);
+ ASSERT_OK(GrantCreateTablePrivilege(kDatabaseName, ""));
+ for (auto idx = 0; idx < num_threads; ++idx) {
+ ASSERT_OK(CreateTable(Substitute("$0.$1", kDatabaseName, idx), ""));
+ ASSERT_OK(GrantAlterTablePrivilege(kDatabaseName, Substitute("$0", idx)));
+ }
+
+ vector<Status> threads_task_status(num_threads);
+ {
+ atomic<bool> stopped(false);
+ vector<thread> threads;
+
+ SCOPED_CLEANUP({
+ stopped = true;
+ for (auto& thread : threads) {
+ thread.join();
+ }
+ });
+
+ for (auto idx = 0; idx < num_threads; ++idx) {
+ const auto thread_idx = idx;
+ threads.emplace_back([&, thread_idx] () {
+ const auto table_name = Substitute("$0.$1", kDatabaseName, thread_idx);
+
+ while (!stopped) {
+ SleepFor(MonoDelta::FromMicroseconds((rand() % 2 + 1) * thread_idx));
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(
+ client_->NewTableAlterer(table_name));
+ table_alterer->DropColumn("int8_val");
+
+ auto s = table_alterer->Alter();
+ if (!s.ok()) {
+ threads_task_status[thread_idx] = s;
+ return;
+ }
+ }
+
+ SleepFor(MonoDelta::FromMicroseconds((rand() % 3 + 1) * thread_idx));
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(
+ client_->NewTableAlterer(table_name));
+ table_alterer->AddColumn("int8_val")->Type(KuduColumnSchema::INT8);
+ auto s = table_alterer->Alter();
+ if (!s.ok()) {
+ threads_task_status[thread_idx] = s;
+ return;
+ }
+ }
+ }
+ });
+ }
+
+ const auto time_beg = MonoTime::Now();
+ const auto time_end = time_beg + run_interval;
+ while (MonoTime::Now() < time_end) {
+ SleepFor(MonoDelta::FromMilliseconds((rand() % 3)));
+ ASSERT_OK(ResetCache());
+ }
+ }
+ for (auto idx = 0; idx < threads_task_status.size(); ++idx) {
+ SCOPED_TRACE(Substitute("results for thread $0", idx));
+ const auto& s = threads_task_status[idx];
+ EXPECT_TRUE(s.ok()) << s.ToString();
+ }
+}
+
+// Basic test to verify access control and functionality of
+// the ResetAuthzCache(); integration with Sentry is not enabled.
+class AuthzCacheControlTest : public ExternalMiniClusterITestBase {
+ public:
+ void SetUp() override {
+ ExternalMiniClusterITestBase::SetUp();
+ cluster::ExternalMiniClusterOptions opts;
+ opts.enable_kerberos = true;
+ StartClusterWithOpts(opts);
+ }
+
+ // Utility method to call master's ResetAuthzCache RPC under the credentials
+ // of the specified 'user'. The credentials should have been set appropriately
+ // before calling this method (e.g., call kinit if necessary).
+ Status ResetCache(const string& user,
+ RpcController* ctl,
+ ResetAuthzCacheResponsePB* resp) {
+ RETURN_NOT_OK(cluster_->kdc()->Kinit(user));
+ std::shared_ptr<MasterServiceProxy> proxy = cluster_->master_proxy();
+ UserCredentials user_credentials;
+ user_credentials.set_real_user(user);
+ proxy->set_user_credentials(user_credentials);
+
+ ResetAuthzCacheRequestPB req;
+ return proxy->ResetAuthzCache(req, resp, ctl);
+ }
+};
+
+TEST_F(AuthzCacheControlTest, ResetCacheNoSentryIntegration) {
+ // Non-admin users (i.e. those not in --superuser_acl) are not allowed
+ // to reset authz cache.
+ for (const auto& user : { "test-user", "joe-interloper", }) {
+ ASSERT_OK(cluster_->kdc()->Kinit(user));
+ ResetAuthzCacheResponsePB resp;
+ RpcController ctl;
+ auto s = ResetCache(user, &ctl, &resp);
+ ASSERT_TRUE(s.IsRemoteError()) << s.ToString();
+ ASSERT_FALSE(resp.has_error());
+
+ const auto* err_status = ctl.error_response();
+ ASSERT_NE(nullptr, err_status);
+ ASSERT_TRUE(err_status->has_code());
+ ASSERT_EQ(rpc::ErrorStatusPB::ERROR_APPLICATION, err_status->code());
+ ASSERT_STR_CONTAINS(err_status->message(),
+ "unauthorized access to method: ResetAuthzCache");
+ }
+
+ // The cache can be reset with credentials of a super-user. However, in
+ // case if integration with Sentry is not enabled, the AuthzProvider
+ // doesn't have any cache.
+ {
+ ResetAuthzCacheResponsePB resp;
+ RpcController ctl;
+ const auto s = ResetCache("test-admin", &ctl, &resp);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(nullptr, ctl.error_response());
+ ASSERT_TRUE(resp.has_error()) << resp.error().DebugString();
+ const auto app_s = StatusFromPB(resp.error().status());
+ ASSERT_TRUE(app_s.IsNotSupported()) << app_s.ToString();
+ ASSERT_STR_CONTAINS(app_s.ToString(),
+ "provider does not have privileges cache");
+ }
+}
+
} // namespace kudu
diff --git a/src/kudu/master/authz_provider.h b/src/kudu/master/authz_provider.h
index c75456b..d768518 100644
--- a/src/kudu/master/authz_provider.h
+++ b/src/kudu/master/authz_provider.h
@@ -46,6 +46,11 @@ class AuthzProvider {
// Stops the AuthzProvider instance.
virtual void Stop() = 0;
+ // Reset the underlying cache (if any), invalidating all cached entries.
+ // Returns Status::NotSupported() if the provider doesn't support resetting
+ // its cache.
+ virtual Status ResetCache() = 0;
+
// Checks if the table creation is authorized for the given user.
// If the table is being created with a different owner than the user,
// then more strict privilege is required.
diff --git a/src/kudu/master/catalog_manager.h b/src/kudu/master/catalog_manager.h
index 94dffc5..b7c8290 100644
--- a/src/kudu/master/catalog_manager.h
+++ b/src/kudu/master/catalog_manager.h
@@ -697,6 +697,10 @@ class CatalogManager : public tserver::TabletReplicaLookupIf {
return hms_catalog_.get();
}
+ master::AuthzProvider* authz_provider() const {
+ return authz_provider_.get();
+ }
+
// Returns the normalized form of the provided table name.
//
// If the HMS integration is configured and the table name is a valid HMS
diff --git a/src/kudu/master/default_authz_provider.h b/src/kudu/master/default_authz_provider.h
index 47509b0..bca2dd6 100644
--- a/src/kudu/master/default_authz_provider.h
+++ b/src/kudu/master/default_authz_provider.h
@@ -38,6 +38,10 @@ class DefaultAuthzProvider : public AuthzProvider {
void Stop() override {}
+ Status ResetCache() override {
+ return Status::NotSupported("provider does not have privileges cache");
+ }
+
Status AuthorizeCreateTable(const std::string& /*table_name*/,
const std::string& /*user*/,
const std::string& /*owner*/) override WARN_UNUSED_RESULT {
diff --git a/src/kudu/master/master.cc b/src/kudu/master/master.cc
index 519699b..5d90381 100644
--- a/src/kudu/master/master.cc
+++ b/src/kudu/master/master.cc
@@ -200,10 +200,10 @@ Status Master::StartAsync() {
RETURN_NOT_OK(maintenance_manager_->Start());
gscoped_ptr<ServiceIf> impl(new MasterServiceImpl(this));
- gscoped_ptr<ServiceIf> consensus_service(new ConsensusServiceImpl(
- this, catalog_manager_.get()));
- gscoped_ptr<ServiceIf> tablet_copy_service(new TabletCopyServiceImpl(
- this, catalog_manager_.get()));
+ gscoped_ptr<ServiceIf> consensus_service(
+ new ConsensusServiceImpl(this, catalog_manager_.get()));
+ gscoped_ptr<ServiceIf> tablet_copy_service(
+ new TabletCopyServiceImpl(this, catalog_manager_.get()));
RETURN_NOT_OK(RegisterService(std::move(impl)));
RETURN_NOT_OK(RegisterService(std::move(consensus_service)));
@@ -216,7 +216,6 @@ Status Master::StartAsync() {
// Start initializing the catalog manager.
RETURN_NOT_OK(init_pool_->SubmitClosure(Bind(&Master::InitCatalogManagerTask,
Unretained(this))));
-
state_ = kRunning;
return Status::OK();
@@ -342,7 +341,7 @@ Status GetMasterEntryForHost(const shared_ptr<rpc::Messenger>& messenger,
} // anonymous namespace
-Status Master::ListMasters(std::vector<ServerEntryPB>* masters) const {
+Status Master::ListMasters(vector<ServerEntryPB>* masters) const {
if (!opts_.IsDistributed()) {
ServerEntryPB local_entry;
local_entry.mutable_instance_id()->CopyFrom(catalog_manager_->NodeInstance());
@@ -385,7 +384,7 @@ Status Master::ListMasters(std::vector<ServerEntryPB>* masters) const {
return Status::OK();
}
-Status Master::GetMasterHostPorts(std::vector<HostPortPB>* hostports) const {
+Status Master::GetMasterHostPorts(vector<HostPortPB>* hostports) const {
auto consensus = catalog_manager_->master_consensus();
if (!consensus) {
return Status::IllegalState("consensus not running");
diff --git a/src/kudu/master/master.proto b/src/kudu/master/master.proto
index 30cbcc8..89afecd 100644
--- a/src/kudu/master/master.proto
+++ b/src/kudu/master/master.proto
@@ -783,6 +783,15 @@ message ReplaceTabletResponsePB {
optional bytes replacement_tablet_id = 2;
}
+// ResetAuthzCache{Request/Result}PB: reset the authz privileges cache,
+// clearing it of all existing entries.
+message ResetAuthzCacheRequestPB {
+}
+
+message ResetAuthzCacheResponsePB {
+ optional MasterErrorPB error = 1;
+}
+
enum MasterFeatures {
UNKNOWN_FEATURE = 0;
// The master supports creating tables with non-covering range partitions.
@@ -865,6 +874,10 @@ service MasterService {
rpc ReplaceTablet(ReplaceTabletRequestPB) returns (ReplaceTabletResponsePB) {
option (kudu.rpc.authz_method) = "AuthorizeSuperUser";
}
+ rpc ResetAuthzCache(ResetAuthzCacheRequestPB) returns
+ (ResetAuthzCacheResponsePB) {
+ option (kudu.rpc.authz_method) = "AuthorizeSuperUser";
+ }
// Master->Master RPCs
// ------------------------------------------------------------
diff --git a/src/kudu/master/master_service.cc b/src/kudu/master/master_service.cc
index 5b44196..eaab0b9 100644
--- a/src/kudu/master/master_service.cc
+++ b/src/kudu/master/master_service.cc
@@ -33,8 +33,10 @@
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/metadata.pb.h"
#include "kudu/consensus/replica_management.pb.h"
+#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"
@@ -596,6 +598,17 @@ void MasterServiceImpl::ReplaceTablet(const ReplaceTabletRequestPB* req,
rpc->RespondSuccess();
}
+void MasterServiceImpl::ResetAuthzCache(
+ const ResetAuthzCacheRequestPB* /* req */,
+ ResetAuthzCacheResponsePB* resp,
+ rpc::RpcContext* rpc) {
+ LOG(INFO) << Substitute("request to reset authz privileges cache from $0",
+ rpc->requestor_string());
+ CheckRespErrorOrSetUnknown(
+ server_->catalog_manager()->authz_provider()->ResetCache(), 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 3f1ebfd..5d6846e 100644
--- a/src/kudu/master/master_service.h
+++ b/src/kudu/master/master_service.h
@@ -20,7 +20,6 @@
#include <cstdint>
#include "kudu/gutil/macros.h"
-#include "kudu/gutil/port.h"
#include "kudu/master/master.service.h"
namespace google {
@@ -68,6 +67,8 @@ class PingRequestPB;
class PingResponsePB;
class ReplaceTabletRequestPB;
class ReplaceTabletResponsePB;
+class ResetAuthzCacheRequestPB;
+class ResetAuthzCacheResponsePB;
class TSHeartbeatRequestPB;
class TSHeartbeatResponsePB;
@@ -159,6 +160,10 @@ class MasterServiceImpl : public MasterServiceIf {
ReplaceTabletResponsePB* resp,
rpc::RpcContext* rpc) override;
+ void ResetAuthzCache(const ResetAuthzCacheRequestPB* req,
+ ResetAuthzCacheResponsePB* resp,
+ rpc::RpcContext* rpc) override;
+
bool SupportsFeature(uint32_t feature) const override;
private:
diff --git a/src/kudu/master/sentry_authz_provider-test.cc b/src/kudu/master/sentry_authz_provider-test.cc
index f06e2a4..e7304ab 100644
--- a/src/kudu/master/sentry_authz_provider-test.cc
+++ b/src/kudu/master/sentry_authz_provider-test.cc
@@ -1346,9 +1346,9 @@ TEST_P(TestConcurrentRequests, SuccessResponses) {
// (kNumRequestThreads / 2) of actual RPC requests to Sentry might be reached
// and the assertion below would be triggered. For example, the OS scheduler
// might de-schedule the majority of the threads spawned above for a time
- // greater than it takes to complete an RPC to Sentry, and the de-scheduling
+ // longer than it takes to complete an RPC to Sentry, and that de-scheduling
// might happen exactly prior the point when the 'earlier-running' thread
- // added itself into a queue record designed to track concurrent requests.
+ // added itself into a queue designed to track concurrent requests.
// Essentially, that's about 'freezing' all incoming requests just before the
// queueing point, and then awakening them one by one, so no more than one
// thread is registered in the queue at any time.
diff --git a/src/kudu/master/sentry_authz_provider.cc b/src/kudu/master/sentry_authz_provider.cc
index 12f1cc1..74c88b6 100644
--- a/src/kudu/master/sentry_authz_provider.cc
+++ b/src/kudu/master/sentry_authz_provider.cc
@@ -113,6 +113,10 @@ void SentryAuthzProvider::Stop() {
fetcher_.Stop();
}
+Status SentryAuthzProvider::ResetCache() {
+ return fetcher_.ResetCache();
+}
+
bool SentryAuthzProvider::IsEnabled() {
return !FLAGS_sentry_service_rpc_addresses.empty();
}
diff --git a/src/kudu/master/sentry_authz_provider.h b/src/kudu/master/sentry_authz_provider.h
index a537e00..aa4a23b 100644
--- a/src/kudu/master/sentry_authz_provider.h
+++ b/src/kudu/master/sentry_authz_provider.h
@@ -56,6 +56,8 @@ class SentryAuthzProvider : public AuthzProvider {
void Stop() override;
+ Status ResetCache() override WARN_UNUSED_RESULT;
+
// Returns true if the SentryAuthzProvider should be enabled.
static bool IsEnabled();
diff --git a/src/kudu/master/sentry_privileges_fetcher.cc b/src/kudu/master/sentry_privileges_fetcher.cc
index a497651..27984e0 100644
--- a/src/kudu/master/sentry_privileges_fetcher.cc
+++ b/src/kudu/master/sentry_privileges_fetcher.cc
@@ -23,7 +23,6 @@
#include <iterator>
#include <memory>
#include <mutex>
-#include <ostream>
#include <type_traits>
#include <unordered_map>
#include <vector>
@@ -153,6 +152,7 @@ using sentry::TListSentryPrivilegesResponse;
using sentry::TSentryAuthorizable;
using sentry::TSentryGrantOption;
using sentry::TSentryPrivilege;
+using std::make_shared;
using std::shared_ptr;
using std::string;
using std::unique_ptr;
@@ -447,11 +447,7 @@ void SentryPrivilegesBranch::DoInit(
SentryAction::Action action;
if (!SentryPrivilegesFetcher::SentryPrivilegeIsWellFormed(
privilege_resp, authorizable, &scope, &action)) {
- if (VLOG_IS_ON(1)) {
- std::ostringstream os;
- privilege_resp.printTo(os);
- VLOG(1) << Substitute("ignoring privilege response: $0", os.str());
- }
+ VLOG(1) << "ignoring privilege response: " << privilege_resp;
continue;
}
const auto& db = privilege_resp.dbName;
@@ -467,11 +463,9 @@ void SentryPrivilegesBranch::DoInit(
(privilege_resp.grantOption == TSentryGrantOption::ENABLED);
}
if (VLOG_IS_ON(1)) {
- std::ostringstream os;
- privilege_resp.printTo(os);
if (action != SentryAction::ALL && action != SentryAction::OWNER &&
privilege_resp.grantOption == TSentryGrantOption::ENABLED) {
- VLOG(1) << "ignoring ENABLED grant option for unexpected action: "
+ VLOG(1) << "ignoring ENABLED grant option for unknown action: "
<< static_cast<int16_t>(action);
}
}
@@ -490,6 +484,11 @@ SentryPrivilegesFetcher::SentryPrivilegesFetcher(
}
Status SentryPrivilegesFetcher::Start() {
+ // The semantics of SentryAuthzProvider's Start()/Stop() don't guarantee
+ // immutability of the Sentry service's end-point between restarts. So, since
+ // the information in the cache might become irrelevant after restarting
+ // 'sentry_client_' with different Sentry address, it makes sense to clear
+ // the cache of all accumulated entries.
ResetCache();
vector<HostPort> addresses;
@@ -521,6 +520,29 @@ void SentryPrivilegesFetcher::Stop() {
sentry_client_.Stop();
}
+Status SentryPrivilegesFetcher::ResetCache() {
+ const auto cache_capacity_bytes =
+ FLAGS_sentry_privileges_cache_capacity_mb * 1024 * 1024;
+ shared_ptr<AuthzInfoCache> new_cache;
+ if (cache_capacity_bytes != 0) {
+ const auto ttl_sec = FLAGS_authz_token_validity_seconds *
+ FLAGS_sentry_privileges_cache_ttl_factor;
+ new_cache = make_shared<AuthzInfoCache>(
+ cache_capacity_bytes, MonoDelta::FromSeconds(ttl_sec));
+ if (metric_entity_) {
+ unique_ptr<SentryPrivilegesCacheMetrics> metrics(
+ new SentryPrivilegesCacheMetrics(metric_entity_));
+ new_cache->SetMetrics(std::move(metrics));
+ }
+ }
+ {
+ std::lock_guard<rw_spinlock> l(cache_lock_);
+ cache_ = new_cache;
+ }
+
+ return Status::OK();
+}
+
Status SentryPrivilegesFetcher::GetSentryPrivileges(
SentryAuthorizableScope::Scope requested_scope,
const string& table_name,
@@ -548,11 +570,22 @@ Status SentryPrivilegesFetcher::GetSentryPrivileges(
// Do not query Sentry for authz scopes narrower than 'TABLE'.
const auto& key = aggregate_key.GetKey(SentryAuthorizableScope::TABLE);
const auto& key_seq = aggregate_key.key_sequence();
+
+ // Copy the shared pointer to the cache. That's necessary because:
+ // * the cache_ member may be reset by concurrent ResetCache()
+ // * TTLCache is based on Cache that doesn't allow for outstanding handles
+ // if the cache itself destructed (in this case, goes out of scope).
+ shared_ptr<AuthzInfoCache> cache;
+ {
+ shared_lock<rw_spinlock> l(cache_lock_);
+ cache = cache_;
+ }
vector<typename AuthzInfoCache::EntryHandle> handles;
handles.reserve(AuthzInfoKey::kKeySequenceMaxSize);
- if (PREDICT_TRUE(cache_)) {
+
+ if (PREDICT_TRUE(cache)) {
for (const auto& e : key_seq) {
- auto handle = cache_->Get(e);
+ auto handle = cache->Get(e);
VLOG(3) << Substitute("'$0': '$1' key lookup", key, e);
if (!handle) {
continue;
@@ -589,7 +622,7 @@ Status SentryPrivilegesFetcher::GetSentryPrivileges(
pending_request.callbacks.emplace_back(sync.AsStatusCallback());
if (is_first_request) {
DCHECK(!pending_request.result);
- pending_request.result = std::make_shared<SentryPrivilegesBranch>();
+ pending_request.result = make_shared<SentryPrivilegesBranch>();
}
fetched_privileges = pending_request.result;
}
@@ -602,7 +635,7 @@ Status SentryPrivilegesFetcher::GetSentryPrivileges(
const auto s = FetchPrivilegesFromSentry(FLAGS_kudu_service_name,
user, authorizable,
fetched_privileges.get());
- if (s.ok() && PREDICT_TRUE(cache_)) {
+ if (s.ok() && PREDICT_TRUE(cache)) {
// Put the result into the cache. Negative results (i.e. errors) are not
// cached. Split the information on privileges into at most two cache
// entries, for authorizables of scope:
@@ -622,7 +655,7 @@ Status SentryPrivilegesFetcher::GetSentryPrivileges(
SentryAuthorizableScope::DATABASE);
const auto result_footprint = result_ptr->memory_footprint() +
key.capacity();
- cache_->Put(key, std::move(result_ptr), result_footprint);
+ cache->Put(key, std::move(result_ptr), result_footprint);
VLOG(2) << Substitute(
"added entry of size $0 bytes for key '$1' (server-database scope)",
result_footprint, key);
@@ -642,7 +675,7 @@ Status SentryPrivilegesFetcher::GetSentryPrivileges(
SentryAuthorizableScope::TABLE);
const auto result_footprint = result_ptr->memory_footprint() +
key.capacity();
- cache_->Put(key, std::move(result_ptr), result_footprint);
+ cache->Put(key, std::move(result_ptr), result_footprint);
VLOG(2) << Substitute(
"added entry of size $0 bytes for key '$1' (table-column scope)",
result_footprint, key);
@@ -829,24 +862,5 @@ Status SentryPrivilegesFetcher::FetchPrivilegesFromSentry(
return Status::OK();
}
-Status SentryPrivilegesFetcher::ResetCache() {
- const auto cache_capacity_bytes =
- FLAGS_sentry_privileges_cache_capacity_mb * 1024 * 1024;
- if (cache_capacity_bytes == 0) {
- cache_.reset();
- } else {
- const auto ttl_sec = FLAGS_authz_token_validity_seconds *
- FLAGS_sentry_privileges_cache_ttl_factor;
- cache_.reset(new AuthzInfoCache(cache_capacity_bytes,
- MonoDelta::FromSeconds(ttl_sec)));
- if (metric_entity_) {
- unique_ptr<SentryPrivilegesCacheMetrics> metrics(
- new SentryPrivilegesCacheMetrics(metric_entity_));
- cache_->SetMetrics(std::move(metrics));
- }
- }
- return Status::OK();
-}
-
} // namespace master
} // namespace kudu
diff --git a/src/kudu/master/sentry_privileges_fetcher.h b/src/kudu/master/sentry_privileges_fetcher.h
index bb17edd..05b5f3a 100644
--- a/src/kudu/master/sentry_privileges_fetcher.h
+++ b/src/kudu/master/sentry_privileges_fetcher.h
@@ -159,6 +159,10 @@ class SentryPrivilegesFetcher {
Status Start();
void Stop();
+ // Resets the authz cache. In addition to lifecycle-related methods like
+ // Start(), this method is also used by SentryAuthzProvider::ResetCache().
+ Status ResetCache();
+
// Fetches the user's privileges from Sentry for the authorizable specified
// by the given table and scope. The result privileges might be served
// from the cache, if caching is enabled and corresponding entry exists
@@ -209,14 +213,6 @@ class SentryPrivilegesFetcher {
const ::sentry::TSentryAuthorizable& authorizable,
SentryPrivilegesBranch* result);
- // Resets the authz cache. In addition to lifecycle-related methods like
- // Start(), this method is also used by tests: if the authz information
- // has been updated by the test, the cache needs to be invalidated.
- //
- // NOTE: this method is not thread-safe and should not be called along with
- // concurrent authz requests.
- Status ResetCache();
-
// Metric entity for registering metric gauges/counters.
scoped_refptr<MetricEntity> metric_entity_;
@@ -224,8 +220,18 @@ class SentryPrivilegesFetcher {
thrift::HaClient<sentry::SentryClient> sentry_client_;
// The TTL cache to store information on privileges received from Sentry.
+ // The instance is wrapped into std::shared_ptr to handle operations with
+ // cache items along with concurrent requests to reset the instance.
typedef TTLCache<std::string, SentryPrivilegesBranch> AuthzInfoCache;
- std::unique_ptr<AuthzInfoCache> cache_;
+ std::shared_ptr<AuthzInfoCache> cache_;
+
+ // Synchronization primitive to guard access to the cache in the presence
+ // of operations with cache items and concurrent requests to reset the cache.
+ // An alternative would be to use std::atomic_load and std::atomic_exchange,
+ // but:
+ // * They're higher overhead operations than just using a spinlock.
+ // * They're not available in all C++11-compatible compilers.
+ rw_spinlock cache_lock_;
// Utility dictionary to keep track of requests sent to Sentry. Access is
// guarded by pending_requests_lock_. The key corresponds to the set of
diff --git a/src/kudu/mini-cluster/external_mini_cluster.h b/src/kudu/mini-cluster/external_mini_cluster.h
index 7f23c73..d9ed431 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.h
+++ b/src/kudu/mini-cluster/external_mini_cluster.h
@@ -397,8 +397,6 @@ class ExternalMiniCluster : public MiniCluster {
std::string GetLogPath(const std::string& daemon_id) const;
private:
- FRIEND_TEST(MasterFailoverTest, TestKillAnyMaster);
-
Status StartMasters();
Status StartSentry();