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/05/05 03:17:56 UTC
[1/2] ambari git commit: AMBARI-10918. Ambari with DB as oracle is
failing to deploy. (swagle)
Repository: ambari
Updated Branches:
refs/heads/trunk ce3e14370 -> 5226ae1be
AMBARI-10918. Ambari with DB as oracle is failing to deploy. (swagle)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/0c39d4ef
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/0c39d4ef
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/0c39d4ef
Branch: refs/heads/trunk
Commit: 0c39d4ef5343c8fb055cd2581538ae26d9a77fba
Parents: ce3e143
Author: Siddharth Wagle <sw...@hortonworks.com>
Authored: Mon May 4 17:52:40 2015 -0700
Committer: Siddharth Wagle <sw...@hortonworks.com>
Committed: Mon May 4 17:52:40 2015 -0700
----------------------------------------------------------------------
.../ambari/server/orm/dao/ServiceConfigDAO.java | 7 ++++---
.../ambari/server/orm/entities/WidgetEntity.java | 15 ++++++++++++---
.../ambari/server/upgrade/UpgradeCatalog210.java | 6 +++---
.../src/main/resources/Ambari-DDL-MySQL-CREATE.sql | 4 ++--
.../src/main/resources/Ambari-DDL-Oracle-CREATE.sql | 4 ++--
.../main/resources/Ambari-DDL-Postgres-CREATE.sql | 4 ++--
.../Ambari-DDL-Postgres-EMBEDDED-CREATE.sql | 4 ++--
7 files changed, 27 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/0c39d4ef/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/ServiceConfigDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/ServiceConfigDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/ServiceConfigDAO.java
index 8f8e196..db0817b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/ServiceConfigDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/ServiceConfigDAO.java
@@ -18,6 +18,7 @@
package org.apache.ambari.server.orm.dao;
+import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -212,13 +213,13 @@ public class ServiceConfigDAO {
*/
@RequiresSession
public Long findNextServiceConfigVersion(long clusterId, String serviceName) {
- TypedQuery<Long> query = entityManagerProvider.get().createNamedQuery(
- "ServiceConfigEntity.findNextServiceConfigVersion", Long.class);
+ TypedQuery<Number> query = entityManagerProvider.get().createNamedQuery(
+ "ServiceConfigEntity.findNextServiceConfigVersion", Number.class);
query.setParameter("clusterId", clusterId);
query.setParameter("serviceName", serviceName);
- return daoUtils.selectSingle(query);
+ return daoUtils.selectSingle(query).longValue();
}
@Transactional
http://git-wip-us.apache.org/repos/asf/ambari/blob/0c39d4ef/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/WidgetEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/WidgetEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/WidgetEntity.java
index 397d845..b65a87b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/WidgetEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/WidgetEntity.java
@@ -17,13 +17,16 @@
*/
package org.apache.ambari.server.orm.entities;
+import javax.persistence.Basic;
import javax.persistence.CascadeType;
import javax.persistence.Column;
import javax.persistence.Entity;
+import javax.persistence.FetchType;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.JoinColumn;
+import javax.persistence.Lob;
import javax.persistence.ManyToOne;
import javax.persistence.NamedQueries;
import javax.persistence.NamedQuery;
@@ -67,7 +70,9 @@ public class WidgetEntity {
@Column(name = "widget_type", nullable = false, length = 255)
private String widgetType;
- @Column(name = "metrics", length = 32672)
+ @Lob
+ @Basic(fetch = FetchType.LAZY)
+ @Column(name = "metrics")
private String metrics;
@Column(name = "time_created", nullable = false, length = 255)
@@ -85,10 +90,14 @@ public class WidgetEntity {
@Column(name = "scope", length = 255)
private String scope;
- @Column(name = "widget_values", length = 4000)
+ @Lob
+ @Basic(fetch = FetchType.LAZY)
+ @Column(name = "widget_values")
private String widgetValues;
- @Column(name = "properties", length = 4000)
+ @Lob
+ @Basic(fetch = FetchType.LAZY)
+ @Column(name = "properties")
private String properties;
@Column(name = "cluster_id", nullable = false)
http://git-wip-us.apache.org/repos/asf/ambari/blob/0c39d4ef/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog210.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog210.java b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog210.java
index dc7f7a9..0845e4b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog210.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog210.java
@@ -498,14 +498,14 @@ public class UpgradeCatalog210 extends AbstractUpgradeCatalog {
columns.add(new DBColumnInfo("id", Long.class, null, null, false));
columns.add(new DBColumnInfo("widget_name", String.class, 255, null, false));
columns.add(new DBColumnInfo("widget_type", String.class, 255, null, false));
- columns.add(new DBColumnInfo("metrics", String.class, 32672, null, true));
+ columns.add(new DBColumnInfo("metrics", byte[].class, null, null, true));
columns.add(new DBColumnInfo("time_created", Long.class, null, null, false));
columns.add(new DBColumnInfo("author", String.class, 255, null, true));
columns.add(new DBColumnInfo("description", String.class, 255, null, true));
columns.add(new DBColumnInfo("display_name", String.class, 255, null, true));
columns.add(new DBColumnInfo("scope", String.class, 255, null, true));
- columns.add(new DBColumnInfo("widget_values", String.class, 4000, null, true));
- columns.add(new DBColumnInfo("properties", String.class, 4000, null, true));
+ columns.add(new DBColumnInfo("widget_values", byte[].class, null, null, true));
+ columns.add(new DBColumnInfo("properties", byte[].class, null, null, true));
columns.add(new DBColumnInfo("cluster_id", Long.class, null, null, false));
dbAccessor.createTable(WIDGET_TABLE, columns, "id");
http://git-wip-us.apache.org/repos/asf/ambari/blob/0c39d4ef/ambari-server/src/main/resources/Ambari-DDL-MySQL-CREATE.sql
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/Ambari-DDL-MySQL-CREATE.sql b/ambari-server/src/main/resources/Ambari-DDL-MySQL-CREATE.sql
index 7f00b10..d32f3cd 100644
--- a/ambari-server/src/main/resources/Ambari-DDL-MySQL-CREATE.sql
+++ b/ambari-server/src/main/resources/Ambari-DDL-MySQL-CREATE.sql
@@ -551,8 +551,8 @@ CREATE TABLE widget (
description VARCHAR(255),
display_name VARCHAR(255),
scope VARCHAR(255),
- widget_values VARCHAR(4000),
- properties VARCHAR(4000),
+ widget_values LONGTEXT,
+ properties LONGTEXT,
cluster_id BIGINT NOT NULL,
PRIMARY KEY(id)
);
http://git-wip-us.apache.org/repos/asf/ambari/blob/0c39d4ef/ambari-server/src/main/resources/Ambari-DDL-Oracle-CREATE.sql
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/Ambari-DDL-Oracle-CREATE.sql b/ambari-server/src/main/resources/Ambari-DDL-Oracle-CREATE.sql
index c0313e5..4317c83 100644
--- a/ambari-server/src/main/resources/Ambari-DDL-Oracle-CREATE.sql
+++ b/ambari-server/src/main/resources/Ambari-DDL-Oracle-CREATE.sql
@@ -541,8 +541,8 @@ CREATE TABLE widget (
description VARCHAR2(255),
display_name VARCHAR2(255),
scope VARCHAR2(255),
- widget_values VARCHAR2(4000),
- properties VARCHAR2(4000),
+ widget_values CLOB,
+ properties CLOB,
cluster_id NUMBER(19) NOT NULL,
PRIMARY KEY(id)
);
http://git-wip-us.apache.org/repos/asf/ambari/blob/0c39d4ef/ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql b/ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql
index 94e4884..e3cef5d 100644
--- a/ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql
+++ b/ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql
@@ -550,8 +550,8 @@ CREATE TABLE widget (
description VARCHAR(255),
display_name VARCHAR(255),
scope VARCHAR(255),
- widget_values VARCHAR(4000),
- properties VARCHAR(4000),
+ widget_values TEXT,
+ properties TEXT,
cluster_id BIGINT NOT NULL,
PRIMARY KEY(id)
);
http://git-wip-us.apache.org/repos/asf/ambari/blob/0c39d4ef/ambari-server/src/main/resources/Ambari-DDL-Postgres-EMBEDDED-CREATE.sql
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/Ambari-DDL-Postgres-EMBEDDED-CREATE.sql b/ambari-server/src/main/resources/Ambari-DDL-Postgres-EMBEDDED-CREATE.sql
index 54c8c40..77186a1 100644
--- a/ambari-server/src/main/resources/Ambari-DDL-Postgres-EMBEDDED-CREATE.sql
+++ b/ambari-server/src/main/resources/Ambari-DDL-Postgres-EMBEDDED-CREATE.sql
@@ -626,8 +626,8 @@ CREATE TABLE ambari.widget (
description VARCHAR(255),
display_name VARCHAR(255),
scope VARCHAR(255),
- widget_values VARCHAR(4000),
- properties VARCHAR(4000),
+ widget_values TEXT,
+ properties TEXT,
cluster_id BIGINT NOT NULL,
PRIMARY KEY(id)
);
[2/2] ambari git commit: AMBARI-10904. Provide a configurable timeout
setting on MetricsTimelineSink.emitMetrics. (swagle)
Posted by sw...@apache.org.
AMBARI-10904. Provide a configurable timeout setting on MetricsTimelineSink.emitMetrics. (swagle)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/5226ae1b
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/5226ae1b
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/5226ae1b
Branch: refs/heads/trunk
Commit: 5226ae1be4ceafb0ca4544a48d69c85edbc4a410
Parents: 0c39d4e
Author: Siddharth Wagle <sw...@hortonworks.com>
Authored: Mon May 4 18:06:44 2015 -0700
Committer: Siddharth Wagle <sw...@hortonworks.com>
Committed: Mon May 4 18:06:44 2015 -0700
----------------------------------------------------------------------
.../timeline/AbstractTimelineMetricsSink.java | 11 +++-
.../cache/HandleConnectExceptionTest.java | 6 ++
.../sink/flume/FlumeTimelineMetricsSink.java | 13 +++--
.../timeline/HadoopTimelineMetricsSink.java | 8 +++
.../kafka/KafkaTimelineMetricsReporter.java | 60 +++++++++++---------
.../storm/StormTimelineMetricsReporter.java | 9 +++
.../sink/storm/StormTimelineMetricsSink.java | 14 ++++-
7 files changed, 85 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/5226ae1b/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 fd4cacd..4b93f50 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
@@ -17,26 +17,28 @@
*/
package org.apache.hadoop.metrics2.sink.timeline;
-import java.io.IOException;
-import java.net.ConnectException;
-
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.StringRequestEntity;
+import org.apache.commons.httpclient.params.HttpMethodParams;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.jackson.map.AnnotationIntrospector;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.annotate.JsonSerialize;
import org.codehaus.jackson.xc.JaxbAnnotationIntrospector;
+import java.io.IOException;
+import java.net.ConnectException;
public abstract class AbstractTimelineMetricsSink {
public static final String TAGS_FOR_PREFIX_PROPERTY_PREFIX = "tagsForPrefix.";
public static final String MAX_METRIC_ROW_CACHE_SIZE = "maxRowCacheSize";
public static final String METRICS_SEND_INTERVAL = "sendInterval";
+ public static final String METRICS_POST_TIMEOUT_SECONDS = "timeout";
public static final String COLLECTOR_HOST_PROPERTY = "collector";
public static final String COLLECTOR_PORT_PROPERTY = "port";
+ protected static final int DEFAULT_POST_TIMEOUT_SECONDS = 10;
protected final Log LOG;
private HttpClient httpClient = new HttpClient();
@@ -63,6 +65,7 @@ public abstract class AbstractTimelineMetricsSink {
PostMethod postMethod = new PostMethod(connectUrl);
postMethod.setRequestEntity(requestEntity);
+ postMethod.setParameter(HttpMethodParams.SO_TIMEOUT, String.valueOf(getTimeoutSeconds() * 1000));
int statusCode = httpClient.executeMethod(postMethod);
if (statusCode != 200) {
LOG.info("Unable to POST metrics to collector, " + connectUrl);
@@ -79,4 +82,6 @@ public abstract class AbstractTimelineMetricsSink {
}
abstract protected String getCollectorUri();
+
+ abstract protected int getTimeoutSeconds();
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/5226ae1b/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/HandleConnectExceptionTest.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/HandleConnectExceptionTest.java b/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/HandleConnectExceptionTest.java
index 2786e3c..4f9b93e 100644
--- a/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/HandleConnectExceptionTest.java
+++ b/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/HandleConnectExceptionTest.java
@@ -68,6 +68,12 @@ public class HandleConnectExceptionTest {
protected String getCollectorUri() {
return COLLECTOR_URL;
}
+
+ @Override
+ protected int getTimeoutSeconds() {
+ return 10;
+ }
+
@Override
public void emitMetrics(TimelineMetrics metrics) throws IOException {
super.emitMetrics(metrics);
http://git-wip-us.apache.org/repos/asf/ambari/blob/5226ae1b/ambari-metrics/ambari-metrics-flume-sink/src/main/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSink.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-flume-sink/src/main/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSink.java b/ambari-metrics/ambari-metrics-flume-sink/src/main/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSink.java
index a6137af..1d4c739 100644
--- a/ambari-metrics/ambari-metrics-flume-sink/src/main/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSink.java
+++ b/ambari-metrics/ambari-metrics-flume-sink/src/main/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSink.java
@@ -24,16 +24,15 @@ import org.apache.flume.Context;
import org.apache.flume.FlumeException;
import org.apache.flume.instrumentation.MonitorService;
import org.apache.flume.instrumentation.util.JMXPollUtil;
+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.AbstractTimelineMetricsSink;
import org.apache.hadoop.metrics2.sink.timeline.UnableToConnectException;
import org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache;
import org.apache.hadoop.metrics2.sink.timeline.configuration.Configuration;
import java.io.IOException;
import java.net.InetAddress;
-import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collections;
@@ -45,8 +44,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-
-
public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implements MonitorService {
private String collectorUri;
private TimelineMetricsCache metricsCache;
@@ -55,6 +52,7 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
private String hostname;
private final static String COUNTER_METRICS_PROPERTY = "counters";
private final Set<String> counterMetrics = new HashSet<String>();
+ private int timeoutSeconds = 10;
@Override
public void start() {
@@ -83,6 +81,8 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
throw new FlumeException("Could not identify hostname.", e);
}
Configuration configuration = new Configuration("/flume-metrics2.properties");
+ timeoutSeconds = Integer.parseInt(configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS,
+ String.valueOf(DEFAULT_POST_TIMEOUT_SECONDS)));
int maxRowCacheSize = Integer.parseInt(configuration.getProperty(MAX_METRIC_ROW_CACHE_SIZE,
String.valueOf(TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT)));
int metricsSendInterval = Integer.parseInt(configuration.getProperty(METRICS_SEND_INTERVAL,
@@ -102,6 +102,11 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
return collectorUri;
}
+ @Override
+ protected int getTimeoutSeconds() {
+ return timeoutSeconds;
+ }
+
public void setPollFrequency(long pollFrequency) {
this.pollFrequency = pollFrequency;
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/5226ae1b/ambari-metrics/ambari-metrics-hadoop-sink/src/main/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSink.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-sink/src/main/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSink.java b/ambari-metrics/ambari-metrics-hadoop-sink/src/main/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSink.java
index 12230f5..2d171d9 100644
--- a/ambari-metrics/ambari-metrics-hadoop-sink/src/main/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSink.java
+++ b/ambari-metrics/ambari-metrics-hadoop-sink/src/main/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSink.java
@@ -56,6 +56,7 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
private String collectorUri;
private static final String SERVICE_NAME_PREFIX = "serviceName-prefix";
private static final String SERVICE_NAME = "serviceName";
+ private int timeoutSeconds = 10;
@Override
public void init(SubsetConfiguration conf) {
@@ -91,6 +92,8 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
LOG.info("Collector Uri: " + collectorUri);
+ timeoutSeconds = conf.getInt(METRICS_POST_TIMEOUT_SECONDS, DEFAULT_POST_TIMEOUT_SECONDS);
+
int maxRowCacheSize = conf.getInt(MAX_METRIC_ROW_CACHE_SIZE,
TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT);
int metricsSendInterval = conf.getInt(METRICS_SEND_INTERVAL,
@@ -151,6 +154,11 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
}
@Override
+ protected int getTimeoutSeconds() {
+ return timeoutSeconds;
+ }
+
+ @Override
public void putMetrics(MetricsRecord record) {
try {
String recordName = record.name();
http://git-wip-us.apache.org/repos/asf/ambari/blob/5226ae1b/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 cc365bd..00a58ee 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
@@ -18,27 +18,6 @@
package org.apache.hadoop.metrics2.sink.kafka;
-import java.io.IOException;
-import java.net.InetAddress;
-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 com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Counter;
import com.yammer.metrics.core.Gauge;
@@ -51,9 +30,31 @@ import com.yammer.metrics.core.MetricsRegistry;
import com.yammer.metrics.core.Summarizable;
import com.yammer.metrics.core.Timer;
import com.yammer.metrics.stats.Snapshot;
+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 java.io.IOException;
+import java.net.InetAddress;
+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 static org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS;
+import static org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT;
-public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink implements KafkaMetricsReporter,
- KafkaTimelineMetricsReporterMBean {
+public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
+ implements KafkaMetricsReporter, KafkaTimelineMetricsReporterMBean {
private final static Log LOG = LogFactory.getLog(KafkaTimelineMetricsReporter.class);
@@ -72,12 +73,18 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink im
private String hostname;
private TimelineScheduledReporter reporter;
private TimelineMetricsCache metricsCache;
+ private int timeoutSeconds = 10;
@Override
protected String getCollectorUri() {
return collectorUri;
}
+ @Override
+ protected int getTimeoutSeconds() {
+ return timeoutSeconds;
+ }
+
public void setMetricsCache(TimelineMetricsCache metricsCache) {
this.metricsCache = metricsCache;
}
@@ -93,10 +100,9 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink im
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)));
+ timeoutSeconds = props.getInt(METRICS_POST_TIMEOUT_SECONDS, DEFAULT_POST_TIMEOUT_SECONDS);
+ 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);
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));
http://git-wip-us.apache.org/repos/asf/ambari/blob/5226ae1b/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 89fc2ca..04b4b90 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
@@ -49,6 +49,7 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
private String collectorUri;
private NimbusClient nimbusClient;
private String applicationId;
+ private int timeoutSeconds;
public StormTimelineMetricsReporter() {
@@ -60,6 +61,11 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
}
@Override
+ protected int getTimeoutSeconds() {
+ return timeoutSeconds;
+ }
+
+ @Override
public void prepare(Map conf) {
LOG.info("Preparing Storm Metrics Reporter");
try {
@@ -75,6 +81,9 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
this.nimbusClient = NimbusClient.getConfiguredClient(stormConf);
String collectorHostname = cf.get(COLLECTOR_HOST).toString();
String port = cf.get(COLLECTOR_PORT).toString();
+ timeoutSeconds = cf.get(METRICS_POST_TIMEOUT_SECONDS) != null ?
+ Integer.parseInt(cf.get(METRICS_POST_TIMEOUT_SECONDS).toString()) :
+ DEFAULT_POST_TIMEOUT_SECONDS;
applicationId = cf.get(APP_ID).toString();
collectorUri = "http://" + collectorHostname + ":" + port + "/ws/v1/timeline/metrics";
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/ambari/blob/5226ae1b/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 767695b..b7827bb 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
@@ -39,10 +39,13 @@ import java.util.Collection;
import java.util.List;
import java.util.Map;
+import static org.apache.hadoop.metrics2.sink.timeline.cache.TimelineMetricsCache.*;
+
public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implements IMetricsConsumer {
private String collectorUri;
private TimelineMetricsCache metricsCache;
private String hostname;
+ private int timeoutSeconds;
@Override
protected String getCollectorUri() {
@@ -50,6 +53,11 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
}
@Override
+ protected int getTimeoutSeconds() {
+ return timeoutSeconds;
+ }
+
+ @Override
public void prepare(Map map, Object o, TopologyContext topologyContext, IErrorReporter iErrorReporter) {
LOG.info("Preparing Storm Metrics Sink");
try {
@@ -59,10 +67,12 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
throw new RuntimeException("Could not identify hostname.", e);
}
Configuration configuration = new Configuration("/storm-metrics2.properties");
+ timeoutSeconds = Integer.parseInt(configuration.getProperty(METRICS_POST_TIMEOUT_SECONDS,
+ String.valueOf(DEFAULT_POST_TIMEOUT_SECONDS)));
int maxRowCacheSize = Integer.parseInt(configuration.getProperty(MAX_METRIC_ROW_CACHE_SIZE,
- String.valueOf(TimelineMetricsCache.MAX_RECS_PER_NAME_DEFAULT)));
+ String.valueOf(MAX_RECS_PER_NAME_DEFAULT)));
int metricsSendInterval = Integer.parseInt(configuration.getProperty(METRICS_SEND_INTERVAL,
- String.valueOf(TimelineMetricsCache.MAX_EVICTION_TIME_MILLIS)));
+ String.valueOf(MAX_EVICTION_TIME_MILLIS)));
metricsCache = new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval);
collectorUri = "http://" + configuration.getProperty(COLLECTOR_HOST_PROPERTY) + ":" + configuration.getProperty(COLLECTOR_PORT_PROPERTY) + "/ws/v1/timeline/metrics";
}