Repository: incubator-ranger Updated Branches: refs/heads/master b15d6fa8a -> 7729373de
RANGER-1133 - Add tests for the Storm plugin Signed-off-by: Selvamohan Neethiraj <[email protected]> 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 <[email protected]> Authored: Tue Aug 2 12:25:55 2016 +0100 Committer: Selvamohan Neethiraj <[email protected]> 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
