Repository: ambari Updated Branches: refs/heads/branch-2.5 613876528 -> 4b36286fd
AMBARI-19732. Allow all sinks a config override to point to a different ZK quorum. (swagle) Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/4b36286f Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/4b36286f Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/4b36286f Branch: refs/heads/branch-2.5 Commit: 4b36286fd84f5811e34d35f81c245c778e01a718 Parents: 6138765 Author: Siddharth Wagle <[email protected]> Authored: Thu Jan 26 14:00:18 2017 -0800 Committer: Siddharth Wagle <[email protected]> Committed: Thu Jan 26 14:00:24 2017 -0800 ---------------------------------------------------------------------- .../metrics2/sink/timeline/AbstractTimelineMetricsSink.java | 1 + .../metrics2/sink/kafka/KafkaTimelineMetricsReporter.java | 6 ++++-- .../metrics2/sink/storm/StormTimelineMetricsReporter.java | 9 ++++++++- .../metrics2/sink/storm/StormTimelineMetricsReporter.java | 6 +++++- .../metrics2/sink/storm/StormTimelineMetricsSink.java | 6 +++++- 5 files changed, 23 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/4b36286f/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java ---------------------------------------------------------------------- diff --git a/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java index 9bc3be5..f5a02e4 100644 --- a/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java +++ b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java @@ -70,6 +70,7 @@ public abstract class AbstractTimelineMetricsSink { public static final String COLLECTOR_PROTOCOL = "protocol"; public static final String COLLECTOR_PORT = "port"; public static final String ZOOKEEPER_QUORUM = "zookeeper.quorum"; + public static final String COLLECTOR_ZOOKEEPER_QUORUM = "metrics.zookeeper.quorum"; public static final int DEFAULT_POST_TIMEOUT_SECONDS = 10; public static final String SKIP_COUNTER_TRANSFROMATION = "skipCounterDerivative"; public static final String RPC_METRIC_PREFIX = "metric.rpc"; http://git-wip-us.apache.org/repos/asf/ambari/blob/4b36286f/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java ---------------------------------------------------------------------- diff --git a/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java index 5892599..b9ca9f5 100644 --- a/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java +++ b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java @@ -86,7 +86,7 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink private TimelineScheduledReporter reporter; private TimelineMetricsCache metricsCache; private int timeoutSeconds = 10; - private String zookeeperQuorum; + private String zookeeperQuorum = null; private String[] excludedMetricsPrefixes; private String[] includedMetricsPrefixes; @@ -155,7 +155,9 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink int metricsSendInterval = props.getInt(TIMELINE_METRICS_SEND_INTERVAL_PROPERTY, MAX_EVICTION_TIME_MILLIS); int maxRowCacheSize = props.getInt(TIMELINE_METRICS_MAX_ROW_CACHE_SIZE_PROPERTY, MAX_RECS_PER_NAME_DEFAULT); - zookeeperQuorum = props.getString("zookeeper.connect"); + zookeeperQuorum = props.containsKey(COLLECTOR_ZOOKEEPER_QUORUM) ? + props.getString(COLLECTOR_ZOOKEEPER_QUORUM) : props.getString("zookeeper.connect"); + metricCollectorPort = props.getString(TIMELINE_PORT_PROPERTY, TIMELINE_DEFAULT_PORT); collectorHosts = parseHostsStringIntoCollection(props.getString(TIMELINE_HOSTS_PROPERTY, TIMELINE_DEFAULT_HOST)); metricCollectorProtocol = props.getString(TIMELINE_PROTOCOL_PROPERTY, TIMELINE_DEFAULT_PROTOCOL); http://git-wip-us.apache.org/repos/asf/ambari/blob/4b36286f/ambari-metrics/ambari-metrics-storm-sink-legacy/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java ---------------------------------------------------------------------- diff --git a/ambari-metrics/ambari-metrics-storm-sink-legacy/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java b/ambari-metrics/ambari-metrics-storm-sink-legacy/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java index 95e5ba1..454b2c8 100644 --- a/ambari-metrics/ambari-metrics-storm-sink-legacy/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java +++ b/ambari-metrics/ambari-metrics-storm-sink-legacy/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java @@ -24,6 +24,8 @@ import backtype.storm.generated.TopologySummary; import backtype.storm.metric.IClusterReporter; import backtype.storm.utils.NimbusClient; import backtype.storm.utils.Utils; + +import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.Validate; import org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink; import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; @@ -114,7 +116,12 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink collectorHosts = parseHostsStringIntoCollection(cf.get(COLLECTOR_HOSTS_PROPERTY).toString()); protocol = cf.get(COLLECTOR_PROTOCOL) != null ? cf.get(COLLECTOR_PROTOCOL).toString() : "http"; port = cf.get(COLLECTOR_PORT) != null ? cf.get(COLLECTOR_PORT).toString() : "6188"; - zkQuorum = cf.get(ZOOKEEPER_QUORUM) != null ? cf.get(ZOOKEEPER_QUORUM).toString() : null; + Object zkQuorumObj = cf.get(COLLECTOR_ZOOKEEPER_QUORUM); + if (zkQuorumObj != null) { + zkQuorum = zkQuorumObj.toString(); + } else { + zkQuorum = cf.get(ZOOKEEPER_QUORUM) != null ? cf.get(ZOOKEEPER_QUORUM).toString() : null; + } if (collectorUri.toLowerCase().startsWith("https://";)) { String trustStorePath = cf.get(SSL_KEYSTORE_PATH_PROPERTY).toString().trim(); http://git-wip-us.apache.org/repos/asf/ambari/blob/4b36286f/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java ---------------------------------------------------------------------- diff --git a/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java b/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java index 9329aae..1de125b 100644 --- a/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java +++ b/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java @@ -18,6 +18,7 @@ package org.apache.hadoop.metrics2.sink.storm; +import org.apache.commons.lang.StringUtils; import org.apache.commons.lang3.ClassUtils; import org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink; import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric; @@ -81,6 +82,7 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink protected String getCollectorPort() { return port; } + @Override protected String getHostname() { return hostname; @@ -106,7 +108,9 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink collectorHosts = parseHostsStringIntoCollection(conf.getProperty(COLLECTOR_HOSTS_PROPERTY).toString()); port = conf.getProperty(COLLECTOR_PORT) != null ? conf.getProperty(COLLECTOR_PORT) : "6188"; protocol = conf.getProperty(COLLECTOR_PROTOCOL, "http"); - zkQuorum = conf.getProperty(ZOOKEEPER_QUORUM) != null ? conf.getProperty(ZOOKEEPER_QUORUM) : null; + + zkQuorum = StringUtils.isEmpty(conf.getProperty(COLLECTOR_ZOOKEEPER_QUORUM)) ? + conf.getProperty(ZOOKEEPER_QUORUM) : conf.getProperty(COLLECTOR_ZOOKEEPER_QUORUM); timeoutSeconds = conf.getProperty(METRICS_POST_TIMEOUT_SECONDS) != null ? Integer.parseInt(conf.getProperty(METRICS_POST_TIMEOUT_SECONDS).toString()) : http://git-wip-us.apache.org/repos/asf/ambari/blob/4b36286f/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsSink.java ---------------------------------------------------------------------- diff --git a/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsSink.java b/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsSink.java index 3c5655f..c70fff7 100644 --- a/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsSink.java +++ b/ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsSink.java @@ -18,6 +18,7 @@ package org.apache.hadoop.metrics2.sink.storm; +import org.apache.commons.lang.StringUtils; import org.apache.storm.Constants; import org.apache.storm.metric.api.IMetricsConsumer; import org.apache.storm.task.IErrorReporter; @@ -126,7 +127,10 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem applicationId = configuration.getProperty(CLUSTER_REPORTER_APP_ID, DEFAULT_CLUSTER_REPORTER_APP_ID); metricsCache = new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval); collectorHosts = parseHostsStringIntoCollection(configuration.getProperty(COLLECTOR_HOSTS_PROPERTY)); - zkQuorum = configuration.getProperty("zookeeper.quorum"); + + zkQuorum = StringUtils.isEmpty(configuration.getProperty(COLLECTOR_ZOOKEEPER_QUORUM)) ? + configuration.getProperty("zookeeper.quorum") : configuration.getProperty(COLLECTOR_ZOOKEEPER_QUORUM); + protocol = configuration.getProperty(COLLECTOR_PROTOCOL, "http"); port = configuration.getProperty(COLLECTOR_PORT, "6188");
