FLUME-2442 Need an alternative to providing clear text passwords in flume config
Adding an interface with 3 implementations to provide functionality at the configuration level to replace variables/keys from external sources. This component is capable of hiding sensitive information or injecting generated data into the configuration. The implementation affects only the configuration layer so existing components (sinks/sources/channels/etc) do not have to change and new components can already have it through the configuration. New custom implementations can be easily added even in plugin form. Each implementation has unit tests in their module and an integration test in the flume-ng-tests module. This closes #197 Reviewers: Miklos Csanady, Denes Arvay (Ferenc Szabo via Denes Arvay) Project: http://git-wip-us.apache.org/repos/asf/flume/repo Commit: http://git-wip-us.apache.org/repos/asf/flume/commit/beb11e59 Tree: http://git-wip-us.apache.org/repos/asf/flume/tree/beb11e59 Diff: http://git-wip-us.apache.org/repos/asf/flume/diff/beb11e59 Branch: refs/heads/trunk Commit: beb11e5988b9306eb6e211e0149d65f96d481b0f Parents: 7f85df9 Author: Ferenc Szabo <fsz...@cloudera.com> Authored: Thu Feb 22 09:07:09 2018 +0100 Committer: Denes Arvay <de...@apache.org> Committed: Thu Feb 22 09:12:30 2018 +0100 ---------------------------------------------------------------------- flume-ng-auth/pom.xml | 1 - .../flume-ng-config-filter-api/pom.xml | 30 + .../configfilter/AbstractConfigFilter.java | 36 + .../apache/flume/configfilter/ConfigFilter.java | 59 ++ .../pom.xml | 50 ++ .../EnvironmentVariableConfigFilter.java | 34 + .../TestEnvironmentVariableConfigFilter.java | 55 ++ .../pom.xml | 89 +++ .../ExternalProcessConfigFilter.java | 109 +++ .../TestExternalProcessConfigFilter.java | 67 ++ .../src/test/resources/test.sh | 11 + .../src/test/resources/test_error.sh | 5 + .../pom.xml | 55 ++ .../HadoopCredentialStoreConfigFilter.java | 77 ++ .../TestHadoopCredentialStoreConfigFilter.java | 183 +++++ .../src/test/resources/test-password.txt | 1 + .../src/test/resources/test-password2.txt | 1 + flume-ng-configfilters/pom.xml | 39 + flume-ng-configuration/pom.xml | 5 + .../src/main/java/org/apache/flume/Context.java | 12 +- .../flume/conf/BasicConfigurationConstants.java | 3 + .../flume/conf/ComponentConfiguration.java | 3 +- .../conf/ComponentConfigurationFactory.java | 4 + .../flume/conf/ComponentWithClassName.java | 21 + .../apache/flume/conf/ConfigFilterFactory.java | 68 ++ .../apache/flume/conf/FlumeConfiguration.java | 729 ++++++++++++------- .../conf/channel/ChannelConfiguration.java | 9 +- .../channel/ChannelSelectorConfiguration.java | 8 +- .../flume/conf/channel/ChannelSelectorType.java | 10 +- .../apache/flume/conf/channel/ChannelType.java | 10 +- .../configfilter/ConfigFilterConfiguration.java | 73 ++ .../conf/configfilter/ConfigFilterType.java | 37 + .../flume/conf/sink/SinkConfiguration.java | 8 +- .../flume/conf/sink/SinkGroupConfiguration.java | 2 +- .../conf/sink/SinkProcessorConfiguration.java | 2 - .../flume/conf/sink/SinkProcessorType.java | 10 +- .../org/apache/flume/conf/sink/SinkType.java | 10 +- .../flume/conf/source/SourceConfiguration.java | 4 +- .../apache/flume/conf/source/SourceType.java | 10 +- .../flume/conf/TestAgentConfiguration.java | 235 ++++++ .../flume/conf/TestFlumeConfiguration.java | 109 ++- .../TestFlumeConfigurationConfigFilter.java | 73 ++ .../channel/MemoryChannelConfiguration.java | 27 + ...onmentVariableConfigFilterConfiguration.java | 27 + .../conf/configfilter/MockConfigFilter.java | 42 ++ .../flume/conf/sink/NullSinkConfiguration.java | 27 + .../conf/source/jms/JMSSourceConfiguration.java | 29 + flume-ng-dist/pom.xml | 16 + flume-ng-doc/sphinx/FlumeUserGuide.rst | 172 ++++- flume-ng-tests/pom.xml | 7 + .../flume/test/agent/TestConfigFilters.java | 176 +++++ .../apache/flume/test/util/StagedInstall.java | 9 +- pom.xml | 27 +- 53 files changed, 2584 insertions(+), 332 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-auth/pom.xml ---------------------------------------------------------------------- diff --git a/flume-ng-auth/pom.xml b/flume-ng-auth/pom.xml index 3f47c48..e07e881 100644 --- a/flume-ng-auth/pom.xml +++ b/flume-ng-auth/pom.xml @@ -103,7 +103,6 @@ limitations under the License. </exclusion> </exclusions> </dependency> - <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-config-filter-api/pom.xml ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-config-filter-api/pom.xml b/flume-ng-configfilters/flume-ng-config-filter-api/pom.xml new file mode 100644 index 0000000..c085495 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-config-filter-api/pom.xml @@ -0,0 +1,30 @@ +<!-- +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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>flume-ng-configfilters</artifactId> + <groupId>org.apache.flume</groupId> + <version>1.9.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>flume-ng-config-filter-api</artifactId> + <name>Flume NG Config Filters API</name> + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/AbstractConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/AbstractConfigFilter.java b/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/AbstractConfigFilter.java new file mode 100644 index 0000000..1028dff --- /dev/null +++ b/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/AbstractConfigFilter.java @@ -0,0 +1,36 @@ +/* + * 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.flume.configfilter; + +/** + * A base implementation of the common methods for Configuration filters + */ +public abstract class AbstractConfigFilter implements ConfigFilter { + + private String name; + + @Override + public void setName(String name) { + this.name = name; + } + + @Override + public String getName() { + return name; + } + +} http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/ConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/ConfigFilter.java b/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/ConfigFilter.java new file mode 100644 index 0000000..ddaab4b --- /dev/null +++ b/flume-ng-configfilters/flume-ng-config-filter-api/src/main/java/org/apache/flume/configfilter/ConfigFilter.java @@ -0,0 +1,59 @@ +/* + * 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.flume.configfilter; + +import java.util.Map; + +/** + * ConfigFilter is a tool for replacing sensitive or generated data in Flume configuration + * + */ +public interface ConfigFilter { + + /** + * Filter method that returns the value associated with the given key + * + * @param key the key to look up in the concrete implementations + * @return the value represented by the key + */ + String filter(String key); + + /** + * Sets the component name. Required by the configuration management. + * + * @param name + */ + void setName(String name); + + + /** + * Returns the component name. Required by the configuration management. + * + * @return String the component name + */ + String getName(); + + /** + * A method to configure the component + * + * @param configuration The map of configuration options needed by concrete implementations. + */ + void initializeWithConfiguration(Map<String, String> configuration); +} http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml new file mode 100644 index 0000000..fbc11c2 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml @@ -0,0 +1,50 @@ +<!-- +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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>flume-ng-configfilters</artifactId> + <groupId>org.apache.flume</groupId> + <version>1.9.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>flume-ng-environment-variable-config-filter</artifactId> + <name>Flume NG Environment Variable Config Filter</name> + + <dependencies> + <dependency> + <groupId>org.apache.flume</groupId> + <artifactId>flume-ng-config-filter-api</artifactId> + </dependency> + <dependency> + <groupId>com.github.stefanbirkner</groupId> + <artifactId>system-rules</artifactId> + <version>1.17.0</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <version>4.8.2</version> + <scope>test</scope> + </dependency> + </dependencies> + + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/main/java/org/apache/flume/configfilter/EnvironmentVariableConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/main/java/org/apache/flume/configfilter/EnvironmentVariableConfigFilter.java b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/main/java/org/apache/flume/configfilter/EnvironmentVariableConfigFilter.java new file mode 100644 index 0000000..1cd40a3 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/main/java/org/apache/flume/configfilter/EnvironmentVariableConfigFilter.java @@ -0,0 +1,34 @@ +/* + * 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.flume.configfilter; + +import java.util.Map; + +public class EnvironmentVariableConfigFilter extends AbstractConfigFilter { + + @Override + public String filter(String key) { + return System.getenv(key); + } + + @Override + public void initializeWithConfiguration(Map<String, String> configuration) { + //NO-OP + } +} http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/test/java/org/apache/flume/configfilter/TestEnvironmentVariableConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/test/java/org/apache/flume/configfilter/TestEnvironmentVariableConfigFilter.java b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/test/java/org/apache/flume/configfilter/TestEnvironmentVariableConfigFilter.java new file mode 100644 index 0000000..d9e36ec --- /dev/null +++ b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/src/test/java/org/apache/flume/configfilter/TestEnvironmentVariableConfigFilter.java @@ -0,0 +1,55 @@ +/* + * 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.flume.configfilter; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.contrib.java.lang.system.EnvironmentVariables; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class TestEnvironmentVariableConfigFilter { + + public static final String MY_PASSWORD_KEY = "my_password_key"; + public static final String MY_PASSWORD_KEY_2 = "my_password_key2"; + public static final String FILTERED = "filtered"; + public static final String FILTERED_2 = "filtered2"; + @Rule + public final EnvironmentVariables environmentVariables + = new EnvironmentVariables(); + + @Test + public void filter() { + environmentVariables.set(MY_PASSWORD_KEY, FILTERED); + environmentVariables.set(MY_PASSWORD_KEY_2, FILTERED_2); + + ConfigFilter configFilter = new EnvironmentVariableConfigFilter(); + + assertEquals(FILTERED, configFilter.filter(MY_PASSWORD_KEY)); + assertEquals(FILTERED_2, configFilter.filter(MY_PASSWORD_KEY_2)); + } + + @Test + public void filterUnknownKey() { + ConfigFilter configFilter = new EnvironmentVariableConfigFilter(); + + assertNull(configFilter.filter("unknown")); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-external-process-config-filter/pom.xml ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-external-process-config-filter/pom.xml b/flume-ng-configfilters/flume-ng-external-process-config-filter/pom.xml new file mode 100644 index 0000000..5b28166 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-external-process-config-filter/pom.xml @@ -0,0 +1,89 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- +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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>flume-ng-configfilters</artifactId> + <groupId>org.apache.flume</groupId> + <version>1.9.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>flume-ng-external-process-config-filter</artifactId> + <name>Flume NG External Process Config Filter</name> + + <dependencies> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flume</groupId> + <artifactId>flume-ng-config-filter-api</artifactId> + </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <version>4.8.2</version> + <scope>test</scope> + </dependency> + </dependencies> + <profiles> + <!-- workaround fro test resource permission errors --> + <profile> + <id>unix</id> + <activation> + <os> + <family>unix</family> + </os> + </activation> + <build> + <plugins> + <plugin> + <groupId>org.codehaus.mojo</groupId> + <artifactId>exec-maven-plugin</artifactId> + <executions> + <execution> + <id>fix-resource-permissions</id> + <goals> + <goal>exec</goal> + </goals> + <phase>process-test-resources</phase> + <configuration> + <executable>/bin/sh</executable> + <arguments> + <argument>-c</argument> + <argument> + set -x + + SRC="${basedir}/src/test/resources" + DST="${project.build.directory}/test-classes" + + find "$$SRC" -printf "%P\0" | xargs -0 -I {} chmod --reference="$$SRC/{}" -f "$$DST/{}" + </argument> + </arguments> + </configuration> + </execution> + </executions> + </plugin> + </plugins> + </build> + </profile> + </profiles> +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-external-process-config-filter/src/main/java/org/apache/flume/configfilter/ExternalProcessConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-external-process-config-filter/src/main/java/org/apache/flume/configfilter/ExternalProcessConfigFilter.java b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/main/java/org/apache/flume/configfilter/ExternalProcessConfigFilter.java new file mode 100644 index 0000000..4ca9a4d --- /dev/null +++ b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/main/java/org/apache/flume/configfilter/ExternalProcessConfigFilter.java @@ -0,0 +1,109 @@ +/* + * 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.flume.configfilter; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.Charset; +import java.nio.charset.UnsupportedCharsetException; +import java.util.Arrays; +import java.util.Map; +import java.util.Scanner; + +public class ExternalProcessConfigFilter extends AbstractConfigFilter { + + private static final Logger LOGGER = LoggerFactory.getLogger(ExternalProcessConfigFilter.class); + + private static final String COMMAND_KEY = "command"; + private static final String CHARSET_KEY = "charset"; + private static final String CHARSET_DEFAULT = "UTF-8"; + + Charset charset; + private String command; + + @Override + public String filter(String key) { + try { + return execCommand(key); + } catch (InterruptedException | IllegalStateException | IOException ex) { + LOGGER.error("Error while reading value for key {}: ", key, ex); + } + return null; + } + + @Override + public void initializeWithConfiguration(Map<String, String> configuration) { + String charsetName = configuration.getOrDefault(CHARSET_KEY, CHARSET_DEFAULT); + try { + charset = Charset.forName(charsetName); + } catch (UnsupportedCharsetException ex) { + throw new RuntimeException("Unsupported charset: " + charsetName, ex); + } + + command = configuration.get(COMMAND_KEY); + if (command == null) { + throw new IllegalArgumentException(COMMAND_KEY + " must be set for " + + "ExternalProcessConfigFilter"); + } + + } + + private String execCommand(String key) throws IOException, InterruptedException { + String[] split = command.split("\\s+"); + int newLength = split.length + 1; + String[] commandParts = Arrays.copyOf(split, newLength); + commandParts[newLength - 1] = key; + Process p = Runtime.getRuntime().exec(commandParts); + p.waitFor(); + if (p.exitValue() != 0) { + String stderr; + try { + stderr = getResultFromStream(p.getErrorStream()); + } catch (Throwable t) { + stderr = null; + } + throw new IllegalStateException( + String.format("Process (%s) exited with non-zero (%s) status code. Sterr: %s", + this.command, p.exitValue(), stderr)); + } + + + return getResultFromStream(p.getInputStream()); + } + + private String getResultFromStream(InputStream inputStream) { + try (Scanner scanner = new Scanner(inputStream, charset.name())) { + String result = null; + if (scanner.hasNextLine()) { + result = scanner.nextLine(); + if (scanner.hasNextLine()) { + LOGGER.warn("External process has more than one line of output. " + + "Only the first line is used."); + } + } else { + LOGGER.warn("External process has not produced any output."); + } + + return result; + } + } +} http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java new file mode 100644 index 0000000..efedc00 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java @@ -0,0 +1,67 @@ +/* + * 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.flume.configfilter; + +import org.apache.flume.configfilter.ExternalProcessConfigFilter; +import org.junit.Before; +import org.junit.Test; + +import java.util.HashMap; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class TestExternalProcessConfigFilter { + + public static final String MY_PASSWORD_KEY = "my_password_key"; + public static final String MY_PASSWORD_KEY_2 = "my_password_key2"; + public static final String FILTERED = "filtered"; + public static final String FILTERED_2 = "filtered2"; + private ExternalProcessConfigFilter configFilter; + + @Before + public void setUp() { + configFilter = new ExternalProcessConfigFilter(); + } + + @Test + public void filterOk() { + String file = Thread.currentThread().getContextClassLoader() + .getResource("test.sh").getFile(); + + + HashMap<String, String> configuration = new HashMap<>(); + configuration.put("command", file); + configFilter.initializeWithConfiguration(configuration); + assertEquals(FILTERED, configFilter.filter(MY_PASSWORD_KEY)); + assertEquals(FILTERED_2, configFilter.filter(MY_PASSWORD_KEY_2)); + + } + + @Test + public void filterError() { + String file = Thread.currentThread().getContextClassLoader() + .getResource("test_error.sh").getFile(); + + HashMap<String, String> configuration = new HashMap<>(); + configuration.put("command", file); + configFilter.initializeWithConfiguration(configuration); + assertNull(configFilter.filter(MY_PASSWORD_KEY)); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test.sh ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test.sh b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test.sh new file mode 100644 index 0000000..4ebd83b --- /dev/null +++ b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash + +if [ $1 = "my_password_key" ]; then + echo "filtered"; +fi + +if [ $1 = "my_password_key2" ]; then + echo "filtered2"; +fi + +exit 0 \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test_error.sh ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test_error.sh b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test_error.sh new file mode 100644 index 0000000..2213ac2 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/resources/test_error.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +echo "Error message from stderr" 1>&2 +echo "Error message from stdout" + +return 123 \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/pom.xml ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/pom.xml b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/pom.xml new file mode 100644 index 0000000..1290069 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/pom.xml @@ -0,0 +1,55 @@ +<!-- +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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>flume-ng-configfilters</artifactId> + <groupId>org.apache.flume</groupId> + <version>1.9.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>flume-ng-hadoop-credential-store-config-filter</artifactId> + <dependencies> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-common</artifactId> + </dependency> + <dependency> + <groupId>org.apache.flume</groupId> + <artifactId>flume-ng-config-filter-api</artifactId> + </dependency> + <!-- test dependencies --> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + </dependency> + <dependency> + <groupId>com.github.stefanbirkner</groupId> + <artifactId>system-rules</artifactId> + <version>1.17.0</version> + <scope>test</scope> + </dependency> + </dependencies> + + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/main/java/org/apache/flume/configfilter/HadoopCredentialStoreConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/main/java/org/apache/flume/configfilter/HadoopCredentialStoreConfigFilter.java b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/main/java/org/apache/flume/configfilter/HadoopCredentialStoreConfigFilter.java new file mode 100644 index 0000000..c8d6870 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/main/java/org/apache/flume/configfilter/HadoopCredentialStoreConfigFilter.java @@ -0,0 +1,77 @@ +/* + * 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.flume.configfilter; + +import java.io.IOException; +import java.util.Map; + +import org.apache.hadoop.conf.Configuration; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class HadoopCredentialStoreConfigFilter extends AbstractConfigFilter { + + private static final Logger LOGGER = LoggerFactory.getLogger( + HadoopCredentialStoreConfigFilter.class); + + static final String PASSWORD_FILE_CONFIG_KEY + = "credstore.java-keystore-provider.password-file"; + static final String CREDENTIAL_PROVIDER_PATH + = "credential.provider.path"; + static final String HADOOP_SECURITY = "hadoop.security."; + private Configuration hadoopConfiguration; + + public void initializeWithConfiguration(Map<String, String> configuration) { + + LOGGER.debug("Initializing hadoop credential store."); + hadoopConfiguration = new Configuration(); + hadoopConfiguration.set( + HADOOP_SECURITY + CREDENTIAL_PROVIDER_PATH, + configuration.get(CREDENTIAL_PROVIDER_PATH) + ); + + String passwordFile = configuration.get(PASSWORD_FILE_CONFIG_KEY); + if (passwordFile != null && !passwordFile.isEmpty()) { + checkPasswordFile(passwordFile); + hadoopConfiguration.set( + HADOOP_SECURITY + PASSWORD_FILE_CONFIG_KEY, passwordFile + ); + } + } + + private void checkPasswordFile(String passwordFile) { + if (Thread.currentThread().getContextClassLoader().getResource(passwordFile) == null) { + LOGGER.error("The java keystore provider password file has to be on the classpath." + + " The password file provided in the configuration cannot be found and will not be used" + ); + } + } + + @Override + public String filter(String key) { + char[] result = null; + try { + result = hadoopConfiguration.getPassword(key); + } catch (IOException e) { + LOGGER.error("Error while reading value for key {}: ", key, e); + } + + return result == null ? null : new String(result); + } + + +} http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/java/org/apache/flume/configfilter/TestHadoopCredentialStoreConfigFilter.java ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/java/org/apache/flume/configfilter/TestHadoopCredentialStoreConfigFilter.java b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/java/org/apache/flume/configfilter/TestHadoopCredentialStoreConfigFilter.java new file mode 100644 index 0000000..55f74d8 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/java/org/apache/flume/configfilter/TestHadoopCredentialStoreConfigFilter.java @@ -0,0 +1,183 @@ +/* + * 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.flume.configfilter; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.security.alias.CredentialShell; +import org.apache.hadoop.util.ToolRunner; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.contrib.java.lang.system.EnvironmentVariables; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.HashMap; + +import static org.apache.flume.configfilter.HadoopCredentialStoreConfigFilter.CREDENTIAL_PROVIDER_PATH; +import static org.apache.flume.configfilter.HadoopCredentialStoreConfigFilter.HADOOP_SECURITY; +import static org.apache.flume.configfilter.HadoopCredentialStoreConfigFilter.PASSWORD_FILE_CONFIG_KEY; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.fail; + +public class TestHadoopCredentialStoreConfigFilter { + + private static String providerPathDefault; + private static String providerPathEnv; + private static String providerPathPwdFile; + + @ClassRule + public static final EnvironmentVariables environmentVariables + = new EnvironmentVariables(); + private static File fileDefault; + private static File fileEnvPassword; + private static File fileFilePassword; + private HadoopCredentialStoreConfigFilter configFilter; + + @BeforeClass + public static void setUpClass() throws Exception { + generateTempFileNames(); + fillCredStoreWithDefaultPassword(); + fillCredStoreWithPasswordFile(); + fillCredStoreWithEnvironmentVariablePassword(); + } + @AfterClass + public static void tearDown() { + fileDefault.deleteOnExit(); + fileEnvPassword.deleteOnExit(); + fileFilePassword.deleteOnExit(); + } + + @Before + public void setUp() { + String[] objects = System.getenv().keySet().toArray(new String[0]); + environmentVariables.clear(objects); + configFilter = new HadoopCredentialStoreConfigFilter(); + } + + @Test + public void filterDefaultPasswordFile() { + HashMap<String, String> configuration = new HashMap<>(); + configuration.put(CREDENTIAL_PROVIDER_PATH, providerPathDefault); + configFilter.initializeWithConfiguration(configuration); + + assertEquals("filtered_default", configFilter.filter("password")); + } + + @Test + public void filterWithEnvPassword() { + environmentVariables.set("HADOOP_CREDSTORE_PASSWORD","envSecret"); + HashMap<String, String> configuration = new HashMap<>(); + configuration.put(CREDENTIAL_PROVIDER_PATH, providerPathEnv); + configFilter.initializeWithConfiguration(configuration); + + assertEquals("filtered_env", configFilter.filter("password")); + } + + @Test + public void filterWithPasswordFile() { + HashMap<String, String> configuration = new HashMap<>(); + configuration.put(CREDENTIAL_PROVIDER_PATH, providerPathPwdFile); + configuration.put(PASSWORD_FILE_CONFIG_KEY, "test-password.txt"); + configFilter.initializeWithConfiguration(configuration); + + assertEquals("filtered_file", configFilter.filter("password")); + } + + @Test + public void filterWithEnvNoPassword() { + HashMap<String, String> configuration = new HashMap<>(); + configuration.put(CREDENTIAL_PROVIDER_PATH, providerPathEnv); + configFilter.initializeWithConfiguration(configuration); + + assertNull(configFilter.filter("password")); + } + + @Test + public void filterErrorWithPasswordFileWrongPassword() { + HashMap<String, String> configuration = new HashMap<>(); + configuration.put(CREDENTIAL_PROVIDER_PATH, providerPathPwdFile); + configuration.put(PASSWORD_FILE_CONFIG_KEY, "test-password2.txt"); + configFilter.initializeWithConfiguration(configuration); + + assertNull(configFilter.filter("password")); + } + + @Test + public void filterErrorWithPasswordFileNoPasswordFile() { + HashMap<String, String> configuration = new HashMap<>(); + configuration.put(CREDENTIAL_PROVIDER_PATH, providerPathPwdFile); + configFilter.initializeWithConfiguration(configuration); + + assertNull(configFilter.filter("password")); + } + + @Test(expected = IllegalArgumentException.class) + public void filterErrorWithNoProvider() { + HashMap<String, String> configuration = new HashMap<>(); + configFilter.initializeWithConfiguration(configuration); + } + + + private static void fillCredStoreWithEnvironmentVariablePassword() throws Exception { + environmentVariables.set("HADOOP_CREDSTORE_PASSWORD", "envSecret"); + + runCommand("create password -value filtered_env -provider " + + providerPathEnv, new Configuration()); + } + + private static void fillCredStoreWithPasswordFile() throws Exception { + Configuration conf = new Configuration(); + conf.set( + HADOOP_SECURITY + PASSWORD_FILE_CONFIG_KEY, + "test-password.txt" + ); + runCommand("create password -value filtered_file -provider " + + providerPathPwdFile, conf); + } + + private static void fillCredStoreWithDefaultPassword() throws Exception { + runCommand("create password -value filtered_default -provider " + + providerPathDefault, new Configuration()); + } + + private static void generateTempFileNames() throws IOException { + fileDefault = Files.createTempFile("test-default-pwd-", ".jceks").toFile(); + boolean deleted = fileDefault.delete(); + fileEnvPassword = Files.createTempFile("test-env-pwd-", ".jceks").toFile(); + deleted &= fileEnvPassword.delete(); + fileFilePassword = Files.createTempFile("test-file-pwd-", ".jceks").toFile(); + deleted &= fileFilePassword.delete(); + if (!deleted) { + fail("Could not delete temporary files"); + } + + providerPathDefault = "jceks://file/" + fileDefault.getAbsolutePath(); + providerPathEnv = "jceks://file/" + fileEnvPassword.getAbsolutePath(); + providerPathPwdFile = "jceks://file/" + fileFilePassword.getAbsolutePath(); + } + + private static void runCommand(String c, Configuration conf) throws Exception { + ToolRunner.run(conf, new CredentialShell(), c.split(" ")); + } + + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password.txt ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password.txt b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password.txt new file mode 100644 index 0000000..db5fd91 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password.txt @@ -0,0 +1 @@ +fileSecret \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password2.txt ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password2.txt b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password2.txt new file mode 100644 index 0000000..2dee0b1 --- /dev/null +++ b/flume-ng-configfilters/flume-ng-hadoop-credential-store-config-filter/src/test/resources/test-password2.txt @@ -0,0 +1 @@ +fileSecret2 \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configfilters/pom.xml ---------------------------------------------------------------------- diff --git a/flume-ng-configfilters/pom.xml b/flume-ng-configfilters/pom.xml new file mode 100644 index 0000000..263d64b --- /dev/null +++ b/flume-ng-configfilters/pom.xml @@ -0,0 +1,39 @@ +<!-- +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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>flume-parent</artifactId> + <groupId>org.apache.flume</groupId> + <version>1.9.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>flume-ng-configfilters</artifactId> + <packaging>pom</packaging> + <name>Flume NG Hadoop Credential Store Config Filter</name> + + <modules> + <module>flume-ng-environment-variable-config-filter</module> + <module>flume-ng-hadoop-credential-store-config-filter</module> + <module>flume-ng-config-filter-api</module> + <module>flume-ng-external-process-config-filter</module> + </modules> + + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configuration/pom.xml ---------------------------------------------------------------------- diff --git a/flume-ng-configuration/pom.xml b/flume-ng-configuration/pom.xml index e431fc8..9fe136c 100644 --- a/flume-ng-configuration/pom.xml +++ b/flume-ng-configuration/pom.xml @@ -76,5 +76,10 @@ limitations under the License. <groupId>org.apache.flume</groupId> <artifactId>flume-ng-sdk</artifactId> </dependency> + <dependency> + <groupId>org.apache.flume</groupId> + <artifactId>flume-ng-config-filter-api</artifactId> + <version>1.9.0-SNAPSHOT</version> + </dependency> </dependencies> </project> http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configuration/src/main/java/org/apache/flume/Context.java ---------------------------------------------------------------------- diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/Context.java b/flume-ng-configuration/src/main/java/org/apache/flume/Context.java index 9b5b659..954baae 100644 --- a/flume-ng-configuration/src/main/java/org/apache/flume/Context.java +++ b/flume-ng-configuration/src/main/java/org/apache/flume/Context.java @@ -22,6 +22,7 @@ package org.apache.flume; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Map.Entry; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; @@ -88,10 +89,11 @@ public class Context { "The given prefix does not end with a period (" + prefix + ")"); Map<String, String> result = Maps.newHashMap(); synchronized (parameters) { - for (String key : parameters.keySet()) { + for (Entry<String, String> entry : parameters.entrySet()) { + String key = entry.getKey(); if (key.startsWith(prefix)) { String name = key.substring(prefix.length()); - result.put(name, parameters.get(key)); + result.put(name, entry.getValue()); } } } @@ -131,7 +133,7 @@ public class Context { public Boolean getBoolean(String key, Boolean defaultValue) { String value = get(key); if (value != null) { - return Boolean.parseBoolean(value.trim()); + return Boolean.valueOf(Boolean.parseBoolean(value.trim())); } return defaultValue; } @@ -160,7 +162,7 @@ public class Context { public Integer getInteger(String key, Integer defaultValue) { String value = get(key); if (value != null) { - return Integer.parseInt(value.trim()); + return Integer.valueOf(Integer.parseInt(value.trim())); } return defaultValue; } @@ -189,7 +191,7 @@ public class Context { public Long getLong(String key, Long defaultValue) { String value = get(key); if (value != null) { - return Long.parseLong(value.trim()); + return Long.valueOf(Long.parseLong(value.trim())); } return defaultValue; } http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configuration/src/main/java/org/apache/flume/conf/BasicConfigurationConstants.java ---------------------------------------------------------------------- diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/conf/BasicConfigurationConstants.java b/flume-ng-configuration/src/main/java/org/apache/flume/conf/BasicConfigurationConstants.java index 9089122..3f23e62 100644 --- a/flume-ng-configuration/src/main/java/org/apache/flume/conf/BasicConfigurationConstants.java +++ b/flume-ng-configuration/src/main/java/org/apache/flume/conf/BasicConfigurationConstants.java @@ -18,6 +18,9 @@ package org.apache.flume.conf; public final class BasicConfigurationConstants { + public static final String CONFIG_CONFIGFILTERS = "configfilters"; + public static final String CONFIG_CONFIGFILTERS_PREFIX = CONFIG_CONFIGFILTERS + "."; + public static final String CONFIG_SOURCES = "sources"; public static final String CONFIG_SOURCES_PREFIX = CONFIG_SOURCES + "."; public static final String CONFIG_SOURCE_CHANNELSELECTOR_PREFIX = "selector."; http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfiguration.java ---------------------------------------------------------------------- diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfiguration.java b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfiguration.java index 477a3e6..b7bab15 100644 --- a/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfiguration.java +++ b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfiguration.java @@ -98,7 +98,7 @@ public abstract class ComponentConfiguration { } public String toString(int indentCount) { - StringBuilder indentSb = new StringBuilder(""); + StringBuilder indentSb = new StringBuilder(); for (int i = 0; i < indentCount; i++) { indentSb.append(FlumeConfiguration.INDENTSTEP); @@ -126,6 +126,7 @@ public abstract class ComponentConfiguration { public enum ComponentType { OTHER(null), + CONFIG_FILTER("ConfigFilter"), SOURCE("Source"), SINK("Sink"), SINK_PROCESSOR("SinkProcessor"), http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfigurationFactory.java ---------------------------------------------------------------------- diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfigurationFactory.java b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfigurationFactory.java index 16860c3..f1e3a8c 100644 --- a/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfigurationFactory.java +++ b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentConfigurationFactory.java @@ -21,6 +21,7 @@ import java.util.Locale; import org.apache.flume.conf.ComponentConfiguration.ComponentType; import org.apache.flume.conf.channel.ChannelConfiguration.ChannelConfigurationType; import org.apache.flume.conf.channel.ChannelSelectorConfiguration.ChannelSelectorConfigurationType; +import org.apache.flume.conf.configfilter.ConfigFilterConfiguration.ConfigFilterConfigurationType; import org.apache.flume.conf.sink.SinkConfiguration.SinkConfigurationType; import org.apache.flume.conf.sink.SinkGroupConfiguration; import org.apache.flume.conf.sink.SinkProcessorConfiguration.SinkProcessorConfigurationType; @@ -47,6 +48,9 @@ public class ComponentConfigurationFactory { case SOURCE: return SourceConfigurationType.valueOf(type.toUpperCase(Locale.ENGLISH)) .getConfiguration(name); + case CONFIG_FILTER: + return ConfigFilterConfigurationType.valueOf(type.toUpperCase(Locale.ENGLISH)) + .getConfiguration(name); case SINK: return SinkConfigurationType.valueOf(type.toUpperCase(Locale.ENGLISH)) .getConfiguration(name); http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentWithClassName.java ---------------------------------------------------------------------- diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentWithClassName.java b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentWithClassName.java new file mode 100644 index 0000000..df1252d --- /dev/null +++ b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ComponentWithClassName.java @@ -0,0 +1,21 @@ +/* + * 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.flume.conf; + +public interface ComponentWithClassName { + String getClassName(); +} http://git-wip-us.apache.org/repos/asf/flume/blob/beb11e59/flume-ng-configuration/src/main/java/org/apache/flume/conf/ConfigFilterFactory.java ---------------------------------------------------------------------- diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/conf/ConfigFilterFactory.java b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ConfigFilterFactory.java new file mode 100644 index 0000000..9c011e9 --- /dev/null +++ b/flume-ng-configuration/src/main/java/org/apache/flume/conf/ConfigFilterFactory.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.flume.conf; + +import com.google.common.base.Preconditions; +import org.apache.flume.FlumeException; +import org.apache.flume.configfilter.ConfigFilter; +import org.apache.flume.conf.configfilter.ConfigFilterType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Locale; + +public class ConfigFilterFactory { + + private static final Logger LOGGER = LoggerFactory + .getLogger(ConfigFilterFactory.class); + + public static ConfigFilter create(String name, String type) throws FlumeException { + Preconditions.checkNotNull(name, "name"); + Preconditions.checkNotNull(type, "type"); + LOGGER.info("Creating instance of configfilter {}, type {}", name, type); + Class<? extends ConfigFilter> aClass = getClass(type); + try { + ConfigFilter configFilter = aClass.newInstance(); + configFilter.setName(name); + return configFilter; + } catch (Exception ex) { + throw new FlumeException("Unable to create configfilter: " + name + + ", type: " + type + ", class: " + aClass.getName(), ex); + } + } + + public static Class<? extends ConfigFilter> getClass(String type) throws FlumeException { + String classname = type; + ConfigFilterType srcType = ConfigFilterType.OTHER; + try { + srcType = ConfigFilterType.valueOf(type.toUpperCase(Locale.ENGLISH)); + } catch (IllegalArgumentException ex) { + LOGGER.debug("Configfilter type {} is a custom type", type); + } + if (srcType != ConfigFilterType.OTHER) { + classname = srcType.getClassName(); + } + try { + return (Class<? extends ConfigFilter>) Class.forName(classname); + } catch (Exception ex) { + throw new FlumeException("Unable to load configfilter type: " + type + + ", class: " + classname, ex); + } + } +}
