You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by ol...@apache.org on 2018/07/23 12:39:42 UTC
[ambari] branch branch-2.7 updated (9c409ff -> c04daaf)
This is an automated email from the ASF dual-hosted git repository.
oleewere pushed a change to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/ambari.git.
from 9c409ff AMBARI-24320. Using the proper command name when creating a pre-initialized ActionExecuteContent instead of the hard coded SET_KEYTAB value (#1834)
new 8dd0150 AMBARI-23079. Log Feeder: support to use load balancer for Solr API (not only cloud client) (#1835)
new c04daaf AMBARI-24329. Log Feeder define default log levels per component.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../api/model/inputconfig/InputDescriptor.java | 3 +
.../inputconfig/impl/InputDescriptorImpl.java | 20 +++++++
.../config/solr/LogLevelFilterManagerSolr.java | 2 +-
.../ambari/logfeeder/plugin/common/ConfigItem.java | 13 +++++
.../logfeeder/common/LogFeederConstants.java | 1 +
.../common/LogFeederSolrClientFactory.java | 64 ++++++++++++++++++++++
.../ambari/logfeeder/conf/ApplicationConfig.java | 17 ++----
.../ambari/logfeeder/conf/LogFeederProps.java | 28 +++++++++-
.../loglevelfilter/LogLevelFilterHandler.java | 20 +++----
.../ambari/logfeeder/output/OutputManagerImpl.java | 17 +++++-
.../apache/ambari/logfeeder/output/OutputSolr.java | 40 +++++++-------
.../ambari/logfeeder/output/OutputManagerTest.java | 6 +-
.../logsearch/model/common/LSServerInput.java | 9 +++
.../test-config/logfeeder/logfeeder.properties | 3 +-
.../input.config-logsearch-docker.json | 5 +-
15 files changed, 197 insertions(+), 51 deletions(-)
create mode 100644 ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederSolrClientFactory.java
[ambari] 02/02: AMBARI-24329. Log Feeder define default log levels
per component.
Posted by ol...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
oleewere pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/ambari.git
commit c04daafbcbbaac4a5a2fe5796e3aa35cfdfe7585
Author: Oliver Szabo <ol...@gmail.com>
AuthorDate: Sun Jul 22 14:50:02 2018 +0200
AMBARI-24329. Log Feeder define default log levels per component.
---
.../api/model/inputconfig/InputDescriptor.java | 3 +++
.../model/inputconfig/impl/InputDescriptorImpl.java | 20 ++++++++++++++++++++
.../loglevelfilter/LogLevelFilterHandler.java | 20 ++++++++++----------
.../ambari/logfeeder/output/OutputManagerImpl.java | 17 +++++++++++++++--
.../ambari/logfeeder/output/OutputManagerTest.java | 6 ++++--
.../ambari/logsearch/model/common/LSServerInput.java | 9 +++++++++
.../shipper-conf/input.config-logsearch-docker.json | 5 ++++-
7 files changed, 65 insertions(+), 15 deletions(-)
diff --git a/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/InputDescriptor.java b/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/InputDescriptor.java
index 71f3cd1..7c00a6b 100644
--- a/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/InputDescriptor.java
+++ b/ambari-logsearch/ambari-logsearch-config-api/src/main/java/org/apache/ambari/logsearch/config/api/model/inputconfig/InputDescriptor.java
@@ -19,6 +19,7 @@
package org.apache.ambari.logsearch.config.api.model.inputconfig;
+import java.util.List;
import java.util.Map;
public interface InputDescriptor {
@@ -53,4 +54,6 @@ public interface InputDescriptor {
String getGroup();
Boolean isInitDefaultFields();
+
+ List<String> getDefaultLogLevels();
}
diff --git a/ambari-logsearch/ambari-logsearch-config-json/src/main/java/org/apache/ambari/logsearch/config/json/model/inputconfig/impl/InputDescriptorImpl.java b/ambari-logsearch/ambari-logsearch-config-json/src/main/java/org/apache/ambari/logsearch/config/json/model/inputconfig/impl/InputDescriptorImpl.java
index 3a04ebc..d175c58 100644
--- a/ambari-logsearch/ambari-logsearch-config-json/src/main/java/org/apache/ambari/logsearch/config/json/model/inputconfig/impl/InputDescriptorImpl.java
+++ b/ambari-logsearch/ambari-logsearch-config-json/src/main/java/org/apache/ambari/logsearch/config/json/model/inputconfig/impl/InputDescriptorImpl.java
@@ -19,6 +19,7 @@
package org.apache.ambari.logsearch.config.json.model.inputconfig.impl;
+import java.util.List;
import java.util.Map;
import org.apache.ambari.logsearch.config.api.ShipperConfigElementDescription;
@@ -200,6 +201,16 @@ public abstract class InputDescriptorImpl implements InputDescriptor {
@SerializedName("init_default_fields")
private Boolean initDefaultFields;
+ @ShipperConfigElementDescription(
+ path = "/input/[]/default_log_levels",
+ type = "list of strings",
+ description = "Use these as default log levels for the input - overrides the global default log levels.",
+ examples = {"default_log_levels: [\"INFO\", \"WARN\"]"}
+ )
+ @Expose
+ @SerializedName("default_log_levels")
+ private List<String> defaultLogLevels;
+
public String getType() {
return type;
}
@@ -329,4 +340,13 @@ public abstract class InputDescriptorImpl implements InputDescriptor {
public void setInitDefaultFields(Boolean initDefaultFields) {
this.initDefaultFields = initDefaultFields;
}
+
+ @Override
+ public List<String> getDefaultLogLevels() {
+ return defaultLogLevels;
+ }
+
+ public void setDefaultLogLevels(List<String> defaultLogLevels) {
+ this.defaultLogLevels = defaultLogLevels;
+ }
}
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/loglevelfilter/LogLevelFilterHandler.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/loglevelfilter/LogLevelFilterHandler.java
index 977e537..8c35d56 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/loglevelfilter/LogLevelFilterHandler.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/loglevelfilter/LogLevelFilterHandler.java
@@ -98,29 +98,29 @@ public class LogLevelFilterHandler implements LogLevelFilterMonitor {
return filters;
}
- public boolean isAllowed(String hostName, String logId, String level) {
+ public boolean isAllowed(String hostName, String logId, String level, List<String> defaultLogLevels) {
if (!logFeederProps.isLogLevelFilterEnabled()) {
return true;
}
- LogLevelFilter logFilter = findLogFilter(logId);
+ LogLevelFilter logFilter = findLogFilter(logId, defaultLogLevels);
List<String> allowedLevels = getAllowedLevels(hostName, logFilter);
return allowedLevels.isEmpty() || allowedLevels.contains(level);
}
- public boolean isAllowed(String jsonBlock, InputMarker inputMarker) {
+ public boolean isAllowed(String jsonBlock, InputMarker inputMarker, List<String> defaultLogLevels) {
if (org.apache.commons.lang3.StringUtils.isEmpty(jsonBlock)) {
return DEFAULT_VALUE;
}
Map<String, Object> jsonObj = LogFeederUtil.toJSONObject(jsonBlock);
- return isAllowed(jsonObj, inputMarker);
+ return isAllowed(jsonObj, inputMarker, defaultLogLevels);
}
- public boolean isAllowed(Map<String, Object> jsonObj, InputMarker inputMarker) {
+ public boolean isAllowed(Map<String, Object> jsonObj, InputMarker inputMarker, List<String> defaultLogLevels) {
if ("audit".equals(inputMarker.getInput().getInputDescriptor().getRowtype()))
return true;
- boolean isAllowed = applyFilter(jsonObj);
+ boolean isAllowed = applyFilter(jsonObj, defaultLogLevels);
if (!isAllowed) {
LOG.trace("Filter block the content :" + LogFeederUtil.getGson().toJson(jsonObj));
}
@@ -128,7 +128,7 @@ public class LogLevelFilterHandler implements LogLevelFilterMonitor {
}
- public boolean applyFilter(Map<String, Object> jsonObj) {
+ public boolean applyFilter(Map<String, Object> jsonObj, List<String> defaultLogLevels) {
if (MapUtils.isEmpty(jsonObj)) {
LOG.warn("Output jsonobj is empty");
return DEFAULT_VALUE;
@@ -138,13 +138,13 @@ public class LogLevelFilterHandler implements LogLevelFilterMonitor {
String logId = (String) jsonObj.get(LogFeederConstants.SOLR_COMPONENT);
String level = (String) jsonObj.get(LogFeederConstants.SOLR_LEVEL);
if (org.apache.commons.lang3.StringUtils.isNotBlank(hostName) && org.apache.commons.lang3.StringUtils.isNotBlank(logId) && org.apache.commons.lang3.StringUtils.isNotBlank(level)) {
- return isAllowed(hostName, logId, level);
+ return isAllowed(hostName, logId, level, defaultLogLevels);
} else {
return DEFAULT_VALUE;
}
}
- private synchronized LogLevelFilter findLogFilter(String logId) {
+ private synchronized LogLevelFilter findLogFilter(String logId, List<String> defaultLogLevels) {
LogLevelFilter logFilter = filters.get(logId);
if (logFilter != null) {
return logFilter;
@@ -153,7 +153,7 @@ public class LogLevelFilterHandler implements LogLevelFilterMonitor {
LOG.info("Filter is not present for log " + logId + ", creating default filter");
LogLevelFilter defaultFilter = new LogLevelFilter();
defaultFilter.setLabel(logId);
- defaultFilter.setDefaultLevels(logFeederProps.getIncludeDefaultLogLevels());
+ defaultFilter.setDefaultLevels(defaultLogLevels);
try {
config.getLogLevelFilterManager().createLogLevelFilter(logFeederProps.getClusterName(), logId, defaultFilter);
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputManagerImpl.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputManagerImpl.java
index 612573b..09951cc 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputManagerImpl.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputManagerImpl.java
@@ -31,6 +31,7 @@ import org.apache.ambari.logfeeder.plugin.manager.OutputManager;
import org.apache.ambari.logfeeder.plugin.output.Output;
import org.apache.ambari.logfeeder.util.LogFeederUtil;
import org.apache.ambari.logsearch.config.api.OutputConfigMonitor;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
@@ -147,7 +148,8 @@ public class OutputManagerImpl extends OutputManager {
jsonObj.put("message_md5", "" + Hashing.md5().hashBytes(logMessage.getBytes()).asLong());
}
}
- if (logLevelFilterHandler.isAllowed(jsonObj, inputMarker)
+ List<String> defaultLogLevels = getDefaultLogLevels(input);
+ if (logLevelFilterHandler.isAllowed(jsonObj, inputMarker, defaultLogLevels)
&& !outputLineFilter.apply(jsonObj, inputMarker.getInput())) {
List<? extends Output> outputList = input.getOutputList();
for (Output output : outputList) {
@@ -160,6 +162,16 @@ public class OutputManagerImpl extends OutputManager {
}
}
+ private List<String> getDefaultLogLevels(Input input) {
+ List<String> defaultLogLevels = logFeederProps.getIncludeDefaultLogLevels();
+ List<String> overrideDefaultLogLevels = input.getInputDescriptor().getDefaultLogLevels();
+ if (CollectionUtils.isNotEmpty(overrideDefaultLogLevels)) {
+ return overrideDefaultLogLevels;
+ } else {
+ return defaultLogLevels;
+ }
+ }
+
@SuppressWarnings("unchecked")
private String truncateLongLogMessage(Map<String, Object> jsonObj, Input input, String logMessage) {
if (logMessage != null && logMessage.getBytes().length > MAX_OUTPUT_SIZE) {
@@ -181,7 +193,8 @@ public class OutputManagerImpl extends OutputManager {
}
public void write(String jsonBlock, InputMarker inputMarker) {
- if (logLevelFilterHandler.isAllowed(jsonBlock, inputMarker)) {
+ List<String> defaultLogLevels = getDefaultLogLevels(inputMarker.getInput());
+ if (logLevelFilterHandler.isAllowed(jsonBlock, inputMarker, defaultLogLevels)) {
List<? extends Output> outputList = inputMarker.getInput().getOutputList();
for (Output output : outputList) {
try {
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputManagerTest.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputManagerTest.java
index 1a651ab..9536cf8 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputManagerTest.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputManagerTest.java
@@ -110,7 +110,7 @@ public class OutputManagerTest {
expect(mockInput.isUseEventMD5()).andReturn(false).anyTimes();
expect(mockInput.isGenEventMD5()).andReturn(false).anyTimes();
expect(mockInput.getInputDescriptor()).andReturn(inputDescriptor).anyTimes();
- expect(mockFilter.isAllowed(jsonObj, inputMarker)).andReturn(true).anyTimes();
+ expect(mockFilter.isAllowed(jsonObj, inputMarker, null)).andReturn(true).anyTimes();
expect(mockInput.getCache()).andReturn(null);
expect(mockInput.getOutputList()).andReturn(Arrays.asList(output1, output2, output3));
@@ -121,6 +121,7 @@ public class OutputManagerTest {
replay(output1, output2, output3, mockFilter, mockInput);
OutputManagerImpl manager = new OutputManagerImpl();
+ manager.setLogFeederProps(new LogFeederProps());
manager.setLogLevelFilterHandler(mockFilter);
manager.add(output1);
manager.add(output2);
@@ -146,7 +147,7 @@ public class OutputManagerTest {
LogLevelFilterHandler mockFilter = strictMock(LogLevelFilterHandler.class);
expect(mockInput.getInputDescriptor()).andReturn(inputDescriptor).anyTimes();
- expect(mockFilter.isAllowed(jsonString, inputMarker)).andReturn(true).anyTimes();
+ expect(mockFilter.isAllowed(jsonString, inputMarker, null)).andReturn(true).anyTimes();
expect(mockInput.getOutputList()).andReturn(Arrays.asList(output1, output2, output3));
output1.write(jsonString, inputMarker); expectLastCall();
@@ -157,6 +158,7 @@ public class OutputManagerTest {
OutputManagerImpl manager = new OutputManagerImpl();
manager.setLogLevelFilterHandler(mockFilter);
+ manager.setLogFeederProps(new LogFeederProps());
manager.add(output1);
manager.add(output2);
manager.add(output3);
diff --git a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/model/common/LSServerInput.java b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/model/common/LSServerInput.java
index af28f17..272f443 100644
--- a/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/model/common/LSServerInput.java
+++ b/ambari-logsearch/ambari-logsearch-server/src/main/java/org/apache/ambari/logsearch/model/common/LSServerInput.java
@@ -19,6 +19,7 @@
package org.apache.ambari.logsearch.model.common;
+import java.util.List;
import java.util.Map;
import javax.validation.constraints.NotNull;
@@ -73,6 +74,9 @@ public abstract class LSServerInput {
@JsonProperty("init_default_fields")
private Boolean initDefaultFields;
+
+ @JsonProperty("default_log_levels")
+ private List<String> defaultLogLevels;
public LSServerInput() {}
@@ -92,6 +96,7 @@ public abstract class LSServerInput {
this.cacheDedupInterval = inputDescriptor.getCacheDedupInterval();
this.isEnabled = inputDescriptor.isEnabled();
this.initDefaultFields = inputDescriptor.isInitDefaultFields();
+ this.defaultLogLevels = inputDescriptor.getDefaultLogLevels();
}
public String getType() {
@@ -153,4 +158,8 @@ public abstract class LSServerInput {
public Boolean getInitDefaultFields() {
return initDefaultFields;
}
+
+ public List<String> getDefaultLogLevels() {
+ return defaultLogLevels;
+ }
}
diff --git a/ambari-logsearch/docker/test-config/logfeeder/shipper-conf/input.config-logsearch-docker.json b/ambari-logsearch/docker/test-config/logfeeder/shipper-conf/input.config-logsearch-docker.json
index a420960..04a3adc 100644
--- a/ambari-logsearch/docker/test-config/logfeeder/shipper-conf/input.config-logsearch-docker.json
+++ b/ambari-logsearch/docker/test-config/logfeeder/shipper-conf/input.config-logsearch-docker.json
@@ -3,7 +3,10 @@
{
"type": "logsearch_server",
"rowtype": "service",
- "docker": "true"
+ "docker": "true",
+ "default_log_levels" : [
+ "FATAL", "ERROR", "WARN", "INFO", "DEBUG"
+ ]
}
],
"filter": [
[ambari] 01/02: AMBARI-23079. Log Feeder: support to use load
balancer for Solr API (not only cloud client) (#1835)
Posted by ol...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
oleewere pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/ambari.git
commit 8dd0150cd364f293c555e09fa47ad9e022b59d1f
Author: Olivér Szabó <ol...@gmail.com>
AuthorDate: Mon Jul 23 09:33:46 2018 +0200
AMBARI-23079. Log Feeder: support to use load balancer for Solr API (not only cloud client) (#1835)
---
.../config/solr/LogLevelFilterManagerSolr.java | 2 +-
.../ambari/logfeeder/plugin/common/ConfigItem.java | 13 +++++
.../logfeeder/common/LogFeederConstants.java | 1 +
.../common/LogFeederSolrClientFactory.java | 64 ++++++++++++++++++++++
.../ambari/logfeeder/conf/ApplicationConfig.java | 17 ++----
.../ambari/logfeeder/conf/LogFeederProps.java | 28 +++++++++-
.../apache/ambari/logfeeder/output/OutputSolr.java | 40 +++++++-------
.../test-config/logfeeder/logfeeder.properties | 3 +-
8 files changed, 132 insertions(+), 36 deletions(-)
diff --git a/ambari-logsearch/ambari-logsearch-config-solr/src/main/java/org/apache/ambari/logsearch/config/solr/LogLevelFilterManagerSolr.java b/ambari-logsearch/ambari-logsearch-config-solr/src/main/java/org/apache/ambari/logsearch/config/solr/LogLevelFilterManagerSolr.java
index 8d8976b..0eabead 100644
--- a/ambari-logsearch/ambari-logsearch-config-solr/src/main/java/org/apache/ambari/logsearch/config/solr/LogLevelFilterManagerSolr.java
+++ b/ambari-logsearch/ambari-logsearch-config-solr/src/main/java/org/apache/ambari/logsearch/config/solr/LogLevelFilterManagerSolr.java
@@ -130,7 +130,7 @@ public class LogLevelFilterManagerSolr implements LogLevelFilterManager {
}
}
} catch (Exception e) {
- LOG.error("Error during getting log level filters: {}", e);
+ LOG.error("Error during getting log level filters: {}", e.getMessage());
}
logLevelFilterMap.setFilter(logLevelFilterTreeMap);
return logLevelFilterMap;
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder-plugin-api/src/main/java/org/apache/ambari/logfeeder/plugin/common/ConfigItem.java b/ambari-logsearch/ambari-logsearch-logfeeder-plugin-api/src/main/java/org/apache/ambari/logfeeder/plugin/common/ConfigItem.java
index 1cbbfd5..5b50a7e 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder-plugin-api/src/main/java/org/apache/ambari/logfeeder/plugin/common/ConfigItem.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder-plugin-api/src/main/java/org/apache/ambari/logfeeder/plugin/common/ConfigItem.java
@@ -111,6 +111,19 @@ public abstract class ConfigItem<PROP_TYPE extends LogFeederProperties> implemen
this.drain = drain;
}
+ public List<String> getListValue(String key) {
+ return getListValue(key, null);
+ }
+
+ public List<String> getListValue(String key, List<String> defaultValue) {
+ Object value = configs.get(key);
+ if (value != null) {
+ return (List<String>)value;
+ } else {
+ return defaultValue;
+ }
+ }
+
public String getStringValue(String property) {
return getStringValue(property, null);
}
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederConstants.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederConstants.java
index 10e38f9..251b4fc 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederConstants.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederConstants.java
@@ -102,5 +102,6 @@ public class LogFeederConstants {
public static final boolean MONITOR_SOLR_FILTER_STORAGE_DEFAULT = true;
public static final String SOLR_ZK_CONNECTION_STRING = "logfeeder.solr.zk_connect_string";
+ public static final String SOLR_URLS = "logfeeder.solr.urls";
}
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederSolrClientFactory.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederSolrClientFactory.java
new file mode 100644
index 0000000..cf94fb5
--- /dev/null
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/common/LogFeederSolrClientFactory.java
@@ -0,0 +1,64 @@
+/*
+ * 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 org.apache.commons.lang3.StringUtils;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.impl.HttpSolrClient;
+import org.apache.solr.client.solrj.impl.LBHttpSolrClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class LogFeederSolrClientFactory {
+
+ private static final Logger logger = LoggerFactory.getLogger(LogFeederSolrClientFactory.class);
+
+ public SolrClient createSolrClient(String zkConnectionString, String[] solrUrls, String collection) {
+ logger.info("Creating solr client ...");
+ logger.info("Using collection=" + collection);
+ if (solrUrls != null && solrUrls.length > 0) {
+ logger.info("Using lbHttpSolrClient with urls: {}",
+ StringUtils.join(appendTo("/" + collection, solrUrls), ","));
+ LBHttpSolrClient.Builder builder = new LBHttpSolrClient.Builder();
+ builder.withBaseSolrUrls(solrUrls);
+ return builder.build();
+ } else {
+ logger.info("Using zookeepr. zkConnectString=" + zkConnectionString);
+ CloudSolrClient.Builder builder = new CloudSolrClient.Builder();
+ builder.withZkHost(zkConnectionString);
+ CloudSolrClient solrClient = builder.build();
+ solrClient.setDefaultCollection(collection);
+ return solrClient;
+ }
+ }
+
+ private String[] appendTo(String toAppend, String... appendees) {
+ for (int i = 0; i < appendees.length; i++) {
+ appendees[i] = appendees[i] + toAppend;
+ }
+ return appendees;
+ }
+
+}
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/ApplicationConfig.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/ApplicationConfig.java
index b431464..ccae373 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/ApplicationConfig.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/ApplicationConfig.java
@@ -19,6 +19,7 @@
package org.apache.ambari.logfeeder.conf;
import com.google.common.collect.Maps;
+import org.apache.ambari.logfeeder.common.LogFeederSolrClientFactory;
import org.apache.ambari.logfeeder.docker.DockerContainerRegistry;
import org.apache.ambari.logfeeder.common.LogFeederConstants;
import org.apache.ambari.logfeeder.input.InputConfigUploader;
@@ -38,9 +39,7 @@ import org.apache.ambari.logsearch.config.local.LogSearchConfigLogFeederLocal;
import org.apache.ambari.logsearch.config.solr.LogLevelFilterManagerSolr;
import org.apache.ambari.logsearch.config.solr.LogLevelFilterUpdaterSolr;
import org.apache.ambari.logsearch.config.zookeeper.LogSearchConfigLogFeederZK;
-import org.apache.commons.lang3.StringUtils;
import org.apache.solr.client.solrj.SolrClient;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
@@ -69,7 +68,7 @@ public class ApplicationConfig {
}
@Bean
- @DependsOn("logSearchConfigLogFeeder")
+ @DependsOn({"logSearchConfigLogFeeder", "propertyConfigurer"})
public ConfigHandler configHandler() throws Exception {
return new ConfigHandler(logSearchConfigLogFeeder());
}
@@ -95,15 +94,9 @@ public class ApplicationConfig {
@Bean
public LogLevelFilterManager logLevelFilterManager() {
if (logFeederProps.isSolrFilterStorage()) {
- if (StringUtils.isNotEmpty(logFeederProps.getSolrZkConnectString())) {
- CloudSolrClient.Builder builder = new CloudSolrClient.Builder();
- builder.withZkHost(logFeederProps.getSolrZkConnectString());
- CloudSolrClient solrClient = builder.build();
- solrClient.setDefaultCollection("history");
- return new LogLevelFilterManagerSolr(solrClient);
- } else {
- return null; // TODO: use lb http client
- }
+ SolrClient solrClient = new LogFeederSolrClientFactory().createSolrClient(
+ logFeederProps.getSolrZkConnectString(), logFeederProps.getSolrUrls(), "history");
+ return new LogLevelFilterManagerSolr(solrClient);
} else { // no default filter manager
return null;
}
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/LogFeederProps.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/LogFeederProps.java
index 8f73e2b..12408b4 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/LogFeederProps.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/conf/LogFeederProps.java
@@ -21,6 +21,7 @@ package org.apache.ambari.logfeeder.conf;
import org.apache.ambari.logfeeder.common.LogFeederConstants;
import org.apache.ambari.logfeeder.plugin.common.LogFeederProperties;
import org.apache.ambari.logsearch.config.api.LogSearchPropertyDescription;
+import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.AbstractEnvironment;
@@ -179,6 +180,15 @@ public class LogFeederProps implements LogFeederProperties {
@Value("${" + LogFeederConstants.SOLR_ZK_CONNECTION_STRING + ":}")
private String solrZkConnectString;
+ @LogSearchPropertyDescription(
+ name = LogFeederConstants.SOLR_URLS,
+ description = "Comma separated solr urls (with protocol and port), override "+ LogFeederConstants.SOLR_ZK_CONNECTION_STRING + " config",
+ examples = {"https://localhost1:8983/solr,https://localhost2:8983"},
+ sources = {LogFeederConstants.LOGFEEDER_PROPERTIES_FILE}
+ )
+ @Value("${" + LogFeederConstants.SOLR_URLS + ":}")
+ private String solrUrlsStr;
+
@Inject
private LogEntryCacheConfig logEntryCacheConfig;
@@ -285,7 +295,7 @@ public class LogFeederProps implements LogFeederProperties {
}
public boolean isUseLocalConfigs() {
- return useLocalConfigs;
+ return this.useLocalConfigs;
}
public void setUseLocalConfigs(boolean useLocalConfigs) {
@@ -316,6 +326,21 @@ public class LogFeederProps implements LogFeederProperties {
this.solrFilterMonitor = solrFilterMonitor;
}
+ public String getSolrUrlsStr() {
+ return this.solrUrlsStr;
+ }
+
+ public void setSolrUrlsStr(String solrUrlsStr) {
+ this.solrUrlsStr = solrUrlsStr;
+ }
+
+ public String[] getSolrUrls() {
+ if (StringUtils.isNotBlank(this.solrUrlsStr)) {
+ return this.solrUrlsStr.split(",");
+ }
+ return null;
+ }
+
@PostConstruct
public void init() {
properties = new Properties();
@@ -331,4 +356,5 @@ public class LogFeederProps implements LogFeederProperties {
throw new IllegalArgumentException("Cannot find logfeeder.properties on the classpath");
}
}
+
}
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputSolr.java b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputSolr.java
index 041c1bd..6b27553 100644
--- a/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputSolr.java
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/output/OutputSolr.java
@@ -19,11 +19,13 @@
package org.apache.ambari.logfeeder.output;
+import org.apache.ambari.logfeeder.common.LogFeederSolrClientFactory;
import org.apache.ambari.logfeeder.conf.LogFeederProps;
import org.apache.ambari.logfeeder.plugin.input.InputMarker;
import org.apache.ambari.logfeeder.plugin.output.Output;
import org.apache.ambari.logfeeder.util.DateUtil;
import org.apache.ambari.logfeeder.util.LogFeederUtil;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
@@ -76,6 +78,7 @@ public class OutputSolr extends Output<LogFeederProps, InputMarker> {
private String splitMode;
private int splitInterval;
private String zkConnectString;
+ private String[] solrUrls = null;
private int maxIntervalMS;
private int workers;
private int maxBufferSize;
@@ -121,8 +124,14 @@ public class OutputSolr extends Output<LogFeederProps, InputMarker> {
type = getStringValue("type");
zkConnectString = getStringValue("zk_connect_string");
- if (StringUtils.isEmpty(zkConnectString)) {
- throw new Exception("For solr output the zk_connect_string property need to be set");
+ List<String> solrUrlsList = getListValue("solr_urls");
+
+ if (StringUtils.isBlank(zkConnectString) && CollectionUtils.isEmpty(solrUrlsList)) {
+ throw new Exception("For solr output the zk_connect_string or solr_urls property need to be set");
+ }
+
+ if (CollectionUtils.isNotEmpty(solrUrlsList)) {
+ solrUrls = solrUrlsList.toArray(new String[0]);
}
skipLogtime = getBooleanValue("skip_logtime", DEFAULT_SKIP_LOGTIME);
@@ -176,42 +185,31 @@ public class OutputSolr extends Output<LogFeederProps, InputMarker> {
private void createSolrWorkers() throws Exception, MalformedURLException {
for (int count = 0; count < workers; count++) {
- CloudSolrClient solrClient = getSolrClient(count);
+ SolrClient solrClient = getSolrClient(count);
createSolrWorkerThread(count, solrClient);
}
}
- CloudSolrClient getSolrClient(int count) throws Exception, MalformedURLException {
- CloudSolrClient solrClient = createSolrClient();
+ private SolrClient getSolrClient(int count) throws Exception, MalformedURLException {
+ SolrClient solrClient = new LogFeederSolrClientFactory().createSolrClient(zkConnectString, solrUrls, collection);
pingSolr(count, solrClient);
-
- return solrClient;
- }
-
- private CloudSolrClient createSolrClient() throws Exception {
- LOG.info("Using zookeepr. zkConnectString=" + zkConnectString);
- LOG.info("Using collection=" + collection);
-
- CloudSolrClient solrClient = new CloudSolrClient.Builder().withZkHost(zkConnectString).build();
- solrClient.setDefaultCollection(collection);
return solrClient;
}
- private void pingSolr(int count, CloudSolrClient solrClient) {
+ private void pingSolr(int count, SolrClient solrClient) {
try {
- LOG.info("Pinging Solr server. zkConnectString=" + zkConnectString);
+ LOG.info("Pinging Solr server.");
SolrPingResponse response = solrClient.ping();
if (response.getStatus() == 0) {
LOG.info("Ping to Solr server is successful for worker=" + count);
} else {
LOG.warn(
- String.format("Ping to Solr server failed. It would check again. worker=%d, zkConnectString=%s, collection=%s, " +
- "response=%s", count, zkConnectString, collection, response));
+ String.format("Ping to Solr server failed. It would check again. worker=%d, collection=%s, " +
+ "response=%s", count, collection, response));
}
} catch (Throwable t) {
LOG.warn(String.format(
- "Ping to Solr server failed. It would check again. worker=%d, zkConnectString=%s, collection=%s", count,
- zkConnectString, collection), t);
+ "Ping to Solr server failed. It would check again. worker=%d, collection=%s", count, collection), t);
}
}
diff --git a/ambari-logsearch/docker/test-config/logfeeder/logfeeder.properties b/ambari-logsearch/docker/test-config/logfeeder/logfeeder.properties
index 20aed68..bd59765 100644
--- a/ambari-logsearch/docker/test-config/logfeeder/logfeeder.properties
+++ b/ambari-logsearch/docker/test-config/logfeeder/logfeeder.properties
@@ -21,7 +21,6 @@ logfeeder.config.files=shipper-conf/global.config.json,\
shipper-conf/output.config.json
logfeeder.log.filter.enable=true
logfeeder.solr.config.interval=5
-logfeeder.solr.core.config.name=history
logfeeder.solr.zk_connect_string=localhost:9983
logfeeder.cache.enabled=true
logfeeder.cache.size=100
@@ -31,5 +30,7 @@ logfeeder.cache.last.dedup.enabled=true
logsearch.config.zk_connect_string=localhost:9983
logfeeder.include.default.level=FATAL,ERROR,WARN,INFO,DEBUG,TRACE,UNKNOWN
logfeeder.docker.registry.enabled=true
+logfeeder.solr.core.config.name=history
+#logfeeder.solr.urls=http://solr:8983/solr
#logfeeder.configs.local.enabled=true
#logfeeder.configs.filter.solr.enabled=true
\ No newline at end of file