Repository: ambari Updated Branches: refs/heads/branch-feature-AMBARI-20859 fd3220099 -> 12ae25915
AMBARI-21210 Add ability to Log Search to test a log entry if it is parseable (mgergely) Change-Id: I545d7ade76beb3b8ce75dda9d66ca1c36b38a691 Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/943c1b0d Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/943c1b0d Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/943c1b0d Branch: refs/heads/branch-feature-AMBARI-20859 Commit: 943c1b0d375a2d92b75be1be7158e8643692b1df Parents: 0dbc797 Author: Miklos Gergely <[email protected]> Authored: Fri Jun 23 16:03:51 2017 +0200 Committer: Miklos Gergely <[email protected]> Committed: Fri Jun 23 16:03:51 2017 +0200 ---------------------------------------------------------------------- .../logsearch/config/api/LogSearchConfig.java | 8 + .../model/inputconfig/FilterGrokDescriptor.java | 2 + .../config/api/LogSearchConfigClass1.java | 5 + .../config/api/LogSearchConfigClass2.java | 5 + .../config/zookeeper/LogSearchConfigZK.java | 9 +- .../impl/FilterGrokDescriptorImpl.java | 1 + .../ambari-logsearch-logfeeder/pom.xml | 5 + .../org/apache/ambari/logfeeder/LogFeeder.java | 55 +++++- .../ambari/logfeeder/LogFeederCommandLine.java | 168 +++++++++++++++++++ .../ambari/logfeeder/common/ConfigHandler.java | 26 +++ .../logfeeder/common/LogEntryParseTester.java | 127 ++++++++++++++ .../ambari/logfeeder/filter/FilterGrok.java | 5 +- .../apache/ambari/logfeeder/input/Input.java | 2 +- .../apache/ambari/logfeeder/util/AliasUtil.java | 6 +- .../apache/ambari/logfeeder/util/FileUtil.java | 23 +-- .../ambari/logfeeder/util/LogFeederUtil.java | 26 +-- .../src/main/scripts/run.sh | 4 +- .../logconfig/LogConfigHandlerTest.java | 2 +- .../logfeeder/metrics/MetrcisManagerTest.java | 128 -------------- .../logfeeder/metrics/MetricsManagerTest.java | 128 ++++++++++++++ .../ambari-logsearch-server/pom.xml | 27 +++ .../ambari/logsearch/doc/DocConstants.java | 1 + .../logsearch/manager/ShipperConfigManager.java | 24 +++ .../logsearch/rest/ShipperConfigResource.java | 12 ++ 24 files changed, 605 insertions(+), 194 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/LogSearchConfig.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/LogSearchConfig.java b/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/LogSearchConfig.java index 4cbf21f..ad1f5d4 100644 --- a/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/LogSearchConfig.java +++ b/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/LogSearchConfig.java @@ -66,6 +66,14 @@ public interface LogSearchConfig extends Closeable { boolean inputConfigExists(String clusterName, String serviceName) throws Exception; /** + * Returns the global configurations of a cluster. Will be used only in SERVER mode. + * + * @param clusterName The name of the cluster where the service is looked for. + * @return The global configurations of the cluster if it exists, null otherwise. + */ + String getGlobalConfigs(String clusterName); + + /** * Returns the input configuration of a service in a cluster. Will be used only in SERVER mode. * * @param clusterName The name of the cluster where the service is looked for. http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/FilterGrokDescriptor.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/FilterGrokDescriptor.java b/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/FilterGrokDescriptor.java index e85ce97..039e1ff 100644 --- a/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/FilterGrokDescriptor.java +++ b/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/FilterGrokDescriptor.java @@ -25,4 +25,6 @@ public interface FilterGrokDescriptor extends FilterDescriptor { String getMultilinePattern(); String getMessagePattern(); + + void setMultilinePattern(String multilinePattern); } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass1.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass1.java b/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass1.java index d7e3c0a..7309382 100644 --- a/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass1.java +++ b/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass1.java @@ -53,6 +53,11 @@ public class LogSearchConfigClass1 implements LogSearchConfig { } @Override + public String getGlobalConfigs(String clusterName) { + return null; + } + + @Override public InputConfig getInputConfig(String clusterName, String serviceName) { return null; } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass2.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass2.java b/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass2.java index 198c133..f83eeef 100644 --- a/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass2.java +++ b/ambari-logsearch/ambari-logsearch-config-api/src/test/java/org/apache/ambari/logsearch/config/api/LogSearchConfigClass2.java @@ -53,6 +53,11 @@ public class LogSearchConfigClass2 implements LogSearchConfig { } @Override + public String getGlobalConfigs(String clusterName) { + return null; + } + + @Override public InputConfig getInputConfig(String clusterName, String serviceName) { return null; } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/LogSearchConfigZK.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/LogSearchConfigZK.java b/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/LogSearchConfigZK.java index 4d10a5b..26375e1 100644 --- a/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/LogSearchConfigZK.java +++ b/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/LogSearchConfigZK.java @@ -261,9 +261,14 @@ public class LogSearchConfigZK implements LogSearchConfig { } @Override - public InputConfig getInputConfig(String clusterName, String serviceName) { + public String getGlobalConfigs(String clusterName) { String globalConfigNodePath = String.format("%s/%s/global", root, clusterName); - String globalConfigData = new String(cache.getCurrentData(globalConfigNodePath).getData()); + return new String(cache.getCurrentData(globalConfigNodePath).getData()); + } + + @Override + public InputConfig getInputConfig(String clusterName, String serviceName) { + String globalConfigData = getGlobalConfigs(clusterName); JsonArray globalConfigs = (JsonArray) new JsonParser().parse(globalConfigData); InputAdapter.setGlobalConfigs(globalConfigs); http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/model/inputconfig/impl/FilterGrokDescriptorImpl.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/model/inputconfig/impl/FilterGrokDescriptorImpl.java b/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/model/inputconfig/impl/FilterGrokDescriptorImpl.java index 7f40b7f..995f76b 100644 --- a/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/model/inputconfig/impl/FilterGrokDescriptorImpl.java +++ b/ambari-logsearch/ambari-logsearch-config-zookeeper/src/main/java/org/apache/ambari/logsearch/config/zookeeper/model/inputconfig/impl/FilterGrokDescriptorImpl.java @@ -51,6 +51,7 @@ public class FilterGrokDescriptorImpl extends FilterDescriptorImpl implements Fi return multilinePattern; } + @Override public void setMultilinePattern(String multilinePattern) { this.multilinePattern = multilinePattern; } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/pom.xml ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/pom.xml b/ambari-logsearch/ambari-logsearch-logfeeder/pom.xml index ce784cb..ae2150e 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/pom.xml +++ b/ambari-logsearch/ambari-logsearch-logfeeder/pom.xml @@ -47,6 +47,11 @@ <version>${project.version}</version> </dependency> <dependency> + <groupId>commons-cli</groupId> + <artifactId>commons-cli</artifactId> + <version>1.3.1</version> + </dependency> + <dependency> <groupId>commons-codec</groupId> <artifactId>commons-codec</artifactId> </dependency> http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeeder.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeeder.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeeder.java index 8d7c69f..e7b6edc 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeeder.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeeder.java @@ -19,14 +19,19 @@ package org.apache.ambari.logfeeder; +import java.io.File; +import java.nio.charset.Charset; import java.util.ArrayList; import java.util.List; +import java.util.Map; import org.apache.ambari.logfeeder.common.ConfigHandler; +import org.apache.ambari.logfeeder.common.LogEntryParseTester; import org.apache.ambari.logsearch.config.api.LogSearchConfig; import org.apache.ambari.logsearch.config.api.LogSearchConfigFactory; import org.apache.ambari.logsearch.config.api.LogSearchConfig.Component; import org.apache.ambari.logsearch.config.zookeeper.LogSearchConfigZK; +import org.apache.commons.io.FileUtils; import org.apache.ambari.logfeeder.input.InputConfigUploader; import org.apache.ambari.logfeeder.input.InputManager; import org.apache.ambari.logfeeder.loglevelfilter.LogLevelFilterHandler; @@ -35,7 +40,10 @@ import org.apache.ambari.logfeeder.metrics.MetricsManager; import org.apache.ambari.logfeeder.util.LogFeederUtil; import org.apache.ambari.logfeeder.util.SSLUtil; import com.google.common.collect.Maps; +import com.google.gson.GsonBuilder; + import org.apache.hadoop.util.ShutdownHookManager; +import org.apache.log4j.LogManager; import org.apache.log4j.Logger; public class LogFeeder { @@ -44,6 +52,8 @@ public class LogFeeder { private static final int LOGFEEDER_SHUTDOWN_HOOK_PRIORITY = 30; private static final int CHECKPOINT_CLEAN_INTERVAL_MS = 24 * 60 * 60 * 60 * 1000; // 24 hours + private final LogFeederCommandLine cli; + private ConfigHandler configHandler = new ConfigHandler(); private LogSearchConfig config; @@ -54,7 +64,9 @@ public class LogFeeder { private boolean isLogfeederCompleted = false; private Thread statLoggerThread = null; - private LogFeeder() {} + private LogFeeder(LogFeederCommandLine cli) { + this.cli = cli; + } public void run() { try { @@ -165,15 +177,40 @@ public class LogFeeder { } } - public static void main(String[] args) { + public void test() { try { - LogFeederUtil.loadProperties("logfeeder.properties", args); - } catch (Throwable t) { - LOG.warn("Could not load logfeeder properites"); - System.exit(1); + LogManager.shutdown(); + String testLogEntry = cli.getTestLogEntry(); + String testShipperConfig = FileUtils.readFileToString(new File(cli.getTestShipperConfig()), Charset.defaultCharset()); + List<String> testGlobalConfigs = new ArrayList<>(); + for (String testGlobalConfigFile : cli.getTestGlobalConfigs().split(",")) { + testGlobalConfigs.add(FileUtils.readFileToString(new File(testGlobalConfigFile), Charset.defaultCharset())); + } + String testLogId = cli.getTestLogId(); + Map<String, Object> result = new LogEntryParseTester(testLogEntry, testShipperConfig, testGlobalConfigs, testLogId).parse(); + String parsedLogEntry = new GsonBuilder().setPrettyPrinting().create().toJson(result); + System.out.println("The result of the parsing is:\n" + parsedLogEntry); + } catch (Exception e) { + System.out.println("Exception occurred, could not test if log entry is parseable"); + e.printStackTrace(System.out); + } + } + + public static void main(String[] args) { + LogFeederCommandLine cli = new LogFeederCommandLine(args); + + LogFeeder logFeeder = new LogFeeder(cli); + + if (cli.isMonitor()) { + try { + LogFeederUtil.loadProperties("logfeeder.properties"); + } catch (Throwable t) { + LOG.warn("Could not load logfeeder properites"); + System.exit(1); + } + logFeeder.run(); + } else if (cli.isTest()) { + logFeeder.test(); } - - LogFeeder logFeeder = new LogFeeder(); - logFeeder.run(); } } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeederCommandLine.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeederCommandLine.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeederCommandLine.java new file mode 100644 index 0000000..d996f98 --- /dev/null +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/LogFeederCommandLine.java @@ -0,0 +1,168 @@ +/* + * 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.ambari.logfeeder; + +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.CommandLineParser; +import org.apache.commons.cli.DefaultParser; +import org.apache.commons.cli.HelpFormatter; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.Options; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class LogFeederCommandLine { + + private static final Logger LOG = LoggerFactory.getLogger(LogFeederCommandLine.class); + + private static final String MONITOR_COMMAND = "monitor"; + + private static final String TEST_COMMAND = "test"; + private static final String TEST_LOG_ENTRY_OPTION = "test-log-entry"; + private static final String TEST_SHIPPER_CONFIG_OPTION = "test-shipper-config"; + private static final String TEST_GLOBAL_CONFIG_OPTION = "test-global-config"; + private static final String TEST_LOG_ID_OPTION = "test-log-id"; + + private static final String COMMAND_LINE_SYNTAX = "java org.apache.ambari.logfeeder.LogFeeder -(monitor|test) [args]"; + + private CommandLine cli; + + public LogFeederCommandLine(String[] args) { + Options options = new Options(); + HelpFormatter helpFormatter = new HelpFormatter(); + helpFormatter.setDescPadding(10); + helpFormatter.setWidth(200); + + Option helpOption = Option.builder("h") + .longOpt("help") + .desc("Print commands") + .build(); + + Option monitorOption = Option.builder("m") + .longOpt(MONITOR_COMMAND) + .desc("Monitor log files") + .build(); + + Option testOption = Option.builder("t") + .longOpt(TEST_COMMAND) + .desc("Test if log entry is parseable") + .build(); + + Option testLogEntryOption = Option.builder("tle") + .longOpt(TEST_LOG_ENTRY_OPTION) + .hasArg() + .desc("Log entry to test if it's parseable") + .build(); + + Option testShipperConfOption = Option.builder("tsc") + .longOpt(TEST_SHIPPER_CONFIG_OPTION) + .hasArg() + .desc("Shipper configuration file for testing if log entry is parseable") + .build(); + + Option testGlobalConfOption = Option.builder("tgc") + .longOpt(TEST_GLOBAL_CONFIG_OPTION) + .hasArg() + .desc("Global configuration files (comma separated list) for testing if log entry is parseable") + .build(); + + Option testLogIdOption = Option.builder("tli") + .longOpt(TEST_LOG_ID_OPTION) + .hasArg() + .desc("The id of the log to test") + .build(); + + options.addOption(helpOption); + options.addOption(monitorOption); + options.addOption(testOption); + options.addOption(testLogEntryOption); + options.addOption(testShipperConfOption); + options.addOption(testGlobalConfOption); + options.addOption(testLogIdOption); + + try { + CommandLineParser cmdLineParser = new DefaultParser(); + cli = cmdLineParser.parse(options, args); + + if (cli.hasOption('h')) { + helpFormatter.printHelp(COMMAND_LINE_SYNTAX, options); + System.exit(0); + } + String command = ""; + if (cli.hasOption("m")) { + command = MONITOR_COMMAND; + } else if (cli.hasOption("t")) { + command = TEST_COMMAND; + validateRequiredOptions(cli, command, testLogEntryOption, testShipperConfOption); + } else { + List<String> commands = Arrays.asList(MONITOR_COMMAND, TEST_COMMAND); + helpFormatter.printHelp(COMMAND_LINE_SYNTAX, options); + LOG.error(String.format("One of the supported commands is required (%s)", StringUtils.join(commands, "|"))); + System.exit(1); + } + } catch (Exception e) { + LOG.error("Error parsing command line parameters", e); + helpFormatter.printHelp(COMMAND_LINE_SYNTAX, options); + System.exit(1); + } + } + + private static void validateRequiredOptions(CommandLine cli, String command, Option... optionsToValidate) { + List<String> requiredOptions = new ArrayList<>(); + for (Option opt : optionsToValidate) { + if (!cli.hasOption(opt.getOpt())) { + requiredOptions.add(opt.getOpt()); + } + } + if (!requiredOptions.isEmpty()) { + throw new IllegalArgumentException( + String.format("The following options required for '%s' : %s", command, StringUtils.join(requiredOptions, ","))); + } + } + + public boolean isMonitor() { + return cli.hasOption('m'); + } + + public boolean isTest() { + return cli.hasOption('t'); + } + + public String getTestLogEntry() { + return cli.getOptionValue("tle"); + } + + public String getTestShipperConfig() { + return cli.getOptionValue("tsc"); + } + + public String getTestGlobalConfigs() { + return cli.getOptionValue("tgc"); + } + + public String getTestLogId() { + return cli.getOptionValue("tli"); + } +} http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/ConfigHandler.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/ConfigHandler.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/ConfigHandler.java index 726ff27..25669d9 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/ConfigHandler.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/ConfigHandler.java @@ -51,6 +51,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.ambari.logfeeder.util.AliasUtil.AliasType; import org.apache.ambari.logsearch.config.api.InputConfigMonitor; import org.apache.ambari.logsearch.config.api.model.inputconfig.FilterDescriptor; +import org.apache.ambari.logsearch.config.api.model.inputconfig.FilterGrokDescriptor; import org.apache.ambari.logsearch.config.api.model.inputconfig.InputConfig; import org.apache.ambari.logsearch.config.api.model.inputconfig.InputDescriptor; import org.apache.ambari.logsearch.config.zookeeper.model.inputconfig.impl.FilterDescriptorImpl; @@ -171,6 +172,31 @@ public class ConfigHandler implements InputConfigMonitor { inputManager.removeInputsForService(serviceName); } + public Input getTestInput(InputConfig inputConfig, String logId) { + for (InputDescriptor inputDescriptor : inputConfig.getInput()) { + if (inputDescriptor.getType().equals(logId)) { + inputConfigList.add(inputDescriptor); + break; + } + } + if (inputConfigList.isEmpty()) { + throw new IllegalArgumentException("Log Id " + logId + " was not found in shipper configuriaton"); + } + + for (FilterDescriptor filterDescriptor : inputConfig.getFilter()) { + if ("grok".equals(filterDescriptor.getFilter())) { + // Thus ensure that the log entry passed will be parsed immediately + ((FilterGrokDescriptor)filterDescriptor).setMultilinePattern(null); + } + filterConfigList.add(filterDescriptor); + } + loadInputs("test"); + loadFilters("test"); + List<Input> inputList = inputManager.getInputList("test"); + + return inputList != null && inputList.size() == 1 ? inputList.get(0) : null; + } + @SuppressWarnings("unchecked") public void loadConfigs(String configData) throws Exception { Type type = new TypeToken<Map<String, Object>>() {}.getType(); http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogEntryParseTester.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogEntryParseTester.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogEntryParseTester.java new file mode 100644 index 0000000..97bc3a2 --- /dev/null +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogEntryParseTester.java @@ -0,0 +1,127 @@ +/* + * 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.ambari.logfeeder.common; + +import java.io.File; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.ambari.logfeeder.input.Input; +import org.apache.ambari.logfeeder.input.InputMarker; +import org.apache.ambari.logfeeder.output.Output; +import org.apache.ambari.logsearch.config.api.model.inputconfig.InputConfig; +import org.apache.ambari.logsearch.config.zookeeper.model.inputconfig.impl.InputConfigGson; +import org.apache.ambari.logsearch.config.zookeeper.model.inputconfig.impl.InputConfigImpl; + +import com.google.common.collect.ImmutableMap; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; + +public class LogEntryParseTester { + + private final String logEntry; + private final String shipperConfig; + private final List<JsonObject> globalConfigs; + private final String logId; + + public LogEntryParseTester(String logEntry, String shipperConfig, String globalConfigsJson, String logId) { + this.logEntry = logEntry; + this.shipperConfig = shipperConfig; + this.globalConfigs = new ArrayList<>(); + this.logId = logId; + + JsonParser jsonParser = new JsonParser(); + JsonArray globalConfigArray = jsonParser.parse(globalConfigsJson).getAsJsonArray(); + for (JsonElement e : globalConfigArray) { + globalConfigs.add(e.getAsJsonObject()); + } + } + + public LogEntryParseTester(String logEntry, String shipperConfig, List<String> globalConfigJsons, String logId) { + this.logEntry = logEntry; + this.shipperConfig = shipperConfig; + this.globalConfigs = new ArrayList<>(); + this.logId = logId; + + JsonParser jsonParser = new JsonParser(); + for (String globalConfig : globalConfigJsons) { + JsonObject globalConfigObject = jsonParser.parse(globalConfig).getAsJsonObject(); + globalConfigs.add(globalConfigObject.get("global").getAsJsonObject()); + } + } + + public Map<String, Object> parse() throws Exception { + InputConfig inputConfig = getInputConfig(); + ConfigHandler configHandler = new ConfigHandler(); + Input input = configHandler.getTestInput(inputConfig, logId); + final Map<String, Object> result = new HashMap<>(); + input.init(); + input.addOutput(new Output() { + @Override + public void write(String block, InputMarker inputMarker) throws Exception { + } + + @Override + public void copyFile(File inputFile, InputMarker inputMarker) throws UnsupportedOperationException { + } + + @Override + public void write(Map<String, Object> jsonObj, InputMarker inputMarker) { + result.putAll(jsonObj); + } + }); + input.outputLine(logEntry, new InputMarker(input, null, 0)); + + return result.isEmpty() ? + ImmutableMap.of("errorMessage", (Object)"Could not parse test log entry") : + result; + } + + private InputConfig getInputConfig() { + JsonParser jsonParser = new JsonParser(); + JsonElement shipperConfigJson = jsonParser.parse(shipperConfig); + for (JsonObject globalConfig : globalConfigs) { + for (Map.Entry<String, JsonElement> typeEntry : shipperConfigJson.getAsJsonObject().entrySet()) { + for (JsonElement e : typeEntry.getValue().getAsJsonArray()) { + merge(globalConfig, e.getAsJsonObject()); + } + } + } + return InputConfigGson.gson.fromJson(shipperConfigJson, InputConfigImpl.class); + } + + private void merge(JsonObject source, JsonObject target) { + for (Map.Entry<String, JsonElement> e : source.entrySet()) { + if (!target.has(e.getKey())) { + target.add(e.getKey(), e.getValue()); + } else { + if (e.getValue().isJsonObject()) { + JsonObject valueJson = (JsonObject)e.getValue(); + merge(valueJson, target.get(e.getKey()).getAsJsonObject()); + } + } + } + } + +} http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/filter/FilterGrok.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/filter/FilterGrok.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/filter/FilterGrok.java index 70aea65..50247e2 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/filter/FilterGrok.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/filter/FilterGrok.java @@ -19,7 +19,7 @@ package org.apache.ambari.logfeeder.filter; -import java.io.BufferedInputStream; +import java.io.InputStream; import java.io.InputStreamReader; import java.lang.reflect.Type; import java.util.ArrayList; @@ -139,8 +139,7 @@ public class FilterGrok extends Filter { InputStreamReader grokPatternsReader = null; LOG.info("Loading pattern file " + GROK_PATTERN_FILE); try { - BufferedInputStream fileInputStream = - (BufferedInputStream) this.getClass().getClassLoader().getResourceAsStream(GROK_PATTERN_FILE); + InputStream fileInputStream = getClass().getClassLoader().getResourceAsStream(GROK_PATTERN_FILE); if (fileInputStream == null) { LOG.fatal("Couldn't load grok-patterns file " + GROK_PATTERN_FILE + ". Things will not work"); return false; http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/input/Input.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/input/Input.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/input/Input.java index 27d16c4..c36f96b 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/input/Input.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/input/Input.java @@ -176,7 +176,7 @@ public abstract class Input extends ConfigItem implements Runnable { */ abstract void start() throws Exception; - protected void outputLine(String line, InputMarker marker) { + public void outputLine(String line, InputMarker marker) { statMetric.value++; readBytesMetric.value += (line.length()); http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/AliasUtil.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/AliasUtil.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/AliasUtil.java index 5049b62..3c48aa2 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/AliasUtil.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/AliasUtil.java @@ -18,7 +18,6 @@ */ package org.apache.ambari.logfeeder.util; -import java.io.File; import java.util.HashMap; import org.apache.ambari.logfeeder.filter.Filter; @@ -36,10 +35,7 @@ public class AliasUtil { private static HashMap<String, Object> aliasMap = null; static { - File jsonFile = FileUtil.getFileFromClasspath(ALIAS_CONFIG_JSON); - if (jsonFile != null) { - aliasMap = FileUtil.readJsonFromFile(jsonFile); - } + aliasMap = FileUtil.getJsonFileContentFromClassPath(ALIAS_CONFIG_JSON); } public static enum AliasType { http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/FileUtil.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/FileUtil.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/FileUtil.java index 94d6558..90d1df6 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/FileUtil.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/FileUtil.java @@ -21,7 +21,7 @@ package org.apache.ambari.logfeeder.util; import java.io.File; import java.io.IOException; -import java.net.URL; +import java.io.InputStream; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; @@ -67,24 +67,11 @@ public class FileUtil { } return file.toString(); } - - public static File getFileFromClasspath(String filename) { - URL fileCompleteUrl = Thread.currentThread().getContextClassLoader().getResource(filename); - LOG.debug("File Complete URI :" + fileCompleteUrl); - File file = null; - try { - file = new File(fileCompleteUrl.toURI()); - } catch (Exception exception) { - LOG.debug(exception.getMessage(), exception.getCause()); - } - return file; - } - - public static HashMap<String, Object> readJsonFromFile(File jsonFile) { + + public static HashMap<String, Object> getJsonFileContentFromClassPath(String fileName) { ObjectMapper mapper = new ObjectMapper(); - try { - HashMap<String, Object> jsonmap = mapper.readValue(jsonFile, new TypeReference<HashMap<String, Object>>() {}); - return jsonmap; + try (InputStream inputStream = FileUtil.class.getClassLoader().getResourceAsStream(fileName)) { + return mapper.readValue(inputStream, new TypeReference<HashMap<String, Object>>() {}); } catch (IOException e) { LOG.error(e, e.getCause()); } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/LogFeederUtil.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/LogFeederUtil.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/LogFeederUtil.java index d8a1fbb..0fdc21f 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/LogFeederUtil.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/util/LogFeederUtil.java @@ -84,7 +84,7 @@ public class LogFeederUtil { /** * This method will read the properties from System, followed by propFile and finally from the map */ - public static void loadProperties(String propFile, String[] propNVList) throws Exception { + public static void loadProperties(String propFile) throws Exception { LOG.info("Loading properties. propFile=" + propFile); props = new Properties(System.getProperties()); boolean propLoaded = false; @@ -122,30 +122,6 @@ public class LogFeederUtil { if (!propLoaded) { LOG.fatal("Properties file is not loaded."); throw new Exception("Properties not loaded"); - } else { - updatePropertiesFromMap(propNVList); - } - } - - private static void updatePropertiesFromMap(String[] nvList) { - if (nvList == null) { - return; - } - LOG.info("Trying to load additional proeprties from argument paramters. nvList.length=" + nvList.length); - for (String nv : nvList) { - LOG.info("Passed nv=" + nv); - if (nv.startsWith("-") && nv.length() > 1) { - nv = nv.substring(1); - LOG.info("Stripped nv=" + nv); - int i = nv.indexOf("="); - if (nv.length() > i) { - LOG.info("Candidate nv=" + nv); - String name = nv.substring(0, i); - String value = nv.substring(i + 1); - LOG.info("Adding property from argument to properties. name=" + name + ", value=" + value); - props.put(name, value); - } - } } } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/main/scripts/run.sh ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/scripts/run.sh b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/scripts/run.sh index 53cd17f..22e1cb2 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/scripts/run.sh +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/scripts/run.sh @@ -95,9 +95,9 @@ if [ $foreground -eq 0 ]; then echo "Starting logfeeder. Output file=$LOGFILE pid_file=$PID_FILE" #LOGFEEDER_CLI_CLASSPATH=set -x - nohup $JAVA -cp "$LOGFEEDER_CLI_CLASSPATH:$LOGFEEDER_CONF_DIR:$script_dir/libs/*:$script_dir/classes" $LOGFEEDER_GC_OPTS $LOGFEEDER_JAVA_MEM $LOGFEEDER_JAVA_OPTS $JMX org.apache.ambari.logfeeder.LogFeeder $* > $LOGFILE 2>&1 & +nohup $JAVA -cp "$LOGFEEDER_CLI_CLASSPATH:$LOGFEEDER_CONF_DIR:$script_dir/libs/*:$script_dir/classes" $LOGFEEDER_GC_OPTS $LOGFEEDER_JAVA_MEM $LOGFEEDER_JAVA_OPTS $JMX org.apache.ambari.logfeeder.LogFeeder --monitor $* > $LOGFILE 2>&1 & echo $! > $PID_FILE else - $JAVA -cp "$LOGFEEDER_CLI_CLASSPATH:$LOGFEEDER_CONF_DIR:$script_dir/libs/*:$script_dir/classes" $LOGFEEDER_JAVA_MEM $LOGFEEDER_JAVA_OPTS $JMX org.apache.ambari.logfeeder.LogFeeder $* +$JAVA -cp "$LOGFEEDER_CLI_CLASSPATH:$LOGFEEDER_CONF_DIR:$script_dir/libs/*:$script_dir/classes" $LOGFEEDER_JAVA_MEM $LOGFEEDER_JAVA_OPTS $JMX org.apache.ambari.logfeeder.LogFeeder --monitor $* fi http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/logconfig/LogConfigHandlerTest.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/logconfig/LogConfigHandlerTest.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/logconfig/LogConfigHandlerTest.java index 4123dad..c07035b 100644 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/logconfig/LogConfigHandlerTest.java +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/logconfig/LogConfigHandlerTest.java @@ -61,7 +61,7 @@ public class LogConfigHandlerTest { @BeforeClass public static void init() throws Exception { - LogFeederUtil.loadProperties("logfeeder.properties", null); + LogFeederUtil.loadProperties("logfeeder.properties"); LogSearchConfig config = strictMock(LogSearchConfig.class); config.createLogLevelFilter(anyString(), anyString(), anyObject(LogLevelFilter.class)); http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetrcisManagerTest.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetrcisManagerTest.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetrcisManagerTest.java deleted file mode 100644 index 8ee6d00..0000000 --- a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetrcisManagerTest.java +++ /dev/null @@ -1,128 +0,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. - */ - -package org.apache.ambari.logfeeder.metrics; - -import static org.easymock.EasyMock.*; -import static org.junit.Assert.*; -import org.easymock.Capture; -import org.easymock.CaptureType; -import org.easymock.EasyMock; - -import java.lang.reflect.Field; -import java.util.Arrays; -import java.util.List; -import java.util.TreeMap; - -import org.apache.ambari.logfeeder.util.LogFeederUtil; -import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; -import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; - -public class MetrcisManagerTest { - - private MetricsManager manager; - private LogFeederAMSClient mockClient; - private Capture<TimelineMetrics> capture; - - @BeforeClass - public static void loadProperties() throws Exception { - LogFeederUtil.loadProperties("logfeeder.properties", null); - } - - @Before - public void init() throws Exception { - manager = new MetricsManager(); - manager.init(); - - mockClient = strictMock(LogFeederAMSClient.class); - Field f = MetricsManager.class.getDeclaredField("amsClient"); - f.setAccessible(true); - f.set(manager, mockClient); - - capture = EasyMock.newCapture(CaptureType.FIRST); - mockClient.emitMetrics(EasyMock.capture(capture)); - EasyMock.expectLastCall().andReturn(true).once(); - - replay(mockClient); - } - - @Test - public void testMetricManager_pointInTime() throws Exception { - MetricData metricCount1 = new MetricData("metric1", true); - metricCount1.value = 123; - metricCount1.prevPublishValue = 0; - metricCount1.publishCount = 0; - - manager.useMetrics(Arrays.asList(metricCount1)); - - verify(mockClient); - - TimelineMetrics metrics = capture.getValue(); - List<TimelineMetric> metricList = metrics.getMetrics(); - assertEquals(metricList.size(), 1); - - TimelineMetric metric = metricList.get(0); - assertEquals(metric.getHostName(), "test_host_name"); - assertEquals(metric.getAppId(), "logfeeder"); - assertEquals(metric.getMetricName(), "metric1"); - assertEquals(metric.getType(), "Long"); - - TreeMap<Long, Double> values = metric.getMetricValues(); - assertEquals(values.size(), 1); - assertEquals(values.firstEntry().getValue(), Double.valueOf(123.0)); - } - - @Test - public void testMetricManager_notPointInTime() throws Exception { - MetricData metricCount1 = new MetricData("metric1", false); - metricCount1.value = 123; - metricCount1.prevPublishValue = 0; - metricCount1.publishCount = 0; - - MetricData metricCount2 = new MetricData("metric1", false); - metricCount2.value = 123; - metricCount2.prevPublishValue = 100; - metricCount2.publishCount = 0; - - MetricData metricCount3 = new MetricData("metric1", false); // not included due to decrease of count - metricCount3.value = 99; - metricCount3.prevPublishValue = 100; - metricCount3.publishCount = 1; - - manager.useMetrics(Arrays.asList(metricCount1, metricCount2, metricCount3)); - - verify(mockClient); - - TimelineMetrics metrics = capture.getValue(); - List<TimelineMetric> metricList = metrics.getMetrics(); - assertEquals(metricList.size(), 1); - - TimelineMetric metric = metricList.get(0); - assertEquals(metric.getHostName(), "test_host_name"); - assertEquals(metric.getAppId(), "logfeeder"); - assertEquals(metric.getMetricName(), "metric1"); - assertEquals(metric.getType(), "Long"); - - TreeMap<Long, Double> values = metric.getMetricValues(); - assertEquals(values.size(), 1); - assertEquals(values.firstEntry().getValue(), Double.valueOf(146.0)); - } -} http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetricsManagerTest.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetricsManagerTest.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetricsManagerTest.java new file mode 100644 index 0000000..1461352 --- /dev/null +++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/metrics/MetricsManagerTest.java @@ -0,0 +1,128 @@ +/* + * 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.ambari.logfeeder.metrics; + +import static org.easymock.EasyMock.*; +import static org.junit.Assert.*; +import org.easymock.Capture; +import org.easymock.CaptureType; +import org.easymock.EasyMock; + +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.List; +import java.util.TreeMap; + +import org.apache.ambari.logfeeder.util.LogFeederUtil; +import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; +import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +public class MetricsManagerTest { + + private MetricsManager manager; + private LogFeederAMSClient mockClient; + private Capture<TimelineMetrics> capture; + + @BeforeClass + public static void loadProperties() throws Exception { + LogFeederUtil.loadProperties("logfeeder.properties"); + } + + @Before + public void init() throws Exception { + manager = new MetricsManager(); + manager.init(); + + mockClient = strictMock(LogFeederAMSClient.class); + Field f = MetricsManager.class.getDeclaredField("amsClient"); + f.setAccessible(true); + f.set(manager, mockClient); + + capture = EasyMock.newCapture(CaptureType.FIRST); + mockClient.emitMetrics(EasyMock.capture(capture)); + EasyMock.expectLastCall().andReturn(true).once(); + + replay(mockClient); + } + + @Test + public void testMetricManager_pointInTime() throws Exception { + MetricData metricCount1 = new MetricData("metric1", true); + metricCount1.value = 123; + metricCount1.prevPublishValue = 0; + metricCount1.publishCount = 0; + + manager.useMetrics(Arrays.asList(metricCount1)); + + verify(mockClient); + + TimelineMetrics metrics = capture.getValue(); + List<TimelineMetric> metricList = metrics.getMetrics(); + assertEquals(metricList.size(), 1); + + TimelineMetric metric = metricList.get(0); + assertEquals(metric.getHostName(), "test_host_name"); + assertEquals(metric.getAppId(), "logfeeder"); + assertEquals(metric.getMetricName(), "metric1"); + assertEquals(metric.getType(), "Long"); + + TreeMap<Long, Double> values = metric.getMetricValues(); + assertEquals(values.size(), 1); + assertEquals(values.firstEntry().getValue(), Double.valueOf(123.0)); + } + + @Test + public void testMetricManager_notPointInTime() throws Exception { + MetricData metricCount1 = new MetricData("metric1", false); + metricCount1.value = 123; + metricCount1.prevPublishValue = 0; + metricCount1.publishCount = 0; + + MetricData metricCount2 = new MetricData("metric1", false); + metricCount2.value = 123; + metricCount2.prevPublishValue = 100; + metricCount2.publishCount = 0; + + MetricData metricCount3 = new MetricData("metric1", false); // not included due to decrease of count + metricCount3.value = 99; + metricCount3.prevPublishValue = 100; + metricCount3.publishCount = 1; + + manager.useMetrics(Arrays.asList(metricCount1, metricCount2, metricCount3)); + + verify(mockClient); + + TimelineMetrics metrics = capture.getValue(); + List<TimelineMetric> metricList = metrics.getMetrics(); + assertEquals(metricList.size(), 1); + + TimelineMetric metric = metricList.get(0); + assertEquals(metric.getHostName(), "test_host_name"); + assertEquals(metric.getAppId(), "logfeeder"); + assertEquals(metric.getMetricName(), "metric1"); + assertEquals(metric.getType(), "Long"); + + TreeMap<Long, Double> values = metric.getMetricValues(); + assertEquals(values.size(), 1); + assertEquals(values.firstEntry().getValue(), Double.valueOf(146.0)); + } +} http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-server/pom.xml ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-server/pom.xml b/ambari-logsearch/ambari-logsearch-server/pom.xml index 7cd90eb..fc4029b 100755 --- a/ambari-logsearch/ambari-logsearch-server/pom.xml +++ b/ambari-logsearch/ambari-logsearch-server/pom.xml @@ -601,6 +601,33 @@ </dependency> <dependency> <groupId>org.apache.ambari</groupId> + <artifactId>ambari-logsearch-logfeeder</artifactId> + <version>${project.version}</version> + <exclusions> + <exclusion> + <groupId>com.sun.jersey</groupId> + <artifactId>jetty-util</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jersey</groupId> + <artifactId>jersey-core</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jersey</groupId> + <artifactId>jersey-json</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jersey</groupId> + <artifactId>jersey-server</artifactId> + </exclusion> + <exclusion> + <groupId>javax.servlet</groupId> + <artifactId>servlet-api</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.ambari</groupId> <artifactId>ambari-metrics-common</artifactId> <version>${project.version}</version> </dependency> http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/doc/DocConstants.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/doc/DocConstants.java b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/doc/DocConstants.java index 2ab5f0a..73de0ee 100644 --- a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/doc/DocConstants.java +++ b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/doc/DocConstants.java @@ -120,6 +120,7 @@ public class DocConstants { public static final String GET_SERVICE_NAMES_OD = "Get service names"; public static final String GET_SHIPPER_CONFIG_OD = "Get shipper config"; public static final String SET_SHIPPER_CONFIG_OD = "Set shipper config"; + public static final String TEST_SHIPPER_CONFIG_OD = "Test shipper config"; public static final String GET_LOG_LEVEL_FILTER_OD = "Get log level filter"; public static final String UPDATE_LOG_LEVEL_FILTER_OD = "Update log level filter"; } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/manager/ShipperConfigManager.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/manager/ShipperConfigManager.java b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/manager/ShipperConfigManager.java index a0db92f..2c143c0 100644 --- a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/manager/ShipperConfigManager.java +++ b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/manager/ShipperConfigManager.java @@ -20,7 +20,10 @@ package org.apache.ambari.logsearch.manager; import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.ambari.logfeeder.common.LogEntryParseTester; import org.apache.ambari.logsearch.config.api.model.inputconfig.InputConfig; import org.apache.ambari.logsearch.configurer.LogSearchConfigConfigurer; import org.apache.ambari.logsearch.model.common.LSServerInputConfig; @@ -33,6 +36,9 @@ import com.google.common.collect.ImmutableMap; import javax.annotation.PostConstruct; import javax.inject.Inject; import javax.inject.Named; +import javax.validation.ConstraintViolation; +import javax.validation.Validation; +import javax.validation.Validator; import javax.ws.rs.core.MediaType; import javax.ws.rs.core.Response; @@ -92,6 +98,24 @@ public class ShipperConfigManager extends JsonManagerBase { } } + public Map<String, Object> testShipperConfig(String shipperConfig, String logId, String testEntry, String clusterName) { + try { + LSServerInputConfig inputConfigValidate = new ObjectMapper().readValue(shipperConfig, LSServerInputConfig.class); + Validator validator = Validation.buildDefaultValidatorFactory().getValidator(); + Set<ConstraintViolation<LSServerInputConfig>> violations = validator.validate(inputConfigValidate); + if (!violations.isEmpty()) { + throw new IllegalArgumentException("Error validating shipper config:\n" + violations); + } + + String globalConfigs = LogSearchConfigConfigurer.getConfig().getGlobalConfigs(clusterName); + LogEntryParseTester tester = new LogEntryParseTester(testEntry, shipperConfig, globalConfigs, logId); + return tester.parse(); + } catch (Exception e) { + Map<String, Object> errorResponse = ImmutableMap.of("errorMessage", (Object)e.toString()); + return errorResponse; + } + } + public LSServerLogLevelFilterMap getLogLevelFilters(String clusterName) { return new LSServerLogLevelFilterMap(LogSearchConfigConfigurer.getConfig().getLogLevelFilters(clusterName)); } http://git-wip-us.apache.org/repos/asf/ambari/blob/943c1b0d/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/rest/ShipperConfigResource.java ---------------------------------------------------------------------- diff --git a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/rest/ShipperConfigResource.java b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/rest/ShipperConfigResource.java index 71da326..d8b1441 100644 --- a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/rest/ShipperConfigResource.java +++ b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/rest/ShipperConfigResource.java @@ -23,6 +23,7 @@ import javax.inject.Inject; import javax.inject.Named; import javax.validation.Valid; import javax.validation.executable.ValidateOnExecution; +import javax.ws.rs.FormParam; import javax.ws.rs.GET; import javax.ws.rs.POST; import javax.ws.rs.PUT; @@ -40,11 +41,13 @@ import org.apache.ambari.logsearch.model.common.LSServerLogLevelFilterMap; import org.springframework.context.annotation.Scope; import java.util.List; +import java.util.Map; import static org.apache.ambari.logsearch.doc.DocConstants.ShipperConfigOperationDescriptions.GET_LOG_LEVEL_FILTER_OD; import static org.apache.ambari.logsearch.doc.DocConstants.ShipperConfigOperationDescriptions.GET_SERVICE_NAMES_OD; import static org.apache.ambari.logsearch.doc.DocConstants.ShipperConfigOperationDescriptions.GET_SHIPPER_CONFIG_OD; import static org.apache.ambari.logsearch.doc.DocConstants.ShipperConfigOperationDescriptions.SET_SHIPPER_CONFIG_OD; +import static org.apache.ambari.logsearch.doc.DocConstants.ShipperConfigOperationDescriptions.TEST_SHIPPER_CONFIG_OD; import static org.apache.ambari.logsearch.doc.DocConstants.ShipperConfigOperationDescriptions.UPDATE_LOG_LEVEL_FILTER_OD; @Api(value = "shipper", description = "Shipper config operations") @@ -93,6 +96,15 @@ public class ShipperConfigResource { return shipperConfigManager.setInputConfig(clusterName, serviceName, request); } + @POST + @Path("/input/{clusterName}/test") + @Produces({"application/json"}) + @ApiOperation(TEST_SHIPPER_CONFIG_OD) + public Map<String, Object> testShipperConfig(@FormParam("shipper_config") String shipperConfig, @FormParam("log_id") String logId, + @FormParam("test_entry") String testEntry, @PathParam("clusterName") String clusterName) { + return shipperConfigManager.testShipperConfig(shipperConfig, logId, testEntry, clusterName); + } + @GET @Path("/filters/{clusterName}/level") @Produces({"application/json"})
