You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by sn...@apache.org on 2016/08/29 14:25:02 UTC
incubator-ranger git commit: RANGER-1133 - Add tests for the Storm
plugin
Repository: incubator-ranger
Updated Branches:
refs/heads/master b15d6fa8a -> 7729373de
RANGER-1133 - Add tests for the Storm plugin
Signed-off-by: Selvamohan Neethiraj <sn...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-ranger/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ranger/commit/7729373d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ranger/tree/7729373d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ranger/diff/7729373d
Branch: refs/heads/master
Commit: 7729373dea5a560f5cb4b24433d12b25a9dc7519
Parents: b15d6fa
Author: Colm O hEigeartaigh <co...@apache.org>
Authored: Tue Aug 2 12:25:55 2016 +0100
Committer: Selvamohan Neethiraj <sn...@apache.org>
Committed: Mon Aug 29 10:13:59 2016 -0400
----------------------------------------------------------------------
.../ranger/plugin/util/RangerResourceTrie.java | 2 +-
.../security/listener/SpringEventListener.java | 3 +-
storm-agent/pom.xml | 27 +-
.../storm/RangerAdminClientImpl.java | 84 +++++
.../storm/StormRangerAuthorizerTest.java | 190 +++++++++++
.../authorization/storm/WordCounterBolt.java | 66 ++++
.../ranger/authorization/storm/WordSpout.java | 68 ++++
.../test/resources/ranger-storm-security.xml | 45 +++
.../src/test/resources/storm-policies.json | 337 +++++++++++++++++++
storm-agent/src/test/resources/storm.yaml | 289 ++++++++++++++++
storm-agent/src/test/resources/words.txt | 27 ++
11 files changed, 1132 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
----------------------------------------------------------------------
diff --git a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
index 809c07e..a5ffd1a 100644
--- a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
+++ b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
@@ -59,7 +59,7 @@ public class RangerResourceTrie {
this.resourceName = resourceDef.getName();
this.optIgnoreCase = strIgnoreCase != null ? Boolean.parseBoolean(strIgnoreCase) : false;
- this.optWildcard = strWildcard != null ? Boolean.parseBoolean(strWildcard) : false;;
+ this.optWildcard = strWildcard != null ? Boolean.parseBoolean(strWildcard) : false;
this.wildcardChars = optWildcard ? DEFAULT_WILDCARD_CHARS : "";
this.root = new TrieNode(Character.valueOf((char)0));
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
----------------------------------------------------------------------
diff --git a/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java b/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
index 5876445..29a35cf 100644
--- a/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
+++ b/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
@@ -50,9 +50,8 @@ public class SpringEventListener implements
process((AuthenticationFailureBadCredentialsEvent) event);
} else if (event instanceof AuthenticationFailureDisabledEvent) {
process((AuthenticationFailureDisabledEvent) event);
- } else {
- // igonre all other events
}
+ // igonre all other events
} catch (Exception e) {
logger.error("Exception in Spring Event Listener.", e);
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/pom.xml
----------------------------------------------------------------------
diff --git a/storm-agent/pom.xml b/storm-agent/pom.xml
index b975a92..d49c2fe 100644
--- a/storm-agent/pom.xml
+++ b/storm-agent/pom.xml
@@ -35,6 +35,12 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>${storm.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
@@ -93,11 +99,26 @@
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore</artifactId>
<version>${httpcomponents.httpcore.version}</version>
- </dependency>
- <dependency>
+ </dependency>
+ <dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>${commons.codec.version}</version>
- </dependency>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
</dependencies>
+ <build>
+ <testResources>
+ <testResource>
+ <directory>src/test/resources</directory>
+ <includes>
+ <include>**/*</include>
+ </includes>
+ <filtering>true</filtering>
+ </testResource>
+ </testResources>
+ </build>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java
new file mode 100644
index 0000000..e6c289a
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java
@@ -0,0 +1,84 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.io.File;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.util.List;
+
+import org.apache.ranger.admin.client.RangerAdminClient;
+import org.apache.ranger.plugin.util.GrantRevokeRequest;
+import org.apache.ranger.plugin.util.ServicePolicies;
+import org.apache.ranger.plugin.util.ServiceTags;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+
+/**
+ * A test implementation of the RangerAdminClient interface that just reads policies in from a file and returns them
+ */
+public class RangerAdminClientImpl implements RangerAdminClient {
+ private static final Logger LOG = LoggerFactory.getLogger(RangerAdminClientImpl.class);
+ private final static String cacheFilename = "storm-policies.json";
+ private Gson gson;
+
+ public void init(String serviceName, String appId, String configPropertyPrefix) {
+ Gson gson = null;
+ try {
+ gson = new GsonBuilder().setDateFormat("yyyyMMdd-HH:mm:ss.SSS-Z").setPrettyPrinting().create();
+ } catch(Throwable excp) {
+ LOG.error("RangerAdminClientImpl: failed to create GsonBuilder object", excp);
+ }
+ this.gson = gson;
+ }
+
+ public ServicePolicies getServicePoliciesIfUpdated(long lastKnownVersion) throws Exception {
+
+ String basedir = System.getProperty("basedir");
+ if (basedir == null) {
+ basedir = new File(".").getCanonicalPath();
+ }
+
+ java.nio.file.Path cachePath = FileSystems.getDefault().getPath(basedir, "/src/test/resources/" + cacheFilename);
+ byte[] cacheBytes = Files.readAllBytes(cachePath);
+
+ return gson.fromJson(new String(cacheBytes), ServicePolicies.class);
+ }
+
+ public void grantAccess(GrantRevokeRequest request) throws Exception {
+
+ }
+
+ public void revokeAccess(GrantRevokeRequest request) throws Exception {
+
+ }
+
+ public ServiceTags getServiceTagsIfUpdated(long lastKnownVersion) throws Exception {
+ return null;
+
+ }
+
+ public List<String> getTagTypes(String tagTypePattern) throws Exception {
+ return null;
+ }
+
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java
new file mode 100644
index 0000000..0c249c5
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.security.Principal;
+import java.security.PrivilegedExceptionAction;
+
+import javax.security.auth.Subject;
+
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.generated.RebalanceOptions;
+import org.apache.storm.topology.TopologyBuilder;
+import org.junit.Assert;
+
+/**
+ * A simple test that wires a WordSpout + WordCounterBolt into a topology and runs it. The "RangerStormAuthorizer" takes care of authorization.
+ * The policies state that "bob" can do anything with the "word-count" topology. In addition, "bob" can create/kill the "temp*" topologies, but do
+ * nothing else.
+ */
+public class StormRangerAuthorizerTest {
+
+ private static LocalCluster cluster;
+
+ @org.junit.BeforeClass
+ public static void setup() throws Exception {
+ cluster = new LocalCluster();
+
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+
+ // bob can create a new topology
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ cluster.submitTopology("word-count", conf, builder.createTopology());
+ return null;
+ }
+ });
+
+ }
+
+ @org.junit.AfterClass
+ public static void cleanup() throws Exception {
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ cluster.killTopology("word-count");
+ return null;
+ }
+ });
+
+ cluster.shutdown();
+ System.clearProperty("storm.conf.file");
+ }
+
+ // "bob" can't create topologies other than "word-count" and "temp*"
+ @org.junit.Test
+ public void testCreateTopologyBob() throws Exception {
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ try {
+ cluster.submitTopology("word-count2", conf, builder.createTopology());
+ Assert.fail("Authorization failure expected");
+ } catch (Throwable ex) {
+ // expected
+ }
+
+ return null;
+ }
+ });
+ }
+
+ @org.junit.Test
+ public void testTopologyActivation() throws Exception {
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+
+ // Deactivate "word-count"
+ cluster.deactivate("word-count");
+
+ // Create a new topology called "temp1"
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+ cluster.submitTopology("temp1", conf, builder.createTopology());
+
+ // Try to deactivate "temp1"
+ try {
+ cluster.deactivate("temp1");
+ Assert.fail("Authorization failure expected");
+ } catch (Throwable ex) {
+ // expected
+ }
+
+ // Re-activate "word-count"
+ cluster.activate("word-count");
+
+ // Kill temp1
+ cluster.killTopology("temp1");
+
+ return null;
+ }
+ });
+ }
+
+ @org.junit.Test
+ public void testTopologyRebalancing() throws Exception {
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ RebalanceOptions options = new RebalanceOptions();
+
+ // Create a new topology called "temp2"
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+ cluster.submitTopology("temp2", conf, builder.createTopology());
+
+ // Try to rebalance "temp2"
+ try {
+ cluster.rebalance("temp2", options);
+ Assert.fail("Authorization failure expected");
+ } catch (Throwable ex) {
+ // expected
+ }
+
+ // Kill temp2
+ cluster.killTopology("temp2");
+
+ return null;
+ }
+ });
+ }
+
+
+ private static class SimplePrincipal implements Principal {
+
+ private final String name;
+
+ public SimplePrincipal(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java
new file mode 100644
index 0000000..0e327c7
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java
@@ -0,0 +1,66 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+
+/**
+ * A Storm Bolt which reads in a word and counts it + outputs the word + current count
+ */
+public class WordCounterBolt extends BaseRichBolt {
+ private OutputCollector outputCollector;
+ private Map<String, Integer> countMap = new HashMap<>();
+
+ @Override
+ public void execute(Tuple tuple) {
+ String word = tuple.getString(0);
+
+ int count = 0;
+ if (countMap.containsKey(word)) {
+ count = countMap.get(word);
+ count++;
+ }
+ count++;
+ countMap.put(word, count);
+
+ outputCollector.emit(new Values(word, count));
+ outputCollector.ack(tuple);
+
+ }
+
+ @Override
+ public void prepare(Map arg0, TopologyContext arg1, OutputCollector outputCollector) {
+ this.outputCollector = outputCollector;
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("word", "count"));
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java
new file mode 100644
index 0000000..5f0b2cf
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java
@@ -0,0 +1,68 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.storm.shade.org.apache.commons.io.IOUtils;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+/**
+ * A Storm Spout which reads in words.txt + emits a word from it (sequentially)
+ */
+public class WordSpout extends BaseRichSpout {
+ private final List<String> words;
+ private SpoutOutputCollector collector;
+ private int line = 0;
+
+ public WordSpout() throws Exception {
+ java.io.File inputFile = new java.io.File(WordSpout.class.getResource("../../../../../words.txt").toURI());
+ words = IOUtils.readLines(new java.io.FileInputStream(inputFile));
+ }
+
+ @Override
+ public void nextTuple() {
+ if (line < words.size()) {
+ String lineVal = words.get(line++);
+ while (lineVal.startsWith("#") && line < words.size()) {
+ lineVal = words.get(line++);
+ }
+ if (lineVal != null) {
+ collector.emit(new Values(lineVal.trim()));
+ }
+ }
+ }
+
+ @Override
+ public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("word"));
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/ranger-storm-security.xml
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/ranger-storm-security.xml b/storm-agent/src/test/resources/ranger-storm-security.xml
new file mode 100644
index 0000000..adff2b9
--- /dev/null
+++ b/storm-agent/src/test/resources/ranger-storm-security.xml
@@ -0,0 +1,45 @@
+<?xml version="1.0"?>
+<!--
+ 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.
+-->
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+<configuration xmlns:xi="http://www.w3.org/2001/XInclude">
+
+ <property>
+ <name>ranger.plugin.storm.service.name</name>
+ <value>StormTest</value>
+ <description>
+ Name of the Ranger service containing policies for this SampleApp instance
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.plugin.storm.policy.source.impl</name>
+ <value>org.apache.ranger.authorization.storm.RangerAdminClientImpl</value>
+ <description>
+ Policy source.
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.plugin.storm.policy.cache.dir</name>
+ <value>${project.build.directory}</value>
+ <description>
+ Directory where Ranger policies are cached after successful retrieval from the source
+ </description>
+ </property>
+
+</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/storm-policies.json
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/storm-policies.json b/storm-agent/src/test/resources/storm-policies.json
new file mode 100644
index 0000000..5c04b5d
--- /dev/null
+++ b/storm-agent/src/test/resources/storm-policies.json
@@ -0,0 +1,337 @@
+{
+ "serviceName": "StormTest",
+ "serviceId": 10,
+ "policyVersion": 12,
+ "policyUpdateTime": "20160704-15:53:20.000-+0100",
+ "policies": [
+ {
+ "service": "StormTest",
+ "name": "WordCount",
+ "description": "",
+ "resourceSignature": "25dc26943b5859a6e5f904388cd02830",
+ "isAuditEnabled": true,
+ "resources": {
+ "topology": {
+ "values": [
+ "word-count"
+ ],
+ "isExcludes": false,
+ "isRecursive": false
+ }
+ },
+ "policyItems": [
+ {
+ "accesses": [
+ {
+ "type": "submitTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "fileUpload",
+ "isAllowed": true
+ },
+ {
+ "type": "getNimbusConf",
+ "isAllowed": true
+ },
+ {
+ "type": "getClusterInfo",
+ "isAllowed": true
+ },
+ {
+ "type": "fileDownload",
+ "isAllowed": true
+ },
+ {
+ "type": "killTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "rebalance",
+ "isAllowed": true
+ },
+ {
+ "type": "activate",
+ "isAllowed": true
+ },
+ {
+ "type": "deactivate",
+ "isAllowed": true
+ },
+ {
+ "type": "getTopologyConf",
+ "isAllowed": true
+ },
+ {
+ "type": "getTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "getUserTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "getTopologyInfo",
+ "isAllowed": true
+ },
+ {
+ "type": "uploadNewCredentials",
+ "isAllowed": true
+ }
+ ],
+ "users": [
+ "bob"
+ ],
+ "groups": [],
+ "conditions": [],
+ "delegateAdmin": false
+ }
+ ],
+ "denyPolicyItems": [],
+ "allowExceptions": [],
+ "denyExceptions": [],
+ "dataMaskPolicyItems": [],
+ "rowFilterPolicyItems": [],
+ "id": 40,
+ "guid": "1467386247700_275_2404",
+ "isEnabled": true,
+ "createdBy": "Admin",
+ "updatedBy": "Admin",
+ "createTime": "20160701-16:17:27.000-+0100",
+ "updateTime": "20160704-14:51:01.000-+0100",
+ "version": 6
+ },
+ {
+ "service": "StormTest",
+ "name": "TempPolicy",
+ "description": "",
+ "resourceSignature": "1e4cafdc98da3cec11b565ef03cfab14",
+ "isAuditEnabled": true,
+ "resources": {
+ "topology": {
+ "values": [
+ "temp*"
+ ],
+ "isExcludes": false,
+ "isRecursive": false
+ }
+ },
+ "policyItems": [
+ {
+ "accesses": [
+ {
+ "type": "submitTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "getClusterInfo",
+ "isAllowed": true
+ },
+ {
+ "type": "killTopology",
+ "isAllowed": true
+ }
+ ],
+ "users": [
+ "bob"
+ ],
+ "groups": [],
+ "conditions": [],
+ "delegateAdmin": false
+ }
+ ],
+ "denyPolicyItems": [],
+ "allowExceptions": [],
+ "denyExceptions": [],
+ "dataMaskPolicyItems": [],
+ "rowFilterPolicyItems": [],
+ "id": 42,
+ "guid": "1467641649473_569_2619",
+ "isEnabled": true,
+ "createdBy": "Admin",
+ "updatedBy": "Admin",
+ "createTime": "20160704-15:14:09.000-+0100",
+ "updateTime": "20160704-15:53:20.000-+0100",
+ "version": 5
+ }
+ ],
+ "serviceDef": {
+ "name": "storm",
+ "implClass": "org.apache.ranger.services.storm.RangerServiceStorm",
+ "label": "Storm",
+ "description": "Storm",
+ "options": {},
+ "configs": [
+ {
+ "itemId": 1,
+ "name": "username",
+ "type": "string",
+ "mandatory": true,
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Username"
+ },
+ {
+ "itemId": 2,
+ "name": "password",
+ "type": "password",
+ "mandatory": true,
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Password"
+ },
+ {
+ "itemId": 3,
+ "name": "nimbus.url",
+ "type": "string",
+ "mandatory": true,
+ "defaultValue": "",
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Nimbus URL"
+ },
+ {
+ "itemId": 4,
+ "name": "commonNameForCertificate",
+ "type": "string",
+ "mandatory": false,
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Common Name for Certificate"
+ }
+ ],
+ "resources": [
+ {
+ "itemId": 1,
+ "name": "topology",
+ "type": "string",
+ "level": 10,
+ "mandatory": true,
+ "lookupSupported": true,
+ "recursiveSupported": false,
+ "excludesSupported": true,
+ "matcher": "org.apache.ranger.plugin.resourcematcher.RangerDefaultResourceMatcher",
+ "matcherOptions": {
+ "wildCard": "true",
+ "ignoreCase": "false"
+ },
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Storm Topology",
+ "description": "Storm Topology"
+ }
+ ],
+ "accessTypes": [
+ {
+ "itemId": 1,
+ "name": "submitTopology",
+ "label": "Submit Topology",
+ "impliedGrants": [
+ "fileUpload",
+ "fileDownload"
+ ]
+ },
+ {
+ "itemId": 2,
+ "name": "fileUpload",
+ "label": "File Upload",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 3,
+ "name": "getNimbusConf",
+ "label": "Get Nimbus Conf",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 4,
+ "name": "getClusterInfo",
+ "label": "Get Cluster Info",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 5,
+ "name": "fileDownload",
+ "label": "File Download",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 6,
+ "name": "killTopology",
+ "label": "Kill Topology",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 7,
+ "name": "rebalance",
+ "label": "Rebalance",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 8,
+ "name": "activate",
+ "label": "Activate",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 9,
+ "name": "deactivate",
+ "label": "Deactivate",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 10,
+ "name": "getTopologyConf",
+ "label": "Get Topology Conf",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 11,
+ "name": "getTopology",
+ "label": "Get Topology",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 12,
+ "name": "getUserTopology",
+ "label": "Get User Topology",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 13,
+ "name": "getTopologyInfo",
+ "label": "Get Topology Info",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 14,
+ "name": "uploadNewCredentials",
+ "label": "Upload New Credential",
+ "impliedGrants": []
+ }
+ ],
+ "policyConditions": [],
+ "contextEnrichers": [],
+ "enums": [],
+ "dataMaskDef": {
+ "maskTypes": [],
+ "accessTypes": [],
+ "resources": []
+ },
+ "rowFilterDef": {
+ "accessTypes": [],
+ "resources": []
+ },
+ "id": 6,
+ "guid": "2a60f427-edcf-4e20-834c-a9a267b5b963",
+ "isEnabled": true,
+ "createTime": "20160314-14:39:35.000-+0000",
+ "updateTime": "20160314-14:39:35.000-+0000",
+ "version": 1
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/storm.yaml
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/storm.yaml b/storm-agent/src/test/resources/storm.yaml
new file mode 100644
index 0000000..a306966
--- /dev/null
+++ b/storm-agent/src/test/resources/storm.yaml
@@ -0,0 +1,289 @@
+# 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.
+
+
+########### These all have default values as shown
+########### Additional configuration goes into storm.yaml
+
+java.library.path: "/usr/local/lib:/opt/local/lib:/usr/lib"
+
+### storm.* configs are general configurations
+# the local dir is where jars are kept
+storm.local.dir: "storm-local"
+storm.log4j2.conf.dir: "log4j2"
+storm.zookeeper.servers:
+ - "localhost"
+storm.zookeeper.port: 2181
+storm.zookeeper.root: "/storm"
+storm.zookeeper.session.timeout: 20000
+storm.zookeeper.connection.timeout: 15000
+storm.zookeeper.retry.times: 5
+storm.zookeeper.retry.interval: 1000
+storm.zookeeper.retry.intervalceiling.millis: 30000
+storm.zookeeper.auth.user: null
+storm.zookeeper.auth.password: null
+storm.exhibitor.port: 8080
+storm.exhibitor.poll.uripath: "/exhibitor/v1/cluster/list"
+storm.cluster.mode: "distributed" # can be distributed or local
+storm.local.mode.zmq: false
+storm.thrift.transport: "org.apache.storm.security.auth.SimpleTransportPlugin"
+storm.principal.tolocal: "org.apache.storm.security.auth.DefaultPrincipalToLocal"
+storm.group.mapping.service: "org.apache.storm.security.auth.ShellBasedGroupsMapping"
+storm.group.mapping.service.params: null
+storm.messaging.transport: "org.apache.storm.messaging.netty.Context"
+storm.nimbus.retry.times: 5
+storm.nimbus.retry.interval.millis: 2000
+storm.nimbus.retry.intervalceiling.millis: 60000
+storm.auth.simple-white-list.users: []
+storm.auth.simple-acl.users: []
+storm.auth.simple-acl.users.commands: []
+storm.auth.simple-acl.admins: []
+storm.cluster.state.store: "org.apache.storm.cluster_state.zookeeper_state_factory"
+storm.meta.serialization.delegate: "org.apache.storm.serialization.GzipThriftSerializationDelegate"
+storm.codedistributor.class: "org.apache.storm.codedistributor.LocalFileSystemCodeDistributor"
+storm.workers.artifacts.dir: "workers-artifacts"
+storm.health.check.dir: "healthchecks"
+storm.health.check.timeout.ms: 5000
+
+### nimbus.* configs are for the master
+nimbus.seeds : ["localhost"]
+nimbus.thrift.port: 6627
+nimbus.thrift.threads: 64
+nimbus.thrift.max_buffer_size: 1048576
+nimbus.childopts: "-Xmx1024m"
+nimbus.task.timeout.secs: 30
+nimbus.supervisor.timeout.secs: 60
+nimbus.monitor.freq.secs: 10
+nimbus.cleanup.inbox.freq.secs: 600
+nimbus.inbox.jar.expiration.secs: 3600
+nimbus.code.sync.freq.secs: 120
+nimbus.task.launch.secs: 120
+nimbus.file.copy.expiration.secs: 600
+nimbus.topology.validator: "org.apache.storm.nimbus.DefaultTopologyValidator"
+topology.min.replication.count: 1
+topology.max.replication.wait.time.sec: 60
+nimbus.credential.renewers.freq.secs: 600
+nimbus.impersonation.authorizer: "org.apache.storm.security.auth.authorizer.ImpersonationAuthorizer"
+# Plug in ranger nimbus.authorizer here
+nimbus.authorizer: "org.apache.ranger.authorization.storm.authorizer.RangerStormAuthorizer"
+nimbus.queue.size: 100000
+scheduler.display.resource: false
+
+### ui.* configs are for the master
+ui.host: 0.0.0.0
+ui.port: 8080
+ui.childopts: "-Xmx768m"
+ui.actions.enabled: true
+ui.filter: null
+ui.filter.params: null
+ui.users: null
+ui.header.buffer.bytes: 4096
+ui.http.creds.plugin: org.apache.storm.security.auth.DefaultHttpCredentialsPlugin
+
+logviewer.port: 8000
+logviewer.childopts: "-Xmx128m"
+logviewer.cleanup.age.mins: 10080
+logviewer.appender.name: "A1"
+logviewer.max.sum.worker.logs.size.mb: 4096
+logviewer.max.per.worker.logs.size.mb: 2048
+
+logs.users: null
+
+drpc.port: 3772
+drpc.worker.threads: 64
+drpc.max_buffer_size: 1048576
+drpc.queue.size: 128
+drpc.invocations.port: 3773
+drpc.invocations.threads: 64
+drpc.request.timeout.secs: 600
+drpc.childopts: "-Xmx768m"
+drpc.http.port: 3774
+drpc.https.port: -1
+drpc.https.keystore.password: ""
+drpc.https.keystore.type: "JKS"
+drpc.http.creds.plugin: org.apache.storm.security.auth.DefaultHttpCredentialsPlugin
+drpc.authorizer.acl.filename: "drpc-auth-acl.yaml"
+drpc.authorizer.acl.strict: false
+
+transactional.zookeeper.root: "/transactional"
+transactional.zookeeper.servers: null
+transactional.zookeeper.port: null
+
+## blobstore configs
+supervisor.blobstore.class: "org.apache.storm.blobstore.NimbusBlobStore"
+supervisor.blobstore.download.thread.count: 5
+supervisor.blobstore.download.max_retries: 3
+supervisor.localizer.cache.target.size.mb: 10240
+supervisor.localizer.cleanup.interval.ms: 600000
+
+nimbus.blobstore.class: "org.apache.storm.blobstore.LocalFsBlobStore"
+nimbus.blobstore.expiration.secs: 600
+
+storm.blobstore.inputstream.buffer.size.bytes: 65536
+client.blobstore.class: "org.apache.storm.blobstore.NimbusBlobStore"
+storm.blobstore.replication.factor: 3
+
+### supervisor.* configs are for node supervisors
+# Define the amount of workers that can be run on this machine. Each worker is assigned a port to use for communication
+supervisor.slots.ports:
+ - 6700
+ - 6701
+ - 6702
+ - 6703
+supervisor.childopts: "-Xmx256m"
+supervisor.run.worker.as.user: false
+#how long supervisor will wait to ensure that a worker process is started
+supervisor.worker.start.timeout.secs: 120
+#how long between heartbeats until supervisor considers that worker dead and tries to restart it
+supervisor.worker.timeout.secs: 30
+#how many seconds to sleep for before shutting down threads on worker
+supervisor.worker.shutdown.sleep.secs: 1
+#how frequently the supervisor checks on the status of the processes it's monitoring and restarts if necessary
+supervisor.monitor.frequency.secs: 3
+#how frequently the supervisor heartbeats to the cluster state (for nimbus)
+supervisor.heartbeat.frequency.secs: 5
+supervisor.enable: true
+supervisor.supervisors: []
+supervisor.supervisors.commands: []
+supervisor.memory.capacity.mb: 3072.0
+#By convention 1 cpu core should be about 100, but this can be adjusted if needed
+# using 100 makes it simple to set the desired value to the capacity measurement
+# for single threaded bolts
+supervisor.cpu.capacity: 400.0
+
+### worker.* configs are for task workers
+worker.heap.memory.mb: 768
+worker.childopts: "-Xmx%HEAP-MEM%m -XX:+PrintGCDetails -Xloggc:artifacts/gc.log -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=1M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=artifacts/heapdump"
+worker.gc.childopts: ""
+
+# Unlocking commercial features requires a special license from Oracle.
+# See http://www.oracle.com/technetwork/java/javase/terms/products/index.html
+# For this reason, profiler features are disabled by default.
+worker.profiler.enabled: false
+worker.profiler.childopts: "-XX:+UnlockCommercialFeatures -XX:+FlightRecorder"
+worker.profiler.command: "flight.bash"
+worker.heartbeat.frequency.secs: 1
+
+# check whether dynamic log levels can be reset from DEBUG to INFO in workers
+worker.log.level.reset.poll.secs: 30
+
+# control how many worker receiver threads we need per worker
+topology.worker.receiver.thread.count: 1
+
+task.heartbeat.frequency.secs: 3
+task.refresh.poll.secs: 10
+task.credentials.poll.secs: 30
+
+# now should be null by default
+topology.backpressure.enable: true
+backpressure.disruptor.high.watermark: 0.9
+backpressure.disruptor.low.watermark: 0.4
+
+zmq.threads: 1
+zmq.linger.millis: 5000
+zmq.hwm: 0
+
+
+storm.messaging.netty.server_worker_threads: 1
+storm.messaging.netty.client_worker_threads: 1
+storm.messaging.netty.buffer_size: 5242880 #5MB buffer
+# Since nimbus.task.launch.secs and supervisor.worker.start.timeout.secs are 120, other workers should also wait at least that long before giving up on connecting to the other worker. The reconnection period need also be bigger than storm.zookeeper.session.timeout(default is 20s), so that we can abort the reconnection when the target worker is dead.
+storm.messaging.netty.max_retries: 300
+storm.messaging.netty.max_wait_ms: 1000
+storm.messaging.netty.min_wait_ms: 100
+
+# If the Netty messaging layer is busy(netty internal buffer not writable), the Netty client will try to batch message as more as possible up to the size of storm.messaging.netty.transfer.batch.size bytes, otherwise it will try to flush message as soon as possible to reduce latency.
+storm.messaging.netty.transfer.batch.size: 262144
+# Sets the backlog value to specify when the channel binds to a local address
+storm.messaging.netty.socket.backlog: 500
+
+# By default, the Netty SASL authentication is set to false. Users can override and set it true for a specific topology.
+storm.messaging.netty.authentication: false
+
+# Default plugin to use for automatic network topology discovery
+storm.network.topography.plugin: org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping
+
+# default number of seconds group mapping service will cache user group
+storm.group.mapping.service.cache.duration.secs: 120
+
+### topology.* configs are for specific executing storms
+topology.enable.message.timeouts: true
+topology.debug: false
+topology.workers: 1
+topology.acker.executors: null
+topology.eventlogger.executors: 0
+topology.tasks: null
+# maximum amount of time a message has to complete before it's considered failed
+topology.message.timeout.secs: 30
+topology.multilang.serializer: "org.apache.storm.multilang.JsonSerializer"
+topology.shellbolt.max.pending: 100
+topology.skip.missing.kryo.registrations: false
+topology.max.task.parallelism: null
+topology.max.spout.pending: null
+topology.state.synchronization.timeout.secs: 60
+topology.stats.sample.rate: 0.05
+topology.builtin.metrics.bucket.size.secs: 60
+topology.fall.back.on.java.serialization: true
+topology.worker.childopts: null
+topology.worker.logwriter.childopts: "-Xmx64m"
+topology.executor.receive.buffer.size: 1024 #batched
+topology.executor.send.buffer.size: 1024 #individual messages
+topology.transfer.buffer.size: 1024 # batched
+topology.tick.tuple.freq.secs: null
+topology.worker.shared.thread.pool.size: 4
+topology.spout.wait.strategy: "org.apache.storm.spout.SleepSpoutWaitStrategy"
+topology.sleep.spout.wait.strategy.time.ms: 1
+topology.error.throttle.interval.secs: 10
+topology.max.error.report.per.interval: 5
+topology.kryo.factory: "org.apache.storm.serialization.DefaultKryoFactory"
+topology.tuple.serializer: "org.apache.storm.serialization.types.ListDelegateSerializer"
+topology.trident.batch.emit.interval.millis: 500
+topology.testing.always.try.serialize: false
+topology.classpath: null
+topology.environment: null
+topology.bolts.outgoing.overflow.buffer.enable: false
+topology.disruptor.wait.timeout.millis: 1000
+topology.disruptor.batch.size: 100
+topology.disruptor.batch.timeout.millis: 1
+topology.disable.loadaware: false
+topology.state.checkpoint.interval.ms: 1000
+
+# Configs for Resource Aware Scheduler
+# topology priority describing the importance of the topology in decreasing importance starting from 0 (i.e. 0 is the highest priority and the priority importance decreases as the priority number increases).
+# Recommended range of 0-29 but no hard limit set.
+topology.priority: 29
+topology.component.resources.onheap.memory.mb: 128.0
+topology.component.resources.offheap.memory.mb: 0.0
+topology.component.cpu.pcore.percent: 10.0
+topology.worker.max.heap.size.mb: 768.0
+topology.scheduler.strategy: "org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy"
+resource.aware.scheduler.eviction.strategy: "org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy"
+resource.aware.scheduler.priority.strategy: "org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy"
+
+dev.zookeeper.path: "/tmp/dev-storm-zookeeper"
+
+pacemaker.host: "localhost"
+pacemaker.port: 6699
+pacemaker.base.threads: 10
+pacemaker.max.threads: 50
+pacemaker.thread.timeout: 10
+pacemaker.childopts: "-Xmx1024m"
+pacemaker.auth.method: "NONE"
+pacemaker.kerberos.users: []
+
+#default storm daemon metrics reporter plugins
+storm.daemon.metrics.reporter.plugins:
+ - "org.apache.storm.daemon.metrics.reporters.JmxPreparableReporter"
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/words.txt
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/words.txt b/storm-agent/src/test/resources/words.txt
new file mode 100644
index 0000000..c7725df
--- /dev/null
+++ b/storm-agent/src/test/resources/words.txt
@@ -0,0 +1,27 @@
+\ufeff# 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.
+the
+and
+storm
+random
+word
+spout
+test
+apache
+the
+spoon
+and
+the
\ No newline at end of file
Re: incubator-ranger git commit: RANGER-1133 - Add tests for the
Storm plugin
Posted by Colm O hEigeartaigh <co...@apache.org>.
Hi Madhan,
The build errors only occur with JDK 1.7, which is why I didn't spot them.
Upgrading Storm from 1.0.0 -> 1.0.2 fixes the problem, which suggests that
the build errors are due to a bug in Storm. Any objections if I perform
this upgrade?
Colm.
On Mon, Aug 29, 2016 at 6:47 PM, Madhan Neethiraj <
mneethiraj@hortonworks.com> wrote:
> Colm,
>
> Afer this commit, master build fails with the following error. Can you
> please review?
>
> Thanks,
> Madhan
>
> 19276 [Thread-12] ERROR o.a.s.event - Error when processing event
> org.apache.storm.generated.AuthorizationException
> at org.apache.storm.blobstore.BlobStoreAclHandler.hasPermissions(BlobStoreAclHandler.java:292)
> ~[storm-core-1.0.0.jar:1.0.0]
> at org.apache.storm.blobstore.LocalFsBlobStore.getBlob(LocalFsBlobStore.java:240)
> ~[storm-core-1.0.0.jar:1.0.0]
> at org.apache.storm.blobstore.BlobStore.readBlobTo(BlobStore.java:271)
> ~[storm-core-1.0.0.jar:1.0.0]
> at org.apache.storm.daemon.supervisor$fn__9351$fn__9352.invoke(supervisor.clj:1176)
> ~[storm-core-1.0.0.jar:1.0.0]
> at org.apache.storm.daemon.supervisor$fn__9351.invoke(supervisor.clj:1172)
> ~[storm-core-1.0.0.jar:1.0.0]
> at clojure.lang.MultiFn.invoke(MultiFn.java:243)
> ~[clojure-1.7.0.jar:?]
> at org.apache.storm.daemon.supervisor$mk_synchronize_
> supervisor$this__9070$fn__9088.invoke(supervisor.clj:582)
> ~[storm-core-1.0.0.jar:1.0.0]
> at org.apache.storm.daemon.supervisor$mk_synchronize_
> supervisor$this__9070.invoke(supervisor.clj:581)
> ~[storm-core-1.0.0.jar:1.0.0]
> at org.apache.storm.event$event_manager$fn__8622.invoke(event.clj:40)
> [storm-core-1.0.0.jar:1.0.0]
> at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
> at java.lang.Thread.run(Thread.java:745) [?:1.7.0_71]
> 19277 [Curator-Framework-0] INFO o.a.s.s.o.a.c.f.i.CuratorFrameworkImpl
> - backgroundOperationsLoop exiting
> 19278 [ProcessThread(sid:0 cport:-1):] INFO o.a.s.s.o.a.z.s.PrepRequestProcessor
> - Processed session termination for sessionid: 0x156d760f67c0010
> 19279 [main] INFO o.a.s.s.o.a.z.ZooKeeper - Session: 0x156d760f67c0010
> closed
> 19279 [main-EventThread] INFO o.a.s.s.o.a.z.ClientCnxn - EventThread shut
> down
> 19279 [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2000] INFO
> o.a.s.s.o.a.z.s.NIOServerCnxn - Closed socket connection for client
> /fe80:0:0:0:0:0:0:1%1:56076 which had sessionid 0x156d760f67c0010
> 19279 [main] INFO o.a.s.cluster - setup-path/blobstore/temp2-2-
> 1472492282-stormcode.ser/10.22.8.99:6627-1
> 19283 [Thread-12] ERROR o.a.s.util - Halting process: ("Error when
> processing an event")
> java.lang.RuntimeException: ("Error when processing an event")
> at org.apache.storm.util$exit_process_BANG_.doInvoke(util.clj:341)
> [storm-core-1.0.0.jar:1.0.0]
> at clojure.lang.RestFn.invoke(RestFn.java:423)
> [clojure-1.7.0.jar:?]
> at org.apache.storm.event$event_manager$fn__8622.invoke(event.clj:48)
> [storm-core-1.0.0.jar:1.0.0]
> at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
> at java.lang.Thread.run(Thread.java:745) [?:1.7.0_71]
>
> Results :
>
> Tests run: 0, Failures: 0, Errors: 0, Skipped: 0
>
>
>
>
>
> On 8/29/16, 7:25 AM, "sneethir@apache.org" <sn...@apache.org> wrote:
>
> Repository: incubator-ranger
> Updated Branches:
> refs/heads/master b15d6fa8a -> 7729373de
>
>
> RANGER-1133 - Add tests for the Storm plugin
>
> Signed-off-by: Selvamohan Neethiraj <sn...@apache.org>
>
>
> Project: http://git-wip-us.apache.org/repos/asf/incubator-ranger/repo
> Commit: http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> commit/7729373d
> Tree: http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> tree/7729373d
> Diff: http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> diff/7729373d
>
> Branch: refs/heads/master
> Commit: 7729373dea5a560f5cb4b24433d12b25a9dc7519
> Parents: b15d6fa
> Author: Colm O hEigeartaigh <co...@apache.org>
> Authored: Tue Aug 2 12:25:55 2016 +0100
> Committer: Selvamohan Neethiraj <sn...@apache.org>
> Committed: Mon Aug 29 10:13:59 2016 -0400
>
> ----------------------------------------------------------------------
> .../ranger/plugin/util/RangerResourceTrie.java | 2 +-
> .../security/listener/SpringEventListener.java | 3 +-
> storm-agent/pom.xml | 27 +-
> .../storm/RangerAdminClientImpl.java | 84 +++++
> .../storm/StormRangerAuthorizerTest.java | 190 +++++++++++
> .../authorization/storm/WordCounterBolt.java | 66 ++++
> .../ranger/authorization/storm/WordSpout.java | 68 ++++
> .../test/resources/ranger-storm-security.xml | 45 +++
> .../src/test/resources/storm-policies.json | 337
> +++++++++++++++++++
> storm-agent/src/test/resources/storm.yaml | 289
> ++++++++++++++++
> storm-agent/src/test/resources/words.txt | 27 ++
> 11 files changed, 1132 insertions(+), 6 deletions(-)
> ----------------------------------------------------------------------
>
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/agents-common/src/main/java/org/apache/ranger/plugin/util/
> RangerResourceTrie.java
> ----------------------------------------------------------------------
> diff --git a/agents-common/src/main/java/
> org/apache/ranger/plugin/util/RangerResourceTrie.java
> b/agents-common/src/main/java/org/apache/ranger/plugin/util/
> RangerResourceTrie.java
> index 809c07e..a5ffd1a 100644
> --- a/agents-common/src/main/java/org/apache/ranger/plugin/util/
> RangerResourceTrie.java
> +++ b/agents-common/src/main/java/org/apache/ranger/plugin/util/
> RangerResourceTrie.java
> @@ -59,7 +59,7 @@ public class RangerResourceTrie {
>
> this.resourceName = resourceDef.getName();
> this.optIgnoreCase = strIgnoreCase != null ?
> Boolean.parseBoolean(strIgnoreCase) : false;
> - this.optWildcard = strWildcard != null ?
> Boolean.parseBoolean(strWildcard) : false;;
> + this.optWildcard = strWildcard != null ?
> Boolean.parseBoolean(strWildcard) : false;
> this.wildcardChars = optWildcard ? DEFAULT_WILDCARD_CHARS :
> "";
> this.root = new TrieNode(Character.valueOf((
> char)0));
>
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/security-admin/src/main/java/org/apache/
> ranger/security/listener/SpringEventListener.java
> ----------------------------------------------------------------------
> diff --git a/security-admin/src/main/java/org/apache/ranger/
> security/listener/SpringEventListener.java b/security-admin/src/main/
> java/org/apache/ranger/security/listener/SpringEventListener.java
> index 5876445..29a35cf 100644
> --- a/security-admin/src/main/java/org/apache/ranger/
> security/listener/SpringEventListener.java
> +++ b/security-admin/src/main/java/org/apache/ranger/
> security/listener/SpringEventListener.java
> @@ -50,9 +50,8 @@ public class SpringEventListener implements
> process((AuthenticationFailureBadCredentialsEvent) event);
> } else if (event instanceof AuthenticationFailureDisabledEvent)
> {
> process((AuthenticationFailureDisabledEvent) event);
> - } else {
> - // igonre all other events
> }
> + // igonre all other events
>
> } catch (Exception e) {
> logger.error("Exception in Spring Event Listener.", e);
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/pom.xml
> ----------------------------------------------------------------------
> diff --git a/storm-agent/pom.xml b/storm-agent/pom.xml
> index b975a92..d49c2fe 100644
> --- a/storm-agent/pom.xml
> +++ b/storm-agent/pom.xml
> @@ -35,6 +35,12 @@
> <groupId>org.apache.storm</groupId>
> <artifactId>storm-core</artifactId>
> <version>${storm.version}</version>
> + <exclusions>
> + <exclusion>
> + <groupId>org.slf4j</groupId>
> + <artifactId>log4j-over-slf4j</artifactId>
> + </exclusion>
> + </exclusions>
> </dependency>
> <dependency>
> <groupId>org.apache.ranger</groupId>
> @@ -93,11 +99,26 @@
> <groupId>org.apache.httpcomponents</groupId>
> <artifactId>httpcore</artifactId>
> <version>${httpcomponents.
> httpcore.version}</version>
> - </dependency>
> - <dependency>
> + </dependency>
> + <dependency>
> <groupId>commons-codec</groupId>
> <artifactId>commons-codec</artifactId>
> <version>${commons.codec.version}</version>
> - </dependency>
> + </dependency>
> + <dependency>
> + <groupId>junit</groupId>
> + <artifactId>junit</artifactId>
> + </dependency>
> </dependencies>
> + <build>
> + <testResources>
> + <testResource>
> + <directory>src/test/resources</directory>
> + <includes>
> + <include>**/*</include>
> + </includes>
> + <filtering>true</filtering>
> + </testResource>
> + </testResources>
> + </build>
> </project>
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/RangerAdminClientImpl.java
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/RangerAdminClientImpl.java
> b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/
> RangerAdminClientImpl.java
> new file mode 100644
> index 0000000..e6c289a
> --- /dev/null
> +++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/
> RangerAdminClientImpl.java
> @@ -0,0 +1,84 @@
> +/*
> + * 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.ranger.authorization.storm;
> +
> +import java.io.File;
> +import java.nio.file.FileSystems;
> +import java.nio.file.Files;
> +import java.util.List;
> +
> +import org.apache.ranger.admin.client.RangerAdminClient;
> +import org.apache.ranger.plugin.util.GrantRevokeRequest;
> +import org.apache.ranger.plugin.util.ServicePolicies;
> +import org.apache.ranger.plugin.util.ServiceTags;
> +import org.slf4j.Logger;
> +import org.slf4j.LoggerFactory;
> +
> +import com.google.gson.Gson;
> +import com.google.gson.GsonBuilder;
> +
> +/**
> + * A test implementation of the RangerAdminClient interface that just
> reads policies in from a file and returns them
> + */
> +public class RangerAdminClientImpl implements RangerAdminClient {
> + private static final Logger LOG = LoggerFactory.getLogger(
> RangerAdminClientImpl.class);
> + private final static String cacheFilename = "storm-policies.json";
> + private Gson gson;
> +
> + public void init(String serviceName, String appId, String
> configPropertyPrefix) {
> + Gson gson = null;
> + try {
> + gson = new GsonBuilder().setDateFormat("
> yyyyMMdd-HH:mm:ss.SSS-Z").setPrettyPrinting().create();
> + } catch(Throwable excp) {
> + LOG.error("RangerAdminClientImpl: failed to create
> GsonBuilder object", excp);
> + }
> + this.gson = gson;
> + }
> +
> + public ServicePolicies getServicePoliciesIfUpdated(long
> lastKnownVersion) throws Exception {
> +
> + String basedir = System.getProperty("basedir");
> + if (basedir == null) {
> + basedir = new File(".").getCanonicalPath();
> + }
> +
> + java.nio.file.Path cachePath = FileSystems.getDefault().getPath(basedir,
> "/src/test/resources/" + cacheFilename);
> + byte[] cacheBytes = Files.readAllBytes(cachePath);
> +
> + return gson.fromJson(new String(cacheBytes),
> ServicePolicies.class);
> + }
> +
> + public void grantAccess(GrantRevokeRequest request) throws
> Exception {
> +
> + }
> +
> + public void revokeAccess(GrantRevokeRequest request) throws
> Exception {
> +
> + }
> +
> + public ServiceTags getServiceTagsIfUpdated(long lastKnownVersion)
> throws Exception {
> + return null;
> +
> + }
> +
> + public List<String> getTagTypes(String tagTypePattern) throws
> Exception {
> + return null;
> + }
> +
> +
> +}
> \ No newline at end of file
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/StormRangerAuthorizerTest.java
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/StormRangerAuthorizerTest.java
> b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/
> StormRangerAuthorizerTest.java
> new file mode 100644
> index 0000000..0c249c5
> --- /dev/null
> +++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/
> StormRangerAuthorizerTest.java
> @@ -0,0 +1,190 @@
> +/*
> + * 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.ranger.authorization.storm;
> +
> +import java.security.Principal;
> +import java.security.PrivilegedExceptionAction;
> +
> +import javax.security.auth.Subject;
> +
> +import org.apache.storm.Config;
> +import org.apache.storm.LocalCluster;
> +import org.apache.storm.generated.RebalanceOptions;
> +import org.apache.storm.topology.TopologyBuilder;
> +import org.junit.Assert;
> +
> +/**
> + * A simple test that wires a WordSpout + WordCounterBolt into a
> topology and runs it. The "RangerStormAuthorizer" takes care of
> authorization.
> + * The policies state that "bob" can do anything with the
> "word-count" topology. In addition, "bob" can create/kill the "temp*"
> topologies, but do
> + * nothing else.
> + */
> +public class StormRangerAuthorizerTest {
> +
> + private static LocalCluster cluster;
> +
> + @org.junit.BeforeClass
> + public static void setup() throws Exception {
> + cluster = new LocalCluster();
> +
> + final Config conf = new Config();
> + conf.setDebug(true);
> +
> + final TopologyBuilder builder = new TopologyBuilder();
> + builder.setSpout("words", new WordSpout());
> + builder.setBolt("counter", new WordCounterBolt()).
> shuffleGrouping("words");
> +
> + // bob can create a new topology
> + final Subject subject = new Subject();
> + subject.getPrincipals().add(new SimplePrincipal("bob"));
> + Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
> + public Void run() throws Exception {
> + cluster.submitTopology("word-count", conf,
> builder.createTopology());
> + return null;
> + }
> + });
> +
> + }
> +
> + @org.junit.AfterClass
> + public static void cleanup() throws Exception {
> + final Subject subject = new Subject();
> + subject.getPrincipals().add(new SimplePrincipal("bob"));
> + Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
> + public Void run() throws Exception {
> + cluster.killTopology("word-count");
> + return null;
> + }
> + });
> +
> + cluster.shutdown();
> + System.clearProperty("storm.conf.file");
> + }
> +
> + // "bob" can't create topologies other than "word-count" and
> "temp*"
> + @org.junit.Test
> + public void testCreateTopologyBob() throws Exception {
> + final Config conf = new Config();
> + conf.setDebug(true);
> +
> + final TopologyBuilder builder = new TopologyBuilder();
> + builder.setSpout("words", new WordSpout());
> + builder.setBolt("counter", new WordCounterBolt()).
> shuffleGrouping("words");
> +
> + final Subject subject = new Subject();
> + subject.getPrincipals().add(new SimplePrincipal("bob"));
> + Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
> + public Void run() throws Exception {
> + try {
> + cluster.submitTopology("word-count2", conf,
> builder.createTopology());
> + Assert.fail("Authorization failure expected");
> + } catch (Throwable ex) {
> + // expected
> + }
> +
> + return null;
> + }
> + });
> + }
> +
> + @org.junit.Test
> + public void testTopologyActivation() throws Exception {
> + final Subject subject = new Subject();
> + subject.getPrincipals().add(new SimplePrincipal("bob"));
> + Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
> + public Void run() throws Exception {
> +
> + // Deactivate "word-count"
> + cluster.deactivate("word-count");
> +
> + // Create a new topology called "temp1"
> + final Config conf = new Config();
> + conf.setDebug(true);
> +
> + final TopologyBuilder builder = new TopologyBuilder();
> + builder.setSpout("words", new WordSpout());
> + builder.setBolt("counter", new WordCounterBolt()).
> shuffleGrouping("words");
> + cluster.submitTopology("temp1", conf,
> builder.createTopology());
> +
> + // Try to deactivate "temp1"
> + try {
> + cluster.deactivate("temp1");
> + Assert.fail("Authorization failure expected");
> + } catch (Throwable ex) {
> + // expected
> + }
> +
> + // Re-activate "word-count"
> + cluster.activate("word-count");
> +
> + // Kill temp1
> + cluster.killTopology("temp1");
> +
> + return null;
> + }
> + });
> + }
> +
> + @org.junit.Test
> + public void testTopologyRebalancing() throws Exception {
> + final Subject subject = new Subject();
> + subject.getPrincipals().add(new SimplePrincipal("bob"));
> + Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
> + public Void run() throws Exception {
> + RebalanceOptions options = new RebalanceOptions();
> +
> + // Create a new topology called "temp2"
> + final Config conf = new Config();
> + conf.setDebug(true);
> +
> + final TopologyBuilder builder = new TopologyBuilder();
> + builder.setSpout("words", new WordSpout());
> + builder.setBolt("counter", new WordCounterBolt()).
> shuffleGrouping("words");
> + cluster.submitTopology("temp2", conf,
> builder.createTopology());
> +
> + // Try to rebalance "temp2"
> + try {
> + cluster.rebalance("temp2", options);
> + Assert.fail("Authorization failure expected");
> + } catch (Throwable ex) {
> + // expected
> + }
> +
> + // Kill temp2
> + cluster.killTopology("temp2");
> +
> + return null;
> + }
> + });
> + }
> +
> +
> + private static class SimplePrincipal implements Principal {
> +
> + private final String name;
> +
> + public SimplePrincipal(String name) {
> + this.name = name;
> + }
> +
> + @Override
> + public String getName() {
> + return name;
> + }
> +
> + }
> +}
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/WordCounterBolt.java
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/WordCounterBolt.java b/storm-agent/src/test/java/
> org/apache/ranger/authorization/storm/WordCounterBolt.java
> new file mode 100644
> index 0000000..0e327c7
> --- /dev/null
> +++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/
> WordCounterBolt.java
> @@ -0,0 +1,66 @@
> +/*
> + * 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.ranger.authorization.storm;
> +
> +import java.util.HashMap;
> +import java.util.Map;
> +
> +import org.apache.storm.task.OutputCollector;
> +import org.apache.storm.task.TopologyContext;
> +import org.apache.storm.topology.OutputFieldsDeclarer;
> +import org.apache.storm.topology.base.BaseRichBolt;
> +import org.apache.storm.tuple.Fields;
> +import org.apache.storm.tuple.Tuple;
> +import org.apache.storm.tuple.Values;
> +
> +/**
> + * A Storm Bolt which reads in a word and counts it + outputs the
> word + current count
> + */
> +public class WordCounterBolt extends BaseRichBolt {
> + private OutputCollector outputCollector;
> + private Map<String, Integer> countMap = new HashMap<>();
> +
> + @Override
> + public void execute(Tuple tuple) {
> + String word = tuple.getString(0);
> +
> + int count = 0;
> + if (countMap.containsKey(word)) {
> + count = countMap.get(word);
> + count++;
> + }
> + count++;
> + countMap.put(word, count);
> +
> + outputCollector.emit(new Values(word, count));
> + outputCollector.ack(tuple);
> +
> + }
> +
> + @Override
> + public void prepare(Map arg0, TopologyContext arg1,
> OutputCollector outputCollector) {
> + this.outputCollector = outputCollector;
> + }
> +
> + @Override
> + public void declareOutputFields(OutputFieldsDeclarer declarer) {
> + declarer.declare(new Fields("word", "count"));
> + }
> +
> +
> +}
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/WordSpout.java
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/WordSpout.java b/storm-agent/src/test/java/
> org/apache/ranger/authorization/storm/WordSpout.java
> new file mode 100644
> index 0000000..5f0b2cf
> --- /dev/null
> +++ b/storm-agent/src/test/java/org/apache/ranger/
> authorization/storm/WordSpout.java
> @@ -0,0 +1,68 @@
> +/*
> + * 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.ranger.authorization.storm;
> +
> +import java.util.List;
> +import java.util.Map;
> +
> +import org.apache.storm.shade.org.apache.commons.io.IOUtils;
> +import org.apache.storm.spout.SpoutOutputCollector;
> +import org.apache.storm.task.TopologyContext;
> +import org.apache.storm.topology.OutputFieldsDeclarer;
> +import org.apache.storm.topology.base.BaseRichSpout;
> +import org.apache.storm.tuple.Fields;
> +import org.apache.storm.tuple.Values;
> +
> +/**
> + * A Storm Spout which reads in words.txt + emits a word from it
> (sequentially)
> + */
> +public class WordSpout extends BaseRichSpout {
> + private final List<String> words;
> + private SpoutOutputCollector collector;
> + private int line = 0;
> +
> + public WordSpout() throws Exception {
> + java.io.File inputFile = new java.io.File(WordSpout.class.
> getResource("../../../../../words.txt").toURI());
> + words = IOUtils.readLines(new java.io.FileInputStream(
> inputFile));
> + }
> +
> + @Override
> + public void nextTuple() {
> + if (line < words.size()) {
> + String lineVal = words.get(line++);
> + while (lineVal.startsWith("#") && line < words.size()) {
> + lineVal = words.get(line++);
> + }
> + if (lineVal != null) {
> + collector.emit(new Values(lineVal.trim()));
> + }
> + }
> + }
> +
> + @Override
> + public void open(Map arg0, TopologyContext arg1,
> SpoutOutputCollector collector) {
> + this.collector = collector;
> + }
> +
> + @Override
> + public void declareOutputFields(OutputFieldsDeclarer declarer) {
> + declarer.declare(new Fields("word"));
> + }
> +
> +
> +}
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/resources/ranger-storm-security.xml
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/resources/ranger-storm-security.xml
> b/storm-agent/src/test/resources/ranger-storm-security.xml
> new file mode 100644
> index 0000000..adff2b9
> --- /dev/null
> +++ b/storm-agent/src/test/resources/ranger-storm-security.xml
> @@ -0,0 +1,45 @@
> +<?xml version="1.0"?>
> +<!--
> + 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.
> +-->
> +<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
> +<configuration xmlns:xi="http://www.w3.org/2001/XInclude">
> +
> + <property>
> + <name>ranger.plugin.storm.service.name</name>
> + <value>StormTest</value>
> + <description>
> + Name of the Ranger service containing policies for
> this SampleApp instance
> + </description>
> + </property>
> +
> + <property>
> + <name>ranger.plugin.storm.policy.source.impl</name>
> + <value>org.apache.ranger.authorization.storm.
> RangerAdminClientImpl</value>
> + <description>
> + Policy source.
> + </description>
> + </property>
> +
> + <property>
> + <name>ranger.plugin.storm.policy.cache.dir</name>
> + <value>${project.build.directory}</value>
> + <description>
> + Directory where Ranger policies are cached after
> successful retrieval from the source
> + </description>
> + </property>
> +
> +</configuration>
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/resources/storm-policies.json
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/resources/storm-policies.json
> b/storm-agent/src/test/resources/storm-policies.json
> new file mode 100644
> index 0000000..5c04b5d
> --- /dev/null
> +++ b/storm-agent/src/test/resources/storm-policies.json
> @@ -0,0 +1,337 @@
> +{
> + "serviceName": "StormTest",
> + "serviceId": 10,
> + "policyVersion": 12,
> + "policyUpdateTime": "20160704-15:53:20.000-+0100",
> + "policies": [
> + {
> + "service": "StormTest",
> + "name": "WordCount",
> + "description": "",
> + "resourceSignature": "25dc26943b5859a6e5f904388cd02830",
> + "isAuditEnabled": true,
> + "resources": {
> + "topology": {
> + "values": [
> + "word-count"
> + ],
> + "isExcludes": false,
> + "isRecursive": false
> + }
> + },
> + "policyItems": [
> + {
> + "accesses": [
> + {
> + "type": "submitTopology",
> + "isAllowed": true
> + },
> + {
> + "type": "fileUpload",
> + "isAllowed": true
> + },
> + {
> + "type": "getNimbusConf",
> + "isAllowed": true
> + },
> + {
> + "type": "getClusterInfo",
> + "isAllowed": true
> + },
> + {
> + "type": "fileDownload",
> + "isAllowed": true
> + },
> + {
> + "type": "killTopology",
> + "isAllowed": true
> + },
> + {
> + "type": "rebalance",
> + "isAllowed": true
> + },
> + {
> + "type": "activate",
> + "isAllowed": true
> + },
> + {
> + "type": "deactivate",
> + "isAllowed": true
> + },
> + {
> + "type": "getTopologyConf",
> + "isAllowed": true
> + },
> + {
> + "type": "getTopology",
> + "isAllowed": true
> + },
> + {
> + "type": "getUserTopology",
> + "isAllowed": true
> + },
> + {
> + "type": "getTopologyInfo",
> + "isAllowed": true
> + },
> + {
> + "type": "uploadNewCredentials",
> + "isAllowed": true
> + }
> + ],
> + "users": [
> + "bob"
> + ],
> + "groups": [],
> + "conditions": [],
> + "delegateAdmin": false
> + }
> + ],
> + "denyPolicyItems": [],
> + "allowExceptions": [],
> + "denyExceptions": [],
> + "dataMaskPolicyItems": [],
> + "rowFilterPolicyItems": [],
> + "id": 40,
> + "guid": "1467386247700_275_2404",
> + "isEnabled": true,
> + "createdBy": "Admin",
> + "updatedBy": "Admin",
> + "createTime": "20160701-16:17:27.000-+0100",
> + "updateTime": "20160704-14:51:01.000-+0100",
> + "version": 6
> + },
> + {
> + "service": "StormTest",
> + "name": "TempPolicy",
> + "description": "",
> + "resourceSignature": "1e4cafdc98da3cec11b565ef03cfab14",
> + "isAuditEnabled": true,
> + "resources": {
> + "topology": {
> + "values": [
> + "temp*"
> + ],
> + "isExcludes": false,
> + "isRecursive": false
> + }
> + },
> + "policyItems": [
> + {
> + "accesses": [
> + {
> + "type": "submitTopology",
> + "isAllowed": true
> + },
> + {
> + "type": "getClusterInfo",
> + "isAllowed": true
> + },
> + {
> + "type": "killTopology",
> + "isAllowed": true
> + }
> + ],
> + "users": [
> + "bob"
> + ],
> + "groups": [],
> + "conditions": [],
> + "delegateAdmin": false
> + }
> + ],
> + "denyPolicyItems": [],
> + "allowExceptions": [],
> + "denyExceptions": [],
> + "dataMaskPolicyItems": [],
> + "rowFilterPolicyItems": [],
> + "id": 42,
> + "guid": "1467641649473_569_2619",
> + "isEnabled": true,
> + "createdBy": "Admin",
> + "updatedBy": "Admin",
> + "createTime": "20160704-15:14:09.000-+0100",
> + "updateTime": "20160704-15:53:20.000-+0100",
> + "version": 5
> + }
> + ],
> + "serviceDef": {
> + "name": "storm",
> + "implClass": "org.apache.ranger.services.
> storm.RangerServiceStorm",
> + "label": "Storm",
> + "description": "Storm",
> + "options": {},
> + "configs": [
> + {
> + "itemId": 1,
> + "name": "username",
> + "type": "string",
> + "mandatory": true,
> + "validationRegEx": "",
> + "validationMessage": "",
> + "uiHint": "",
> + "label": "Username"
> + },
> + {
> + "itemId": 2,
> + "name": "password",
> + "type": "password",
> + "mandatory": true,
> + "validationRegEx": "",
> + "validationMessage": "",
> + "uiHint": "",
> + "label": "Password"
> + },
> + {
> + "itemId": 3,
> + "name": "nimbus.url",
> + "type": "string",
> + "mandatory": true,
> + "defaultValue": "",
> + "validationRegEx": "",
> + "validationMessage": "",
> + "uiHint": "",
> + "label": "Nimbus URL"
> + },
> + {
> + "itemId": 4,
> + "name": "commonNameForCertificate",
> + "type": "string",
> + "mandatory": false,
> + "validationRegEx": "",
> + "validationMessage": "",
> + "uiHint": "",
> + "label": "Common Name for Certificate"
> + }
> + ],
> + "resources": [
> + {
> + "itemId": 1,
> + "name": "topology",
> + "type": "string",
> + "level": 10,
> + "mandatory": true,
> + "lookupSupported": true,
> + "recursiveSupported": false,
> + "excludesSupported": true,
> + "matcher": "org.apache.ranger.plugin.resourcematcher.
> RangerDefaultResourceMatcher",
> + "matcherOptions": {
> + "wildCard": "true",
> + "ignoreCase": "false"
> + },
> + "validationRegEx": "",
> + "validationMessage": "",
> + "uiHint": "",
> + "label": "Storm Topology",
> + "description": "Storm Topology"
> + }
> + ],
> + "accessTypes": [
> + {
> + "itemId": 1,
> + "name": "submitTopology",
> + "label": "Submit Topology",
> + "impliedGrants": [
> + "fileUpload",
> + "fileDownload"
> + ]
> + },
> + {
> + "itemId": 2,
> + "name": "fileUpload",
> + "label": "File Upload",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 3,
> + "name": "getNimbusConf",
> + "label": "Get Nimbus Conf",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 4,
> + "name": "getClusterInfo",
> + "label": "Get Cluster Info",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 5,
> + "name": "fileDownload",
> + "label": "File Download",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 6,
> + "name": "killTopology",
> + "label": "Kill Topology",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 7,
> + "name": "rebalance",
> + "label": "Rebalance",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 8,
> + "name": "activate",
> + "label": "Activate",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 9,
> + "name": "deactivate",
> + "label": "Deactivate",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 10,
> + "name": "getTopologyConf",
> + "label": "Get Topology Conf",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 11,
> + "name": "getTopology",
> + "label": "Get Topology",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 12,
> + "name": "getUserTopology",
> + "label": "Get User Topology",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 13,
> + "name": "getTopologyInfo",
> + "label": "Get Topology Info",
> + "impliedGrants": []
> + },
> + {
> + "itemId": 14,
> + "name": "uploadNewCredentials",
> + "label": "Upload New Credential",
> + "impliedGrants": []
> + }
> + ],
> + "policyConditions": [],
> + "contextEnrichers": [],
> + "enums": [],
> + "dataMaskDef": {
> + "maskTypes": [],
> + "accessTypes": [],
> + "resources": []
> + },
> + "rowFilterDef": {
> + "accessTypes": [],
> + "resources": []
> + },
> + "id": 6,
> + "guid": "2a60f427-edcf-4e20-834c-a9a267b5b963",
> + "isEnabled": true,
> + "createTime": "20160314-14:39:35.000-+0000",
> + "updateTime": "20160314-14:39:35.000-+0000",
> + "version": 1
> + }
> +}
> \ No newline at end of file
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/resources/storm.yaml
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/resources/storm.yaml
> b/storm-agent/src/test/resources/storm.yaml
> new file mode 100644
> index 0000000..a306966
> --- /dev/null
> +++ b/storm-agent/src/test/resources/storm.yaml
> @@ -0,0 +1,289 @@
> +# 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.
> +
> +
> +########### These all have default values as shown
> +########### Additional configuration goes into storm.yaml
> +
> +java.library.path: "/usr/local/lib:/opt/local/lib:/usr/lib"
> +
> +### storm.* configs are general configurations
> +# the local dir is where jars are kept
> +storm.local.dir: "storm-local"
> +storm.log4j2.conf.dir: "log4j2"
> +storm.zookeeper.servers:
> + - "localhost"
> +storm.zookeeper.port: 2181
> +storm.zookeeper.root: "/storm"
> +storm.zookeeper.session.timeout: 20000
> +storm.zookeeper.connection.timeout: 15000
> +storm.zookeeper.retry.times: 5
> +storm.zookeeper.retry.interval: 1000
> +storm.zookeeper.retry.intervalceiling.millis: 30000
> +storm.zookeeper.auth.user: null
> +storm.zookeeper.auth.password: null
> +storm.exhibitor.port: 8080
> +storm.exhibitor.poll.uripath: "/exhibitor/v1/cluster/list"
> +storm.cluster.mode: "distributed" # can be distributed or local
> +storm.local.mode.zmq: false
> +storm.thrift.transport: "org.apache.storm.security.
> auth.SimpleTransportPlugin"
> +storm.principal.tolocal: "org.apache.storm.security.
> auth.DefaultPrincipalToLocal"
> +storm.group.mapping.service: "org.apache.storm.security.
> auth.ShellBasedGroupsMapping"
> +storm.group.mapping.service.params: null
> +storm.messaging.transport: "org.apache.storm.messaging.netty.Context"
> +storm.nimbus.retry.times: 5
> +storm.nimbus.retry.interval.millis: 2000
> +storm.nimbus.retry.intervalceiling.millis: 60000
> +storm.auth.simple-white-list.users: []
> +storm.auth.simple-acl.users: []
> +storm.auth.simple-acl.users.commands: []
> +storm.auth.simple-acl.admins: []
> +storm.cluster.state.store: "org.apache.storm.cluster_
> state.zookeeper_state_factory"
> +storm.meta.serialization.delegate: "org.apache.storm.serialization.
> GzipThriftSerializationDelegate"
> +storm.codedistributor.class: "org.apache.storm.codedistributor.
> LocalFileSystemCodeDistributor"
> +storm.workers.artifacts.dir: "workers-artifacts"
> +storm.health.check.dir: "healthchecks"
> +storm.health.check.timeout.ms: 5000
> +
> +### nimbus.* configs are for the master
> +nimbus.seeds : ["localhost"]
> +nimbus.thrift.port: 6627
> +nimbus.thrift.threads: 64
> +nimbus.thrift.max_buffer_size: 1048576
> +nimbus.childopts: "-Xmx1024m"
> +nimbus.task.timeout.secs: 30
> +nimbus.supervisor.timeout.secs: 60
> +nimbus.monitor.freq.secs: 10
> +nimbus.cleanup.inbox.freq.secs: 600
> +nimbus.inbox.jar.expiration.secs: 3600
> +nimbus.code.sync.freq.secs: 120
> +nimbus.task.launch.secs: 120
> +nimbus.file.copy.expiration.secs: 600
> +nimbus.topology.validator: "org.apache.storm.nimbus.
> DefaultTopologyValidator"
> +topology.min.replication.count: 1
> +topology.max.replication.wait.time.sec: 60
> +nimbus.credential.renewers.freq.secs: 600
> +nimbus.impersonation.authorizer: "org.apache.storm.security.
> auth.authorizer.ImpersonationAuthorizer"
> +# Plug in ranger nimbus.authorizer here
> +nimbus.authorizer: "org.apache.ranger.authorization.storm.authorizer.
> RangerStormAuthorizer"
> +nimbus.queue.size: 100000
> +scheduler.display.resource: false
> +
> +### ui.* configs are for the master
> +ui.host: 0.0.0.0
> +ui.port: 8080
> +ui.childopts: "-Xmx768m"
> +ui.actions.enabled: true
> +ui.filter: null
> +ui.filter.params: null
> +ui.users: null
> +ui.header.buffer.bytes: 4096
> +ui.http.creds.plugin: org.apache.storm.security.auth.
> DefaultHttpCredentialsPlugin
> +
> +logviewer.port: 8000
> +logviewer.childopts: "-Xmx128m"
> +logviewer.cleanup.age.mins: 10080
> +logviewer.appender.name: "A1"
> +logviewer.max.sum.worker.logs.size.mb: 4096
> +logviewer.max.per.worker.logs.size.mb: 2048
> +
> +logs.users: null
> +
> +drpc.port: 3772
> +drpc.worker.threads: 64
> +drpc.max_buffer_size: 1048576
> +drpc.queue.size: 128
> +drpc.invocations.port: 3773
> +drpc.invocations.threads: 64
> +drpc.request.timeout.secs: 600
> +drpc.childopts: "-Xmx768m"
> +drpc.http.port: 3774
> +drpc.https.port: -1
> +drpc.https.keystore.password: ""
> +drpc.https.keystore.type: "JKS"
> +drpc.http.creds.plugin: org.apache.storm.security.auth.
> DefaultHttpCredentialsPlugin
> +drpc.authorizer.acl.filename: "drpc-auth-acl.yaml"
> +drpc.authorizer.acl.strict: false
> +
> +transactional.zookeeper.root: "/transactional"
> +transactional.zookeeper.servers: null
> +transactional.zookeeper.port: null
> +
> +## blobstore configs
> +supervisor.blobstore.class: "org.apache.storm.blobstore.
> NimbusBlobStore"
> +supervisor.blobstore.download.thread.count: 5
> +supervisor.blobstore.download.max_retries: 3
> +supervisor.localizer.cache.target.size.mb: 10240
> +supervisor.localizer.cleanup.interval.ms: 600000
> +
> +nimbus.blobstore.class: "org.apache.storm.blobstore.LocalFsBlobStore"
> +nimbus.blobstore.expiration.secs: 600
> +
> +storm.blobstore.inputstream.buffer.size.bytes: 65536
> +client.blobstore.class: "org.apache.storm.blobstore.NimbusBlobStore"
> +storm.blobstore.replication.factor: 3
> +
> +### supervisor.* configs are for node supervisors
> +# Define the amount of workers that can be run on this machine. Each
> worker is assigned a port to use for communication
> +supervisor.slots.ports:
> + - 6700
> + - 6701
> + - 6702
> + - 6703
> +supervisor.childopts: "-Xmx256m"
> +supervisor.run.worker.as.user: false
> +#how long supervisor will wait to ensure that a worker process is
> started
> +supervisor.worker.start.timeout.secs: 120
> +#how long between heartbeats until supervisor considers that worker
> dead and tries to restart it
> +supervisor.worker.timeout.secs: 30
> +#how many seconds to sleep for before shutting down threads on worker
> +supervisor.worker.shutdown.sleep.secs: 1
> +#how frequently the supervisor checks on the status of the processes
> it's monitoring and restarts if necessary
> +supervisor.monitor.frequency.secs: 3
> +#how frequently the supervisor heartbeats to the cluster state (for
> nimbus)
> +supervisor.heartbeat.frequency.secs: 5
> +supervisor.enable: true
> +supervisor.supervisors: []
> +supervisor.supervisors.commands: []
> +supervisor.memory.capacity.mb: 3072.0
> +#By convention 1 cpu core should be about 100, but this can be
> adjusted if needed
> +# using 100 makes it simple to set the desired value to the capacity
> measurement
> +# for single threaded bolts
> +supervisor.cpu.capacity: 400.0
> +
> +### worker.* configs are for task workers
> +worker.heap.memory.mb: 768
> +worker.childopts: "-Xmx%HEAP-MEM%m -XX:+PrintGCDetails
> -Xloggc:artifacts/gc.log -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps
> -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=1M
> -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=artifacts/heapdump"
> +worker.gc.childopts: ""
> +
> +# Unlocking commercial features requires a special license from
> Oracle.
> +# See http://www.oracle.com/technetwork/java/javase/terms/
> products/index.html
> +# For this reason, profiler features are disabled by default.
> +worker.profiler.enabled: false
> +worker.profiler.childopts: "-XX:+UnlockCommercialFeatures
> -XX:+FlightRecorder"
> +worker.profiler.command: "flight.bash"
> +worker.heartbeat.frequency.secs: 1
> +
> +# check whether dynamic log levels can be reset from DEBUG to INFO in
> workers
> +worker.log.level.reset.poll.secs: 30
> +
> +# control how many worker receiver threads we need per worker
> +topology.worker.receiver.thread.count: 1
> +
> +task.heartbeat.frequency.secs: 3
> +task.refresh.poll.secs: 10
> +task.credentials.poll.secs: 30
> +
> +# now should be null by default
> +topology.backpressure.enable: true
> +backpressure.disruptor.high.watermark: 0.9
> +backpressure.disruptor.low.watermark: 0.4
> +
> +zmq.threads: 1
> +zmq.linger.millis: 5000
> +zmq.hwm: 0
> +
> +
> +storm.messaging.netty.server_worker_threads: 1
> +storm.messaging.netty.client_worker_threads: 1
> +storm.messaging.netty.buffer_size: 5242880 #5MB buffer
> +# Since nimbus.task.launch.secs and supervisor.worker.start.timeout.secs
> are 120, other workers should also wait at least that long before giving up
> on connecting to the other worker. The reconnection period need also be
> bigger than storm.zookeeper.session.timeout(default is 20s), so that we
> can abort the reconnection when the target worker is dead.
> +storm.messaging.netty.max_retries: 300
> +storm.messaging.netty.max_wait_ms: 1000
> +storm.messaging.netty.min_wait_ms: 100
> +
> +# If the Netty messaging layer is busy(netty internal buffer not
> writable), the Netty client will try to batch message as more as possible
> up to the size of storm.messaging.netty.transfer.batch.size bytes,
> otherwise it will try to flush message as soon as possible to reduce
> latency.
> +storm.messaging.netty.transfer.batch.size: 262144
> +# Sets the backlog value to specify when the channel binds to a local
> address
> +storm.messaging.netty.socket.backlog: 500
> +
> +# By default, the Netty SASL authentication is set to false. Users
> can override and set it true for a specific topology.
> +storm.messaging.netty.authentication: false
> +
> +# Default plugin to use for automatic network topology discovery
> +storm.network.topography.plugin: org.apache.storm.networktopography.
> DefaultRackDNSToSwitchMapping
> +
> +# default number of seconds group mapping service will cache user
> group
> +storm.group.mapping.service.cache.duration.secs: 120
> +
> +### topology.* configs are for specific executing storms
> +topology.enable.message.timeouts: true
> +topology.debug: false
> +topology.workers: 1
> +topology.acker.executors: null
> +topology.eventlogger.executors: 0
> +topology.tasks: null
> +# maximum amount of time a message has to complete before it's
> considered failed
> +topology.message.timeout.secs: 30
> +topology.multilang.serializer: "org.apache.storm.multilang.
> JsonSerializer"
> +topology.shellbolt.max.pending: 100
> +topology.skip.missing.kryo.registrations: false
> +topology.max.task.parallelism: null
> +topology.max.spout.pending: null
> +topology.state.synchronization.timeout.secs: 60
> +topology.stats.sample.rate: 0.05
> +topology.builtin.metrics.bucket.size.secs: 60
> +topology.fall.back.on.java.serialization: true
> +topology.worker.childopts: null
> +topology.worker.logwriter.childopts: "-Xmx64m"
> +topology.executor.receive.buffer.size: 1024 #batched
> +topology.executor.send.buffer.size: 1024 #individual messages
> +topology.transfer.buffer.size: 1024 # batched
> +topology.tick.tuple.freq.secs: null
> +topology.worker.shared.thread.pool.size: 4
> +topology.spout.wait.strategy: "org.apache.storm.spout.
> SleepSpoutWaitStrategy"
> +topology.sleep.spout.wait.strategy.time.ms: 1
> +topology.error.throttle.interval.secs: 10
> +topology.max.error.report.per.interval: 5
> +topology.kryo.factory: "org.apache.storm.serialization.
> DefaultKryoFactory"
> +topology.tuple.serializer: "org.apache.storm.serialization.types.
> ListDelegateSerializer"
> +topology.trident.batch.emit.interval.millis: 500
> +topology.testing.always.try.serialize: false
> +topology.classpath: null
> +topology.environment: null
> +topology.bolts.outgoing.overflow.buffer.enable: false
> +topology.disruptor.wait.timeout.millis: 1000
> +topology.disruptor.batch.size: 100
> +topology.disruptor.batch.timeout.millis: 1
> +topology.disable.loadaware: false
> +topology.state.checkpoint.interval.ms: 1000
> +
> +# Configs for Resource Aware Scheduler
> +# topology priority describing the importance of the topology in
> decreasing importance starting from 0 (i.e. 0 is the highest priority and
> the priority importance decreases as the priority number increases).
> +# Recommended range of 0-29 but no hard limit set.
> +topology.priority: 29
> +topology.component.resources.onheap.memory.mb: 128.0
> +topology.component.resources.offheap.memory.mb: 0.0
> +topology.component.cpu.pcore.percent: 10.0
> +topology.worker.max.heap.size.mb: 768.0
> +topology.scheduler.strategy: "org.apache.storm.scheduler.
> resource.strategies.scheduling.DefaultResourceAwareStrategy"
> +resource.aware.scheduler.eviction.strategy:
> "org.apache.storm.scheduler.resource.strategies.eviction.
> DefaultEvictionStrategy"
> +resource.aware.scheduler.priority.strategy:
> "org.apache.storm.scheduler.resource.strategies.priority.
> DefaultSchedulingPriorityStrategy"
> +
> +dev.zookeeper.path: "/tmp/dev-storm-zookeeper"
> +
> +pacemaker.host: "localhost"
> +pacemaker.port: 6699
> +pacemaker.base.threads: 10
> +pacemaker.max.threads: 50
> +pacemaker.thread.timeout: 10
> +pacemaker.childopts: "-Xmx1024m"
> +pacemaker.auth.method: "NONE"
> +pacemaker.kerberos.users: []
> +
> +#default storm daemon metrics reporter plugins
> +storm.daemon.metrics.reporter.plugins:
> + - "org.apache.storm.daemon.metrics.reporters.
> JmxPreparableReporter"
>
> http://git-wip-us.apache.org/repos/asf/incubator-ranger/
> blob/7729373d/storm-agent/src/test/resources/words.txt
> ----------------------------------------------------------------------
> diff --git a/storm-agent/src/test/resources/words.txt
> b/storm-agent/src/test/resources/words.txt
> new file mode 100644
> index 0000000..c7725df
> --- /dev/null
> +++ b/storm-agent/src/test/resources/words.txt
> @@ -0,0 +1,27 @@
> +???# 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.
> +the
> +and
> +storm
> +random
> +word
> +spout
> +test
> +apache
> +the
> +spoon
> +and
> +the
> \ No newline at end of file
>
>
>
>
>
>
>
--
Colm O hEigeartaigh
Talend Community Coder
http://coders.talend.com
Re: incubator-ranger git commit: RANGER-1133 - Add tests for the
Storm plugin
Posted by Madhan Neethiraj <mn...@hortonworks.com>.
Colm,
Afer this commit, master build fails with the following error. Can you please review?
Thanks,
Madhan
19276 [Thread-12] ERROR o.a.s.event - Error when processing event
org.apache.storm.generated.AuthorizationException
at org.apache.storm.blobstore.BlobStoreAclHandler.hasPermissions(BlobStoreAclHandler.java:292) ~[storm-core-1.0.0.jar:1.0.0]
at org.apache.storm.blobstore.LocalFsBlobStore.getBlob(LocalFsBlobStore.java:240) ~[storm-core-1.0.0.jar:1.0.0]
at org.apache.storm.blobstore.BlobStore.readBlobTo(BlobStore.java:271) ~[storm-core-1.0.0.jar:1.0.0]
at org.apache.storm.daemon.supervisor$fn__9351$fn__9352.invoke(supervisor.clj:1176) ~[storm-core-1.0.0.jar:1.0.0]
at org.apache.storm.daemon.supervisor$fn__9351.invoke(supervisor.clj:1172) ~[storm-core-1.0.0.jar:1.0.0]
at clojure.lang.MultiFn.invoke(MultiFn.java:243) ~[clojure-1.7.0.jar:?]
at org.apache.storm.daemon.supervisor$mk_synchronize_supervisor$this__9070$fn__9088.invoke(supervisor.clj:582) ~[storm-core-1.0.0.jar:1.0.0]
at org.apache.storm.daemon.supervisor$mk_synchronize_supervisor$this__9070.invoke(supervisor.clj:581) ~[storm-core-1.0.0.jar:1.0.0]
at org.apache.storm.event$event_manager$fn__8622.invoke(event.clj:40) [storm-core-1.0.0.jar:1.0.0]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:745) [?:1.7.0_71]
19277 [Curator-Framework-0] INFO o.a.s.s.o.a.c.f.i.CuratorFrameworkImpl - backgroundOperationsLoop exiting
19278 [ProcessThread(sid:0 cport:-1):] INFO o.a.s.s.o.a.z.s.PrepRequestProcessor - Processed session termination for sessionid: 0x156d760f67c0010
19279 [main] INFO o.a.s.s.o.a.z.ZooKeeper - Session: 0x156d760f67c0010 closed
19279 [main-EventThread] INFO o.a.s.s.o.a.z.ClientCnxn - EventThread shut down
19279 [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2000] INFO o.a.s.s.o.a.z.s.NIOServerCnxn - Closed socket connection for client /fe80:0:0:0:0:0:0:1%1:56076 which had sessionid 0x156d760f67c0010
19279 [main] INFO o.a.s.cluster - setup-path/blobstore/temp2-2-1472492282-stormcode.ser/10.22.8.99:6627-1
19283 [Thread-12] ERROR o.a.s.util - Halting process: ("Error when processing an event")
java.lang.RuntimeException: ("Error when processing an event")
at org.apache.storm.util$exit_process_BANG_.doInvoke(util.clj:341) [storm-core-1.0.0.jar:1.0.0]
at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.7.0.jar:?]
at org.apache.storm.event$event_manager$fn__8622.invoke(event.clj:48) [storm-core-1.0.0.jar:1.0.0]
at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
at java.lang.Thread.run(Thread.java:745) [?:1.7.0_71]
Results :
Tests run: 0, Failures: 0, Errors: 0, Skipped: 0
On 8/29/16, 7:25 AM, "sneethir@apache.org" <sn...@apache.org> wrote:
Repository: incubator-ranger
Updated Branches:
refs/heads/master b15d6fa8a -> 7729373de
RANGER-1133 - Add tests for the Storm plugin
Signed-off-by: Selvamohan Neethiraj <sn...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-ranger/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ranger/commit/7729373d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ranger/tree/7729373d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ranger/diff/7729373d
Branch: refs/heads/master
Commit: 7729373dea5a560f5cb4b24433d12b25a9dc7519
Parents: b15d6fa
Author: Colm O hEigeartaigh <co...@apache.org>
Authored: Tue Aug 2 12:25:55 2016 +0100
Committer: Selvamohan Neethiraj <sn...@apache.org>
Committed: Mon Aug 29 10:13:59 2016 -0400
----------------------------------------------------------------------
.../ranger/plugin/util/RangerResourceTrie.java | 2 +-
.../security/listener/SpringEventListener.java | 3 +-
storm-agent/pom.xml | 27 +-
.../storm/RangerAdminClientImpl.java | 84 +++++
.../storm/StormRangerAuthorizerTest.java | 190 +++++++++++
.../authorization/storm/WordCounterBolt.java | 66 ++++
.../ranger/authorization/storm/WordSpout.java | 68 ++++
.../test/resources/ranger-storm-security.xml | 45 +++
.../src/test/resources/storm-policies.json | 337 +++++++++++++++++++
storm-agent/src/test/resources/storm.yaml | 289 ++++++++++++++++
storm-agent/src/test/resources/words.txt | 27 ++
11 files changed, 1132 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
----------------------------------------------------------------------
diff --git a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
index 809c07e..a5ffd1a 100644
--- a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
+++ b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerResourceTrie.java
@@ -59,7 +59,7 @@ public class RangerResourceTrie {
this.resourceName = resourceDef.getName();
this.optIgnoreCase = strIgnoreCase != null ? Boolean.parseBoolean(strIgnoreCase) : false;
- this.optWildcard = strWildcard != null ? Boolean.parseBoolean(strWildcard) : false;;
+ this.optWildcard = strWildcard != null ? Boolean.parseBoolean(strWildcard) : false;
this.wildcardChars = optWildcard ? DEFAULT_WILDCARD_CHARS : "";
this.root = new TrieNode(Character.valueOf((char)0));
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
----------------------------------------------------------------------
diff --git a/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java b/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
index 5876445..29a35cf 100644
--- a/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
+++ b/security-admin/src/main/java/org/apache/ranger/security/listener/SpringEventListener.java
@@ -50,9 +50,8 @@ public class SpringEventListener implements
process((AuthenticationFailureBadCredentialsEvent) event);
} else if (event instanceof AuthenticationFailureDisabledEvent) {
process((AuthenticationFailureDisabledEvent) event);
- } else {
- // igonre all other events
}
+ // igonre all other events
} catch (Exception e) {
logger.error("Exception in Spring Event Listener.", e);
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/pom.xml
----------------------------------------------------------------------
diff --git a/storm-agent/pom.xml b/storm-agent/pom.xml
index b975a92..d49c2fe 100644
--- a/storm-agent/pom.xml
+++ b/storm-agent/pom.xml
@@ -35,6 +35,12 @@
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>${storm.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.ranger</groupId>
@@ -93,11 +99,26 @@
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore</artifactId>
<version>${httpcomponents.httpcore.version}</version>
- </dependency>
- <dependency>
+ </dependency>
+ <dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>${commons.codec.version}</version>
- </dependency>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
</dependencies>
+ <build>
+ <testResources>
+ <testResource>
+ <directory>src/test/resources</directory>
+ <includes>
+ <include>**/*</include>
+ </includes>
+ <filtering>true</filtering>
+ </testResource>
+ </testResources>
+ </build>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java
new file mode 100644
index 0000000..e6c289a
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/RangerAdminClientImpl.java
@@ -0,0 +1,84 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.io.File;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.util.List;
+
+import org.apache.ranger.admin.client.RangerAdminClient;
+import org.apache.ranger.plugin.util.GrantRevokeRequest;
+import org.apache.ranger.plugin.util.ServicePolicies;
+import org.apache.ranger.plugin.util.ServiceTags;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+
+/**
+ * A test implementation of the RangerAdminClient interface that just reads policies in from a file and returns them
+ */
+public class RangerAdminClientImpl implements RangerAdminClient {
+ private static final Logger LOG = LoggerFactory.getLogger(RangerAdminClientImpl.class);
+ private final static String cacheFilename = "storm-policies.json";
+ private Gson gson;
+
+ public void init(String serviceName, String appId, String configPropertyPrefix) {
+ Gson gson = null;
+ try {
+ gson = new GsonBuilder().setDateFormat("yyyyMMdd-HH:mm:ss.SSS-Z").setPrettyPrinting().create();
+ } catch(Throwable excp) {
+ LOG.error("RangerAdminClientImpl: failed to create GsonBuilder object", excp);
+ }
+ this.gson = gson;
+ }
+
+ public ServicePolicies getServicePoliciesIfUpdated(long lastKnownVersion) throws Exception {
+
+ String basedir = System.getProperty("basedir");
+ if (basedir == null) {
+ basedir = new File(".").getCanonicalPath();
+ }
+
+ java.nio.file.Path cachePath = FileSystems.getDefault().getPath(basedir, "/src/test/resources/" + cacheFilename);
+ byte[] cacheBytes = Files.readAllBytes(cachePath);
+
+ return gson.fromJson(new String(cacheBytes), ServicePolicies.class);
+ }
+
+ public void grantAccess(GrantRevokeRequest request) throws Exception {
+
+ }
+
+ public void revokeAccess(GrantRevokeRequest request) throws Exception {
+
+ }
+
+ public ServiceTags getServiceTagsIfUpdated(long lastKnownVersion) throws Exception {
+ return null;
+
+ }
+
+ public List<String> getTagTypes(String tagTypePattern) throws Exception {
+ return null;
+ }
+
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java
new file mode 100644
index 0000000..0c249c5
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/StormRangerAuthorizerTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.security.Principal;
+import java.security.PrivilegedExceptionAction;
+
+import javax.security.auth.Subject;
+
+import org.apache.storm.Config;
+import org.apache.storm.LocalCluster;
+import org.apache.storm.generated.RebalanceOptions;
+import org.apache.storm.topology.TopologyBuilder;
+import org.junit.Assert;
+
+/**
+ * A simple test that wires a WordSpout + WordCounterBolt into a topology and runs it. The "RangerStormAuthorizer" takes care of authorization.
+ * The policies state that "bob" can do anything with the "word-count" topology. In addition, "bob" can create/kill the "temp*" topologies, but do
+ * nothing else.
+ */
+public class StormRangerAuthorizerTest {
+
+ private static LocalCluster cluster;
+
+ @org.junit.BeforeClass
+ public static void setup() throws Exception {
+ cluster = new LocalCluster();
+
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+
+ // bob can create a new topology
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ cluster.submitTopology("word-count", conf, builder.createTopology());
+ return null;
+ }
+ });
+
+ }
+
+ @org.junit.AfterClass
+ public static void cleanup() throws Exception {
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ cluster.killTopology("word-count");
+ return null;
+ }
+ });
+
+ cluster.shutdown();
+ System.clearProperty("storm.conf.file");
+ }
+
+ // "bob" can't create topologies other than "word-count" and "temp*"
+ @org.junit.Test
+ public void testCreateTopologyBob() throws Exception {
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ try {
+ cluster.submitTopology("word-count2", conf, builder.createTopology());
+ Assert.fail("Authorization failure expected");
+ } catch (Throwable ex) {
+ // expected
+ }
+
+ return null;
+ }
+ });
+ }
+
+ @org.junit.Test
+ public void testTopologyActivation() throws Exception {
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+
+ // Deactivate "word-count"
+ cluster.deactivate("word-count");
+
+ // Create a new topology called "temp1"
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+ cluster.submitTopology("temp1", conf, builder.createTopology());
+
+ // Try to deactivate "temp1"
+ try {
+ cluster.deactivate("temp1");
+ Assert.fail("Authorization failure expected");
+ } catch (Throwable ex) {
+ // expected
+ }
+
+ // Re-activate "word-count"
+ cluster.activate("word-count");
+
+ // Kill temp1
+ cluster.killTopology("temp1");
+
+ return null;
+ }
+ });
+ }
+
+ @org.junit.Test
+ public void testTopologyRebalancing() throws Exception {
+ final Subject subject = new Subject();
+ subject.getPrincipals().add(new SimplePrincipal("bob"));
+ Subject.doAs(subject, new PrivilegedExceptionAction<Void>() {
+ public Void run() throws Exception {
+ RebalanceOptions options = new RebalanceOptions();
+
+ // Create a new topology called "temp2"
+ final Config conf = new Config();
+ conf.setDebug(true);
+
+ final TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("words", new WordSpout());
+ builder.setBolt("counter", new WordCounterBolt()).shuffleGrouping("words");
+ cluster.submitTopology("temp2", conf, builder.createTopology());
+
+ // Try to rebalance "temp2"
+ try {
+ cluster.rebalance("temp2", options);
+ Assert.fail("Authorization failure expected");
+ } catch (Throwable ex) {
+ // expected
+ }
+
+ // Kill temp2
+ cluster.killTopology("temp2");
+
+ return null;
+ }
+ });
+ }
+
+
+ private static class SimplePrincipal implements Principal {
+
+ private final String name;
+
+ public SimplePrincipal(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java
new file mode 100644
index 0000000..0e327c7
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordCounterBolt.java
@@ -0,0 +1,66 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+
+/**
+ * A Storm Bolt which reads in a word and counts it + outputs the word + current count
+ */
+public class WordCounterBolt extends BaseRichBolt {
+ private OutputCollector outputCollector;
+ private Map<String, Integer> countMap = new HashMap<>();
+
+ @Override
+ public void execute(Tuple tuple) {
+ String word = tuple.getString(0);
+
+ int count = 0;
+ if (countMap.containsKey(word)) {
+ count = countMap.get(word);
+ count++;
+ }
+ count++;
+ countMap.put(word, count);
+
+ outputCollector.emit(new Values(word, count));
+ outputCollector.ack(tuple);
+
+ }
+
+ @Override
+ public void prepare(Map arg0, TopologyContext arg1, OutputCollector outputCollector) {
+ this.outputCollector = outputCollector;
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("word", "count"));
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java
new file mode 100644
index 0000000..5f0b2cf
--- /dev/null
+++ b/storm-agent/src/test/java/org/apache/ranger/authorization/storm/WordSpout.java
@@ -0,0 +1,68 @@
+/*
+ * 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.ranger.authorization.storm;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.storm.shade.org.apache.commons.io.IOUtils;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+/**
+ * A Storm Spout which reads in words.txt + emits a word from it (sequentially)
+ */
+public class WordSpout extends BaseRichSpout {
+ private final List<String> words;
+ private SpoutOutputCollector collector;
+ private int line = 0;
+
+ public WordSpout() throws Exception {
+ java.io.File inputFile = new java.io.File(WordSpout.class.getResource("../../../../../words.txt").toURI());
+ words = IOUtils.readLines(new java.io.FileInputStream(inputFile));
+ }
+
+ @Override
+ public void nextTuple() {
+ if (line < words.size()) {
+ String lineVal = words.get(line++);
+ while (lineVal.startsWith("#") && line < words.size()) {
+ lineVal = words.get(line++);
+ }
+ if (lineVal != null) {
+ collector.emit(new Values(lineVal.trim()));
+ }
+ }
+ }
+
+ @Override
+ public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("word"));
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/ranger-storm-security.xml
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/ranger-storm-security.xml b/storm-agent/src/test/resources/ranger-storm-security.xml
new file mode 100644
index 0000000..adff2b9
--- /dev/null
+++ b/storm-agent/src/test/resources/ranger-storm-security.xml
@@ -0,0 +1,45 @@
+<?xml version="1.0"?>
+<!--
+ 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.
+-->
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+<configuration xmlns:xi="http://www.w3.org/2001/XInclude">
+
+ <property>
+ <name>ranger.plugin.storm.service.name</name>
+ <value>StormTest</value>
+ <description>
+ Name of the Ranger service containing policies for this SampleApp instance
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.plugin.storm.policy.source.impl</name>
+ <value>org.apache.ranger.authorization.storm.RangerAdminClientImpl</value>
+ <description>
+ Policy source.
+ </description>
+ </property>
+
+ <property>
+ <name>ranger.plugin.storm.policy.cache.dir</name>
+ <value>${project.build.directory}</value>
+ <description>
+ Directory where Ranger policies are cached after successful retrieval from the source
+ </description>
+ </property>
+
+</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/storm-policies.json
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/storm-policies.json b/storm-agent/src/test/resources/storm-policies.json
new file mode 100644
index 0000000..5c04b5d
--- /dev/null
+++ b/storm-agent/src/test/resources/storm-policies.json
@@ -0,0 +1,337 @@
+{
+ "serviceName": "StormTest",
+ "serviceId": 10,
+ "policyVersion": 12,
+ "policyUpdateTime": "20160704-15:53:20.000-+0100",
+ "policies": [
+ {
+ "service": "StormTest",
+ "name": "WordCount",
+ "description": "",
+ "resourceSignature": "25dc26943b5859a6e5f904388cd02830",
+ "isAuditEnabled": true,
+ "resources": {
+ "topology": {
+ "values": [
+ "word-count"
+ ],
+ "isExcludes": false,
+ "isRecursive": false
+ }
+ },
+ "policyItems": [
+ {
+ "accesses": [
+ {
+ "type": "submitTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "fileUpload",
+ "isAllowed": true
+ },
+ {
+ "type": "getNimbusConf",
+ "isAllowed": true
+ },
+ {
+ "type": "getClusterInfo",
+ "isAllowed": true
+ },
+ {
+ "type": "fileDownload",
+ "isAllowed": true
+ },
+ {
+ "type": "killTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "rebalance",
+ "isAllowed": true
+ },
+ {
+ "type": "activate",
+ "isAllowed": true
+ },
+ {
+ "type": "deactivate",
+ "isAllowed": true
+ },
+ {
+ "type": "getTopologyConf",
+ "isAllowed": true
+ },
+ {
+ "type": "getTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "getUserTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "getTopologyInfo",
+ "isAllowed": true
+ },
+ {
+ "type": "uploadNewCredentials",
+ "isAllowed": true
+ }
+ ],
+ "users": [
+ "bob"
+ ],
+ "groups": [],
+ "conditions": [],
+ "delegateAdmin": false
+ }
+ ],
+ "denyPolicyItems": [],
+ "allowExceptions": [],
+ "denyExceptions": [],
+ "dataMaskPolicyItems": [],
+ "rowFilterPolicyItems": [],
+ "id": 40,
+ "guid": "1467386247700_275_2404",
+ "isEnabled": true,
+ "createdBy": "Admin",
+ "updatedBy": "Admin",
+ "createTime": "20160701-16:17:27.000-+0100",
+ "updateTime": "20160704-14:51:01.000-+0100",
+ "version": 6
+ },
+ {
+ "service": "StormTest",
+ "name": "TempPolicy",
+ "description": "",
+ "resourceSignature": "1e4cafdc98da3cec11b565ef03cfab14",
+ "isAuditEnabled": true,
+ "resources": {
+ "topology": {
+ "values": [
+ "temp*"
+ ],
+ "isExcludes": false,
+ "isRecursive": false
+ }
+ },
+ "policyItems": [
+ {
+ "accesses": [
+ {
+ "type": "submitTopology",
+ "isAllowed": true
+ },
+ {
+ "type": "getClusterInfo",
+ "isAllowed": true
+ },
+ {
+ "type": "killTopology",
+ "isAllowed": true
+ }
+ ],
+ "users": [
+ "bob"
+ ],
+ "groups": [],
+ "conditions": [],
+ "delegateAdmin": false
+ }
+ ],
+ "denyPolicyItems": [],
+ "allowExceptions": [],
+ "denyExceptions": [],
+ "dataMaskPolicyItems": [],
+ "rowFilterPolicyItems": [],
+ "id": 42,
+ "guid": "1467641649473_569_2619",
+ "isEnabled": true,
+ "createdBy": "Admin",
+ "updatedBy": "Admin",
+ "createTime": "20160704-15:14:09.000-+0100",
+ "updateTime": "20160704-15:53:20.000-+0100",
+ "version": 5
+ }
+ ],
+ "serviceDef": {
+ "name": "storm",
+ "implClass": "org.apache.ranger.services.storm.RangerServiceStorm",
+ "label": "Storm",
+ "description": "Storm",
+ "options": {},
+ "configs": [
+ {
+ "itemId": 1,
+ "name": "username",
+ "type": "string",
+ "mandatory": true,
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Username"
+ },
+ {
+ "itemId": 2,
+ "name": "password",
+ "type": "password",
+ "mandatory": true,
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Password"
+ },
+ {
+ "itemId": 3,
+ "name": "nimbus.url",
+ "type": "string",
+ "mandatory": true,
+ "defaultValue": "",
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Nimbus URL"
+ },
+ {
+ "itemId": 4,
+ "name": "commonNameForCertificate",
+ "type": "string",
+ "mandatory": false,
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Common Name for Certificate"
+ }
+ ],
+ "resources": [
+ {
+ "itemId": 1,
+ "name": "topology",
+ "type": "string",
+ "level": 10,
+ "mandatory": true,
+ "lookupSupported": true,
+ "recursiveSupported": false,
+ "excludesSupported": true,
+ "matcher": "org.apache.ranger.plugin.resourcematcher.RangerDefaultResourceMatcher",
+ "matcherOptions": {
+ "wildCard": "true",
+ "ignoreCase": "false"
+ },
+ "validationRegEx": "",
+ "validationMessage": "",
+ "uiHint": "",
+ "label": "Storm Topology",
+ "description": "Storm Topology"
+ }
+ ],
+ "accessTypes": [
+ {
+ "itemId": 1,
+ "name": "submitTopology",
+ "label": "Submit Topology",
+ "impliedGrants": [
+ "fileUpload",
+ "fileDownload"
+ ]
+ },
+ {
+ "itemId": 2,
+ "name": "fileUpload",
+ "label": "File Upload",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 3,
+ "name": "getNimbusConf",
+ "label": "Get Nimbus Conf",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 4,
+ "name": "getClusterInfo",
+ "label": "Get Cluster Info",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 5,
+ "name": "fileDownload",
+ "label": "File Download",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 6,
+ "name": "killTopology",
+ "label": "Kill Topology",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 7,
+ "name": "rebalance",
+ "label": "Rebalance",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 8,
+ "name": "activate",
+ "label": "Activate",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 9,
+ "name": "deactivate",
+ "label": "Deactivate",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 10,
+ "name": "getTopologyConf",
+ "label": "Get Topology Conf",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 11,
+ "name": "getTopology",
+ "label": "Get Topology",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 12,
+ "name": "getUserTopology",
+ "label": "Get User Topology",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 13,
+ "name": "getTopologyInfo",
+ "label": "Get Topology Info",
+ "impliedGrants": []
+ },
+ {
+ "itemId": 14,
+ "name": "uploadNewCredentials",
+ "label": "Upload New Credential",
+ "impliedGrants": []
+ }
+ ],
+ "policyConditions": [],
+ "contextEnrichers": [],
+ "enums": [],
+ "dataMaskDef": {
+ "maskTypes": [],
+ "accessTypes": [],
+ "resources": []
+ },
+ "rowFilterDef": {
+ "accessTypes": [],
+ "resources": []
+ },
+ "id": 6,
+ "guid": "2a60f427-edcf-4e20-834c-a9a267b5b963",
+ "isEnabled": true,
+ "createTime": "20160314-14:39:35.000-+0000",
+ "updateTime": "20160314-14:39:35.000-+0000",
+ "version": 1
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/storm.yaml
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/storm.yaml b/storm-agent/src/test/resources/storm.yaml
new file mode 100644
index 0000000..a306966
--- /dev/null
+++ b/storm-agent/src/test/resources/storm.yaml
@@ -0,0 +1,289 @@
+# 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.
+
+
+########### These all have default values as shown
+########### Additional configuration goes into storm.yaml
+
+java.library.path: "/usr/local/lib:/opt/local/lib:/usr/lib"
+
+### storm.* configs are general configurations
+# the local dir is where jars are kept
+storm.local.dir: "storm-local"
+storm.log4j2.conf.dir: "log4j2"
+storm.zookeeper.servers:
+ - "localhost"
+storm.zookeeper.port: 2181
+storm.zookeeper.root: "/storm"
+storm.zookeeper.session.timeout: 20000
+storm.zookeeper.connection.timeout: 15000
+storm.zookeeper.retry.times: 5
+storm.zookeeper.retry.interval: 1000
+storm.zookeeper.retry.intervalceiling.millis: 30000
+storm.zookeeper.auth.user: null
+storm.zookeeper.auth.password: null
+storm.exhibitor.port: 8080
+storm.exhibitor.poll.uripath: "/exhibitor/v1/cluster/list"
+storm.cluster.mode: "distributed" # can be distributed or local
+storm.local.mode.zmq: false
+storm.thrift.transport: "org.apache.storm.security.auth.SimpleTransportPlugin"
+storm.principal.tolocal: "org.apache.storm.security.auth.DefaultPrincipalToLocal"
+storm.group.mapping.service: "org.apache.storm.security.auth.ShellBasedGroupsMapping"
+storm.group.mapping.service.params: null
+storm.messaging.transport: "org.apache.storm.messaging.netty.Context"
+storm.nimbus.retry.times: 5
+storm.nimbus.retry.interval.millis: 2000
+storm.nimbus.retry.intervalceiling.millis: 60000
+storm.auth.simple-white-list.users: []
+storm.auth.simple-acl.users: []
+storm.auth.simple-acl.users.commands: []
+storm.auth.simple-acl.admins: []
+storm.cluster.state.store: "org.apache.storm.cluster_state.zookeeper_state_factory"
+storm.meta.serialization.delegate: "org.apache.storm.serialization.GzipThriftSerializationDelegate"
+storm.codedistributor.class: "org.apache.storm.codedistributor.LocalFileSystemCodeDistributor"
+storm.workers.artifacts.dir: "workers-artifacts"
+storm.health.check.dir: "healthchecks"
+storm.health.check.timeout.ms: 5000
+
+### nimbus.* configs are for the master
+nimbus.seeds : ["localhost"]
+nimbus.thrift.port: 6627
+nimbus.thrift.threads: 64
+nimbus.thrift.max_buffer_size: 1048576
+nimbus.childopts: "-Xmx1024m"
+nimbus.task.timeout.secs: 30
+nimbus.supervisor.timeout.secs: 60
+nimbus.monitor.freq.secs: 10
+nimbus.cleanup.inbox.freq.secs: 600
+nimbus.inbox.jar.expiration.secs: 3600
+nimbus.code.sync.freq.secs: 120
+nimbus.task.launch.secs: 120
+nimbus.file.copy.expiration.secs: 600
+nimbus.topology.validator: "org.apache.storm.nimbus.DefaultTopologyValidator"
+topology.min.replication.count: 1
+topology.max.replication.wait.time.sec: 60
+nimbus.credential.renewers.freq.secs: 600
+nimbus.impersonation.authorizer: "org.apache.storm.security.auth.authorizer.ImpersonationAuthorizer"
+# Plug in ranger nimbus.authorizer here
+nimbus.authorizer: "org.apache.ranger.authorization.storm.authorizer.RangerStormAuthorizer"
+nimbus.queue.size: 100000
+scheduler.display.resource: false
+
+### ui.* configs are for the master
+ui.host: 0.0.0.0
+ui.port: 8080
+ui.childopts: "-Xmx768m"
+ui.actions.enabled: true
+ui.filter: null
+ui.filter.params: null
+ui.users: null
+ui.header.buffer.bytes: 4096
+ui.http.creds.plugin: org.apache.storm.security.auth.DefaultHttpCredentialsPlugin
+
+logviewer.port: 8000
+logviewer.childopts: "-Xmx128m"
+logviewer.cleanup.age.mins: 10080
+logviewer.appender.name: "A1"
+logviewer.max.sum.worker.logs.size.mb: 4096
+logviewer.max.per.worker.logs.size.mb: 2048
+
+logs.users: null
+
+drpc.port: 3772
+drpc.worker.threads: 64
+drpc.max_buffer_size: 1048576
+drpc.queue.size: 128
+drpc.invocations.port: 3773
+drpc.invocations.threads: 64
+drpc.request.timeout.secs: 600
+drpc.childopts: "-Xmx768m"
+drpc.http.port: 3774
+drpc.https.port: -1
+drpc.https.keystore.password: ""
+drpc.https.keystore.type: "JKS"
+drpc.http.creds.plugin: org.apache.storm.security.auth.DefaultHttpCredentialsPlugin
+drpc.authorizer.acl.filename: "drpc-auth-acl.yaml"
+drpc.authorizer.acl.strict: false
+
+transactional.zookeeper.root: "/transactional"
+transactional.zookeeper.servers: null
+transactional.zookeeper.port: null
+
+## blobstore configs
+supervisor.blobstore.class: "org.apache.storm.blobstore.NimbusBlobStore"
+supervisor.blobstore.download.thread.count: 5
+supervisor.blobstore.download.max_retries: 3
+supervisor.localizer.cache.target.size.mb: 10240
+supervisor.localizer.cleanup.interval.ms: 600000
+
+nimbus.blobstore.class: "org.apache.storm.blobstore.LocalFsBlobStore"
+nimbus.blobstore.expiration.secs: 600
+
+storm.blobstore.inputstream.buffer.size.bytes: 65536
+client.blobstore.class: "org.apache.storm.blobstore.NimbusBlobStore"
+storm.blobstore.replication.factor: 3
+
+### supervisor.* configs are for node supervisors
+# Define the amount of workers that can be run on this machine. Each worker is assigned a port to use for communication
+supervisor.slots.ports:
+ - 6700
+ - 6701
+ - 6702
+ - 6703
+supervisor.childopts: "-Xmx256m"
+supervisor.run.worker.as.user: false
+#how long supervisor will wait to ensure that a worker process is started
+supervisor.worker.start.timeout.secs: 120
+#how long between heartbeats until supervisor considers that worker dead and tries to restart it
+supervisor.worker.timeout.secs: 30
+#how many seconds to sleep for before shutting down threads on worker
+supervisor.worker.shutdown.sleep.secs: 1
+#how frequently the supervisor checks on the status of the processes it's monitoring and restarts if necessary
+supervisor.monitor.frequency.secs: 3
+#how frequently the supervisor heartbeats to the cluster state (for nimbus)
+supervisor.heartbeat.frequency.secs: 5
+supervisor.enable: true
+supervisor.supervisors: []
+supervisor.supervisors.commands: []
+supervisor.memory.capacity.mb: 3072.0
+#By convention 1 cpu core should be about 100, but this can be adjusted if needed
+# using 100 makes it simple to set the desired value to the capacity measurement
+# for single threaded bolts
+supervisor.cpu.capacity: 400.0
+
+### worker.* configs are for task workers
+worker.heap.memory.mb: 768
+worker.childopts: "-Xmx%HEAP-MEM%m -XX:+PrintGCDetails -Xloggc:artifacts/gc.log -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=1M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=artifacts/heapdump"
+worker.gc.childopts: ""
+
+# Unlocking commercial features requires a special license from Oracle.
+# See http://www.oracle.com/technetwork/java/javase/terms/products/index.html
+# For this reason, profiler features are disabled by default.
+worker.profiler.enabled: false
+worker.profiler.childopts: "-XX:+UnlockCommercialFeatures -XX:+FlightRecorder"
+worker.profiler.command: "flight.bash"
+worker.heartbeat.frequency.secs: 1
+
+# check whether dynamic log levels can be reset from DEBUG to INFO in workers
+worker.log.level.reset.poll.secs: 30
+
+# control how many worker receiver threads we need per worker
+topology.worker.receiver.thread.count: 1
+
+task.heartbeat.frequency.secs: 3
+task.refresh.poll.secs: 10
+task.credentials.poll.secs: 30
+
+# now should be null by default
+topology.backpressure.enable: true
+backpressure.disruptor.high.watermark: 0.9
+backpressure.disruptor.low.watermark: 0.4
+
+zmq.threads: 1
+zmq.linger.millis: 5000
+zmq.hwm: 0
+
+
+storm.messaging.netty.server_worker_threads: 1
+storm.messaging.netty.client_worker_threads: 1
+storm.messaging.netty.buffer_size: 5242880 #5MB buffer
+# Since nimbus.task.launch.secs and supervisor.worker.start.timeout.secs are 120, other workers should also wait at least that long before giving up on connecting to the other worker. The reconnection period need also be bigger than storm.zookeeper.session.timeout(default is 20s), so that we can abort the reconnection when the target worker is dead.
+storm.messaging.netty.max_retries: 300
+storm.messaging.netty.max_wait_ms: 1000
+storm.messaging.netty.min_wait_ms: 100
+
+# If the Netty messaging layer is busy(netty internal buffer not writable), the Netty client will try to batch message as more as possible up to the size of storm.messaging.netty.transfer.batch.size bytes, otherwise it will try to flush message as soon as possible to reduce latency.
+storm.messaging.netty.transfer.batch.size: 262144
+# Sets the backlog value to specify when the channel binds to a local address
+storm.messaging.netty.socket.backlog: 500
+
+# By default, the Netty SASL authentication is set to false. Users can override and set it true for a specific topology.
+storm.messaging.netty.authentication: false
+
+# Default plugin to use for automatic network topology discovery
+storm.network.topography.plugin: org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping
+
+# default number of seconds group mapping service will cache user group
+storm.group.mapping.service.cache.duration.secs: 120
+
+### topology.* configs are for specific executing storms
+topology.enable.message.timeouts: true
+topology.debug: false
+topology.workers: 1
+topology.acker.executors: null
+topology.eventlogger.executors: 0
+topology.tasks: null
+# maximum amount of time a message has to complete before it's considered failed
+topology.message.timeout.secs: 30
+topology.multilang.serializer: "org.apache.storm.multilang.JsonSerializer"
+topology.shellbolt.max.pending: 100
+topology.skip.missing.kryo.registrations: false
+topology.max.task.parallelism: null
+topology.max.spout.pending: null
+topology.state.synchronization.timeout.secs: 60
+topology.stats.sample.rate: 0.05
+topology.builtin.metrics.bucket.size.secs: 60
+topology.fall.back.on.java.serialization: true
+topology.worker.childopts: null
+topology.worker.logwriter.childopts: "-Xmx64m"
+topology.executor.receive.buffer.size: 1024 #batched
+topology.executor.send.buffer.size: 1024 #individual messages
+topology.transfer.buffer.size: 1024 # batched
+topology.tick.tuple.freq.secs: null
+topology.worker.shared.thread.pool.size: 4
+topology.spout.wait.strategy: "org.apache.storm.spout.SleepSpoutWaitStrategy"
+topology.sleep.spout.wait.strategy.time.ms: 1
+topology.error.throttle.interval.secs: 10
+topology.max.error.report.per.interval: 5
+topology.kryo.factory: "org.apache.storm.serialization.DefaultKryoFactory"
+topology.tuple.serializer: "org.apache.storm.serialization.types.ListDelegateSerializer"
+topology.trident.batch.emit.interval.millis: 500
+topology.testing.always.try.serialize: false
+topology.classpath: null
+topology.environment: null
+topology.bolts.outgoing.overflow.buffer.enable: false
+topology.disruptor.wait.timeout.millis: 1000
+topology.disruptor.batch.size: 100
+topology.disruptor.batch.timeout.millis: 1
+topology.disable.loadaware: false
+topology.state.checkpoint.interval.ms: 1000
+
+# Configs for Resource Aware Scheduler
+# topology priority describing the importance of the topology in decreasing importance starting from 0 (i.e. 0 is the highest priority and the priority importance decreases as the priority number increases).
+# Recommended range of 0-29 but no hard limit set.
+topology.priority: 29
+topology.component.resources.onheap.memory.mb: 128.0
+topology.component.resources.offheap.memory.mb: 0.0
+topology.component.cpu.pcore.percent: 10.0
+topology.worker.max.heap.size.mb: 768.0
+topology.scheduler.strategy: "org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy"
+resource.aware.scheduler.eviction.strategy: "org.apache.storm.scheduler.resource.strategies.eviction.DefaultEvictionStrategy"
+resource.aware.scheduler.priority.strategy: "org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy"
+
+dev.zookeeper.path: "/tmp/dev-storm-zookeeper"
+
+pacemaker.host: "localhost"
+pacemaker.port: 6699
+pacemaker.base.threads: 10
+pacemaker.max.threads: 50
+pacemaker.thread.timeout: 10
+pacemaker.childopts: "-Xmx1024m"
+pacemaker.auth.method: "NONE"
+pacemaker.kerberos.users: []
+
+#default storm daemon metrics reporter plugins
+storm.daemon.metrics.reporter.plugins:
+ - "org.apache.storm.daemon.metrics.reporters.JmxPreparableReporter"
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/7729373d/storm-agent/src/test/resources/words.txt
----------------------------------------------------------------------
diff --git a/storm-agent/src/test/resources/words.txt b/storm-agent/src/test/resources/words.txt
new file mode 100644
index 0000000..c7725df
--- /dev/null
+++ b/storm-agent/src/test/resources/words.txt
@@ -0,0 +1,27 @@
+???# 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.
+the
+and
+storm
+random
+word
+spout
+test
+apache
+the
+spoon
+and
+the
\ No newline at end of file