You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by sw...@apache.org on 2015/01/16 21:44:28 UTC
ambari git commit: AMBARI-9185. Add Kafka metric sink implementation
to enable sink to AMS.
Repository: ambari
Updated Branches:
refs/heads/trunk 15dbe3578 -> 8d27ac2b7
AMBARI-9185. Add Kafka metric sink implementation to enable sink to AMS.
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/8d27ac2b
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/8d27ac2b
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/8d27ac2b
Branch: refs/heads/trunk
Commit: 8d27ac2b77abdeeb61fd8d9bc6b51ce113fcd900
Parents: 15dbe35
Author: Siddharth Wagle <sw...@hortonworks.com>
Authored: Fri Jan 16 12:37:46 2015 -0800
Committer: Siddharth Wagle <sw...@hortonworks.com>
Committed: Fri Jan 16 12:37:46 2015 -0800
----------------------------------------------------------------------
ambari-metrics/ambari-metrics-assembly/pom.xml | 47 +-
.../src/main/assembly/sink-windows.xml | 8 +
.../src/main/assembly/sink.xml | 8 +
.../src/main/package/rpm/sink/postinstall.sh | 11 +-
.../ambari-metrics-kafka-sink/pom.xml | 163 +++++++
.../src/main/assemblies/empty.xml | 21 +
.../src/main/assemblies/jar-with-common.xml | 34 ++
.../kafka/KafkaTimelineMetricsReporter.java | 448 +++++++++++++++++++
.../KafkaTimelineMetricsReporterMBean.java | 25 ++
.../metrics2/sink/kafka/ScheduledReporter.java | 218 +++++++++
.../kafka/KafkaTimelineMetricsReporterTest.java | 109 +++++
.../sink/kafka/ScheduledReporterTest.java | 105 +++++
ambari-metrics/pom.xml | 7 +-
.../0.8.1.2.2/configuration/kafka-broker.xml | 31 +-
.../KAFKA/0.8.1.2.2/configuration/kafka-env.xml | 6 +
.../KAFKA/0.8.1.2.2/package/scripts/kafka.py | 3 +
.../KAFKA/0.8.1.2.2/package/scripts/params.py | 27 ++
17 files changed, 1259 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-assembly/pom.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-assembly/pom.xml b/ambari-metrics/ambari-metrics-assembly/pom.xml
index 1e1ca8c..60ba30b 100644
--- a/ambari-metrics/ambari-metrics-assembly/pom.xml
+++ b/ambari-metrics/ambari-metrics-assembly/pom.xml
@@ -36,6 +36,7 @@
<hadoop-sink.dir>${project.basedir}/../ambari-metrics-hadoop-sink</hadoop-sink.dir>
<storm-sink.dir>${project.basedir}/../ambari-metrics-storm-sink</storm-sink.dir>
<flume-sink.dir>${project.basedir}/../ambari-metrics-flume-sink</flume-sink.dir>
+ <kafka-sink.dir>${project.basedir}/../ambari-metrics-kafka-sink</kafka-sink.dir>
<python.ver>python >= 2.6</python.ver>
<deb.python.ver>python (>= 2.6)</deb.python.ver>
<deb.architecture>amd64</deb.architecture>
@@ -43,6 +44,7 @@
<hadoop.sink.jar>ambari-metrics-hadoop-sink-with-common-${project.version}.jar</hadoop.sink.jar>
<storm.sink.jar>ambari-metrics-storm-sink-with-common-${project.version}.jar</storm.sink.jar>
<flume.sink.jar>ambari-metrics-flume-sink-with-common-${project.version}.jar</flume.sink.jar>
+ <kafka.sink.jar>ambari-metrics-kafka-sink-with-common-${project.version}.jar</kafka.sink.jar>
</properties>
<build>
@@ -447,7 +449,6 @@
<location>${hadoop-sink.dir}/target/ambari-metrics-hadoop-sink-with-common-${project.version}.jar</location>
</source>
</sources>
-
</mapping>
<mapping>
<directory>/usr/lib/flume/lib</directory>
@@ -465,6 +466,22 @@
</source>
</sources>
</mapping>
+ <mapping>
+ <directory>/usr/lib/ambari-metrics-kafka-sink</directory>
+ <sources>
+ <source>
+ <location>${kafka-sink.dir}/target/${kafka.sink.jar}</location>
+ </source>
+ </sources>
+ </mapping>
+ <mapping>
+ <directory>/usr/lib/ambari-metrics-kafka-sink/lib</directory>
+ <sources>
+ <source>
+ <location>${kafka-sink.dir}/target/lib</location>
+ </source>
+ </sources>
+ </mapping>
</mappings>
</configuration>
@@ -595,6 +612,7 @@
<path>/var/log/ambari-metrics-collector</path>
<path>/var/lib/ambari-metrics-collector</path>
<path>/usr/lib/ambari-metrics-hadoop-sink</path>
+ <path>/usr/lib/ambari-metrics-kafka-sink</path>
<path>/usr/lib/flume/lib</path>
<path>/usr/lib/storm/lib</path>
</paths>
@@ -771,6 +789,28 @@
</mapper>
</data>
+ <!-- kafka sink -->
+
+ <data>
+ <src>${kafka-sink.dir}/target/${kafka.sink.jar}</src>
+ <type>file</type>
+ <mapper>
+ <type>perm</type>
+ <filemode>644</filemode>
+ <dirmode>755</dirmode>
+ <prefix>/usr/lib/ambari-metrics-kafka-sink</prefix>
+ </mapper>
+ </data>
+ <data>
+ <src>${kafka-sink.dir}/target/lib</src>
+ <type>file</type>
+ <mapper>
+ <type>perm</type>
+ <filemode>644</filemode>
+ <dirmode>755</dirmode>
+ <prefix>/usr/lib/ambari-metrics-kafka-sink/lib</prefix>
+ </mapper>
+ </data>
</dataSet>
</configuration>
</plugin>
@@ -1063,6 +1103,11 @@
</dependency>
<dependency>
<groupId>org.apache.ambari</groupId>
+ <artifactId>ambari-metrics-kafka-sink</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ambari</groupId>
<artifactId>ambari-metrics-host-monitoring</artifactId>
<version>${project.version}</version>
<type>pom</type>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink-windows.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink-windows.xml b/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink-windows.xml
index 0a36fac..38a8093 100644
--- a/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink-windows.xml
+++ b/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink-windows.xml
@@ -37,6 +37,10 @@
<directory>${storm-sink.dir}/src/main/conf</directory>
<outputDirectory>hadoop-sink/conf</outputDirectory>
</fileSet>
+ <fileSet>
+ <directory>${kafka-sink.dir}/target/lib</directory>
+ <outputDirectory>hadoop-sink/lib</outputDirectory>
+ </fileSet>
</fileSets>
<files>
@@ -53,6 +57,10 @@
<outputDirectory>hadoop-sink</outputDirectory>
</file>
<file>
+ <source>${kafka-sink.dir}/target/ambari-metrics-kafka-sink-with-common-${project.version}.jar</source>
+ <outputDirectory>hadoop-sink</outputDirectory>
+ </file>
+ <file>
<source>${basedir}/src/main/package/msi/sink.wxs</source>
<outputDirectory>../../</outputDirectory>
<filtered>true</filtered>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink.xml b/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink.xml
index 2426904..4a3b7c5 100644
--- a/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink.xml
+++ b/ambari-metrics/ambari-metrics-assembly/src/main/assembly/sink.xml
@@ -38,6 +38,10 @@
<directory>${storm-sink.dir}/src/main/conf</directory>
<outputDirectory>hadoop-sink/conf</outputDirectory>
</fileSet>
+ <fileSet>
+ <directory>${kafka-sink.dir}/target/lib</directory>
+ <outputDirectory>hadoop-sink/lib</outputDirectory>
+ </fileSet>
</fileSets>
<files>
@@ -53,6 +57,10 @@
<source>${storm-sink.dir}/target/ambari-metrics-storm-sink-with-common-${project.version}.jar</source>
<outputDirectory>hadoop-sink</outputDirectory>
</file>
+ <file>
+ <source>${kafka-sink.dir}/target/ambari-metrics-kafka-sink-with-common-${project.version}.jar</source>
+ <outputDirectory>hadoop-sink</outputDirectory>
+ </file>
</files>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-assembly/src/main/package/rpm/sink/postinstall.sh
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-assembly/src/main/package/rpm/sink/postinstall.sh b/ambari-metrics/ambari-metrics-assembly/src/main/package/rpm/sink/postinstall.sh
index 1955680..e87b9f0 100644
--- a/ambari-metrics/ambari-metrics-assembly/src/main/package/rpm/sink/postinstall.sh
+++ b/ambari-metrics/ambari-metrics-assembly/src/main/package/rpm/sink/postinstall.sh
@@ -15,15 +15,20 @@
# limitations under the License
HADOOP_LINK_NAME="/usr/lib/ambari-metrics-hadoop-sink/ambari-metrics-hadoop-sink.jar"
-FLUME_LINK_NAME="/usr/lib/flume/lib/ambari-metrics-flume-sink.jar"
HADOOP_SINK_JAR="/usr/lib/ambari-metrics-hadoop-sink/${hadoop.sink.jar}"
+
+FLUME_LINK_NAME="/usr/lib/flume/lib/ambari-metrics-flume-sink.jar"
FLUME_SINK_JAR="/usr/lib/flume/lib/${flume.sink.jar}"
+
+KAFKA_LINK_NAME="/usr/lib/ambari-metrics-kafka-sink/ambari-metrics-kafka-sink.jar"
+KAFKA_SINK_JAR="/usr/lib/ambari-metrics-kafka-sink/${kafka.sink.jar}"
+
#link for storm jar not required with current loading
#STORM_SINK_JAR="/usr/lib/storm/lib/${storm.sink.jar}"
#STORM_LINK_NAME="/usr/lib/storm/lib/ambari-metrics-storm-sink.jar"
-JARS=(${HADOOP_SINK_JAR} ${FLUME_SINK_JAR})
-LINKS=(${HADOOP_LINK_NAME} ${FLUME_LINK_NAME})
+JARS=(${HADOOP_SINK_JAR} ${FLUME_SINK_JAR} ${KAFKA_SINK_JAR})
+LINKS=(${HADOOP_LINK_NAME} ${FLUME_LINK_NAME} ${KAFKA_LINK_NAME})
for index in ${!LINKS[*]}
do
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-kafka-sink/pom.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-kafka-sink/pom.xml b/ambari-metrics/ambari-metrics-kafka-sink/pom.xml
new file mode 100644
index 0000000..55c6f07
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/pom.xml
@@ -0,0 +1,163 @@
+<?<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+ http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>ambari-metrics</artifactId>
+ <groupId>org.apache.ambari</groupId>
+ <version>0.1.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>ambari-metrics-kafka-sink</artifactId>
+ <version>0.1.0-SNAPSHOT</version>
+ <packaging>jar</packaging>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <version>2.9</version>
+ <executions>
+ <execution>
+ <id>copy-dependencies</id>
+ <phase>package</phase>
+ <goals>
+ <goal>copy-dependencies</goal>
+ </goals>
+ <configuration>
+ <includeArtifactIds>commons-codec,commons-collections,commons-httpclient,commons-lang,commons-logging,guava,jackson-core-asl,jackson-mapper-asl,jackson-xc,hadoop-common</includeArtifactIds>
+ <outputDirectory>${project.build.directory}/lib</outputDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <execution>
+ <configuration>
+ <descriptors>
+ <descriptor>src/main/assemblies/jar-with-common.xml</descriptor>
+ </descriptors>
+ <attach>false</attach>
+ <tarLongFileMode>gnu</tarLongFileMode>
+ <appendAssemblyId>false</appendAssemblyId>
+ <finalName>${project.artifactId}-with-common-${project.version}</finalName>
+ </configuration>
+ <id>build-jar</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.0</version>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>1.8</version>
+ <executions>
+ <execution>
+ <id>parse-version</id>
+ <phase>validate</phase>
+ <goals>
+ <goal>parse-version</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>regex-property</id>
+ <goals>
+ <goal>regex-property</goal>
+ </goals>
+ <configuration>
+ <name>ambariVersion</name>
+ <value>${project.version}</value>
+ <regex>^([0-9]+)\.([0-9]+)\.([0-9]+)(\.|-).*</regex>
+ <replacement>$1.$2.$3</replacement>
+ <failIfNoMatch>false</failIfNoMatch>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>com.github.goldin</groupId>
+ <artifactId>copy-maven-plugin</artifactId>
+ <version>0.2.5</version>
+ <executions>
+ <execution>
+ <id>create-archive</id>
+ <phase>none</phase>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.ambari</groupId>
+ <artifactId>ambari-metrics-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_2.10</artifactId>
+ <version>0.8.1.1</version>
+ </dependency>
+ <dependency>
+ <groupId>com.yammer.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>2.2.0</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ <version>4.10</version>
+ </dependency>
+ <dependency>
+ <groupId>org.easymock</groupId>
+ <artifactId>easymock</artifactId>
+ <version>3.2</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-api-easymock</artifactId>
+ <version>1.4.9</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-module-junit4</artifactId>
+ <version>1.4.9</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>1.9.5</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/empty.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/empty.xml b/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/empty.xml
new file mode 100644
index 0000000..17ff68a
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/empty.xml
@@ -0,0 +1,21 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<assembly>
+ <id>empty</id>
+ <formats/>
+</assembly>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/jar-with-common.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/jar-with-common.xml b/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/jar-with-common.xml
new file mode 100644
index 0000000..4fa8585
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/src/main/assemblies/jar-with-common.xml
@@ -0,0 +1,34 @@
+<?xml version="1.0"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<assembly>
+ <id>jar-with-common</id>
+ <formats>
+ <format>jar</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <dependencySets>
+ <dependencySet>
+ <outputDirectory>/</outputDirectory>
+ <unpack>true</unpack>
+ <includes>
+ <include>org.apache.ambari:ambari-metrics-common</include>
+ <include>org.apache.ambari:ambari-metrics-kafka-sink</include>
+ </includes>
+ </dependencySet>
+ </dependencySets>
+</assembly>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/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
new file mode 100644
index 0000000..762b5f2
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java
@@ -0,0 +1,448 @@
+/**
+ * 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.hadoop.metrics2.sink.kafka;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import kafka.metrics.KafkaMetricsConfig;
+import kafka.metrics.KafkaMetricsReporter;
+import kafka.utils.VerifiableProperties;
+
+import org.apache.commons.lang.ClassUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
+import org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache;
+import org.apache.hadoop.metrics2.util.Servers;
+
+import com.yammer.metrics.Metrics;
+import com.yammer.metrics.core.Counter;
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Metered;
+import com.yammer.metrics.core.Metric;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.MetricProcessor;
+import com.yammer.metrics.core.MetricsRegistry;
+import com.yammer.metrics.core.Timer;
+import com.yammer.metrics.stats.Snapshot;
+
+public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink implements KafkaMetricsReporter,
+ KafkaTimelineMetricsReporterMBean {
+
+ private final static Log LOG = LogFactory.getLog(KafkaTimelineMetricsReporter.class);
+
+ private static final String TIMELINE_METRICS_SEND_INTERVAL_PROPERTY = "kafka.timeline.metrics.sendInterval";
+ private static final String TIMELINE_METRICS_MAX_ROW_CACHE_SIZE_PROPERTY = "kafka.timeline.metrics.maxRowCacheSize";
+ private static final String TIMELINE_HOST_PROPERTY = "kafka.timeline.metrics.host";
+ private static final String TIMELINE_PORT_PROPERTY = "kafka.timeline.metrics.port";
+ private static final String TIMELINE_REPORTER_ENABLED_PROPERTY = "kafka.timeline.metrics.reporter.enabled";
+ private static final String TIMELINE_DEFAULT_HOST = "localhost";
+ private static final String TIMELINE_DEFAULT_PORT = "8188";
+
+ private boolean initialized = false;
+ private boolean running = false;
+ private Object lock = new Object();
+ private String collectorUri;
+ private String hostname;
+ private SocketAddress socketAddress;
+ private TimelineScheduledReporter reporter;
+ private TimelineMetricsCache metricsCache;
+
+ @Override
+ protected SocketAddress getServerSocketAddress() {
+ return socketAddress;
+ }
+
+ @Override
+ protected String getCollectorUri() {
+ return collectorUri;
+ }
+
+ public void setMetricsCache(TimelineMetricsCache metricsCache) {
+ this.metricsCache = metricsCache;
+ }
+
+ public void init(VerifiableProperties props) {
+ synchronized (lock) {
+ if (!initialized) {
+ LOG.info("Initializing Kafka Timeline Metrics Sink");
+ try {
+ hostname = InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ LOG.error("Could not identify hostname.");
+ throw new RuntimeException("Could not identify hostname.", e);
+ }
+ KafkaMetricsConfig metricsConfig = new KafkaMetricsConfig(props);
+ int metricsSendInterval = Integer.parseInt(props.getString(TIMELINE_METRICS_SEND_INTERVAL_PROPERTY,
+ String.valueOf(TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS)));
+ int maxRowCacheSize = Integer.parseInt(props.getString(TIMELINE_METRICS_MAX_ROW_CACHE_SIZE_PROPERTY,
+ String.valueOf(TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT)));
+ String metricCollectorHost = props.getString(TIMELINE_HOST_PROPERTY, TIMELINE_DEFAULT_HOST);
+ String metricCollectorPort = props.getString(TIMELINE_PORT_PROPERTY, TIMELINE_DEFAULT_PORT);
+ setMetricsCache(new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval));
+ collectorUri = "http://" + metricCollectorHost + ":" + metricCollectorPort + "/ws/v1/timeline/metrics";
+ List<InetSocketAddress> socketAddresses = Servers.parse(metricCollectorHost,
+ Integer.parseInt(metricCollectorPort));
+ if (socketAddresses != null && !socketAddresses.isEmpty()) {
+ socketAddress = socketAddresses.get(0);
+ }
+ initializeReporter();
+ if (props.getBoolean(TIMELINE_REPORTER_ENABLED_PROPERTY, false)) {
+ startReporter(metricsConfig.pollingIntervalSecs());
+ }
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("CollectorUri = " + collectorUri);
+ LOG.trace("SocketAddress = " + socketAddress);
+ LOG.trace("MetricsSendInterval = " + metricsSendInterval);
+ LOG.trace("MaxRowCacheSize = " + maxRowCacheSize);
+ }
+ }
+ }
+ }
+
+ public String getMBeanName() {
+ return "kafka:type=org.apache.hadoop.metrics2.sink.kafka.KafkaTimelineMetricsReporter";
+ }
+
+ public synchronized void startReporter(long period) {
+ synchronized (lock) {
+ if (initialized && !running) {
+ reporter.start(period, TimeUnit.SECONDS);
+ running = true;
+ LOG.info(String.format("Started Kafka Timeline metrics reporter with polling period %d seconds", period));
+ }
+ }
+ }
+
+ public synchronized void stopReporter() {
+ synchronized (lock) {
+ if (initialized && running) {
+ reporter.stop();
+ running = false;
+ LOG.info("Stopped Kafka Timeline metrics reporter");
+ initializeReporter();
+ }
+ }
+ }
+
+ private void initializeReporter() {
+ reporter = new TimelineScheduledReporter(Metrics.defaultRegistry(), "timeline-scheduled-reporter",
+ TimeUnit.SECONDS, TimeUnit.MILLISECONDS);
+ initialized = true;
+ }
+
+ interface Context {
+ public List<TimelineMetric> getTimelineMetricList();
+ }
+
+ class TimelineScheduledReporter extends ScheduledReporter implements MetricProcessor<Context> {
+
+ private static final String APP_ID = "kafka_broker";
+ private static final String COUNT_SUFIX = ".count";
+ private static final String ONE_MINUTE_RATE_SUFIX = ".1MinuteRate";
+ private static final String MEAN_SUFIX = ".mean";
+ private static final String MEAN_RATE_SUFIX = ".meanRate";
+ private static final String FIVE_MINUTE_RATE_SUFIX = ".5MinuteRate";
+ private static final String FIFTEEN_MINUTE_RATE_SUFIX = ".15MinuteRate";
+ private static final String MIN_SUFIX = ".min";
+ private static final String MAX_SUFIX = ".max";
+ private static final String MEDIAN_SUFIX = ".median";
+ private static final String STD_DEV_SUFIX = "stddev";
+ private static final String SEVENTY_FIFTH_PERCENTILE_SUFIX = ".75percentile";
+ private static final String NINETY_FIFTH_PERCENTILE_SUFIX = ".95percentile";
+ private static final String NINETY_EIGHTH_PERCENTILE_SUFIX = ".98percentile";
+ private static final String NINETY_NINTH_PERCENTILE_SUFIX = ".99percentile";
+ private static final String NINETY_NINE_POINT_NINE_PERCENTILE_SUFIX = ".999percentile";
+
+ protected TimelineScheduledReporter(MetricsRegistry registry, String name, TimeUnit rateUnit, TimeUnit durationUnit) {
+ super(registry, name, rateUnit, durationUnit);
+ }
+
+ @Override
+ public void report(Set<Entry<MetricName, Metric>> metrics) {
+ final List<TimelineMetric> metricsList = new ArrayList<TimelineMetric>();
+ try {
+ for (Entry<MetricName, Metric> entry : metrics) {
+ final MetricName metricName = entry.getKey();
+ final Metric metric = entry.getValue();
+ Context context = new Context() {
+
+ public List<TimelineMetric> getTimelineMetricList() {
+ return metricsList;
+ }
+
+ };
+ metric.processWith(this, metricName, context);
+ }
+ } catch (Throwable t) {
+ LOG.error("Exception processing Kafka metric", t);
+ }
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Metrics List size: " + metricsList.size());
+ LOG.trace("Metics Set size: " + metrics.size());
+ }
+ if (!metricsList.isEmpty()) {
+ TimelineMetrics timelineMetrics = new TimelineMetrics();
+ timelineMetrics.setMetrics(metricsList);
+ try {
+ emitMetrics(timelineMetrics);
+ } catch (IOException e) {
+ LOG.error("Unexpected error", e);
+ } catch (Throwable t) {
+ LOG.error("Exception emitting metrics", t);
+ }
+ }
+ }
+
+ private TimelineMetric createTimelineMetric(long currentTimeMillis, String component, String attributeName,
+ Number attributeValue) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Creating timeline metric: " + attributeName + " = " + attributeValue + " time = "
+ + currentTimeMillis + " app_id = " + component);
+ }
+ TimelineMetric timelineMetric = new TimelineMetric();
+ timelineMetric.setMetricName(attributeName);
+ timelineMetric.setHostName(hostname);
+ timelineMetric.setAppId(component);
+ timelineMetric.setStartTime(currentTimeMillis);
+ timelineMetric.setType(ClassUtils.getShortCanonicalName(attributeValue, "Number"));
+ timelineMetric.getMetricValues().put(currentTimeMillis, attributeValue.doubleValue());
+ return timelineMetric;
+ }
+
+ @Override
+ public void processMeter(MetricName name, Metered meter, Context context) throws Exception {
+ final long currentTimeMillis = System.currentTimeMillis();
+ final String sanitizedName = sanitizeName(name);
+ final String meterCountName = sanitizedName + COUNT_SUFIX;
+ final TimelineMetric countMetric = createTimelineMetric(currentTimeMillis, APP_ID, meterCountName, meter.count());
+
+ final String meterOneMinuteRateName = sanitizedName + ONE_MINUTE_RATE_SUFIX;
+ final TimelineMetric oneMinuteRateMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ meterOneMinuteRateName, meter.oneMinuteRate());
+
+ final String meterMeanRateName = sanitizedName + MEAN_RATE_SUFIX;
+ final TimelineMetric meanMetric = createTimelineMetric(currentTimeMillis, APP_ID, meterMeanRateName,
+ meter.meanRate());
+
+ final String meterFiveMinuteRateName = sanitizedName + FIVE_MINUTE_RATE_SUFIX;
+ final TimelineMetric fiveMinuteRateMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ meterFiveMinuteRateName, meter.fiveMinuteRate());
+
+ final String meterFifteenMinuteRateName = sanitizedName + FIFTEEN_MINUTE_RATE_SUFIX;
+ final TimelineMetric fifteenMinuteRateMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ meterFifteenMinuteRateName, meter.fifteenMinuteRate());
+
+ metricsCache.putTimelineMetric(countMetric);
+ metricsCache.putTimelineMetric(oneMinuteRateMetric);
+ metricsCache.putTimelineMetric(meanMetric);
+ metricsCache.putTimelineMetric(fiveMinuteRateMetric);
+ metricsCache.putTimelineMetric(fifteenMinuteRateMetric);
+
+ String[] metricNames = new String[] { meterCountName, meterOneMinuteRateName, meterMeanRateName,
+ meterFiveMinuteRateName, meterFifteenMinuteRateName };
+ populateMetricsList(context, metricNames);
+ }
+
+ @Override
+ public void processCounter(MetricName name, Counter counter, Context context) throws Exception {
+ final long currentTimeMillis = System.currentTimeMillis();
+ final String sanitizedName = sanitizeName(name);
+ final String metricCountName = sanitizedName + COUNT_SUFIX;
+ final TimelineMetric metric = createTimelineMetric(currentTimeMillis, APP_ID, metricCountName, counter.count());
+ metricsCache.putTimelineMetric(metric);
+ populateMetricsList(context, metricCountName);
+ }
+
+ @Override
+ public void processHistogram(MetricName name, Histogram histogram, Context context) throws Exception {
+ final long currentTimeMillis = System.currentTimeMillis();
+ final Snapshot snapshot = histogram.getSnapshot();
+ final String sanitizedName = sanitizeName(name);
+
+ final String histogramMinName = sanitizedName + MIN_SUFIX;
+ final TimelineMetric minMetric = createTimelineMetric(currentTimeMillis, APP_ID, histogramMinName,
+ histogram.min());
+
+ final String histogramMaxName = sanitizedName + MAX_SUFIX;
+ final TimelineMetric maxMetric = createTimelineMetric(currentTimeMillis, APP_ID, histogramMaxName,
+ histogram.max());
+
+ final String histogramMeanName = sanitizedName + MEAN_SUFIX;
+ final TimelineMetric meanMetric = createTimelineMetric(currentTimeMillis, APP_ID, histogramMeanName,
+ histogram.mean());
+
+ final String histogramMedianName = sanitizedName + MEDIAN_SUFIX;
+ final TimelineMetric medianMetric = createTimelineMetric(currentTimeMillis, APP_ID, histogramMedianName,
+ snapshot.getMedian());
+
+ final String histogramStdDevName = sanitizedName + STD_DEV_SUFIX;
+ final TimelineMetric stdDevMetric = createTimelineMetric(currentTimeMillis, APP_ID, histogramStdDevName,
+ histogram.stdDev());
+
+ final String histogramSeventyFifthPercentileName = sanitizedName + SEVENTY_FIFTH_PERCENTILE_SUFIX;
+ final TimelineMetric seventyFifthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ histogramSeventyFifthPercentileName, snapshot.get75thPercentile());
+
+ final String histogramNinetyFifthPercentileName = sanitizedName + NINETY_FIFTH_PERCENTILE_SUFIX;
+ final TimelineMetric nintyFifthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ histogramNinetyFifthPercentileName, snapshot.get95thPercentile());
+
+ final String histogramNinetyEighthPercentileName = sanitizedName + NINETY_EIGHTH_PERCENTILE_SUFIX;
+ final TimelineMetric nintyEighthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ histogramNinetyEighthPercentileName, snapshot.get98thPercentile());
+
+ final String histogramNinetyNinethPercentileName = sanitizedName + NINETY_NINTH_PERCENTILE_SUFIX;
+ final TimelineMetric nintyNinthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ histogramNinetyNinethPercentileName, snapshot.get99thPercentile());
+
+ final String histogramNinetyNinePointNinePercentileName = sanitizedName + NINETY_NINE_POINT_NINE_PERCENTILE_SUFIX;
+ final TimelineMetric nintyNinthPointNinePercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ histogramNinetyNinePointNinePercentileName, snapshot.get999thPercentile());
+
+ metricsCache.putTimelineMetric(minMetric);
+ metricsCache.putTimelineMetric(maxMetric);
+ metricsCache.putTimelineMetric(meanMetric);
+ metricsCache.putTimelineMetric(medianMetric);
+ metricsCache.putTimelineMetric(stdDevMetric);
+ metricsCache.putTimelineMetric(seventyFifthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyFifthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyEighthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyNinthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyNinthPointNinePercentileMetric);
+
+ String[] metricNames = new String[] { histogramMaxName, histogramMeanName, histogramMedianName, histogramMinName,
+ histogramNinetyEighthPercentileName, histogramNinetyFifthPercentileName,
+ histogramNinetyNinePointNinePercentileName, histogramNinetyNinethPercentileName,
+ histogramSeventyFifthPercentileName, histogramStdDevName };
+ populateMetricsList(context, metricNames);
+ }
+
+ @Override
+ public void processTimer(MetricName name, Timer timer, Context context) throws Exception {
+ final long currentTimeMillis = System.currentTimeMillis();
+ final Snapshot snapshot = timer.getSnapshot();
+ final String sanitizedName = sanitizeName(name);
+
+ final String timerMinName = sanitizedName + MIN_SUFIX;
+ final TimelineMetric minMetric = createTimelineMetric(currentTimeMillis, APP_ID, timerMinName, timer.min());
+
+ final String timerMaxName = sanitizedName + MAX_SUFIX;
+ final TimelineMetric maxMetric = createTimelineMetric(currentTimeMillis, APP_ID, timerMaxName, timer.max());
+
+ final String timerMeanName = sanitizedName + MEAN_SUFIX;
+ final TimelineMetric meanMetric = createTimelineMetric(currentTimeMillis, APP_ID, timerMeanName, timer.mean());
+
+ final String timerMedianName = sanitizedName + MEDIAN_SUFIX;
+ final TimelineMetric medianMetric = createTimelineMetric(currentTimeMillis, APP_ID, timerMedianName,
+ snapshot.getMedian());
+
+ final String timerStdDevName = sanitizedName + STD_DEV_SUFIX;
+ final TimelineMetric stdDevMetric = createTimelineMetric(currentTimeMillis, APP_ID, timerStdDevName,
+ timer.stdDev());
+
+ final String timerSeventyFifthPercentileName = sanitizedName + SEVENTY_FIFTH_PERCENTILE_SUFIX;
+ final TimelineMetric seventyFifthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ timerSeventyFifthPercentileName, snapshot.get75thPercentile());
+
+ final String timerNinetyFifthPercentileName = sanitizedName + NINETY_FIFTH_PERCENTILE_SUFIX;
+ final TimelineMetric nintyFifthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ timerNinetyFifthPercentileName, snapshot.get95thPercentile());
+
+ final String timerNinetyEighthPercentileName = sanitizedName + NINETY_EIGHTH_PERCENTILE_SUFIX;
+ final TimelineMetric nintyEighthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ timerNinetyEighthPercentileName, snapshot.get98thPercentile());
+
+ final String timerNinetyNinthPercentileName = sanitizedName + NINETY_NINTH_PERCENTILE_SUFIX;
+ final TimelineMetric nintyNinthPercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ timerNinetyNinthPercentileName, snapshot.get99thPercentile());
+
+ final String timerNinetyNinePointNinePercentileName = sanitizedName + NINETY_NINE_POINT_NINE_PERCENTILE_SUFIX;
+ final TimelineMetric nintyNinthPointNinePercentileMetric = createTimelineMetric(currentTimeMillis, APP_ID,
+ timerNinetyNinePointNinePercentileName, snapshot.get999thPercentile());
+
+ metricsCache.putTimelineMetric(minMetric);
+ metricsCache.putTimelineMetric(maxMetric);
+ metricsCache.putTimelineMetric(meanMetric);
+ metricsCache.putTimelineMetric(medianMetric);
+ metricsCache.putTimelineMetric(stdDevMetric);
+ metricsCache.putTimelineMetric(seventyFifthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyFifthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyEighthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyNinthPercentileMetric);
+ metricsCache.putTimelineMetric(nintyNinthPointNinePercentileMetric);
+
+ String[] metricNames = new String[] { timerMaxName, timerMeanName, timerMedianName, timerMinName,
+ timerNinetyEighthPercentileName, timerNinetyFifthPercentileName, timerNinetyNinePointNinePercentileName,
+ timerNinetyNinthPercentileName, timerSeventyFifthPercentileName, timerStdDevName };
+ populateMetricsList(context, metricNames);
+ }
+
+ @Override
+ public void processGauge(MetricName name, Gauge<?> gauge, Context context) throws Exception {
+ final long currentTimeMillis = System.currentTimeMillis();
+ final String sanitizedName = sanitizeName(name);
+ final TimelineMetric metric = createTimelineMetric(currentTimeMillis, APP_ID, sanitizedName,
+ Double.parseDouble(String.valueOf(gauge.value())));
+ metricsCache.putTimelineMetric(metric);
+ populateMetricsList(context, sanitizedName);
+ }
+
+ private void populateMetricsList(Context context, String... metricNames) {
+ for (String metricName : metricNames) {
+ TimelineMetric cachedMetric = metricsCache.getTimelineMetric(metricName);
+ if (cachedMetric != null) {
+ context.getTimelineMetricList().add(cachedMetric);
+ }
+ }
+ }
+
+ protected String sanitizeName(MetricName name) {
+ if (name == null) {
+ return "";
+ }
+ final String qualifiedTypeName = name.getGroup() + "." + name.getType() + "." + name.getName();
+ final String metricName = name.hasScope() ? qualifiedTypeName + '.' + name.getScope() : qualifiedTypeName;
+ final StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < metricName.length(); i++) {
+ final char p = metricName.charAt(i);
+ if (!(p >= 'A' && p <= 'Z') && !(p >= 'a' && p <= 'z') && !(p >= '0' && p <= '9') && (p != '_') && (p != '-')
+ && (p != '.') && (p != '\0')) {
+ sb.append('_');
+ } else {
+ sb.append(p);
+ }
+ }
+ return sb.toString();
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterMBean.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterMBean.java b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterMBean.java
new file mode 100644
index 0000000..7f6c5c9
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterMBean.java
@@ -0,0 +1,25 @@
+/**
+ * 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.hadoop.metrics2.sink.kafka;
+
+import kafka.metrics.KafkaMetricsReporterMBean;
+
+public interface KafkaTimelineMetricsReporterMBean extends KafkaMetricsReporterMBean {
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporter.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporter.java b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporter.java
new file mode 100644
index 0000000..f4f8333
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporter.java
@@ -0,0 +1,218 @@
+/**
+ * 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.hadoop.metrics2.sink.kafka;
+
+import java.io.Closeable;
+import java.util.Locale;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.yammer.metrics.core.Metric;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.MetricsRegistry;
+import com.yammer.metrics.reporting.ConsoleReporter;
+import com.yammer.metrics.reporting.CsvReporter;
+
+/**
+ * The abstract base class for all scheduled reporters (i.e., reporters which
+ * process a registry's metrics periodically).
+ *
+ * @see ConsoleReporter
+ * @see CsvReporter
+ * @see Slf4jReporter
+ */
+public abstract class ScheduledReporter implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ScheduledReporter.class);
+
+ /**
+ * A simple named thread factory.
+ */
+ @SuppressWarnings("NullableProblems")
+ private static class NamedThreadFactory implements ThreadFactory {
+ private final ThreadGroup group;
+ private final AtomicInteger threadNumber = new AtomicInteger(1);
+ private final String namePrefix;
+
+ private NamedThreadFactory(String name) {
+ final SecurityManager s = System.getSecurityManager();
+ this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
+ this.namePrefix = "metrics-" + name + "-thread-";
+ }
+
+ @Override
+ public Thread newThread(Runnable r) {
+ final Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
+ t.setDaemon(true);
+ if (t.getPriority() != Thread.NORM_PRIORITY) {
+ t.setPriority(Thread.NORM_PRIORITY);
+ }
+ return t;
+ }
+ }
+
+ private static final AtomicInteger FACTORY_ID = new AtomicInteger();
+
+ private final MetricsRegistry registry;
+ private final ScheduledExecutorService executor;
+ private final double durationFactor;
+ private final String durationUnit;
+ private final double rateFactor;
+ private final String rateUnit;
+
+ /**
+ * Creates a new {@link ScheduledReporter} instance.
+ *
+ * @param registry
+ * the {@link com.codahale.metrics.MetricRegistry} containing the
+ * metrics this reporter will report
+ * @param name
+ * the reporter's name
+ * @param rateUnit
+ * a unit of time
+ * @param durationUnit
+ * a unit of time
+ */
+ protected ScheduledReporter(MetricsRegistry registry, String name, TimeUnit rateUnit, TimeUnit durationUnit) {
+ this(registry, rateUnit, durationUnit, Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory(name + '-'
+ + FACTORY_ID.incrementAndGet())));
+ }
+
+ /**
+ * Creates a new {@link ScheduledReporter} instance.
+ *
+ * @param registry
+ * the {@link com.codahale.metrics.MetricRegistry} containing the
+ * metrics this reporter will report
+ * @param executor
+ * the executor to use while scheduling reporting of metrics.
+ */
+ protected ScheduledReporter(MetricsRegistry registry, TimeUnit rateUnit, TimeUnit durationUnit,
+ ScheduledExecutorService executor) {
+ this.registry = registry;
+ this.executor = executor;
+ this.rateFactor = rateUnit.toSeconds(1);
+ this.rateUnit = calculateRateUnit(rateUnit);
+ this.durationFactor = 1.0 / durationUnit.toNanos(1);
+ this.durationUnit = durationUnit.toString().toLowerCase(Locale.US);
+ }
+
+ /**
+ * Starts the reporter polling at the given period.
+ *
+ * @param period
+ * the amount of time between polls
+ * @param unit
+ * the unit for {@code period}
+ */
+ public void start(long period, TimeUnit unit) {
+ executor.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ report();
+ } catch (RuntimeException ex) {
+ LOG.error("RuntimeException thrown from {}#report. Exception was suppressed.", ScheduledReporter.this
+ .getClass().getSimpleName(), ex);
+ }
+ }
+ }, period, period, unit);
+ }
+
+ /**
+ * Stops the reporter and shuts down its thread of execution.
+ *
+ * Uses the shutdown pattern from
+ * http://docs.oracle.com/javase/7/docs/api/java
+ * /util/concurrent/ExecutorService.html
+ */
+ public void stop() {
+ executor.shutdown(); // Disable new tasks from being submitted
+ try {
+ // Wait a while for existing tasks to terminate
+ if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
+ executor.shutdownNow(); // Cancel currently executing tasks
+ // Wait a while for tasks to respond to being cancelled
+ if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
+ System.err.println(getClass().getSimpleName() + ": ScheduledExecutorService did not terminate");
+ }
+ }
+ } catch (InterruptedException ie) {
+ // (Re-)Cancel if current thread also interrupted
+ executor.shutdownNow();
+ // Preserve interrupt status
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Stops the reporter and shuts down its thread of execution.
+ */
+ @Override
+ public void close() {
+ stop();
+ }
+
+ /**
+ * Report the current values of all metrics in the registry.
+ */
+ public void report() {
+ synchronized (this) {
+ report(registry.allMetrics().entrySet());
+ }
+ }
+
+ /**
+ * Called periodically by the polling thread. Subclasses should report all the
+ * given metrics.
+ *
+ * @param metrics
+ * all the metrics in the registry
+ */
+ public abstract void report(Set<Entry<MetricName, Metric>> metrics);
+
+ protected String getRateUnit() {
+ return rateUnit;
+ }
+
+ protected String getDurationUnit() {
+ return durationUnit;
+ }
+
+ protected double convertDuration(double duration) {
+ return duration * durationFactor;
+ }
+
+ protected double convertRate(double rate) {
+ return rate * rateFactor;
+ }
+
+ private String calculateRateUnit(TimeUnit unit) {
+ final String s = unit.toString().toLowerCase(Locale.US);
+ return s.substring(0, s.length() - 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterTest.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterTest.java b/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterTest.java
new file mode 100644
index 0000000..67c61e1
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporterTest.java
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.metrics2.sink.kafka;
+
+import static org.mockito.Mockito.mock;
+import static org.powermock.api.easymock.PowerMock.mockStatic;
+import static org.powermock.api.easymock.PowerMock.replay;
+import static org.powermock.api.easymock.PowerMock.verifyAll;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import kafka.utils.VerifiableProperties;
+
+import org.apache.commons.httpclient.HttpClient;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache;
+import org.easymock.EasyMock;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import com.yammer.metrics.Metrics;
+import com.yammer.metrics.core.Counter;
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.Metric;
+import com.yammer.metrics.core.MetricsRegistry;
+import com.yammer.metrics.core.Timer;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ Metrics.class, HttpClient.class })
+@PowerMockIgnore("javax.management.*")
+public class KafkaTimelineMetricsReporterTest {
+
+ private final List<Metric> list = new ArrayList<Metric>();
+ private final MetricsRegistry registry = new MetricsRegistry();
+ @SuppressWarnings("rawtypes")
+ private final Gauge gauge = mock(Gauge.class);
+ private final KafkaTimelineMetricsReporter kafkaTimelineMetricsReporter = new KafkaTimelineMetricsReporter();
+ private VerifiableProperties props;
+
+ @Before
+ public void setUp() throws Exception {
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ Gauge g = registry.newGauge(System.class, "gauge", gauge);
+ Counter counter = registry.newCounter(System.class, "counter");
+ Histogram histogram = registry.newHistogram(System.class, "histogram");
+ Meter meter = registry.newMeter(System.class, "meter", "empty", TimeUnit.MILLISECONDS);
+ Timer timer = registry.newTimer(System.class, "timer");
+ list.add(g);
+ list.add(counter);
+ list.add(histogram);
+ list.add(meter);
+ list.add(timer);
+ Properties properties = new Properties();
+ properties.setProperty("kafka.timeline.metrics.sendInterval", "5900");
+ properties.setProperty("kafka.timeline.metrics.maxRowCacheSize", "10000");
+ properties.setProperty("kafka.timeline.metrics.host", "localhost");
+ properties.setProperty("kafka.timeline.metrics.port", "8188");
+ properties.setProperty("kafka.timeline.metrics.reporter.enabled", "true");
+ props = new VerifiableProperties(properties);
+ }
+
+ @Test
+ public void testReporterStartStop() {
+ mockStatic(Metrics.class);
+ EasyMock.expect(Metrics.defaultRegistry()).andReturn(registry).times(2);
+ TimelineMetricsCache timelineMetricsCache = getTimelineMetricsCache(kafkaTimelineMetricsReporter);
+ kafkaTimelineMetricsReporter.setMetricsCache(timelineMetricsCache);
+ HttpClient httpClient = EasyMock.createNiceMock(HttpClient.class);
+ kafkaTimelineMetricsReporter.setHttpClient(httpClient);
+ replay(Metrics.class, httpClient, timelineMetricsCache);
+ kafkaTimelineMetricsReporter.init(props);
+ kafkaTimelineMetricsReporter.stopReporter();
+ verifyAll();
+ }
+
+ private TimelineMetricsCache getTimelineMetricsCache(KafkaTimelineMetricsReporter kafkaTimelineMetricsReporter) {
+ TimelineMetricsCache timelineMetricsCache = EasyMock.createNiceMock(TimelineMetricsCache.class);
+ kafkaTimelineMetricsReporter.setMetricsCache(timelineMetricsCache);
+ EasyMock.expect(timelineMetricsCache.getTimelineMetric("key1")).andReturn(new TimelineMetric()).once();
+ timelineMetricsCache.putTimelineMetric(EasyMock.anyObject(TimelineMetric.class));
+ EasyMock.expectLastCall().once();
+ return timelineMetricsCache;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporterTest.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporterTest.java b/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporterTest.java
new file mode 100644
index 0000000..41f9126
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-kafka-sink/src/test/java/org/apache/hadoop/metrics2/sink/kafka/ScheduledReporterTest.java
@@ -0,0 +1,105 @@
+/**
+ * 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.hadoop.metrics2.sink.kafka;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.yammer.metrics.core.Counter;
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.Metric;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.MetricsRegistry;
+import com.yammer.metrics.core.Timer;
+
+public class ScheduledReporterTest {
+ private final Gauge gauge = mock(Gauge.class);
+ private final List<Metric> list = new ArrayList<Metric>();
+ private final MetricsRegistry registry = new MetricsRegistry();
+ private final ScheduledReporter reporter = spy(new ScheduledReporter(registry, "example", TimeUnit.SECONDS,
+ TimeUnit.MILLISECONDS) {
+ @Override
+ public void report(Set<Entry<MetricName, Metric>> metrics) {
+ // nothing doing!
+ }
+ });
+
+ @SuppressWarnings("unchecked")
+ @Before
+ public void setUp() throws Exception {
+ Gauge g = registry.newGauge(System.class, "gauge", gauge);
+ Counter counter = registry.newCounter(System.class, "counter");
+ Histogram histogram = registry.newHistogram(System.class, "histogram");
+ Meter meter = registry.newMeter(System.class, "meter", "empty", TimeUnit.MILLISECONDS);
+ Timer timer = registry.newTimer(System.class, "timer");
+ list.add(g);
+ list.add(counter);
+ list.add(histogram);
+ list.add(meter);
+ list.add(timer);
+ reporter.start(200, TimeUnit.MILLISECONDS);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ reporter.stop();
+ }
+
+ @Test
+ public void pollsPeriodically() throws Exception {
+ Thread.sleep(500);
+ verify(reporter, times(2)).report(set(list));
+ }
+
+ private Set<Entry<MetricName, Metric>> set(List<Metric> metrics) {
+ final Map<MetricName, Metric> map = new HashMap<MetricName, Metric>();
+ for (Metric metric : metrics) {
+ String name = null;
+ if (metric instanceof Gauge) {
+ name = "gauge";
+ } else if (metric instanceof Counter) {
+ name = "counter";
+ } else if (metric instanceof Histogram) {
+ name = "histogram";
+ } else if (metric instanceof Meter) {
+ name = "meter";
+ } else if (metric instanceof Timer) {
+ name = "timer";
+ }
+ map.put(new MetricName(System.class, name), metric);
+ }
+ return map.entrySet();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-metrics/pom.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/pom.xml b/ambari-metrics/pom.xml
index d99cc82..22067d9 100644
--- a/ambari-metrics/pom.xml
+++ b/ambari-metrics/pom.xml
@@ -14,11 +14,7 @@
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
- http://maven.apache.org/xsd/maven-4.0.0.xsd">
+--><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<groupId>org.apache.ambari</groupId>
<modelVersion>4.0.0</modelVersion>
@@ -29,6 +25,7 @@
<module>ambari-metrics-common</module>
<module>ambari-metrics-hadoop-sink</module>
<module>ambari-metrics-flume-sink</module>
+ <module>ambari-metrics-kafka-sink</module>
<module>ambari-metrics-storm-sink</module>
<module>ambari-metrics-timelineservice</module>
<module>ambari-metrics-host-monitoring</module>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-broker.xml
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-broker.xml b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-broker.xml
index dc1b6b4..9c11007 100644
--- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-broker.xml
+++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-broker.xml
@@ -101,7 +101,7 @@
<name>num.partitions</name>
<value>1</value>
<description>
- The default number of partitions per topic.
+ The default number of partitions per topic.
</description>
</property>
<property>
@@ -291,9 +291,9 @@
</property>
<property>
<name>kafka.metrics.reporters</name>
- <value>kafka.ganglia.KafkaGangliaMetricsReporter</value>
+ <value>{{kafka_metrics_reporters}}</value>
<description>
- kafka ganglia metrics reporter
+ kafka ganglia metrics reporter and kafka timeline metrics reporter
</description>
</property>
<property>
@@ -318,4 +318,29 @@
<value>kafka</value>
<description>Ganglia group name </description>
</property>
+ <property>
+ <name>kafka.timeline.metrics.reporter.enabled</name>
+ <value>true</value>
+ <description>Kafka timeline metrics reporter enable</description>
+ </property>
+ <property>
+ <name>kafka.timeline.metrics.host</name>
+ <value>{{metric_collector_host}}</value>
+ <description>Timeline host</description>
+ </property>
+ <property>
+ <name>kafka.timeline.metrics.port</name>
+ <value>{{metric_collector_port}}</value>
+ <description>Timeline port</description>
+ </property>
+ <property>
+ <name>kafka.timeline.metrics.reporter.sendInterval</name>
+ <value>5900</value>
+ <description>Timeline metrics reporter send interval</description>
+ </property>
+ <property>
+ <name>kafka.timeline.metrics.maxRowCacheSize</name>
+ <value>10000</value>
+ <description>Timeline metrics reporter send interval</description>
+ </property>
</configuration>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-env.xml
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-env.xml b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-env.xml
index 7ad4396..f322406 100644
--- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-env.xml
+++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/configuration/kafka-env.xml
@@ -50,6 +50,12 @@
# The java implementation to use.
export JAVA_HOME={{java64_home}}
export PATH=$PATH:$JAVA_HOME/bin
+
+# Add kafka sink to classpath and related depenencies
+if [ -e "/usr/lib/ambari-metrics-kafka-sink/ambari-metrics-kafka-sink.jar" ]; then
+ export CLASSPATH=$CLASSPATH:/usr/lib/ambari-metrics-kafka-sink/ambari-metrics-kafka-sink.jar
+ export CLASSPATH=$CLASSPATH:/usr/lib/ambari-metrics-kafka-sink/lib/*
+fi
</value>
</property>
</configuration>
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka.py b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka.py
index c0231a8..33ee47a 100644
--- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka.py
+++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka.py
@@ -35,6 +35,9 @@ def kafka():
kafka_server_config = mutable_config_dict(params.config['configurations']['kafka-broker'])
kafka_server_config['broker.id'] = brokerid
kafka_server_config['host.name'] = params.hostname
+ kafka_server_config['kafka.timeline.metrics.host'] = params.metric_collector_host
+ kafka_server_config['kafka.timeline.metrics.port'] = params.metric_collector_port
+ kafka_server_config['kafka.metrics.reporters'] = params.kafka_metrics_reporters
kafka_data_dir = kafka_server_config['log.dirs']
Directory(filter(None,kafka_data_dir.split(",")),
owner=params.kafka_user,
http://git-wip-us.apache.org/repos/asf/ambari/blob/8d27ac2b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
index 067e537..800ccc4 100644
--- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
+++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
@@ -60,3 +60,30 @@ if (('kafka-log4j' in config['configurations']) and ('content' in config['config
log4j_props = config['configurations']['kafka-log4j']['content']
else:
log4j_props = None
+
+if 'ganglia_server_host' in config['clusterHostInfo'] and \
+ len(config['clusterHostInfo']['ganglia_server_host'])>0:
+ ganglia_installed = True
+ ganglia_server = config['clusterHostInfo']['ganglia_server_host'][0]
+ ganglia_report_interval = 60
+else:
+ ganglia_installed = False
+
+kafka_metrics_reporters=""
+
+if ganglia_installed:
+ kafka_metrics_reporters = "kafka.ganglia.KafkaGangliaMetricsReporter"
+
+ams_collector_hosts = default("/clusterHostInfo/metric_collector_hosts", [])
+has_metric_collector = not len(ams_collector_hosts) == 0
+
+if has_metric_collector:
+ metric_collector_host = ams_collector_hosts[0]
+ metric_collector_port = default("/configurations/ams-site/timeline.metrics.service.webapp.address", "0.0.0.0:8188")
+ if metric_collector_port and metric_collector_port.find(':') != -1:
+ metric_collector_port = metric_collector_port.split(':')[1]
+
+ if not len(kafka_metrics_reporters) == 0:
+ kafka_metrics_reporters = kafka_metrics_reporters + ','
+
+ kafka_metrics_reporters = kafka_metrics_reporters + "org.apache.hadoop.metrics2.sink.kafka.KafkaTimelineMetricsReporter"