You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by gr...@apache.org on 2019/04/23 17:15:28 UTC
[kudu] branch master updated (64cdf30 -> 2b023b9)
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 64cdf30 hybrid_clock: use different message when injecting errors
new 73182b0 master: use AuthzProvider to generate authz tokens
new 6be025f [sentry] enable sentry integration for master stress test
new 2b023b9 [backup] Fix fromMs override option
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../scala/org/apache/kudu/backup/KuduBackup.scala | 14 +-
.../scala/org/apache/kudu/backup/Options.scala | 7 +-
.../scala/org/apache/kudu/backup/SessionIO.scala | 7 +-
.../org/apache/kudu/backup/TestKuduBackup.scala | 50 +-
src/kudu/integration-tests/CMakeLists.txt | 5 +-
.../alter_table-randomized-test.cc | 63 ++-
src/kudu/integration-tests/cluster_itest_util.cc | 119 +++--
src/kudu/integration-tests/cluster_itest_util.h | 12 +
src/kudu/integration-tests/hms_itest-base.h | 12 +-
src/kudu/integration-tests/master-stress-test.cc | 31 +-
.../integration-tests/master_failover-itest.cc | 56 ++-
src/kudu/integration-tests/master_hms-itest.cc | 4 +-
src/kudu/integration-tests/master_sentry-itest.cc | 10 +-
src/kudu/integration-tests/ts_sentry-itest.cc | 542 +++++++++++++++++++++
src/kudu/master/catalog_manager.cc | 32 +-
src/kudu/master/catalog_manager.h | 8 +-
src/kudu/master/master_service.cc | 23 +-
src/kudu/sentry/mini_sentry.cc | 3 +
src/kudu/util/random_util.h | 7 +-
19 files changed, 843 insertions(+), 162 deletions(-)
create mode 100644 src/kudu/integration-tests/ts_sentry-itest.cc
[kudu] 03/03: [backup] Fix fromMs override option
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 2b023b979358cf3716c3b282ff21879bd97cec4a
Author: Grant Henke <gr...@apache.org>
AuthorDate: Mon Apr 22 21:20:00 2019 -0500
[backup] Fix fromMs override option
Fix the logic for deciding when to lookup old backups.
Previously the fromMs argument wasn’t being handled
correctly.
Additionally contains a small change to avoid converting
back and forth from Path to HPath multiple times.
Change-Id: I8a3abc47dd9d1441ba269dfc9405691f79e6615d
Reviewed-on: http://gerrit.cloudera.org:8080/13080
Tested-by: Kudu Jenkins
Reviewed-by: Will Berkeley <wd...@gmail.com>
---
.../scala/org/apache/kudu/backup/KuduBackup.scala | 14 +++---
.../scala/org/apache/kudu/backup/Options.scala | 7 +--
.../scala/org/apache/kudu/backup/SessionIO.scala | 7 ++-
.../org/apache/kudu/backup/TestKuduBackup.scala | 50 ++++++++++++++++++----
4 files changed, 58 insertions(+), 20 deletions(-)
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
index a8d89f6..445bef0 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduBackup.scala
@@ -56,22 +56,26 @@ object KuduBackup {
// Unless we are forcing a full backup or a fromMs was set, find the previous backup and
// use the `to_ms` metadata as the `from_ms` time for this backup.
- if (!tableOptions.forceFull || tableOptions.fromMs != 0) {
- log.info("No fromMs option set, looking for a previous backup.")
+ if (tableOptions.forceFull) {
+ log.info("Performing a full backup, forceFull was set to true")
+ } else if (tableOptions.fromMs == BackupOptions.DefaultFromMS) {
+ log.info(s"Performing an incremental backup, fromMs was set to ${tableOptions.fromMs}")
+ } else {
+ log.info("Looking for a previous backup, forceFull or fromMs options are not set.")
val graph = io.readBackupGraph(tableName)
if (graph.hasFullBackup) {
val base = graph.backupBase
log.info(s"Setting fromMs to ${base.metadata.getToMs} from backup in path: ${base.path}")
tableOptions = tableOptions.copy(fromMs = base.metadata.getToMs)
} else {
- log.info("No full backup was found. Starting a full backup.")
- tableOptions = tableOptions.copy(fromMs = 0)
+ log.info("No previous backup was found. Starting a full backup.")
+ tableOptions = tableOptions.copy(forceFull = true)
}
}
val rdd = new KuduBackupRDD(table, tableOptions, context, session.sparkContext)
val df =
session.sqlContext
- .createDataFrame(rdd, io.dataSchema(table.getSchema, options.isIncremental))
+ .createDataFrame(rdd, io.dataSchema(table.getSchema, tableOptions.isIncremental))
// Write the data to the backup path.
// The backup path contains the timestampMs and should not already exist.
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala
index a57931f..2874eac 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala
@@ -39,7 +39,7 @@ case class BackupOptions(
kuduMasterAddresses: String = InetAddress.getLocalHost.getCanonicalHostName,
toMs: Long = System.currentTimeMillis(),
forceFull: Boolean = BackupOptions.DefaultForceFull,
- fromMs: Long = 0,
+ fromMs: Long = BackupOptions.DefaultFromMS,
format: String = BackupOptions.DefaultFormat,
scanBatchSize: Int = BackupOptions.DefaultScanBatchSize,
scanRequestTimeoutMs: Long = BackupOptions.DefaultScanRequestTimeoutMs,
@@ -47,14 +47,15 @@ case class BackupOptions(
keepAlivePeriodMs: Long = BackupOptions.DefaultKeepAlivePeriodMs)
extends CommonOptions {
- // If not forcing a full backup and fromMs is not zero, this is an incremental backup.
+ // If not forcing a full backup and fromMs is not set, this is an incremental backup.
def isIncremental: Boolean = {
- !forceFull && fromMs != 0
+ !forceFull && fromMs != BackupOptions.DefaultFromMS
}
}
object BackupOptions {
val DefaultForceFull: Boolean = false
+ val DefaultFromMS: Long = 0
val DefaultFormat: String = "parquet"
val DefaultScanBatchSize: Int = 1024 * 1024 * 20 // 20 MiB
val DefaultScanRequestTimeoutMs: Long =
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/SessionIO.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/SessionIO.scala
index 8977174..40f513f 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/SessionIO.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/SessionIO.scala
@@ -161,7 +161,7 @@ class SessionIO(val session: SparkSession, options: CommonOptions) {
if (file.isDirectory) {
val metadataHPath = new HPath(file.getPath, MetadataFileName)
if (fs.exists(metadataHPath)) {
- val metadata = readTableMetadata(Paths.get(metadataHPath.toString))
+ val metadata = readTableMetadata(metadataHPath)
results += ((Paths.get(file.getPath.toString), metadata))
}
}
@@ -176,9 +176,8 @@ class SessionIO(val session: SparkSession, options: CommonOptions) {
* @param metadataPath the path to the metadata file.
* @return the deserialized table metadata.
*/
- private def readTableMetadata(metadataPath: Path): TableMetadataPB = {
- val hPath = new HPath(metadataPath.toString)
- val in = new InputStreamReader(fs.open(hPath), StandardCharsets.UTF_8)
+ private def readTableMetadata(metadataPath: HPath): TableMetadataPB = {
+ val in = new InputStreamReader(fs.open(metadataPath), StandardCharsets.UTF_8)
val json = CharStreams.toString(in)
in.close()
val builder = TableMetadataPB.newBuilder()
diff --git a/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala b/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala
index a707702..8e1c6a2 100644
--- a/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala
+++ b/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala
@@ -19,6 +19,7 @@ package org.apache.kudu.backup
import java.nio.file.Files
import java.nio.file.Path
import java.util
+import java.util.concurrent.TimeUnit
import com.google.common.base.Objects
import org.apache.commons.io.FileUtils
@@ -27,6 +28,7 @@ import org.apache.kudu.client._
import org.apache.kudu.ColumnSchema
import org.apache.kudu.Schema
import org.apache.kudu.spark.kudu._
+import org.apache.kudu.test.CapturingLogAppender
import org.apache.kudu.test.RandomUtils
import org.apache.kudu.util.DataGenerator.DataGeneratorBuilder
import org.apache.kudu.util.HybridTimeUtil
@@ -74,12 +76,12 @@ class TestKuduBackup extends KuduTestSuite {
val rootDir = Files.createTempDirectory("backup")
doBackup(rootDir, Seq(tableName)) // Full backup.
insertRows(table, 100, 100) // Insert more data.
- doBackup(rootDir, Seq(tableName), incremental = true) // Incremental backup.
+ doBackup(rootDir, Seq(tableName)) // Incremental backup.
// Delete rows that span the full and incremental backup.
Range(50, 150).foreach { i =>
deleteRow(i)
}
- doBackup(rootDir, Seq(tableName), incremental = true) // Incremental backup.
+ doBackup(rootDir, Seq(tableName)) // Incremental backup.
doRestore(rootDir, Seq(tableName)) // Restore all the backups.
FileUtils.deleteDirectory(rootDir.toFile)
@@ -89,6 +91,28 @@ class TestKuduBackup extends KuduTestSuite {
}
@Test
+ def TestForceIncrementalBackup() {
+ insertRows(table, 100) // Insert data into the default test table.
+ val beforeMs = getPropagatedTimestampMs
+ insertRows(table, 100, 100) // Insert more data.
+
+ val rootDir = Files.createTempDirectory("backup")
+
+ // Capture the logs to validate job internals.
+ val logs = new CapturingLogAppender()
+
+ // Force an incremental backup without a full backup.
+ // It will use a diff scan and won't check the existing dependency graph.
+ val handle = logs.attach()
+ doBackup(rootDir, Seq(tableName), fromMs = beforeMs) // Incremental backup.
+ handle.close()
+
+ assertTrue(Files.isDirectory(rootDir))
+ assertEquals(1, rootDir.toFile.list().length)
+ assertTrue(logs.getAppendedText.contains("fromMs was set"))
+ }
+
+ @Test
def testSimpleBackupAndRestoreWithSpecialCharacters() {
// Use an Impala-style table name to verify url encoding/decoding of the table name works.
val impalaTableName = "impala::default.test"
@@ -294,7 +318,10 @@ class TestKuduBackup extends KuduTestSuite {
FileUtils.deleteDirectory(rootDir.toFile)
}
- def doBackup(rootDir: Path, tableNames: Seq[String], incremental: Boolean = false): Unit = {
+ def doBackup(
+ rootDir: Path,
+ tableNames: Seq[String],
+ fromMs: Long = BackupOptions.DefaultFromMS): Unit = {
val nowMs = System.currentTimeMillis()
// Log the timestamps to simplify flaky debugging.
@@ -310,11 +337,11 @@ class TestKuduBackup extends KuduTestSuite {
// millisecond value as nowMs (after truncating the micros) the records inserted in the
// microseconds after truncation could be unread.
val backupOptions = new BackupOptions(
- tableNames,
- rootDir.toUri.toString,
- harness.getMasterAddressesAsString,
- nowMs + 1,
- incremental
+ tables = tableNames,
+ rootPath = rootDir.toUri.toString,
+ kuduMasterAddresses = harness.getMasterAddressesAsString,
+ toMs = nowMs + 1,
+ fromMs = fromMs
)
KuduBackup.run(backupOptions, ss)
}
@@ -324,4 +351,11 @@ class TestKuduBackup extends KuduTestSuite {
new RestoreOptions(tableNames, rootDir.toUri.toString, harness.getMasterAddressesAsString)
KuduRestore.run(restoreOptions, ss)
}
+
+ private def getPropagatedTimestampMs: Long = {
+ val propagatedTimestamp = harness.getClient.getLastPropagatedTimestamp
+ val physicalTimeMicros =
+ HybridTimeUtil.HTTimestampToPhysicalAndLogical(propagatedTimestamp).head
+ TimeUnit.MILLISECONDS.convert(physicalTimeMicros, TimeUnit.MICROSECONDS)
+ }
}
[kudu] 01/03: master: use AuthzProvider to generate authz tokens
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 73182b0665e336e2c864235c4aceb75db081dfe2
Author: Andrew Wong <aw...@apache.org>
AuthorDate: Mon Apr 8 17:26:59 2019 -0700
master: use AuthzProvider to generate authz tokens
This patch plugs the AuthzProvider into the master's GetTableSchema
endpoint. This allows for privileges to be returned to clients upon
calling OpenTable().
Change-Id: Ic5404d6437699bc6c7c8bb0e530b202109e8f166
Reviewed-on: http://gerrit.cloudera.org:8080/13013
Tested-by: Andrew Wong <aw...@cloudera.com>
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Reviewed-by: Hao Hao <ha...@cloudera.com>
---
src/kudu/integration-tests/CMakeLists.txt | 1 +
src/kudu/integration-tests/ts_sentry-itest.cc | 542 ++++++++++++++++++++++++++
src/kudu/master/catalog_manager.cc | 32 +-
src/kudu/master/catalog_manager.h | 8 +-
src/kudu/master/master_service.cc | 23 +-
src/kudu/sentry/mini_sentry.cc | 3 +
src/kudu/util/random_util.h | 7 +-
7 files changed, 581 insertions(+), 35 deletions(-)
diff --git a/src/kudu/integration-tests/CMakeLists.txt b/src/kudu/integration-tests/CMakeLists.txt
index fcd0b74..477231d 100644
--- a/src/kudu/integration-tests/CMakeLists.txt
+++ b/src/kudu/integration-tests/CMakeLists.txt
@@ -121,6 +121,7 @@ ADD_KUDU_TEST(token_signer-itest)
ADD_KUDU_TEST(location_assignment-itest
DATA_FILES ../scripts/assign-location.py)
ADD_KUDU_TEST(ts_recovery-itest PROCESSORS 4)
+ADD_KUDU_TEST(ts_sentry-itest NUM_SHARDS 2)
ADD_KUDU_TEST(ts_tablet_manager-itest)
ADD_KUDU_TEST(update_scan_delta_compact-test RUN_SERIAL true)
ADD_KUDU_TEST(webserver-stress-itest RUN_SERIAL true)
diff --git a/src/kudu/integration-tests/ts_sentry-itest.cc b/src/kudu/integration-tests/ts_sentry-itest.cc
new file mode 100644
index 0000000..9c8b3be
--- /dev/null
+++ b/src/kudu/integration-tests/ts_sentry-itest.cc
@@ -0,0 +1,542 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <cstdlib>
+#include <memory>
+#include <string>
+#include <thread>
+#include <type_traits>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include "kudu/client/client.h"
+#include "kudu/client/scan_batch.h"
+#include "kudu/client/schema.h"
+#include "kudu/client/shared_ptr.h"
+#include "kudu/client/write_op.h"
+#include "kudu/common/common.pb.h"
+#include "kudu/common/partial_row.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/hms/hms_client.h"
+#include "kudu/hms/mini_hms.h"
+#include "kudu/integration-tests/hms_itest-base.h"
+#include "kudu/master/sentry_authz_provider-test-base.h"
+#include "kudu/mini-cluster/external_mini_cluster.h"
+#include "kudu/security/test/mini_kdc.h"
+#include "kudu/sentry/mini_sentry.h"
+#include "kudu/sentry/sentry_client.h"
+#include "kudu/sentry/sentry_policy_service_types.h"
+#include "kudu/tablet/transactions/write_transaction.h"
+#include "kudu/thrift/client.h"
+#include "kudu/tools/data_gen_util.h"
+#include "kudu/util/barrier.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/random.h"
+#include "kudu/util/random_util.h"
+#include "kudu/util/scoped_cleanup.h"
+#include "kudu/util/status.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
+
+using kudu::client::sp::shared_ptr;
+using kudu::client::KuduClient;
+using kudu::client::KuduColumnSchema;
+using kudu::client::KuduDelete;
+using kudu::client::KuduInsert;
+using kudu::client::KuduError;
+using kudu::client::KuduUpdate;
+using kudu::client::KuduScanner;
+using kudu::client::KuduScanBatch;
+using kudu::client::KuduSchema;
+using kudu::client::KuduSchemaBuilder;
+using kudu::client::KuduSession;
+using kudu::client::KuduTable;
+using kudu::client::KuduTableAlterer;
+using kudu::client::KuduTableCreator;
+using kudu::cluster::ExternalMiniClusterOptions;
+using kudu::master::AlterRoleGrantPrivilege;
+using kudu::master::CreateRoleAndAddToGroups;
+using kudu::master::GetColumnPrivilege;
+using kudu::master::GetDatabasePrivilege;
+using kudu::master::GetTablePrivilege;
+using kudu::sentry::SentryClient;
+using kudu::tablet::WritePrivileges;
+using kudu::tablet::WritePrivilegeType;
+using kudu::tools::GenerateDataForRow;
+using sentry::TSentryGrantOption;
+using std::pair;
+using std::string;
+using std::thread;
+using std::unique_ptr;
+using std::unordered_map;
+using std::unordered_set;
+using std::vector;
+using strings::Substitute;
+
+namespace kudu {
+
+namespace {
+
+// Encapsulates the set of read and write privileges granted to a user. This is
+// used for easier composability of tests.
+//
+// Note: while full table scan privileges could also be included, leaving this
+// out simplifies the below tests, which are aimed at testing functionality of
+// privileges granted by authz tokens end-to-end; privilege-checking for
+// different actions is tested in more depth elsewhere.
+struct RWPrivileges {
+ // The set of write privileges a user may be granted for a table.
+ WritePrivileges table_write_privileges;
+
+ // The set of column names that the user is authorized to scan.
+ unordered_set<string> column_scan_privileges;
+};
+
+const WritePrivileges kFullPrivileges = {
+ WritePrivilegeType::INSERT,
+ WritePrivilegeType::UPDATE,
+ WritePrivilegeType::DELETE,
+};
+
+// Returns a randomly generated set of read and write privileges, ensuring that
+// it contains at least one read and one write privilege.
+RWPrivileges GeneratePrivileges(const unordered_set<string>& all_cols, ThreadSafeRandom* prng) {
+ WritePrivilegeType write_privilege =
+ SelectRandomElement<WritePrivileges, WritePrivilegeType, ThreadSafeRandom>(
+ kFullPrivileges, prng);
+ vector<string> scan_privileges =
+ SelectRandomSubset<unordered_set<string>, string, ThreadSafeRandom>(
+ all_cols, /*min_to_return*/1, prng);
+ RWPrivileges privileges;
+ privileges.table_write_privileges = WritePrivileges({ write_privilege });
+ privileges.column_scan_privileges =
+ unordered_set<string>(scan_privileges.begin(), scan_privileges.end());
+ return privileges;
+}
+
+// Returns the complentary set of privileges to 'orig_privileges'. This is
+// useful for generating operations that should fail, if a user is granted the
+// privileges in 'orig_privileges'.
+RWPrivileges ComplementaryPrivileges(const unordered_set<string>& all_cols,
+ const RWPrivileges& orig_privileges) {
+ RWPrivileges privileges;
+ for (const auto& wp : kFullPrivileges) {
+ if (!ContainsKey(orig_privileges.table_write_privileges, wp)) {
+ InsertOrDie(&privileges.table_write_privileges, wp);
+ }
+ }
+ for (const auto& col : all_cols) {
+ if (!ContainsKey(orig_privileges.column_scan_privileges, col)) {
+ InsertOrDie(&privileges.column_scan_privileges, col);
+ }
+ }
+ return privileges;
+}
+
+// Performs a write operation to 'table' that should be allowed based on the
+// privileges in 'write_privileges', using 'prng' to determine the operation.
+Status PerformWrite(const WritePrivileges& write_privileges,
+ ThreadSafeRandom* prng,
+ KuduTable* table) {
+ WritePrivilegeType op_type =
+ SelectRandomElement<WritePrivileges, WritePrivilegeType, ThreadSafeRandom>(
+ write_privileges, prng);
+ shared_ptr<KuduSession> session = table->client()->NewSession();
+ const auto unwrap_session_error = [&session] (Status s) {
+ if (s.IsIOError()) {
+ vector<KuduError*> errors;
+ session->GetPendingErrors(&errors, nullptr);
+ ElementDeleter deleter(&errors);
+ CHECK_EQ(1, errors.size());
+ return errors[0]->status();
+ }
+ return s;
+ };
+ // Note: we could test UPSERTs, but it complicates the logic, and UPSERTs are
+ // tested elsewhere anyway.
+ switch (op_type) {
+ case WritePrivilegeType::INSERT: {
+ unique_ptr<KuduInsert> ins(table->NewInsert());
+ GenerateDataForRow(table->schema(), prng->Next32(), prng, ins->mutable_row());
+ return unwrap_session_error(session->Apply(ins.release()));
+ }
+ break;
+ case WritePrivilegeType::UPDATE: {
+ unique_ptr<KuduUpdate> upd(table->NewUpdate());
+ GenerateDataForRow(table->schema(), prng->Next32(), prng, upd->mutable_row());
+ return unwrap_session_error(session->Apply(upd.release()));
+ }
+ break;
+ case WritePrivilegeType::DELETE: {
+ unique_ptr<KuduDelete> del(table->NewDelete());
+ KuduPartialRow* row = del->mutable_row();
+ RETURN_NOT_OK(row->SetInt32(0, prng->Next32()));
+ return unwrap_session_error(session->Apply(del.release()));
+ }
+ break;
+ }
+ return Status::OK();
+}
+
+// Performs a scan operation to 'table' that should be allowed if the user is
+// granted scan privileges on all columns in 'columns'. If provided, uses
+// 'prng' to select a subset of rows to scan; otherwise uses all columns.
+Status PerformScan(const unordered_set<string>& columns,
+ ThreadSafeRandom* prng,
+ KuduTable* table) {
+ vector<string> cols_to_scan = prng ?
+ SelectRandomSubset<unordered_set<string>, string, ThreadSafeRandom>(
+ columns, /*min_to_return*/1, prng) :
+ vector<string>(columns.begin(), columns.end());
+ KuduScanner scanner(table);
+ RETURN_NOT_OK(scanner.SetTimeoutMillis(30000));
+ RETURN_NOT_OK(scanner.SetProjectedColumnNames(cols_to_scan));
+ RETURN_NOT_OK(scanner.Open());
+ while (scanner.HasMoreRows()) {
+ KuduScanBatch batch;
+ RETURN_NOT_OK(scanner.NextBatch(&batch));
+ }
+ return Status::OK();
+}
+
+// Performs an action that should be allowed with the given set of
+// privileges.
+Status PerformAction(const RWPrivileges& privileges,
+ ThreadSafeRandom* prng, KuduTable* table) {
+ bool can_write = !privileges.table_write_privileges.empty();
+ bool can_scan = !privileges.column_scan_privileges.empty();
+ CHECK(can_write || can_scan);
+ // If the user can scan and write, flip a coin for what to do. Otherwise,
+ // just perform whichever it can.
+ bool should_write = (can_write && can_scan && rand() % 2 == 0) ||
+ (can_write && !can_scan);
+ if (should_write) {
+ CHECK(can_write);
+ RETURN_NOT_OK(PerformWrite(privileges.table_write_privileges, prng, table));
+ } else {
+ CHECK(can_scan);
+ RETURN_NOT_OK(PerformScan(privileges.column_scan_privileges, prng, table));
+ }
+ return Status::OK();
+}
+
+} // anonymous namespace
+
+// These tests will use the HMS and Sentry, and thus, are very slow.
+// SKIP_IF_SLOW_NOT_ALLOWED() should be the very first thing called in the body
+// of every test based on this test class.
+class TSSentryITest : public HmsITestBase {
+ public:
+ // Note: groups and users therein are statically provided to MiniSentry (see
+ // mini_sentry.cc). We expect Sentry to be aware of users "user[0-2]".
+ static constexpr int kNumUsers = 3;
+ static constexpr const char* kAdminGroup = "admin";
+
+ static constexpr int kNumTables = 3;
+ static constexpr int kNumColsPerTable = 3;
+ static constexpr const char* kDb = "db";
+ static constexpr const char* kTablePrefix = "table";
+ static constexpr const char* kAdminRole = "kudu-admin";
+
+ static constexpr int kAuthzTokenTTLSecs = 1;
+ static constexpr int kAuthzCacheTTLMultiplier = 3;
+
+ void SetUp() override {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+ for (int u = 0; u < kNumUsers; u++) {
+ users_.emplace_back(Substitute("user$0", u));
+ }
+
+ ExternalMiniClusterOptions opts;
+ opts.enable_kerberos = true;
+ opts.enable_sentry = true;
+ opts.hms_mode = HmsMode::ENABLE_METASTORE_INTEGRATION;
+ // Set a low token timeout so we can ensure retries are working properly.
+ opts.extra_master_flags.emplace_back(Substitute("--authz_token_validity_seconds=$0",
+ kAuthzTokenTTLSecs));
+ opts.extra_master_flags.emplace_back(Substitute("--sentry_privileges_cache_ttl_factor=$0",
+ kAuthzCacheTTLMultiplier));
+ // In addition to our users, we will be using the "kudu" user to perform
+ // administrative tasks like creating tables.
+ opts.extra_master_flags.emplace_back(
+ Substitute("--user_acl=kudu,$0", JoinStrings(users_, ",")));
+ opts.extra_tserver_flags.emplace_back(
+ Substitute("--user_acl=$0", JoinStrings(users_, ",")));
+ opts.extra_tserver_flags.emplace_back("--tserver_enforce_access_control=true");
+ NO_FATALS(StartClusterWithOpts(std::move(opts)));
+ ASSERT_OK(cluster_->kdc()->CreateUserPrincipal("kudu"));
+ ASSERT_OK(cluster_->kdc()->Kinit("kudu"));
+
+ // Set up the HMS client so we can set up a database.
+ thrift::ClientOptions hms_opts;
+ hms_opts.enable_kerberos = true;
+ hms_opts.service_principal = "hive";
+ hms_client_.reset(new hms::HmsClient(cluster_->hms()->address(), hms_opts));
+ ASSERT_OK(hms_client_->Start());
+
+ // Set up the Sentry client so we can set up privileges.
+ thrift::ClientOptions sentry_opts;
+ sentry_opts.enable_kerberos = true;
+ sentry_opts.service_principal = "sentry";
+ sentry_client_.reset(new SentryClient(cluster_->sentry()->address(), sentry_opts));
+ ASSERT_OK(sentry_client_->Start());
+ ASSERT_OK(CreateRoleAndAddToGroups(sentry_client_.get(), kAdminRole, kAdminGroup));
+ ASSERT_OK(AlterRoleGrantPrivilege(sentry_client_.get(), kAdminRole,
+ GetDatabasePrivilege(kDb, "ALL", TSentryGrantOption::DISABLED)));
+
+ // Create the database in the HMS.
+ ASSERT_OK(CreateDatabase(kDb));
+
+ // Create a client as the "kudu" user, who now has admin privileges.
+ ASSERT_OK(cluster_->CreateClient(nullptr, &admin_client_));
+
+ // Finally populate a set of column names to use for our tables.
+ for (int i = 0; i < kNumColsPerTable; i++) {
+ cols_.emplace_back(Substitute("col$0", i));
+ }
+ }
+
+ // Creates a table named 'table_ident' with 'kNumColsPerTable' columns.
+ Status CreateTable(const string& table_ident) {
+ KuduSchema schema;
+ KuduSchemaBuilder b;
+ auto iter = cols_.begin();
+ b.AddColumn(*iter++)->Type(KuduColumnSchema::INT32)->NotNull()->PrimaryKey();
+ while (iter < cols_.end()) {
+ b.AddColumn(*iter++)->Type(KuduColumnSchema::INT32);
+ }
+ RETURN_NOT_OK(b.Build(&schema));
+ unique_ptr<KuduTableCreator> table_creator(client_->NewTableCreator());
+ return table_creator->table_name(table_ident)
+ .schema(&schema)
+ .set_range_partition_columns({ "col0" })
+ .num_replicas(1)
+ .Create();
+ }
+
+ void TearDown() override {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+ HmsITestBase::TearDown();
+ }
+
+ protected:
+ // A Sentry client with which to grant privileges.
+ unique_ptr<SentryClient> sentry_client_;
+
+ // Kudu client with which to perform admin operations.
+ shared_ptr<KuduClient> admin_client_;
+
+ // A list of users that may try to do things.
+ vector<string> users_;
+
+ // A list of columns that each table should have.
+ vector<string> cols_;
+};
+
+// Tests authorizing read and write operations coming from multiple concurrent
+// users for multiple tables.
+TEST_F(TSSentryITest, TestReadsAndWrites) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
+ // First, set up the tables.
+ vector<string> tables;
+ for (int i = 0; i < kNumTables; i++) {
+ string table_name = Substitute("$0$1", kTablePrefix, i);
+ ASSERT_OK(CreateTable(Substitute("$0.$1", kDb, table_name)));
+ tables.emplace_back(std::move(table_name));
+ }
+
+ // Keep track of the privileges that each user has been granted and not been
+ // granted per table.
+ typedef pair<RWPrivileges, RWPrivileges> GrantedNotGrantedPrivileges;
+ typedef unordered_map<string, GrantedNotGrantedPrivileges> TableNameToPrivileges;
+ unordered_map<string, TableNameToPrivileges> user_to_privileges;
+
+ // Set up a bunch of clients for each user.
+ unordered_map<string, vector<shared_ptr<KuduClient>>> user_to_clients;
+ ThreadSafeRandom prng(SeedRandom());
+ unordered_set<string> cols(cols_.begin(), cols_.end());
+ static constexpr int kNumClientsPerUser = 4;
+ for (int i = 0; i < kNumUsers; i++) {
+ const string& user = users_[i];
+ // Register the user with the KDC, and add a role to the user's group
+ // (provided to MiniSentry in mini_sentry.cc).
+ ASSERT_OK(cluster_->kdc()->CreateUserPrincipal(user));
+ ASSERT_OK(cluster_->kdc()->Kinit(user));
+ const string role = Substitute("role$0", i);
+ ASSERT_OK(CreateRoleAndAddToGroups(sentry_client_.get(), role, Substitute("group$0", i)));
+
+ // Set up multiple clients for each user.
+ vector<shared_ptr<KuduClient>> clients;
+ for (int i = 0; i < kNumClientsPerUser; i++) {
+ shared_ptr<KuduClient> client;
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client));
+ clients.emplace_back(std::move(client));
+ }
+ EmplaceOrDie(&user_to_clients, user, std::move(clients));
+
+ // Generate privileges for each user for every table, and grant the
+ // appropriate Sentry privileges.
+ TableNameToPrivileges table_to_privileges;
+ for (const string& table_name : tables) {
+ RWPrivileges granted_privileges = GeneratePrivileges(cols, &prng);
+ for (const auto& wp : granted_privileges.table_write_privileges) {
+ ASSERT_OK(AlterRoleGrantPrivilege(sentry_client_.get(), role,
+ GetTablePrivilege(kDb, table_name, WritePrivilegeToString(wp))));
+ }
+ for (const auto& col : granted_privileges.column_scan_privileges) {
+ ASSERT_OK(AlterRoleGrantPrivilege(sentry_client_.get(), role,
+ GetColumnPrivilege(kDb, table_name, col, "SELECT")));
+ }
+ RWPrivileges not_granted_privileges = ComplementaryPrivileges(cols, granted_privileges);
+ InsertOrDie(&table_to_privileges, table_name,
+ { std::move(granted_privileges), std::move(not_granted_privileges) });
+ }
+ EmplaceOrDie(&user_to_privileges, user, std::move(table_to_privileges));
+ }
+
+ // In parallel, have each user's clients perform a series of operations on a
+ // table for some extended period of time (longer than the token timeout). Do
+ // this for a few tables for each client.
+ static constexpr int kNumOpPeriods = 3;
+ static const MonoDelta kPeriodTime = MonoDelta::FromSeconds(kAuthzTokenTTLSecs * 3);
+ vector<thread> threads;
+ Barrier b(kNumUsers * kNumClientsPerUser);
+ SCOPED_CLEANUP({
+ for (auto& t : threads) {
+ t.join();
+ }
+ });
+ for (const string& user : users_) {
+ // Start a thread for every user that performs a bunch of operations.
+ const auto* const table_to_privileges = FindOrNull(user_to_privileges, user);
+ for (const auto& client_sp : FindOrDie(user_to_clients, user)) {
+ KuduClient* client = client_sp.get();
+ threads.emplace_back([client, table_to_privileges, &b, &tables, &prng] {
+ b.Wait();
+ // Perform a bunch of operations, switching back and forth between
+ // different tables to ensure a client uses the appropriate privileges.
+ for (int i = 0; i < kNumOpPeriods; i++) {
+ const auto& table_name =
+ SelectRandomElement<vector<string>, string, ThreadSafeRandom>(tables, &prng);
+ shared_ptr<KuduTable> table;
+ ASSERT_OK(client->OpenTable(Substitute("$0.$1", kDb, table_name), &table));
+ const MonoTime end_time = MonoTime::Now() + kPeriodTime;
+ while (MonoTime::Now() < end_time) {
+ const auto& privileges = FindOrDie(*table_to_privileges, table_name);
+ const auto& granted_privileges = privileges.first;
+ const auto& non_granted_privileges = privileges.second;
+ // Perform a permitted operation. We might not get an OK status if
+ // e.g. we're inserting a row that already exists, but the operation
+ // should always be permitted.
+ Status s = PerformAction(granted_privileges, &prng, table.get());
+ ASSERT_FALSE(s.IsNotAuthorized()) << s.ToString();
+ ASSERT_STR_NOT_CONTAINS(s.ToString(), "not authorized");
+
+ // Now perform an operation based on the privileges we _don't_ have;
+ // this should always yield authorization errors.
+ s = PerformAction(non_granted_privileges, &prng, table.get());
+ ASSERT_TRUE(s.IsRemoteError()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(), "not authorized");
+ }
+ }
+ });
+ }
+ }
+}
+
+// Test for a couple of scenarios related to alter tables.
+TEST_F(TSSentryITest, TestAlters) {
+ SKIP_IF_SLOW_NOT_ALLOWED();
+
+ static const string kTableName = "table";
+ const string table_ident = Substitute("$0.$1", kDb, kTableName);
+ ASSERT_OK(CreateTable(table_ident));
+
+ const string user = "user0";
+ ASSERT_OK(cluster_->kdc()->CreateUserPrincipal(user));
+ ASSERT_OK(cluster_->kdc()->Kinit(user));
+ const string role = "role0";
+ ASSERT_OK(CreateRoleAndAddToGroups(sentry_client_.get(), role, "group0"));
+
+ shared_ptr<KuduClient> user_client;
+ ASSERT_OK(cluster_->CreateClient(nullptr, &user_client));
+
+ // Note: we only need privileges on the metadata for OpenTable() calls.
+ // METADATA isn't a first-class Sentry privilege and won't get carried over
+ // on table rename, so we just grant INSERT privileges.
+ ASSERT_OK(AlterRoleGrantPrivilege(sentry_client_.get(), role,
+ GetTablePrivilege(kDb, kTableName, "INSERT")));
+
+ // 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(AlterRoleGrantPrivilege(sentry_client_.get(), role,
+ GetColumnPrivilege(kDb, kTableName, new_column, "SELECT")));
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(table_ident));
+ table_alterer->AddColumn(new_column)->Type(KuduColumnSchema::INT32);
+ ASSERT_OK(table_alterer->Alter());
+ }
+ shared_ptr<KuduTable> table;
+ ASSERT_OK(user_client->OpenTable(table_ident, &table));
+ ASSERT_OK(PerformScan({ new_column }, /*prng=*/nullptr, table.get()));
+
+ // Now create another column and grant the user privileges for that column.
+ // Since privileges are cached, even though we've granted privileges, clients
+ // will use the cached privilege and not be authorized for a bit.
+ const string another_column = Substitute("col$0", kNumColsPerTable + 1);
+ ASSERT_OK(AlterRoleGrantPrivilege(sentry_client_.get(), role,
+ GetColumnPrivilege(kDb, kTableName, another_column, "SELECT")));
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(table_ident));
+ table_alterer->AddColumn(another_column)->Type(KuduColumnSchema::INT32);
+ ASSERT_OK(table_alterer->Alter());
+ }
+ ASSERT_OK(user_client->OpenTable(table_ident, &table));
+ Status s = PerformScan({ another_column }, /*prng=*/nullptr, table.get());
+ ASSERT_TRUE(s.IsRemoteError()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(), "not authorized");
+
+ // Wait the full duration of the cache TTL, and an additional full token TTL.
+ // This ensures that the client's token will expire we will get a new one
+ // with the most up-to-date privileges from Sentry.
+ SleepFor(MonoDelta::FromSeconds(kAuthzTokenTTLSecs * (1 + kAuthzCacheTTLMultiplier)));
+ ASSERT_OK(PerformScan({ another_column }, /*prng=*/nullptr, table.get()));
+
+ // Now rename the table to something else. There shouldn't be any privileges
+ // cached for the newly-renamed table, so we should immediately be able to
+ // scan it.
+ const string new_table_ident = Substitute("$0.$1", kDb, "newtable");
+ {
+ unique_ptr<KuduTableAlterer> table_alterer(client_->NewTableAlterer(table_ident));
+ table_alterer->RenameTo(new_table_ident);
+ ASSERT_OK(table_alterer->Alter());
+ }
+ ASSERT_OK(user_client->OpenTable(new_table_ident, &table));
+ ASSERT_OK(PerformScan({ another_column }, nullptr, table.get()));
+}
+
+} // namespace kudu
diff --git a/src/kudu/master/catalog_manager.cc b/src/kudu/master/catalog_manager.cc
index d59c45a..c5b9dcd 100644
--- a/src/kudu/master/catalog_manager.cc
+++ b/src/kudu/master/catalog_manager.cc
@@ -286,6 +286,7 @@ using kudu::rpc::RpcContext;
using kudu::security::Cert;
using kudu::security::DataFormat;
using kudu::security::PrivateKey;
+using kudu::security::TablePrivilegePB;
using kudu::security::TokenSigner;
using kudu::security::TokenSigningPrivateKey;
using kudu::security::TokenSigningPrivateKeyPB;
@@ -2725,7 +2726,8 @@ Status CatalogManager::IsAlterTableDone(const IsAlterTableDoneRequestPB* req,
Status CatalogManager::GetTableSchema(const GetTableSchemaRequestPB* req,
GetTableSchemaResponsePB* resp,
- optional<const string&> user) {
+ optional<const string&> user,
+ const TokenSigner* token_signer) {
leader_lock_.AssertAcquiredForReading();
RETURN_NOT_OK(CheckOnline());
@@ -2742,15 +2744,25 @@ Status CatalogManager::GetTableSchema(const GetTableSchemaRequestPB* req,
&table, &l));
RETURN_NOT_OK(CheckIfTableDeletedOrNotRunning(&l, resp));
- if (l.data().pb.has_fully_applied_schema()) {
- // An AlterTable is in progress; fully_applied_schema is the last
- // schema that has reached every TS.
- CHECK_EQ(SysTablesEntryPB::ALTERING, l.data().pb.state());
- resp->mutable_schema()->CopyFrom(l.data().pb.fully_applied_schema());
- } else {
- // There's no AlterTable, the regular schema is "fully applied".
- resp->mutable_schema()->CopyFrom(l.data().pb.schema());
- }
+ // If fully_applied_schema is set, use it, since an alter is in progress.
+ CHECK(!l.data().pb.has_fully_applied_schema() ||
+ (l.data().pb.state() == SysTablesEntryPB::ALTERING));
+ const SchemaPB& schema_pb = l.data().pb.has_fully_applied_schema() ?
+ l.data().pb.fully_applied_schema() : l.data().pb.schema();
+
+ if (token_signer && user) {
+ TablePrivilegePB table_privilege;
+ table_privilege.set_table_id(table->id());
+ RETURN_NOT_OK(
+ SetupError(authz_provider_->FillTablePrivilegePB(l.data().name(), *user, schema_pb,
+ &table_privilege),
+ resp, MasterErrorPB::UNKNOWN_ERROR));
+ security::SignedTokenPB authz_token;
+ RETURN_NOT_OK(token_signer->GenerateAuthzToken(
+ *user, std::move(table_privilege), &authz_token));
+ *resp->mutable_authz_token() = std::move(authz_token);
+ }
+ resp->mutable_schema()->CopyFrom(schema_pb);
resp->set_num_replicas(l.data().pb.num_replicas());
resp->set_table_id(table->id());
resp->mutable_partition_schema()->CopyFrom(l.data().pb.partition_schema());
diff --git a/src/kudu/master/catalog_manager.h b/src/kudu/master/catalog_manager.h
index 88a0227..94dffc5 100644
--- a/src/kudu/master/catalog_manager.h
+++ b/src/kudu/master/catalog_manager.h
@@ -84,6 +84,7 @@ class RpcContext;
namespace security {
class Cert;
class PrivateKey;
+class TokenSigner;
class TokenSigningPublicKeyPB;
} // namespace security
@@ -576,10 +577,13 @@ class CatalogManager : public tserver::TabletReplicaLookupIf {
boost::optional<const std::string&> user);
// Get the information about the specified table. If 'user' is provided,
- // checks that the user is authorized to get such information.
+ // checks that the user is authorized to get such information. If a token
+ // signer is provided (e.g. authz token generation is enabled), an authz
+ // token will be attached to the response.
Status GetTableSchema(const GetTableSchemaRequestPB* req,
GetTableSchemaResponsePB* resp,
- boost::optional<const std::string&> user);
+ boost::optional<const std::string&> user,
+ const security::TokenSigner* token_signer);
// List all the running tables. If 'user' is provided, checks that the user
// is authorized to get such information, otherwise, only list the tables that
diff --git a/src/kudu/master/master_service.cc b/src/kudu/master/master_service.cc
index bcb244f..5b44196 100644
--- a/src/kudu/master/master_service.cc
+++ b/src/kudu/master/master_service.cc
@@ -431,33 +431,14 @@ void MasterServiceImpl::GetTableSchema(const GetTableSchemaRequestPB* req,
}
Status s = server_->catalog_manager()->GetTableSchema(
- req, resp, make_optional<const string&>(rpc->remote_user().username()));
+ req, resp, make_optional<const string&>(rpc->remote_user().username()),
+ FLAGS_master_support_authz_tokens ? server_->token_signer() : nullptr);
CheckRespErrorOrSetUnknown(s, resp);
if (resp->has_error()) {
// If there was an application error, respond to the RPC.
rpc->RespondSuccess();
return;
}
-
- // TODO(awong): fill this token in with actual privileges from the
- // appropriate AuthzProvider. For now, assume the user has all privileges
- // for the table.
- if (PREDICT_TRUE(FLAGS_master_support_authz_tokens)) {
- SignedTokenPB authz_token;
- TablePrivilegePB table_privilege;
- table_privilege.set_table_id(resp->table_id());
- table_privilege.set_scan_privilege(true);
- table_privilege.set_insert_privilege(true);
- table_privilege.set_update_privilege(true);
- table_privilege.set_delete_privilege(true);
- s = server_->token_signer()->GenerateAuthzToken(rpc->remote_user().username(),
- std::move(table_privilege), &authz_token);
- if (!s.ok()) {
- rpc->RespondFailure(s);
- return;
- }
- *resp->mutable_authz_token() = std::move(authz_token);
- }
rpc->RespondSuccess();
}
diff --git a/src/kudu/sentry/mini_sentry.cc b/src/kudu/sentry/mini_sentry.cc
index caecfa9..cb3bbe1 100644
--- a/src/kudu/sentry/mini_sentry.cc
+++ b/src/kudu/sentry/mini_sentry.cc
@@ -329,6 +329,9 @@ test-admin=admin
test-user=user
kudu=admin
joe-interloper=""
+user0=group0
+user1=group1
+user2=group2
)";
RETURN_NOT_OK(WriteStringToFile(Env::Default(), kUsers, users_ini_path));
diff --git a/src/kudu/util/random_util.h b/src/kudu/util/random_util.h
index 9dfc510..a607051 100644
--- a/src/kudu/util/random_util.h
+++ b/src/kudu/util/random_util.h
@@ -50,10 +50,13 @@ T SelectRandomElement(const Container& c, Rand* r) {
}
// Returns a randomly-selected subset from the container.
+//
+// The results are not stored in a randomized order: the order of results will
+// match their order in the input collection.
template <typename Container, typename T, typename Rand>
std::vector<T> SelectRandomSubset(const Container& c, int min_to_return, Rand* r) {
- CHECK_GT(c.size(), min_to_return);
- int num_to_return = min_to_return + r->Uniform(c.size() - min_to_return);
+ CHECK_GE(c.size(), min_to_return);
+ int num_to_return = min_to_return + r->Uniform(1 + c.size() - min_to_return);
std::vector<T> rand_list;
ReservoirSample(c, num_to_return, std::set<T>{}, r, &rand_list);
return rand_list;
[kudu] 02/03: [sentry] enable sentry integration for master stress
test
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 6be025faa8f2e36d31bcc6c4e4f65324d07a3926
Author: Hao Hao <ha...@cloudera.com>
AuthorDate: Wed Mar 27 23:51:58 2019 -0700
[sentry] enable sentry integration for master stress test
This patch adds more coverage for master authorization enforcement via
enabling Sentry integration for master stress and failover tests.
I looped each of the following tests 2000 times:
alter_table-randomized-test: http://dist-test.cloudera.org/job?job_id=hao.hao.1555900144.107645
master-stress-test: http://dist-test.cloudera.org/job?job_id=hao.hao.1555897983.99429
master_failover-itest: http://dist-test.cloudera.org/job?job_id=hao.hao.1555893155.81719
All of the failures are due to known flakiness (KUDU-2621, KUDU-2774,
KUDU-2779, and KUDU-1358).
Change-Id: Ic48aa1bfd0947c645bb81137bb34e6cdfc088cf4
Reviewed-on: http://gerrit.cloudera.org:8080/12877
Reviewed-by: Andrew Wong <aw...@cloudera.com>
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Tested-by: Hao Hao <ha...@cloudera.com>
---
src/kudu/integration-tests/CMakeLists.txt | 4 +-
.../alter_table-randomized-test.cc | 63 +++++++----
src/kudu/integration-tests/cluster_itest_util.cc | 119 +++++++++++++--------
src/kudu/integration-tests/cluster_itest_util.h | 12 +++
src/kudu/integration-tests/hms_itest-base.h | 12 +--
src/kudu/integration-tests/master-stress-test.cc | 31 ++++--
.../integration-tests/master_failover-itest.cc | 56 ++++++----
src/kudu/integration-tests/master_hms-itest.cc | 4 +-
src/kudu/integration-tests/master_sentry-itest.cc | 10 +-
9 files changed, 204 insertions(+), 107 deletions(-)
diff --git a/src/kudu/integration-tests/CMakeLists.txt b/src/kudu/integration-tests/CMakeLists.txt
index 477231d..97968b2 100644
--- a/src/kudu/integration-tests/CMakeLists.txt
+++ b/src/kudu/integration-tests/CMakeLists.txt
@@ -56,7 +56,7 @@ SET_KUDU_TEST_LINK_LIBS(itest_util)
ADD_KUDU_TEST(all_types-itest
PROCESSORS 4
NUM_SHARDS 8)
-ADD_KUDU_TEST(alter_table-randomized-test NUM_SHARDS 2 PROCESSORS 4)
+ADD_KUDU_TEST(alter_table-randomized-test NUM_SHARDS 3 PROCESSORS 4)
ADD_KUDU_TEST(alter_table-test PROCESSORS 3)
ADD_KUDU_TEST(auth_token_expire-itest)
ADD_KUDU_TEST(authz_token-itest PROCESSORS 2)
@@ -91,7 +91,7 @@ ADD_KUDU_TEST_DEPENDENCIES(master_migration-itest
kudu)
ADD_KUDU_TEST(master_replication-itest)
ADD_KUDU_TEST(master_sentry-itest RUN_SERIAL true NUM_SHARDS 8 PROCESSORS 4)
-ADD_KUDU_TEST(master-stress-test RUN_SERIAL true)
+ADD_KUDU_TEST(master-stress-test RUN_SERIAL true NUM_SHARDS 3)
ADD_KUDU_TEST(multidir_cluster-itest)
ADD_KUDU_TEST(open-readonly-fs-itest PROCESSORS 4)
ADD_KUDU_TEST(raft_config_change-itest)
diff --git a/src/kudu/integration-tests/alter_table-randomized-test.cc b/src/kudu/integration-tests/alter_table-randomized-test.cc
index cfe2847..200b8fc 100644
--- a/src/kudu/integration-tests/alter_table-randomized-test.cc
+++ b/src/kudu/integration-tests/alter_table-randomized-test.cc
@@ -43,32 +43,33 @@
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/integration-tests/cluster_itest_util.h"
#include "kudu/integration-tests/cluster_verifier.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
+#include "kudu/sentry/mini_sentry.h"
#include "kudu/util/monotime.h"
#include "kudu/util/random.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
-namespace kudu {
-
-using client::KuduClient;
-using client::KuduColumnSchema;
-using client::KuduColumnStorageAttributes;
-using client::KuduError;
-using client::KuduScanner;
-using client::KuduSchema;
-using client::KuduSchemaBuilder;
-using client::KuduSession;
-using client::KuduTable;
-using client::KuduTableAlterer;
-using client::KuduTableCreator;
-using client::KuduValue;
-using client::KuduWriteOperation;
-using client::sp::shared_ptr;
-using cluster::ExternalMiniCluster;
-using cluster::ExternalMiniClusterOptions;
+using kudu::client::KuduClient;
+using kudu::client::KuduColumnSchema;
+using kudu::client::KuduColumnStorageAttributes;
+using kudu::client::KuduError;
+using kudu::client::KuduScanner;
+using kudu::client::KuduSchema;
+using kudu::client::KuduSchemaBuilder;
+using kudu::client::KuduSession;
+using kudu::client::KuduTable;
+using kudu::client::KuduTableAlterer;
+using kudu::client::KuduTableCreator;
+using kudu::client::KuduValue;
+using kudu::client::KuduWriteOperation;
+using kudu::client::sp::shared_ptr;
+using kudu::cluster::ExternalMiniCluster;
+using kudu::cluster::ExternalMiniClusterOptions;
+using kudu::itest::SentryMode;
using std::make_pair;
using std::map;
using std::pair;
@@ -78,6 +79,8 @@ using std::vector;
using strings::Substitute;
using strings::SubstituteAndAppend;
+namespace kudu {
+
const char* kTableName = "default.test_table";
const int kMaxColumns = 30;
const uint32_t kMaxRangePartitions = 32;
@@ -94,15 +97,19 @@ const vector <KuduColumnStorageAttributes::EncodingType> kInt32Encodings =
const vector<int32_t> kBlockSizes = {0, 2 * 1024 * 1024,
4 * 1024 * 1024, 8 * 1024 * 1024};
+// Parameterized based on HmsMode and whether or not to enable Sentry integration.
class AlterTableRandomized : public KuduTest,
- public ::testing::WithParamInterface<HmsMode> {
+ public ::testing::WithParamInterface<pair<HmsMode, SentryMode>> {
public:
void SetUp() override {
KuduTest::SetUp();
ExternalMiniClusterOptions opts;
opts.num_tablet_servers = 3;
- opts.hms_mode = GetParam();
+ opts.hms_mode = std::get<0>(GetParam());
+ bool enable_sentry = (std::get<1>(GetParam()) == SentryMode::ENABLED);
+ opts.enable_sentry = enable_sentry;
+ opts.enable_kerberos = enable_sentry;
// This test produces tables with lots of columns. With container preallocation,
// we end up using quite a bit of disk space. So, we disable it.
opts.extra_tserver_flags.emplace_back("--log_container_preallocate_bytes=0");
@@ -110,6 +117,10 @@ class AlterTableRandomized : public KuduTest,
ASSERT_OK(cluster_->Start());
ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
+ if (enable_sentry) {
+ itest::SetupAdministratorPrivileges(cluster_->kdc(),
+ cluster_->sentry()->address());
+ }
}
void TearDown() override {
@@ -140,9 +151,15 @@ class AlterTableRandomized : public KuduTest,
shared_ptr<KuduClient> client_;
};
-// Run the test with the HMS integration enabled and disabled.
-INSTANTIATE_TEST_CASE_P(HmsConfigurations, AlterTableRandomized,
- ::testing::Values(HmsMode::NONE, HmsMode::ENABLE_METASTORE_INTEGRATION));
+// Run the test with the HMS/Sentry integration enabled and disabled. Sentry integration
+// should be only enabled when HMS integration is enabled.
+INSTANTIATE_TEST_CASE_P(HmsSentryConfigurations, AlterTableRandomized, ::testing::ValuesIn(
+ vector<pair<HmsMode, SentryMode>> {
+ { HmsMode::NONE, SentryMode::DISABLED },
+ { HmsMode::ENABLE_METASTORE_INTEGRATION, SentryMode::DISABLED },
+ { HmsMode::ENABLE_METASTORE_INTEGRATION, SentryMode::ENABLED },
+ }
+));
struct RowState {
// We use this special value to denote NULL values.
diff --git a/src/kudu/integration-tests/cluster_itest_util.cc b/src/kudu/integration-tests/cluster_itest_util.cc
index 49a22cb..bdaf403 100644
--- a/src/kudu/integration-tests/cluster_itest_util.cc
+++ b/src/kudu/integration-tests/cluster_itest_util.cc
@@ -42,9 +42,14 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/master.proxy.h"
+#include "kudu/master/sentry_authz_provider-test-base.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/rpc/rpc_header.pb.h"
+#include "kudu/security/test/mini_kdc.h"
+#include "kudu/sentry/sentry_client.h"
+#include "kudu/sentry/sentry_policy_service_types.h"
#include "kudu/tablet/tablet.pb.h"
+#include "kudu/thrift/client.h"
#include "kudu/tserver/tablet_copy.proxy.h"
#include "kudu/tserver/tablet_server_test_util.h"
#include "kudu/tserver/tserver_admin.pb.h"
@@ -62,41 +67,52 @@
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
-namespace kudu {
-namespace itest {
-
+using ::sentry::TSentryGrantOption;
+using ::sentry::TSentryPrivilege;
using boost::optional;
-using client::KuduSchema;
-using client::KuduSchemaBuilder;
-using consensus::BulkChangeConfigRequestPB;
-using consensus::ChangeConfigRequestPB;
-using consensus::ChangeConfigResponsePB;
-using consensus::ConsensusStatePB;
-using consensus::CountVoters;
-using consensus::EXCLUDE_HEALTH_REPORT;
-using consensus::GetConsensusStateRequestPB;
-using consensus::GetConsensusStateResponsePB;
-using consensus::GetLastOpIdRequestPB;
-using consensus::GetLastOpIdResponsePB;
-using consensus::IncludeHealthReport;
-using consensus::LeaderStepDownRequestPB;
-using consensus::LeaderStepDownResponsePB;
-using consensus::OpId;
-using consensus::OpIdType;
-using consensus::RaftPeerPB;
-using consensus::RunLeaderElectionResponsePB;
-using consensus::RunLeaderElectionRequestPB;
-using consensus::VoteRequestPB;
-using consensus::VoteResponsePB;
-using consensus::kInvalidOpIdIndex;
-using master::ListTabletServersResponsePB_Entry;
-using master::MasterServiceProxy;
-using master::TabletLocationsPB;
-using pb_util::SecureDebugString;
-using pb_util::SecureShortDebugString;
+using kudu::client::KuduSchema;
+using kudu::client::KuduSchemaBuilder;
+using kudu::consensus::BulkChangeConfigRequestPB;
+using kudu::consensus::ChangeConfigRequestPB;
+using kudu::consensus::ChangeConfigResponsePB;
+using kudu::consensus::ConsensusStatePB;
+using kudu::consensus::CountVoters;
+using kudu::consensus::EXCLUDE_HEALTH_REPORT;
+using kudu::consensus::GetConsensusStateRequestPB;
+using kudu::consensus::GetConsensusStateResponsePB;
+using kudu::consensus::GetLastOpIdRequestPB;
+using kudu::consensus::GetLastOpIdResponsePB;
+using kudu::consensus::IncludeHealthReport;
+using kudu::consensus::LeaderStepDownRequestPB;
+using kudu::consensus::LeaderStepDownResponsePB;
+using kudu::consensus::OpId;
+using kudu::consensus::OpIdType;
+using kudu::consensus::RaftPeerPB;
+using kudu::consensus::RunLeaderElectionResponsePB;
+using kudu::consensus::RunLeaderElectionRequestPB;
+using kudu::consensus::VoteRequestPB;
+using kudu::consensus::VoteResponsePB;
+using kudu::consensus::kInvalidOpIdIndex;
+using kudu::master::ListTabletServersResponsePB_Entry;
+using kudu::master::MasterServiceProxy;
+using kudu::master::TabletLocationsPB;
+using kudu::pb_util::SecureDebugString;
+using kudu::pb_util::SecureShortDebugString;
+using kudu::rpc::Messenger;
+using kudu::rpc::RpcController;
+using kudu::sentry::SentryClient;
+using kudu::tablet::TabletDataState;
+using kudu::tserver::CreateTsClientProxies;
+using kudu::tserver::ListTabletsResponsePB;
+using kudu::tserver::DeleteTabletRequestPB;
+using kudu::tserver::DeleteTabletResponsePB;
+using kudu::tserver::BeginTabletCopySessionRequestPB;
+using kudu::tserver::BeginTabletCopySessionResponsePB;
+using kudu::tserver::TabletCopyErrorPB;
+using kudu::tserver::TabletServerErrorPB;
+using kudu::tserver::WriteRequestPB;
+using kudu::tserver::WriteResponsePB;
using rapidjson::Value;
-using rpc::Messenger;
-using rpc::RpcController;
using std::min;
using std::shared_ptr;
using std::string;
@@ -104,17 +120,9 @@ using std::unique_ptr;
using std::unordered_map;
using std::vector;
using strings::Substitute;
-using tablet::TabletDataState;
-using tserver::CreateTsClientProxies;
-using tserver::ListTabletsResponsePB;
-using tserver::DeleteTabletRequestPB;
-using tserver::DeleteTabletResponsePB;
-using tserver::BeginTabletCopySessionRequestPB;
-using tserver::BeginTabletCopySessionResponsePB;
-using tserver::TabletCopyErrorPB;
-using tserver::TabletServerErrorPB;
-using tserver::WriteRequestPB;
-using tserver::WriteResponsePB;
+
+namespace kudu {
+namespace itest {
const string& TServerDetails::uuid() const {
return instance_id.permanent_uuid();
@@ -1228,5 +1236,28 @@ Status GetInt64Metric(const HostPort& http_hp,
return Status::NotFound(msg);
}
+Status SetupAdministratorPrivileges(MiniKdc* kdc,
+ const HostPort& address) {
+ DCHECK(kdc);
+ RETURN_NOT_OK(kdc->CreateUserPrincipal("kudu"));
+ RETURN_NOT_OK(kdc->Kinit("kudu"));
+
+ thrift::ClientOptions sentry_opts;
+ sentry_opts.service_principal = "sentry";
+ sentry_opts.enable_kerberos = true;
+ unique_ptr<SentryClient> sentry_client(
+ new SentryClient(address, sentry_opts));
+ RETURN_NOT_OK(sentry_client->Start());
+
+ // Create an admin role for the "admin" group specified in mini_sentry.cc.
+ // Grant this role all privileges for the server so the admin user can
+ // perform any operations required in tests.
+ RETURN_NOT_OK(master::CreateRoleAndAddToGroups(sentry_client.get(), "admin-role", "admin"));
+ TSentryPrivilege privilege = master::GetServerPrivilege("ALL", TSentryGrantOption::DISABLED);
+ RETURN_NOT_OK(master::AlterRoleGrantPrivilege(sentry_client.get(), "admin-role", privilege));
+ return kdc->Kinit("test-admin");
+}
+
+
} // namespace itest
} // namespace kudu
diff --git a/src/kudu/integration-tests/cluster_itest_util.h b/src/kudu/integration-tests/cluster_itest_util.h
index fba9fae..352ed27 100644
--- a/src/kudu/integration-tests/cluster_itest_util.h
+++ b/src/kudu/integration-tests/cluster_itest_util.h
@@ -50,6 +50,7 @@ namespace kudu {
class HostPort;
class MetricEntityPrototype;
class MetricPrototype;
+class MiniKdc;
class MonoDelta;
class Status;
@@ -71,6 +72,12 @@ class Messenger;
namespace itest {
+// Mode to indicate whether external service Sentry is enabled or not.
+enum class SentryMode {
+ DISABLED,
+ ENABLED
+};
+
struct TServerDetails {
NodeInstancePB instance_id;
ServerRegistrationPB registration;
@@ -445,6 +452,11 @@ Status GetInt64Metric(const HostPort& http_hp,
const char* value_field,
int64_t* value);
+// Grants the 'test-admin' user Sentry privileges to perform any operation,
+// using 'kdc' to authenticate with the Sentry instance at 'address'. Once
+// called, the 'test-admin' user will be logged in.
+Status SetupAdministratorPrivileges(MiniKdc* kdc,
+ const HostPort& address);
} // namespace itest
} // namespace kudu
diff --git a/src/kudu/integration-tests/hms_itest-base.h b/src/kudu/integration-tests/hms_itest-base.h
index 57f1a3f..b348031 100644
--- a/src/kudu/integration-tests/hms_itest-base.h
+++ b/src/kudu/integration-tests/hms_itest-base.h
@@ -30,24 +30,24 @@ namespace kudu {
class HmsITestBase : public ExternalMiniClusterITestBase {
public:
- Status StopHms();
- Status StartHms();
+ Status StartHms() WARN_UNUSED_RESULT;
+ Status StopHms() WARN_UNUSED_RESULT;
// Creates a database in the HMS catalog.
- Status CreateDatabase(const std::string& database_name);
+ Status CreateDatabase(const std::string& database_name) WARN_UNUSED_RESULT;
// Creates a table in Kudu.
Status CreateKuduTable(const std::string& database_name,
- const std::string& table_name);
+ const std::string& table_name) WARN_UNUSED_RESULT;
// Renames a table entry in the HMS catalog.
Status RenameHmsTable(const std::string& database_name,
const std::string& old_table_name,
- const std::string& new_table_name);
+ const std::string& new_table_name) WARN_UNUSED_RESULT;
// Drops all columns from a Kudu HMS table entry.
Status AlterHmsTableDropColumns(const std::string& database_name,
- const std::string& table_name);
+ const std::string& table_name) WARN_UNUSED_RESULT;
// Checks that the Kudu table schema and the HMS table entry in their
// respective catalogs are synchronized for a particular table. It also
diff --git a/src/kudu/integration-tests/master-stress-test.cc b/src/kudu/integration-tests/master-stress-test.cc
index 995848d..f5471d6 100644
--- a/src/kudu/integration-tests/master-stress-test.cc
+++ b/src/kudu/integration-tests/master-stress-test.cc
@@ -39,11 +39,13 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/strings/util.h"
#include "kudu/integration-tests/cluster_itest_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/rpc/messenger.h"
#include "kudu/rpc/rpc_controller.h"
+#include "kudu/sentry/mini_sentry.h"
#include "kudu/tablet/tablet.pb.h"
#include "kudu/tools/tool_action_common.h"
#include "kudu/tserver/tserver.pb.h"
@@ -82,6 +84,7 @@ using kudu::cluster::ExternalMaster;
using kudu::cluster::ExternalMiniCluster;
using kudu::cluster::ExternalMiniClusterOptions;
using kudu::itest::ListTablets;
+using kudu::itest::SentryMode;
using kudu::master::ListTablesRequestPB;
using kudu::master::ListTablesResponsePB;
using kudu::master::ReplaceTabletRequestPB;
@@ -91,6 +94,7 @@ using kudu::rpc::Messenger;
using kudu::rpc::MessengerBuilder;
using kudu::rpc::RpcController;
using kudu::tools::LeaderMasterProxy;
+using std::pair;
using std::shared_ptr;
using std::string;
using std::thread;
@@ -103,8 +107,9 @@ namespace kudu {
static const MonoDelta kDefaultAdminTimeout = MonoDelta::FromSeconds(300);
static const MonoDelta kTransientStateBackoff = MonoDelta::FromMilliseconds(50);
-class MasterStressTest : public KuduTest,
- public ::testing::WithParamInterface<HmsMode> {
+// Parameterized based on HmsMode and whether or not to enable Sentry integration.
+class MasterStressTest : public ExternalMiniClusterITestBase,
+ public ::testing::WithParamInterface<pair<HmsMode, SentryMode>> {
public:
MasterStressTest()
: done_(1),
@@ -134,7 +139,10 @@ class MasterStressTest : public KuduTest,
opts.start_process_timeout = MonoDelta::FromSeconds(60);
opts.rpc_negotiation_timeout = MonoDelta::FromSeconds(30);
- opts.hms_mode = GetParam();
+ opts.hms_mode = std::get<0>(GetParam());
+ bool enable_sentry = (std::get<1>(GetParam()) == SentryMode::ENABLED);
+ opts.enable_sentry = enable_sentry;
+ opts.enable_kerberos = enable_sentry;
// Tune down the notification log poll period in order to speed up catalog convergence.
opts.extra_master_flags.emplace_back("--hive_metastore_notification_log_poll_period_seconds=1");
@@ -192,6 +200,11 @@ class MasterStressTest : public KuduTest,
shared_ptr<MasterServiceProxy> m_proxy(
new MasterServiceProxy(cluster_->messenger(), addr, addr.host()));
ASSERT_OK(CreateTabletServerMap(m_proxy, cluster_->messenger(), &ts_map_));
+
+ if (enable_sentry) {
+ itest::SetupAdministratorPrivileges(cluster_->kdc(),
+ cluster_->sentry()->address());
+ }
}
void TearDown() override {
@@ -483,9 +496,15 @@ class MasterStressTest : public KuduTest,
std::unordered_map<string, itest::TServerDetails*> ts_map_;
};
-// Run the test with the HMS integration enabled and disabled.
-INSTANTIATE_TEST_CASE_P(HmsConfigurations, MasterStressTest,
- ::testing::Values(HmsMode::NONE, HmsMode::ENABLE_METASTORE_INTEGRATION));
+// Run the test with the HMS/Sentry integration enabled and disabled. Sentry integration
+// should be only enabled when HMS integration is enabled.
+INSTANTIATE_TEST_CASE_P(HmsSentryConfigurations, MasterStressTest, ::testing::ValuesIn(
+ vector<pair<HmsMode, SentryMode>> {
+ { HmsMode::NONE, SentryMode::DISABLED },
+ { HmsMode::ENABLE_METASTORE_INTEGRATION, SentryMode::DISABLED },
+ { HmsMode::ENABLE_METASTORE_INTEGRATION, SentryMode::ENABLED },
+ }
+));
TEST_P(MasterStressTest, Test) {
OverrideFlagForSlowTests("num_create_table_threads", "10");
diff --git a/src/kudu/integration-tests/master_failover-itest.cc b/src/kudu/integration-tests/master_failover-itest.cc
index cb1af80..9dcaa97 100644
--- a/src/kudu/integration-tests/master_failover-itest.cc
+++ b/src/kudu/integration-tests/master_failover-itest.cc
@@ -20,6 +20,7 @@
#include <ostream> // IWYU pragma: keep
#include <set>
#include <string>
+#include <utility>
#include <vector>
#include <glog/logging.h>
@@ -37,6 +38,7 @@
#include "kudu/integration-tests/cluster_itest_util.h"
#include "kudu/master/sys_catalog.h" // IWYU pragma: keep
#include "kudu/mini-cluster/external_mini_cluster.h"
+#include "kudu/sentry/mini_sentry.h"
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h" // IWYU pragma: keep
@@ -49,6 +51,22 @@
METRIC_DECLARE_entity(server);
METRIC_DECLARE_histogram(handler_latency_kudu_consensus_ConsensusService_GetNodeInstance);
+using kudu::client::sp::shared_ptr;
+using kudu::cluster::ExternalDaemon;
+using kudu::cluster::ExternalMaster;
+using kudu::cluster::ExternalMiniCluster;
+using kudu::cluster::ExternalMiniClusterOptions;
+using kudu::cluster::ScopedResumeExternalDaemon;
+using kudu::itest::GetInt64Metric;
+using kudu::itest::SentryMode;
+using std::pair;
+using std::set;
+using std::string;
+using std::unique_ptr;
+using std::vector;
+using strings::Split;
+using strings::Substitute;
+
namespace kudu {
// Note: this test needs to be in the client namespace in order for
@@ -57,22 +75,9 @@ namespace client {
const int kNumTabletServerReplicas = 3;
-using cluster::ExternalDaemon;
-using cluster::ExternalMaster;
-using cluster::ExternalMiniCluster;
-using cluster::ExternalMiniClusterOptions;
-using cluster::ScopedResumeExternalDaemon;
-using itest::GetInt64Metric;
-using sp::shared_ptr;
-using std::set;
-using std::string;
-using std::unique_ptr;
-using std::vector;
-using strings::Split;
-using strings::Substitute;
-
+// Parameterized based on HmsMode and whether or not to enable Sentry integration.
class MasterFailoverTest : public KuduTest,
- public ::testing::WithParamInterface<HmsMode> {
+ public ::testing::WithParamInterface<pair<HmsMode, SentryMode>> {
public:
enum CreateTableMode {
kWaitForCreate = 0,
@@ -82,7 +87,9 @@ class MasterFailoverTest : public KuduTest,
MasterFailoverTest() {
opts_.num_masters = 3;
opts_.num_tablet_servers = kNumTabletServerReplicas;
- opts_.hms_mode = GetParam();
+ opts_.hms_mode = std::get<0>(GetParam());
+ opts_.enable_sentry = (std::get<1>(GetParam()) == SentryMode::ENABLED);
+ opts_.enable_kerberos = opts_.enable_sentry;
// Reduce various timeouts below as to make the detection of
// leader master failures (specifically, failures as result of
@@ -126,6 +133,11 @@ class MasterFailoverTest : public KuduTest,
// the global operation timeout.
builder.default_admin_operation_timeout(MonoDelta::FromSeconds(90));
ASSERT_OK(cluster_->CreateClient(&builder, &client_));
+
+ if (opts_.enable_sentry) {
+ ASSERT_OK(itest::SetupAdministratorPrivileges(cluster_->kdc(),
+ cluster_->sentry()->address()));
+ }
}
Status CreateTable(const std::string& table_name, CreateTableMode mode) {
@@ -157,9 +169,15 @@ class MasterFailoverTest : public KuduTest,
shared_ptr<KuduClient> client_;
};
-// Run the test with the HMS integration enabled and disabled.
-INSTANTIATE_TEST_CASE_P(HmsConfigurations, MasterFailoverTest,
- ::testing::Values(HmsMode::NONE, HmsMode::ENABLE_METASTORE_INTEGRATION));
+// Run the test with the HMS/Sentry integration enabled and disabled. Sentry integration
+// should be only enabled when HMS integration is enabled.
+INSTANTIATE_TEST_CASE_P(HmsSentryConfigurations, MasterFailoverTest, ::testing::ValuesIn(
+ vector<pair<HmsMode, SentryMode>> {
+ { HmsMode::NONE, SentryMode::DISABLED },
+ { HmsMode::ENABLE_METASTORE_INTEGRATION, SentryMode::DISABLED },
+ { HmsMode::ENABLE_METASTORE_INTEGRATION, SentryMode::ENABLED },
+ }
+));
// Test that synchronous CreateTable (issue CreateTable call and then
// wait until the table has been created) works even when the original
diff --git a/src/kudu/integration-tests/master_hms-itest.cc b/src/kudu/integration-tests/master_hms-itest.cc
index f539c08..0574eba 100644
--- a/src/kudu/integration-tests/master_hms-itest.cc
+++ b/src/kudu/integration-tests/master_hms-itest.cc
@@ -348,12 +348,12 @@ TEST_F(MasterHmsTest, TestNotificationLogListener) {
// Ensure that Kudu can rename a table just after it's been renamed through the HMS.
- RenameHmsTable("default", "a", "b");
+ ASSERT_OK(RenameHmsTable("default", "a", "b"));
table_alterer.reset(client_->NewTableAlterer("default.b")->RenameTo("default.c"));
ASSERT_OK(table_alterer->Alter());
// Ensure that Kudu can drop a table just after it's been renamed through the HMS.
- RenameHmsTable("default", "c", "a");
+ ASSERT_OK(RenameHmsTable("default", "c", "a"));
ASSERT_OK(client_->DeleteTable("default.a"));
// Test concurrent drops from the HMS and Kudu.
diff --git a/src/kudu/integration-tests/master_sentry-itest.cc b/src/kudu/integration-tests/master_sentry-itest.cc
index 0ada81f..a110162 100644
--- a/src/kudu/integration-tests/master_sentry-itest.cc
+++ b/src/kudu/integration-tests/master_sentry-itest.cc
@@ -109,8 +109,8 @@ class SentryITestBase : public HmsITestBase {
return Status::OK();
}
- Status GetTableLocationsWithTableId(const string& table_name,
- optional<const string&> table_id) {
+ Status GetTableLocationsWithTableId(const string& table_name,
+ optional<const string&> table_id) {
const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
std::shared_ptr<MasterServiceProxy> proxy = cluster_->master_proxy();
UserCredentials user_credentials;
@@ -120,7 +120,7 @@ class SentryITestBase : public HmsITestBase {
GetTableLocationsResponsePB table_locations;
return itest::GetTableLocations(proxy, table_name, kTimeout, master::VOTER_REPLICA,
table_id, &table_locations);
- }
+ }
Status GrantCreateTablePrivilege(const string& database_name,
const string& /*table_name*/) {
@@ -324,10 +324,10 @@ class SentryITestBase : public HmsITestBase {
void TearDown() override {
if (sentry_client_) {
- ASSERT_OK(sentry_client_->Stop());
+ ASSERT_OK(sentry_client_->Stop());
}
if (hms_client_) {
- ASSERT_OK(hms_client_->Stop());
+ ASSERT_OK(hms_client_->Stop());
}
HmsITestBase::TearDown();
}