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 2020/03/05 02:56:22 UTC
[kudu] branch master updated (61ddfe3 -> e901424)
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 61ddfe3 blocking_queue: replace gscoped_ptr with unique_ptr
new 148c169 test: re-deflake TabletServerQuiescingITest
new 2ea8478 [java] KUDU-2972: Add Kudu Ranger plugin
new 2d858f6 [build] Use Gradle to build the hms-plugin jar
new e901424 [python] Remove use of dict.iteritems()
The 4 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:
CMakeLists.txt | 19 ++-
build-support/dist_test.py | 4 +-
.../relocate_binaries_for_mini_cluster.py | 4 +-
build-support/run_dist_test.py | 8 +-
java/gradle/dependencies.gradle | 2 +
java/kudu-subprocess/build.gradle | 8 +-
.../subprocess/ranger/RangerProtocolHandler.java | 71 ++++++++
.../RangerSubprocessMain.java} | 18 +-
.../ranger/authorization/RangerKuduAuthorizer.java | 185 +++++++++++++++++++++
...essageTestUtil.java => SubprocessTestUtil.java} | 90 +++++++++-
.../org/apache/kudu/subprocess/TestMessageIO.java | 8 +-
.../kudu/subprocess/echo/TestEchoSubprocess.java | 106 ++----------
.../subprocess/ranger/TestRangerSubprocess.java | 134 +++++++++++++++
.../authorization/TestRangerKuduAuthorizer.java | 81 +++++++++
src/kudu/hms/CMakeLists.txt | 19 ++-
.../tablet_server_quiescing-itest.cc | 5 +-
src/kudu/{rebalance => ranger}/CMakeLists.txt | 34 ++--
src/kudu/ranger/ranger.proto | 73 ++++++++
src/kudu/subprocess/CMakeLists.txt | 9 +-
19 files changed, 723 insertions(+), 155 deletions(-)
create mode 100644 java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java
copy java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/{echo/EchoSubprocessMain.java => ranger/RangerSubprocessMain.java} (62%)
create mode 100644 java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java
rename java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/{MessageTestUtil.java => SubprocessTestUtil.java} (53%)
create mode 100644 java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java
create mode 100644 java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/authorization/TestRangerKuduAuthorizer.java
copy src/kudu/{rebalance => ranger}/CMakeLists.txt (67%)
create mode 100644 src/kudu/ranger/ranger.proto
[kudu] 01/04: test: re-deflake TabletServerQuiescingITest
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 148c169d46d2cdccc7434e827162552279bc3e4a
Author: Andrew Wong <aw...@cloudera.com>
AuthorDate: Wed Mar 4 17:41:32 2020 -0800
test: re-deflake TabletServerQuiescingITest
Commit bc2efa1ae97fcc4e592c93d38592206d48d6f8f5 undid some of the
deflaking done in 6e4dd49 by increasing the number of read threads.
This puts us back at using a single reader, which, while may not always
yield active scanners, should be significantly less flaky.
I ran this 100 times in TSAN mode and saw no failures, compared to the
20% flakiness reported by our test dashboard.
Change-Id: I598c0f0a9a665691bc675920d8dc5a0d272be72e
Reviewed-on: http://gerrit.cloudera.org:8080/15365
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Kudu Jenkins
---
src/kudu/integration-tests/tablet_server_quiescing-itest.cc | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/src/kudu/integration-tests/tablet_server_quiescing-itest.cc b/src/kudu/integration-tests/tablet_server_quiescing-itest.cc
index 921915c..f01cfb1 100644
--- a/src/kudu/integration-tests/tablet_server_quiescing-itest.cc
+++ b/src/kudu/integration-tests/tablet_server_quiescing-itest.cc
@@ -378,8 +378,9 @@ TEST_F(TServerQuiescingITest, TestQuiescingToolBasics) {
auto* ts = cluster_->mini_tablet_server(0);
auto rw_workload = CreateFaultIntolerantRWWorkload();
rw_workload->Setup();
- // Spawn a bunch of read threads so we'll be more likely to see scanners.
- rw_workload->set_num_read_threads(10);
+ // NOTE: if this value is too high, this test can become flaky, since the
+ // degrees of freedom in the number of active scanners will be high.
+ rw_workload->set_num_read_threads(1);
ASSERT_FALSE(ts->server()->quiescing());
const auto& master_addr = cluster_->mini_master()->bound_rpc_addr().ToString();
// First, call the start tool a couple of times.
[kudu] 04/04: [python] Remove use of dict.iteritems()
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 e901424abf7d54f9e3060bf3f6c1a5cc0d42874d
Author: Grant Henke <gr...@apache.org>
AuthorDate: Mon Mar 2 12:46:11 2020 -0600
[python] Remove use of dict.iteritems()
This patch removes the use of dict.iteritems() given it is removed
in Python 3.
Change-Id: I22d0d66cee0b9c940376e6c518504912fd431702
Reviewed-on: http://gerrit.cloudera.org:8080/15359
Tested-by: Kudu Jenkins
Reviewed-by: Grant Henke <gr...@apache.org>
---
build-support/dist_test.py | 4 ++--
build-support/mini-cluster/relocate_binaries_for_mini_cluster.py | 4 ++--
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/build-support/dist_test.py b/build-support/dist_test.py
index 588a396..c32022e 100755
--- a/build-support/dist_test.py
+++ b/build-support/dist_test.py
@@ -341,7 +341,7 @@ def create_archive_input(staging, execution, dep_extractor,
'-e', 'KUDU_ALLOW_SLOW_TESTS=%s' % os.environ.get('KUDU_ALLOW_SLOW_TESTS', 1),
'-e', 'KUDU_COMPRESS_TEST_OUTPUT=%s' % \
os.environ.get('KUDU_COMPRESS_TEST_OUTPUT', 0)]
- for k, v in execution.env.iteritems():
+ for k, v in execution.env.items():
if k == 'KUDU_TEST_TIMEOUT':
# Currently we don't respect the test timeouts specified in ctest, since
# we want to make sure that the dist-test task timeout and the
@@ -397,7 +397,7 @@ def create_task_json(staging,
# Some versions of 'isolate batcharchive' directly list the items in
# the dumped JSON. Others list it in an 'items' dictionary.
items = inmap.get('items', inmap)
- for k, v in items.iteritems():
+ for k, v in items.items():
# The key may be 'foo-test.<shard>'. So, chop off the last component
# to get the test name.
test_name = ".".join(k.split(".")[:-1]) if TEST_SHARD_RE.search(k) else k
diff --git a/build-support/mini-cluster/relocate_binaries_for_mini_cluster.py b/build-support/mini-cluster/relocate_binaries_for_mini_cluster.py
index 562c1f8..a443814 100755
--- a/build-support/mini-cluster/relocate_binaries_for_mini_cluster.py
+++ b/build-support/mini-cluster/relocate_binaries_for_mini_cluster.py
@@ -366,7 +366,7 @@ def relocate_deps_macos(target_src, target_dst, config):
# For each dependency, relocate the path we will search for it and ensure it
# is shipped with the archive.
- for (dep_search_name, dep_src) in target_deps.iteritems():
+ for (dep_search_name, dep_src) in target_deps.items():
# Filter out libs we don't want to archive.
if PAT_MACOS_LIB_EXCLUDE.search(dep_search_name):
continue
@@ -401,7 +401,7 @@ def relocate_sasl2(target_src, config):
"""
# Find the libsasl2 module in our dependencies.
- deps = get_resolved_deps(target_src);
+ deps = get_resolved_deps(target_src)
sasl_lib = None
for dep in deps:
if re.search('libsasl2', dep):
[kudu] 02/04: [java] KUDU-2972: Add Kudu Ranger plugin
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 2ea8478d256b156a9257f5f51ad88a29a36872de
Author: Hao Hao <ha...@cloudera.com>
AuthorDate: Mon Jan 20 22:45:12 2020 -0800
[java] KUDU-2972: Add Kudu Ranger plugin
This commit adds the Kudu Ranger plugin for retrieving authorization
decision from the Ranger service. It also introduces the Ranger subprocess
which utilizes the plugin and communicates the authz decision with
Kudu master via protobuf message.
Change-Id: I0c995ac1a48ebf57667231cd3a82d3794f6ddf8d
Reviewed-on: http://gerrit.cloudera.org:8080/15074
Tested-by: Hao Hao <ha...@cloudera.com>
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Reviewed-by: Andrew Wong <aw...@cloudera.com>
---
CMakeLists.txt | 1 +
java/gradle/dependencies.gradle | 2 +
java/kudu-subprocess/build.gradle | 8 +-
.../subprocess/ranger/RangerProtocolHandler.java | 71 ++++++++
.../subprocess/ranger/RangerSubprocessMain.java | 38 +++++
.../ranger/authorization/RangerKuduAuthorizer.java | 185 +++++++++++++++++++++
...essageTestUtil.java => SubprocessTestUtil.java} | 90 +++++++++-
.../org/apache/kudu/subprocess/TestMessageIO.java | 8 +-
.../kudu/subprocess/echo/TestEchoSubprocess.java | 106 ++----------
.../subprocess/ranger/TestRangerSubprocess.java | 134 +++++++++++++++
.../authorization/TestRangerKuduAuthorizer.java | 81 +++++++++
src/kudu/ranger/CMakeLists.txt | 33 ++++
src/kudu/ranger/ranger.proto | 73 ++++++++
13 files changed, 728 insertions(+), 102 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 21fb38f..b3207f8 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -1417,6 +1417,7 @@ add_subdirectory(src/kudu/integration-tests)
add_subdirectory(src/kudu/kserver)
add_subdirectory(src/kudu/master)
add_subdirectory(src/kudu/mini-cluster)
+add_subdirectory(src/kudu/ranger)
add_subdirectory(src/kudu/rebalance)
add_subdirectory(src/kudu/rpc)
add_subdirectory(src/kudu/security)
diff --git a/java/gradle/dependencies.gradle b/java/gradle/dependencies.gradle
index 4b3801b..022a52b 100755
--- a/java/gradle/dependencies.gradle
+++ b/java/gradle/dependencies.gradle
@@ -51,6 +51,7 @@ versions += [
osdetector : "1.6.2",
parquet : "1.11.0",
protobuf : "3.11.3",
+ ranger : "2.0.0",
scala : "2.11.12",
scalatest : "3.0.8",
scopt : "3.7.1",
@@ -106,6 +107,7 @@ libs += [
protobufJava : "com.google.protobuf:protobuf-java:$versions.protobuf",
protobufJavaUtil : "com.google.protobuf:protobuf-java-util:$versions.protobuf",
protoc : "com.google.protobuf:protoc:$versions.protobuf",
+ rangerPlugin : "org.apache.ranger:ranger-plugins-common:$versions.ranger",
scalaLibrary : "org.scala-lang:scala-library:$versions.scala",
scalap : "org.scala-lang:scalap:$versions.scala",
scalatest : "org.scalatest:scalatest_$versions.scalaBase:$versions.scalatest",
diff --git a/java/kudu-subprocess/build.gradle b/java/kudu-subprocess/build.gradle
index 6188ccf..1e04918 100644
--- a/java/kudu-subprocess/build.gradle
+++ b/java/kudu-subprocess/build.gradle
@@ -19,17 +19,23 @@ apply from: "$rootDir/gradle/protobuf.gradle"
apply from: "$rootDir/gradle/shadow.gradle"
dependencies {
+ compile libs.hadoopCommon
compile libs.protobufJava
compile libs.protobufJavaUtil
- compile libs.hadoopCommon
+ compile libs.rangerPlugin
compile libs.slf4jApi
+ // Workaround for RANGER-2749. Remove once resolved.
+ compile "commons-lang:commons-lang:2.6"
+
+ optional libs.jsr305
optional libs.yetusAnnotations
testCompile project(path: ":kudu-test-utils", configuration: "shadow")
testCompile libs.junit
testCompile libs.log4j
testCompile libs.log4jSlf4jImpl
+ testCompile libs.mockitoCore
}
// Add protobuf files to the proto source set.
diff --git a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java
new file mode 100644
index 0000000..66a6a47
--- /dev/null
+++ b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerProtocolHandler.java
@@ -0,0 +1,71 @@
+// 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.
+
+package org.apache.kudu.subprocess.ranger;
+
+import org.apache.kudu.ranger.Ranger.RangerRequestListPB;
+import org.apache.kudu.ranger.Ranger.RangerResponseListPB;
+import org.apache.kudu.ranger.Ranger.RangerResponsePB;
+import org.apache.kudu.subprocess.ProtocolHandler;
+import org.apache.kudu.subprocess.ranger.authorization.RangerKuduAuthorizer;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Class that sends requests to Ranger and gets authorization decision
+ * (e.g. allow or deny) as a response.
+ */
+@InterfaceAudience.Private
+class RangerProtocolHandler extends ProtocolHandler<RangerRequestListPB,
+ RangerResponseListPB> {
+ private static final Logger LOG = LoggerFactory.getLogger(RangerProtocolHandler.class);
+
+ // The Ranger Kudu authorizer plugin. This field is not final
+ // as it is used in the mock test.
+ @InterfaceAudience.LimitedPrivate("Test")
+ static RangerKuduAuthorizer authz = new RangerKuduAuthorizer();
+
+ RangerProtocolHandler() {
+ authz.init();
+ }
+
+ @Override
+ protected RangerResponseListPB executeRequest(RangerRequestListPB requests) {
+ RangerResponseListPB.Builder responses = RangerResponseListPB.newBuilder();
+ for (RangerAccessResult result : authz.authorize(requests)) {
+ // The result can be null when Ranger plugin fails to load the policies
+ // from the Ranger admin server.
+ // TODO(Hao): add a test for the above case.
+ boolean isAllowed = (result != null && result.getIsAllowed());
+ RangerResponsePB.Builder response = RangerResponsePB.newBuilder();
+ response.setAllowed(isAllowed);
+ responses.addResponses(response);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("RangerAccessRequest [%s] receives result [%s]",
+ result.getAccessRequest().toString(), result.toString()));
+ }
+ }
+ return responses.build();
+ }
+
+ @Override
+ protected Class<RangerRequestListPB> getRequestClass() {
+ return RangerRequestListPB.class;
+ }
+}
diff --git a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerSubprocessMain.java b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerSubprocessMain.java
new file mode 100644
index 0000000..167cd08
--- /dev/null
+++ b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/RangerSubprocessMain.java
@@ -0,0 +1,38 @@
+// 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.
+
+package org.apache.kudu.subprocess.ranger;
+
+import org.apache.kudu.subprocess.SubprocessExecutor;
+import org.apache.yetus.audience.InterfaceAudience;
+
+// The Ranger subprocess that wraps the Kudu Ranger plugin. For the
+// plugin to successfully connect to the Ranger service, configurations
+// such as ranger-kudu-security.xml (and ranger-kudu-policymgr-ssl.xml
+// for SSL connection) are required. To enable auditing in Ranger,
+// ranger-kudu-security.xml is needed. The plugin also requires
+// core-site.xml to use Hadoop UserGroupInformation for user group
+// resolution.
+@InterfaceAudience.Private
+class RangerSubprocessMain {
+
+ public static void main(String[] args) throws Exception {
+ SubprocessExecutor subprocessExecutor = new SubprocessExecutor();
+ RangerProtocolHandler protocolProcessor = new RangerProtocolHandler();
+ subprocessExecutor.run(args, protocolProcessor, /* timeoutMs= */-1);
+ }
+}
diff --git a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java
new file mode 100644
index 0000000..01ebe27
--- /dev/null
+++ b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/ranger/authorization/RangerKuduAuthorizer.java
@@ -0,0 +1,185 @@
+// 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.
+
+package org.apache.kudu.subprocess.ranger.authorization;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.kudu.ranger.Ranger.RangerRequestListPB;
+import org.apache.kudu.ranger.Ranger.RangerRequestPB;
+import org.apache.ranger.plugin.audit.RangerDefaultAuditHandler;
+import org.apache.ranger.plugin.model.RangerServiceDef;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResourceImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class RangerKuduAuthorizer {
+ private static final Logger LOG = LoggerFactory.getLogger(RangerKuduAuthorizer.class);
+ // The following properties need to match the Kudu service def in Ranger
+ // (https://github.com/apache/ranger/blob/master/agents-common/src/main/resources/service-defs/ranger-servicedef-kudu.json).
+ private static final String APP_ID = "kudu";
+ private static final String RANGER_DB_RESOURCE_NAME = "database";
+ private static final String RANGER_TABLE_RESOURCE_NAME = "table";
+ private static final String RANGER_COLUMN_RESOURCE_NAME = "column";
+ private static final String SERVICE_TYPE = "kudu";
+
+ // The Ranger Kudu plugin. This field is not final as it is used in the
+ // mock test.
+ @InterfaceAudience.LimitedPrivate("Test")
+ RangerBasePlugin plugin;
+
+ public RangerKuduAuthorizer() {
+ plugin = new RangerBasePlugin(SERVICE_TYPE, APP_ID);
+ plugin.setResultProcessor(new RangerDefaultAuditHandler());
+ }
+
+ /**
+ * Initializes the Ranger Kudu plugin, which has to be called explicitly
+ * before doing any authorizations.
+ */
+ public void init() {
+ LOG.info("Initializing Ranger Kudu plugin");
+ plugin.init();
+ }
+
+ /**
+ * Authorizes a given <code>RangerRequestListPB</code> in Ranger and returns
+ * a list of <code>RangerAccessResult</code> which contains the authorization
+ * decisions. Note that the order of results is determined by the order of
+ * requests.
+ *
+ * @param requests a RangerRequestListPB
+ * @return a list of RangerAccessResult
+ */
+ @VisibleForTesting
+ public Collection<RangerAccessResult> authorize(RangerRequestListPB requests) {
+ Collection<RangerAccessRequest> rangerRequests = createRequests(requests);
+ // Reject requests if user field is empty.
+ if (!requests.hasUser() || requests.getUser().isEmpty()) {
+ Collection<RangerAccessResult> results = new ArrayList<>();
+ for (RangerAccessRequest request : rangerRequests) {
+ // Create a 'dummy' RangerAccessResult that denies the request (to have
+ // a short cut), instead of sending the request to Ranger.
+ RangerAccessResult result = new RangerAccessResult(
+ /* policyType= */1, APP_ID,
+ new RangerServiceDef(), request);
+ result.setIsAllowed(false);
+ results.add(result);
+ }
+ return results;
+ }
+ return plugin.isAccessAllowed(rangerRequests);
+ }
+
+ /**
+ * Gets a list of authorization decision from Ranger with the specified list
+ * of ranger access request.
+ *
+ * @param requests a list of RangerAccessRequest
+ * @return a list of RangerAccessResult
+ */
+ @VisibleForTesting
+ Collection<RangerAccessResult> authorize(Collection<RangerAccessRequest> requests) {
+ return plugin.isAccessAllowed(requests);
+ }
+
+ /**
+ * Creates a Ranger access request for the specified user who performs
+ * the given action on the resource.
+ *
+ * @param action action to be authorized on the resource. Note that when it
+ * is null, Ranger will match to any valid actions
+ * @param user user who is performing the action
+ * @param groups the set of groups the user belongs to
+ * @param db the database name the action is to be performed on
+ * @param table the table name the action is to be performed on
+ * @param col the column name the action is to be performed on
+ * @return the ranger access request
+ */
+ private static RangerAccessRequestImpl createRequest(
+ @Nullable String action, String user,
+ @Nullable Set<String> groups, @Nullable String db,
+ @Nullable String table, @Nullable String col) {
+ final RangerAccessResourceImpl resource = new RangerAccessResourceImpl();
+ resource.setValue(RANGER_DB_RESOURCE_NAME, db);
+ resource.setValue(RANGER_TABLE_RESOURCE_NAME, table);
+ resource.setValue(RANGER_COLUMN_RESOURCE_NAME, col);
+
+ final RangerAccessRequestImpl request = new RangerAccessRequestImpl();
+ request.setResource(resource);
+ request.setAccessType(action);
+ // Add action as it is used for auditing in Ranger.
+ request.setAction(action);
+ request.setUser(user);
+ request.setUserGroups(groups);
+ return request;
+ }
+
+ /**
+ * Creates a list of <code>RangerAccessRequest</code> for the given
+ * <code>RangerRequestListPB</code>.
+ *
+ * @param requests the given RangerRequestListPB
+ * @return a list of RangerAccessRequest
+ */
+ private static List<RangerAccessRequest> createRequests(RangerRequestListPB requests) {
+ List<RangerAccessRequest> rangerRequests = new ArrayList<>();
+ Preconditions.checkArgument(requests.hasUser());
+ Preconditions.checkArgument(!requests.getUser().isEmpty());
+ final String user = requests.getUser();
+ Set<String> groups = getUserGroups(user);
+ for (RangerRequestPB request : requests.getRequestsList()) {
+ // Action should be lower case to match the Kudu service def in Ranger.
+ String action = request.getAction().name().toLowerCase();
+ String db = request.hasDatabase() ? request.getDatabase() : null;
+ String table = request.hasTable() ? request.getTable() : null;
+ String column = request.hasColumn() ? request.getColumn() : null;
+ rangerRequests.add(createRequest(action, user, groups,
+ db, table, column));
+ }
+ return rangerRequests;
+ }
+
+ /**
+ * Gets the user group mapping from Hadoop. The groups of a user is determined by a
+ * group mapping service provider. See more detail at
+ * https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/GroupsMapping.html.
+ *
+ * @param user the user name
+ * @return the set of groups the user belongs to
+ */
+ private static Set<String> getUserGroups(String user) {
+ Preconditions.checkNotNull(user);
+ Preconditions.checkArgument(!user.isEmpty());
+ UserGroupInformation ugi;
+ ugi = UserGroupInformation.createRemoteUser(user);
+ return new HashSet<>(ugi.getGroups());
+ }
+}
diff --git a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/MessageTestUtil.java b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/SubprocessTestUtil.java
similarity index 53%
rename from java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/MessageTestUtil.java
rename to java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/SubprocessTestUtil.java
index bca288f..b7d44c9 100644
--- a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/MessageTestUtil.java
+++ b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/SubprocessTestUtil.java
@@ -17,11 +17,20 @@
package org.apache.kudu.subprocess;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.OutputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.io.PrintStream;
+import java.io.UnsupportedEncodingException;
+import java.util.function.Function;
import com.google.protobuf.Any;
import com.google.protobuf.Message;
@@ -29,12 +38,85 @@ import com.google.protobuf.Parser;
import org.apache.kudu.subprocess.Subprocess.EchoRequestPB;
import org.apache.kudu.subprocess.Subprocess.SubprocessRequestPB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
- * Utility class of common functions used for testing subprocess
- * message processing.
+ * Utility class of common functions used for testing subprocess.
*/
-public class MessageTestUtil {
+public class SubprocessTestUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(SubprocessTestUtil.class);
+ protected static final String[] NO_ARGS = {""};
+ protected static final int TIMEOUT_MS = 1000;
+
+ // Helper functors that can be passed around to ensure we either see an error
+ // or not.
+ protected static final Function<Throwable, Void> NO_ERR = e -> {
+ LOG.error(String.format("Unexpected error: %s", e.getMessage()));
+ fail();
+ return null;
+ };
+ protected final Function<Throwable, Void> HAS_ERR = e -> {
+ assertTrue(e instanceof KuduSubprocessException);
+ return null;
+ };
+
+ // Pipe that we can write to that will feed requests to the subprocess's
+ // input pipe.
+ protected PipedOutputStream requestSenderPipe;
+
+ // Pipe that we can read from that will receive responses from the
+ // subprocess's output pipe.
+ protected final PipedInputStream responseReceiverPipe = new PipedInputStream();
+
+ public static class PrintStreamWithIOException extends PrintStream {
+ public PrintStreamWithIOException(OutputStream out, boolean autoFlush, String encoding)
+ throws UnsupportedEncodingException {
+ super(out, autoFlush, encoding);
+ }
+
+ @Override
+ public boolean checkError() {
+ // Always say that we've got an error.
+ return true;
+ }
+ }
+
+ // Sends a SubprocessRequestPB to the sender pipe, serializing it as
+ // appropriate.
+ public void sendRequestToPipe(Subprocess.SubprocessRequestPB req) throws IOException {
+ requestSenderPipe.write(SubprocessTestUtil.serializeMessage(req));
+ }
+
+ // Receives a response from the receiver pipe and deserializes it into a
+ // SubprocessResponsePB.
+ public Subprocess.SubprocessResponsePB receiveResponse() throws IOException {
+ BufferedInputStream bufferedInput = new BufferedInputStream(responseReceiverPipe);
+ return SubprocessTestUtil.deserializeMessage(bufferedInput, Subprocess.SubprocessResponsePB.parser());
+ }
+
+ // Sets up and returns a SubprocessExecutor with the given error handler and
+ // IO error injection behavior. The SubprocessExecutor will do IO to and from
+ // 'requestSenderPipe' and 'responseReceiverPipe'.
+ public SubprocessExecutor setUpExecutorIO(Function<Throwable, Void> errorHandler,
+ boolean injectIOError) throws IOException {
+ // Initialize the pipe that we'll push requests to; feed it into the
+ // executor's input pipe.
+ PipedInputStream inputPipe = new PipedInputStream();
+ requestSenderPipe = new PipedOutputStream(inputPipe);
+ System.setIn(inputPipe);
+
+ // Initialize the pipe that the executor will write to; feed it into the
+ // response pipe that we can read from.
+ PipedOutputStream outputPipe = new PipedOutputStream(responseReceiverPipe);
+ if (injectIOError) {
+ System.setOut(new PrintStreamWithIOException(outputPipe, /*autoFlush*/false, "UTF-8"));
+ } else {
+ System.setOut(new PrintStream(outputPipe));
+ }
+ SubprocessExecutor subprocessExecutor = new SubprocessExecutor(errorHandler);
+ return subprocessExecutor;
+ }
/**
* Constructs a SubprocessRequestPB message of echo request with the
@@ -91,7 +173,7 @@ public class MessageTestUtil {
* @return a message
* @throws IOException if an I/O error occurs
*/
- static <T extends Message> T deserializeMessage(byte[] bytes, Parser<T> parser)
+ public static <T extends Message> T deserializeMessage(byte[] bytes, Parser<T> parser)
throws IOException {
ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
return deserializeMessage(new BufferedInputStream(inputStream), parser);
diff --git a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/TestMessageIO.java b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/TestMessageIO.java
index 011f4e9..98fc0b0 100644
--- a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/TestMessageIO.java
+++ b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/TestMessageIO.java
@@ -64,9 +64,9 @@ public class TestMessageIO {
@Test
public void testBasicEchoMessage() throws Exception {
final String data = "data";
- final SubprocessRequestPB request = MessageTestUtil.createEchoSubprocessRequest(data);
- final byte[] message = MessageTestUtil.serializeMessage(request);
- final SubprocessRequestPB actualRequest = MessageTestUtil.deserializeMessage(
+ final SubprocessRequestPB request = SubprocessTestUtil.createEchoSubprocessRequest(data);
+ final byte[] message = SubprocessTestUtil.serializeMessage(request);
+ final SubprocessRequestPB actualRequest = SubprocessTestUtil.deserializeMessage(
message, SubprocessRequestPB.parser());
Assert.assertEquals(request, actualRequest);
}
@@ -79,7 +79,7 @@ public class TestMessageIO {
@Test
public void testSubprocessOutputStream() {
final String data = "data";
- final SubprocessRequestPB request = MessageTestUtil.createEchoSubprocessRequest(data);
+ final SubprocessRequestPB request = SubprocessTestUtil.createEchoSubprocessRequest(data);
final PrintStreamOverload printStreamOverload =
new PrintStreamOverload(new ByteArrayOutputStream());
final BufferedOutputStream out = new BufferedOutputStream(
diff --git a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/echo/TestEchoSubprocess.java b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/echo/TestEchoSubprocess.java
index c0e06c4..66f7ddc 100644
--- a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/echo/TestEchoSubprocess.java
+++ b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/echo/TestEchoSubprocess.java
@@ -17,18 +17,11 @@
package org.apache.kudu.subprocess.echo;
-import java.io.BufferedInputStream;
import java.io.IOException;
-import java.io.OutputStream;
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
-import java.io.PrintStream;
-import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
-import java.util.function.Function;
import org.junit.Assert;
import org.junit.Rule;
@@ -37,62 +30,25 @@ import org.junit.function.ThrowingRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.kudu.subprocess.KuduSubprocessException;
import org.apache.kudu.subprocess.MessageIO;
-import org.apache.kudu.subprocess.MessageTestUtil;
import org.apache.kudu.subprocess.OutboundResponse;
import org.apache.kudu.subprocess.Subprocess.EchoResponsePB;
import org.apache.kudu.subprocess.Subprocess.SubprocessMetricsPB;
-import org.apache.kudu.subprocess.Subprocess.SubprocessRequestPB;
import org.apache.kudu.subprocess.Subprocess.SubprocessResponsePB;
import org.apache.kudu.subprocess.SubprocessExecutor;
+import org.apache.kudu.subprocess.SubprocessTestUtil;
import org.apache.kudu.test.junit.RetryRule;
/**
* Tests for subprocess that handles EchoRequest messages in various conditions.
*/
-public class TestEchoSubprocess {
+public class TestEchoSubprocess extends SubprocessTestUtil {
private static final Logger LOG = LoggerFactory.getLogger(TestEchoSubprocess.class);
- private static final String[] NO_ARGS = {""};
- private static final int TIMEOUT_MS = 1000;
private static final String MESSAGE = "We are one. We are many.";
- // Helper functors that can be passed around to ensure we either see an error
- // or not.
- private static final Function<Throwable, Void> NO_ERR = e -> {
- LOG.error(String.format("Unexpected error: %s", e.getMessage()));
- Assert.fail();
- return null;
- };
- private final Function<Throwable, Void> HAS_ERR = e -> {
- Assert.assertTrue(e instanceof KuduSubprocessException);
- return null;
- };
-
- // Pipe that we can write to that will feed requests to the subprocess's
- // input pipe.
- private PipedOutputStream requestSenderPipe;
-
- // Pipe that we can read from that will receive responses from the
- // subprocess's output pipe.
- private final PipedInputStream responseReceiverPipe = new PipedInputStream();
-
@Rule
public RetryRule retryRule = new RetryRule();
- public static class PrintStreamWithIOException extends PrintStream {
- public PrintStreamWithIOException(OutputStream out, boolean autoFlush, String encoding)
- throws UnsupportedEncodingException {
- super(out, autoFlush, encoding);
- }
-
- @Override
- public boolean checkError() {
- // Always say that we've got an error.
- return true;
- }
- }
-
// Given that executors run multiple threads, the exceptions that we expect
// may not necessarily be the first thrown. This checks for the expected
// error on all thrown exceptions, including suppressed ones.
@@ -119,42 +75,6 @@ public class TestEchoSubprocess {
throw new AssertionError("Didn't throw an exception");
}
- // Sends a SubprocessRequestPB to the sender pipe, serializing it as
- // appropriate.
- void sendRequestToPipe(SubprocessRequestPB req) throws IOException {
- requestSenderPipe.write(MessageTestUtil.serializeMessage(req));
- }
-
- // Receives a response from the receiver pipe and deserializes it into a
- // SubprocessResponsePB.
- SubprocessResponsePB receiveResponse() throws IOException {
- BufferedInputStream bufferedInput = new BufferedInputStream(responseReceiverPipe);
- return MessageTestUtil.deserializeMessage(bufferedInput, SubprocessResponsePB.parser());
- }
-
- // Sets up and returns a SubprocessExecutor with the given error handler and
- // IO error injection behavior. The SubprocessExecutor will do IO to and from
- // 'requestSenderPipe' and 'responseReceiverPipe'.
- SubprocessExecutor setUpExecutorIO(Function<Throwable, Void> errorHandler,
- boolean injectIOError) throws IOException {
- // Initialize the pipe that we'll push requests to; feed it into the
- // executor's input pipe.
- PipedInputStream inputPipe = new PipedInputStream();
- requestSenderPipe = new PipedOutputStream(inputPipe);
- System.setIn(inputPipe);
-
- // Initialize the pipe that the executor will write to; feed it into the
- // response pipe that we can read from.
- PipedOutputStream outputPipe = new PipedOutputStream(responseReceiverPipe);
- if (injectIOError) {
- System.setOut(new PrintStreamWithIOException(outputPipe, /*autoFlush*/false, "UTF-8"));
- } else {
- System.setOut(new PrintStream(outputPipe));
- }
- SubprocessExecutor subprocessExecutor = new SubprocessExecutor(errorHandler);
- return subprocessExecutor;
- }
-
/**
* Test a regular old message. There should be no exceptions of any kind.
* We should also see some metrics that make sense.
@@ -163,7 +83,7 @@ public class TestEchoSubprocess {
public void testBasicMsg() throws Exception {
SubprocessExecutor executor =
setUpExecutorIO(NO_ERR, /*injectIOError*/false);
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE));
executor.run(NO_ARGS, new EchoProtocolHandler(), TIMEOUT_MS);
SubprocessResponsePB spResp = receiveResponse();
@@ -196,9 +116,9 @@ public class TestEchoSubprocess {
SubprocessExecutor executor =
setUpExecutorIO(NO_ERR, /*injectIOError*/false);
final int SLEEP_MS = 200;
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE, SLEEP_MS));
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE, SLEEP_MS));
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE, SLEEP_MS));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE, SLEEP_MS));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE, SLEEP_MS));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE, SLEEP_MS));
// Run the executor with a single parser thread so we can make stronger
// assumptions about timing.
@@ -253,9 +173,9 @@ public class TestEchoSubprocess {
setUpExecutorIO(NO_ERR, /*injectIOError*/false);
final int BLOCK_MS = 200;
executor.blockWriteMs(BLOCK_MS);
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE));
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE));
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE));
executor.run(NO_ARGS, new EchoProtocolHandler(), TIMEOUT_MS);
// In writing the first request, the other two requests should've been
@@ -312,7 +232,7 @@ public class TestEchoSubprocess {
public void testInjectIOException() throws Exception {
SubprocessExecutor executor =
setUpExecutorIO(HAS_ERR, /*injectIOError*/true);
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE));
// NOTE: we don't expect the ExecutionException from the MessageWriter's
// CompletableFuture because, in waiting for completion, the MessageReader
// times out before CompletableFuture.get() is called on the writer.
@@ -330,7 +250,7 @@ public class TestEchoSubprocess {
SubprocessExecutor executor =
setUpExecutorIO(HAS_ERR, /*injectIOError*/false);
executor.interrupt();
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest(MESSAGE));
+ sendRequestToPipe(createEchoSubprocessRequest(MESSAGE));
assertIncludingSuppressedThrows(ExecutionException.class,
"Unable to put the message to the queue",
() -> executor.run(NO_ARGS, new EchoProtocolHandler(), TIMEOUT_MS));
@@ -344,8 +264,8 @@ public class TestEchoSubprocess {
public void testMessageParser() throws Exception {
SubprocessExecutor executor =
setUpExecutorIO(NO_ERR, /*injectIOError*/false);
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest("a"));
- sendRequestToPipe(MessageTestUtil.createEchoSubprocessRequest("b"));
+ sendRequestToPipe(createEchoSubprocessRequest("a"));
+ sendRequestToPipe(createEchoSubprocessRequest("b"));
executor.blockWriteMs(1000);
Assert.assertThrows(TimeoutException.class,
() -> executor.run(NO_ARGS, new EchoProtocolHandler(), /*timeoutMs*/500));
diff --git a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java
new file mode 100644
index 0000000..667fa02
--- /dev/null
+++ b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/TestRangerSubprocess.java
@@ -0,0 +1,134 @@
+// 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.
+
+package org.apache.kudu.subprocess.ranger;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
+import com.google.protobuf.Any;
+import org.apache.kudu.ranger.Ranger.ActionPB;
+import org.apache.kudu.ranger.Ranger.RangerRequestListPB;
+import org.apache.kudu.ranger.Ranger.RangerRequestPB;
+import org.apache.kudu.ranger.Ranger.RangerResponseListPB;
+import org.apache.kudu.subprocess.Subprocess.SubprocessRequestPB;
+import org.apache.kudu.subprocess.SubprocessExecutor;
+import org.apache.kudu.subprocess.SubprocessTestUtil;
+import org.apache.kudu.subprocess.ranger.authorization.RangerKuduAuthorizer;
+import org.apache.kudu.test.junit.RetryRule;
+import org.apache.ranger.plugin.model.RangerServiceDef;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+/**
+ * Tests for the ranger subprocess.
+ */
+public class TestRangerSubprocess extends SubprocessTestUtil {
+
+ @Rule
+ public RetryRule retryRule = new RetryRule();
+
+ private static RangerRequestPB createRangerRequest(ActionPB action, String db,
+ String table, String col) {
+ RangerRequestPB.Builder builder = RangerRequestPB.newBuilder();
+ builder.setAction(action);
+ builder.setDatabase(db);
+ builder.setTable(table);
+ builder.setColumn(col);
+ return builder.build();
+ }
+
+ private static RangerRequestListPB createRangerRequestList(
+ List<RangerRequestPB> requests, String user) {
+ RangerRequestListPB.Builder builder = RangerRequestListPB.newBuilder();
+ builder.addAllRequests(requests);
+ builder.setUser(user);
+ return builder.build();
+ }
+
+ private static SubprocessRequestPB createRangerSubprocessRequest(
+ RangerRequestListPB request) {
+ SubprocessRequestPB.Builder builder = SubprocessRequestPB.newBuilder();
+ builder.setRequest(Any.pack(request));
+ return builder.build();
+ }
+
+ @Before
+ public void mockAuthorizer() {
+ RangerProtocolHandler.authz = Mockito.mock(RangerKuduAuthorizer.class);
+ }
+
+ /**
+ * Sends a list of Ranger request and verifies the response by mocking the authorization
+ * decisions.
+ */
+ @Test
+ public void testBasicRangerMessage() throws Exception {
+ final String user = "Alice";
+ final String db = "db";
+ final String table = "table";
+ final String col = "col";
+ final RangerRequestPB updateRequest = createRangerRequest(ActionPB.UPDATE, db, table, col);
+ final RangerRequestPB selectRequest = createRangerRequest(ActionPB.SELECT, db, table, col);
+ final RangerRequestPB createRequest = createRangerRequest(ActionPB.CREATE, db, table, col);
+ final List<RangerRequestPB> requests = new ArrayList<>();
+ // Send multiple ranger requests in one message.
+ requests.add(updateRequest);
+ requests.add(selectRequest);
+ requests.add(createRequest);
+ final RangerRequestListPB requestList = createRangerRequestList(requests, user);
+ final SubprocessRequestPB subprocessRequest = createRangerSubprocessRequest(requestList);
+
+ // Mock the authorization results.
+ List<RangerAccessResult> rangerResults = new ArrayList<>();
+ final RangerAccessResult positiveResult = new RangerAccessResult(
+ /* policyType= */1, "kudu",
+ new RangerServiceDef(), new RangerAccessRequestImpl());
+ positiveResult.setIsAllowed(true);
+ final RangerAccessResult negativeResult = new RangerAccessResult(
+ /* policyType= */1, "kudu",
+ new RangerServiceDef(), new RangerAccessRequestImpl());
+ negativeResult.setIsAllowed(false);
+ rangerResults.add(positiveResult);
+ rangerResults.add(negativeResult);
+ rangerResults.add(positiveResult);
+ Mockito.when(RangerProtocolHandler.authz.authorize(requestList))
+ .thenReturn(rangerResults);
+
+ SubprocessExecutor executor =
+ setUpExecutorIO(NO_ERR, /*injectIOError*/false);
+ sendRequestToPipe(subprocessRequest);
+ // We expect the executor to time out since it is non cancelable
+ // if no exception encountered.
+ assertThrows(TimeoutException.class,
+ () -> executor.run(NO_ARGS, new RangerProtocolHandler(), TIMEOUT_MS));
+
+ RangerResponseListPB resp = receiveResponse().getResponse().unpack(RangerResponseListPB.class);
+ assertTrue(resp.getResponses(/* index= */0).getAllowed());
+ assertFalse(resp.getResponses(/* index= */1).getAllowed());
+ assertTrue(resp.getResponses(/* index= */2).getAllowed());
+ }
+}
\ No newline at end of file
diff --git a/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/authorization/TestRangerKuduAuthorizer.java b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/authorization/TestRangerKuduAuthorizer.java
new file mode 100644
index 0000000..e90c8ad
--- /dev/null
+++ b/java/kudu-subprocess/src/test/java/org/apache/kudu/subprocess/ranger/authorization/TestRangerKuduAuthorizer.java
@@ -0,0 +1,81 @@
+// 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.
+
+package org.apache.kudu.subprocess.ranger.authorization;
+
+import org.apache.kudu.test.junit.RetryRule;
+import org.apache.ranger.plugin.model.RangerServiceDef;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequest;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for the Ranger authorizer.
+ */
+public class TestRangerKuduAuthorizer {
+
+ @Rule
+ public RetryRule retryRule = new RetryRule();
+
+ /**
+ * Generates a few ranger authorization results and verifies the
+ * Ranger authorizer work as expected.
+ */
+ @Test
+ public void testBasicRangerAuthorizer() {
+ RangerKuduAuthorizer authz = new RangerKuduAuthorizer();
+ authz.plugin = Mockito.mock(RangerBasePlugin.class);
+ // We have to mock RangerAccessRequestImpl as it does not implement equals().
+ // Mock with a positive authz result.
+ RangerAccessRequestImpl mockUpdateRequest = Mockito.mock(RangerAccessRequestImpl.class);
+ final RangerAccessResult updateResult = new RangerAccessResult(
+ /* policyType= */1, "kudu",
+ new RangerServiceDef(), mockUpdateRequest);
+ updateResult.setIsAllowed(true);
+
+ // Mock with a negative authz result.
+ RangerAccessRequestImpl mockCreateRequest = Mockito.mock(RangerAccessRequestImpl.class);
+ final RangerAccessResult createResult = new RangerAccessResult(
+ /* policyType= */1, "kudu",
+ new RangerServiceDef(), mockCreateRequest);
+ createResult.setIsAllowed(false);
+
+ Collection<RangerAccessRequest> requests = new ArrayList<>();
+ requests.add(mockUpdateRequest);
+ requests.add(mockCreateRequest);
+ Collection<RangerAccessResult> results = new ArrayList<>();
+ results.add(updateResult);
+ results.add(createResult);
+
+ Mockito.when(authz.plugin.isAccessAllowed(requests))
+ .thenReturn(results);
+ Iterator<RangerAccessResult> actualResultsIter = authz.authorize(requests).iterator();
+ assertTrue(actualResultsIter.next().getIsAllowed());
+ assertFalse(actualResultsIter.next().getIsAllowed());
+ }
+}
\ No newline at end of file
diff --git a/src/kudu/ranger/CMakeLists.txt b/src/kudu/ranger/CMakeLists.txt
new file mode 100644
index 0000000..6a21976
--- /dev/null
+++ b/src/kudu/ranger/CMakeLists.txt
@@ -0,0 +1,33 @@
+# 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.
+
+#######################################
+# ranger_proto
+#######################################
+
+PROTOBUF_GENERATE_CPP(
+ RANGER_PROTO_SRCS RANGER_PROTO_HDRS RANGER_PROTO_TGTS
+ SOURCE_ROOT ${CMAKE_CURRENT_SOURCE_DIR}/../..
+ BINARY_ROOT ${CMAKE_CURRENT_BINARY_DIR}/../..
+ PROTO_FILES ranger.proto)
+
+add_library(ranger_proto
+ ${RANGER_PROTO_SRCS}
+ ${RANGER_PROTO_HDRS})
+target_link_libraries(ranger_proto
+ protobuf
+)
diff --git a/src/kudu/ranger/ranger.proto b/src/kudu/ranger/ranger.proto
new file mode 100644
index 0000000..4848ea3
--- /dev/null
+++ b/src/kudu/ranger/ranger.proto
@@ -0,0 +1,73 @@
+// 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.
+syntax = "proto2";
+package kudu.ranger;
+option java_package = "org.apache.kudu.ranger";
+
+// Describes the type of action that can be performed in Ranger.
+enum ActionPB {
+ SELECT = 0;
+ INSERT = 1;
+ UPDATE = 2;
+ DELETE = 3;
+ ALTER = 4;
+ CREATE = 5;
+ DROP = 6;
+ ALL = 7;
+ METADATA = 8;
+}
+
+// The following protobuf definitions are used as interface between
+// Kudu master and the Ranger authorization plugin operating in a
+// Java subprocess that is managed by the Kudu master.
+
+// Describes a single Ranger authorization request per user.
+message RangerRequestPB {
+ // Database name the action is to be performed on.
+ optional string database = 1;
+
+ // Table name the action is to be performed on.
+ optional string table = 2;
+
+ // Column name the action is to be performed on.
+ optional string column = 3;
+
+ // Action to be authorized on the resource.
+ optional ActionPB action = 4;
+}
+
+// Describes a single Ranger authorization response for a single user.
+message RangerResponsePB {
+ // Whether or not a request is allowed.
+ optional bool allowed = 1;
+}
+
+// Describes a list of authorization requests sent to the Ranger service
+// for a single user.
+message RangerRequestListPB {
+ repeated RangerRequestPB requests = 1;
+
+ // User performing the action.
+ optional string user = 2;
+}
+
+// Describes a list of authorization descison responded from the Ranger
+// service for a request for a single user. Note that the order of
+// responses is determined by the order of requests in RangerRequestListPB.
+message RangerResponseListPB {
+ repeated RangerResponsePB responses = 1;
+}
[kudu] 03/04: [build] Use Gradle to build the hms-plugin jar
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 2d858f66a0ac91a56128e4e2a90b6c81db05df41
Author: Grant Henke <gr...@apache.org>
AuthorDate: Wed Mar 4 08:35:30 2020 -0600
[build] Use Gradle to build the hms-plugin jar
In an effort to unify how we build Java from cmake/make, this
patch converts the `hms-plugin.jar` compilation from using
`add_jar` to using `add_custom_command` calling a Gradle
command.
In order to ensure mutliple calls to Gradle don’t try and
download the `gradle-wrapper.jar` at the same time, I added an
`init-gradle` target that all Gradle commands should depend on.
I also added `--quiet --console=plain` command line flags to
the Gradle commands to prevent weird Gradle logging within the
make build.
Change-Id: I3312b1605c4f050380264688f9d1b78038608257
Reviewed-on: http://gerrit.cloudera.org:8080/15360
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
CMakeLists.txt | 18 +++++++++++++++++-
build-support/run_dist_test.py | 8 ++++++--
src/kudu/hms/CMakeLists.txt | 19 ++++++++++---------
src/kudu/subprocess/CMakeLists.txt | 9 +++------
4 files changed, 36 insertions(+), 18 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index b3207f8..3a56fb5 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -76,7 +76,6 @@ if (SLASH_POS EQUAL -1)
endfunction()
endif()
-# TODO: can we somehow pass this into the java build?
file(STRINGS "${CMAKE_CURRENT_SOURCE_DIR}/version.txt" KUDU_VERSION_NUMBER)
set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake_modules")
@@ -94,6 +93,23 @@ set(THIRDPARTY_INSTALL_TSAN_DIR ${THIRDPARTY_INSTALL_DIR}/tsan)
# This can be useful to, e.g. spawn a Java subprocess.
set(JAVA_DIR ${CMAKE_CURRENT_SOURCE_DIR}/java)
+# Adds the EXTRA_GRADLE_FLAGS environment variable to a cmake list and calls `separate_arguments`.
+# This solves the problem of escaped spaces in the EXTRA_GRADLE_FLAGS environment variable.
+list(APPEND GRADLE_FLAGS $ENV{EXTRA_GRADLE_FLAGS})
+separate_arguments(GRADLE_FLAGS)
+# We always want Gradle to use the plain console and quiet flag when called from cmake.
+list(APPEND GRADLE_FLAGS --quiet --console=plain)
+list(REMOVE_DUPLICATES GRADLE_FLAGS)
+
+# Initialize Gradle to ensure the wrapper is downloaded.
+# This is important so that multiple calls to Gradle don't try and
+# download the gradle-wrapper.jar at the same time.
+set(GRADLE_WRAPPER_JAR ${JAVA_DIR}/gradle/wrapper/gradle-wrapper.jar)
+add_custom_command(OUTPUT ${GRADLE_WRAPPER_JAR}
+ COMMAND ./gradlew --version ${GRADLE_FLAGS}
+ WORKING_DIRECTORY "${JAVA_DIR}")
+add_custom_target(init_gradle DEPENDS ${GRADLE_WRAPPER_JAR})
+
# Allow "make install" to not depend on all targets.
#
# Must be declared in the top-level CMakeLists.txt.
diff --git a/build-support/run_dist_test.py b/build-support/run_dist_test.py
index d613b44..99b3a78 100755
--- a/build-support/run_dist_test.py
+++ b/build-support/run_dist_test.py
@@ -153,8 +153,8 @@ def main():
env['SENTRY_HOME'] = glob.glob(os.path.join(ROOT, "thirdparty/src/sentry-*"))[0]
env['JAVA_HOME'] = glob.glob("/usr/lib/jvm/java-1.8.0-*")[0]
- # Restore the symlinks to the chrony and subprocess binaries; tests expect to
- # find them in same directory as the test binaries themselves.
+ # Restore the symlinks to the chrony, hms-plugin, and subprocess binaries;
+ # tests expect to find them in same directory as the test binaries themselves.
for bin_path in glob.glob(os.path.join(ROOT, "build/*/bin")):
os.symlink(os.path.join(ROOT, "thirdparty/installed/common/bin/chronyc"),
os.path.join(bin_path, "chronyc"))
@@ -162,6 +162,10 @@ def main():
os.path.join(bin_path, "chronyd"))
os.symlink(
glob.glob(os.path.join(ROOT,
+ "java/kudu-hive/build/libs/kudu-hive*"))[0],
+ os.path.join(bin_path, "hms-plugin.jar"))
+ os.symlink(
+ glob.glob(os.path.join(ROOT,
"java/kudu-subprocess/build/libs/kudu-subprocess*"))[0],
os.path.join(bin_path, "kudu-subprocess.jar"))
diff --git a/src/kudu/hms/CMakeLists.txt b/src/kudu/hms/CMakeLists.txt
index 62a764b..2815f83 100644
--- a/src/kudu/hms/CMakeLists.txt
+++ b/src/kudu/hms/CMakeLists.txt
@@ -63,15 +63,16 @@ execute_process(COMMAND ln -nsf
"${JAVA_HOME}"
"${EXECUTABLE_OUTPUT_PATH}/java-home")
-file(GLOB DEPENDENCY_JARS
- "${CMAKE_SOURCE_DIR}/thirdparty/installed/common/opt/hive/lib/*"
- "${CMAKE_SOURCE_DIR}/thirdparty/installed/common/opt/hadoop/share/hadoop/common/*"
- "${CMAKE_SOURCE_DIR}/thirdparty/installed/common/opt/hadoop/share/hadoop/common/lib/*")
+set(HMS_PLUGIN_JAR ${EXECUTABLE_OUTPUT_PATH}/hms-plugin.jar)
-add_jar(hms-plugin
- "${CMAKE_SOURCE_DIR}/java/kudu-hive/src/main/java/org/apache/kudu/hive/metastore/KuduMetastorePlugin.java"
- INCLUDE_JARS "${DEPENDENCY_JARS}"
- OUTPUT_DIR "${EXECUTABLE_OUTPUT_PATH}")
+add_custom_command(OUTPUT ${HMS_PLUGIN_JAR}
+ COMMAND ./gradlew :kudu-hive:jar ${GRADLE_FLAGS}
+ COMMAND ln -nsf
+ "${JAVA_DIR}/kudu-hive/build/libs/kudu-hive-${KUDU_VERSION_NUMBER}.jar"
+ "${HMS_PLUGIN_JAR}"
+ WORKING_DIRECTORY "${JAVA_DIR}"
+ DEPENDS init_gradle)
+add_custom_target(hms_plugin_jar DEPENDS ${HMS_PLUGIN_JAR})
set(MINI_HMS_SRCS
mini_hms.cc)
@@ -82,7 +83,7 @@ target_link_libraries(mini_hms
krpc
kudu_test_util
kudu_util)
-add_dependencies(mini_hms hms-plugin)
+add_dependencies(mini_hms hms_plugin_jar)
##############################
# hms tests
diff --git a/src/kudu/subprocess/CMakeLists.txt b/src/kudu/subprocess/CMakeLists.txt
index 8f284f5..a7514a4 100644
--- a/src/kudu/subprocess/CMakeLists.txt
+++ b/src/kudu/subprocess/CMakeLists.txt
@@ -37,16 +37,13 @@ target_link_libraries(subprocess_proto
#######################################
set(SUBPROCESS_JAR ${EXECUTABLE_OUTPUT_PATH}/kudu-subprocess.jar)
-# Add the EXTRA_GRADLE_FLAGS environment variable to a cmake list and calls `separate_arguments`.
-# This solves the problem of escaped spaces in the EXTRA_GRADLE_FLAGS environment variable.
-list(APPEND EXTRA_GRADLE_FLAGS $ENV{EXTRA_GRADLE_FLAGS})
-separate_arguments(EXTRA_GRADLE_FLAGS)
add_custom_command(OUTPUT ${SUBPROCESS_JAR}
- COMMAND ./gradlew :kudu-subprocess:jar ${EXTRA_GRADLE_FLAGS}
+ COMMAND ./gradlew :kudu-subprocess:jar ${GRADLE_FLAGS}
COMMAND ln -nsf
"${JAVA_DIR}/kudu-subprocess/build/libs/kudu-subprocess-${KUDU_VERSION_NUMBER}.jar"
"${SUBPROCESS_JAR}"
- WORKING_DIRECTORY "${JAVA_DIR}")
+ WORKING_DIRECTORY "${JAVA_DIR}"
+ DEPENDS init_gradle)
add_custom_target(subprocess_jar DEPENDS ${SUBPROCESS_JAR})
add_library(kudu_subprocess