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 2014/12/02 18:28:23 UTC
[07/30] ambari git commit: AMBARI-5707. Replace Ganglia with high
performant and pluggable Metrics System. (swagle)
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java
index 217e216..5896a88 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackDefinedPropertyProvider.java
@@ -17,25 +17,17 @@
*/
package org.apache.ambari.server.controller.internal;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-
+import com.google.inject.Inject;
+import com.google.inject.Injector;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
-import org.apache.ambari.server.controller.ganglia.GangliaComponentPropertyProvider;
-import org.apache.ambari.server.controller.ganglia.GangliaHostComponentPropertyProvider;
-import org.apache.ambari.server.controller.ganglia.GangliaHostProvider;
-import org.apache.ambari.server.controller.ganglia.GangliaPropertyProvider;
import org.apache.ambari.server.controller.jmx.JMXHostProvider;
import org.apache.ambari.server.controller.jmx.JMXPropertyProvider;
-import org.apache.ambari.server.controller.metrics.MetricsHostProvider;
+import org.apache.ambari.server.controller.metrics.MetricHostProvider;
+import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider;
+import org.apache.ambari.server.controller.metrics.ganglia.GangliaComponentPropertyProvider;
+import org.apache.ambari.server.controller.metrics.ganglia.GangliaHostComponentPropertyProvider;
+import org.apache.ambari.server.controller.metrics.ganglia.GangliaPropertyProvider;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.PropertyProvider;
import org.apache.ambari.server.controller.spi.Request;
@@ -50,8 +42,17 @@ import org.apache.ambari.server.state.stack.MetricDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.inject.Inject;
-import com.google.inject.Injector;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import static org.apache.ambari.server.controller.metrics.MetricsPropertyProvider.MetricsService;
/**
* This class analyzes a service's metrics to determine if additional
@@ -77,11 +78,11 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
private ComponentSSLConfiguration sslConfig = null;
private StreamProvider streamProvider = null;
private JMXHostProvider jmxHostProvider;
- private GangliaHostProvider gangliaHostProvider;
private PropertyProvider defaultJmx = null;
private PropertyProvider defaultGanglia = null;
- private final MetricsHostProvider metricsHostProvider;
+ private final MetricHostProvider metricHostProvider;
+ private MetricsService metricsService = MetricsService.GANGLIA;
/**
* PropertyHelper/AbstractPropertyProvider expect map of maps,
@@ -97,19 +98,17 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
}
public StackDefinedPropertyProvider(Resource.Type type,
- JMXHostProvider jmxHostProvider,
- GangliaHostProvider gangliaHostProvider,
- MetricsHostProvider metricsHostProvider,
- StreamProvider streamProvider,
- String clusterPropertyId,
- String hostPropertyId,
- String componentPropertyId,
- String resourceStatePropertyId,
- PropertyProvider defaultJmxPropertyProvider,
- PropertyProvider defaultGangliaPropertyProvider
- ) {
-
- this.metricsHostProvider = metricsHostProvider;
+ JMXHostProvider jmxHostProvider,
+ MetricHostProvider metricHostProvider,
+ StreamProvider streamProvider,
+ String clusterPropertyId,
+ String hostPropertyId,
+ String componentPropertyId,
+ String resourceStatePropertyId,
+ PropertyProvider defaultJmxPropertyProvider,
+ PropertyProvider defaultGangliaPropertyProvider) {
+
+ this.metricHostProvider = metricHostProvider;
if (null == clusterPropertyId)
throw new NullPointerException("Cluster name property id cannot be null");
@@ -123,7 +122,6 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
componentNamePropertyId = componentPropertyId;
this.resourceStatePropertyId = resourceStatePropertyId;
this.jmxHostProvider = jmxHostProvider;
- this.gangliaHostProvider = gangliaHostProvider;
sslConfig = ComponentSSLConfiguration.instance();
this.streamProvider = streamProvider;
defaultJmx = defaultJmxPropertyProvider;
@@ -131,6 +129,26 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
}
+ public StackDefinedPropertyProvider(Resource.Type type,
+ MetricsService metricsService,
+ JMXHostProvider jmxHostProvider,
+ MetricHostProvider metricHostProvider,
+ StreamProvider streamProvider,
+ String clusterPropertyId,
+ String hostPropertyId,
+ String componentPropertyId,
+ String jmxStatePropertyId,
+ PropertyProvider defaultJmxPropertyProvider,
+ PropertyProvider defaultGangliaPropertyProvider) {
+
+ this(type, jmxHostProvider, metricHostProvider,
+ streamProvider, clusterPropertyId, hostPropertyId, componentPropertyId,
+ jmxStatePropertyId, defaultJmxPropertyProvider, defaultGangliaPropertyProvider);
+
+ this.metricsService = metricsService;
+ }
+
+
@Override
public Set<Resource> populateResources(Set<Resource> resources,
Request request, Predicate predicate) throws SystemException {
@@ -164,7 +182,7 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
jmxMap.put(componentName, getPropertyInfo(m));
} else {
PropertyProvider pp = getDelegate(m,
- streamProvider, metricsHostProvider,
+ streamProvider, metricHostProvider,
clusterNamePropertyId, hostNamePropertyId,
componentNamePropertyId, resourceStatePropertyId,
componentName);
@@ -180,22 +198,19 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
}
if (gangliaMap.size() > 0) {
- GangliaPropertyProvider gpp = type.equals (Resource.Type.Component) ?
- new GangliaComponentPropertyProvider(gangliaMap,
- streamProvider, sslConfig, gangliaHostProvider,
- clusterNamePropertyId, componentNamePropertyId) :
- new GangliaHostComponentPropertyProvider(gangliaMap,
- streamProvider, sslConfig, gangliaHostProvider,
- clusterNamePropertyId, hostNamePropertyId, componentNamePropertyId);
-
- gpp.populateResources(resources, request, predicate);
+ MetricsPropertyProvider propertyProvider =
+ MetricsPropertyProvider.createInstance(metricsService,
+ type, gangliaMap, streamProvider, sslConfig, metricHostProvider,
+ clusterNamePropertyId, hostNamePropertyId, componentNamePropertyId);
+
+ propertyProvider.populateResources(resources, request, predicate);
} else {
defaultGanglia.populateResources(resources, request, predicate);
}
if (jmxMap.size() > 0) {
JMXPropertyProvider jpp = new JMXPropertyProvider(jmxMap, streamProvider,
- jmxHostProvider, metricsHostProvider,
+ jmxHostProvider, metricHostProvider,
clusterNamePropertyId, hostNamePropertyId,
componentNamePropertyId, resourceStatePropertyId);
@@ -289,7 +304,7 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
private PropertyProvider getDelegate(MetricDefinition definition,
StreamProvider streamProvider,
- MetricsHostProvider metricsHostProvider,
+ MetricHostProvider metricsHostProvider,
String clusterNamePropertyId,
String hostNamePropertyId,
String componentNamePropertyId,
@@ -308,7 +323,7 @@ public class StackDefinedPropertyProvider implements PropertyProvider {
* all implementations when modifying constructor interface
*/
Constructor<?> ct = clz.getConstructor(Injector.class, Map.class,
- Map.class, StreamProvider.class, MetricsHostProvider.class,
+ Map.class, StreamProvider.class, MetricHostProvider.class,
String.class, String.class, String.class, String.class, String.class);
Object o = ct.newInstance(
injector,
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/jmx/JMXPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/jmx/JMXPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/jmx/JMXPropertyProvider.java
index 975a479..bbacb0d 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/jmx/JMXPropertyProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/jmx/JMXPropertyProvider.java
@@ -18,26 +18,13 @@
package org.apache.ambari.server.controller.jmx;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
import org.apache.ambari.server.controller.internal.PropertyInfo;
-import org.apache.ambari.server.controller.metrics.MetricsHostProvider;
-import org.apache.ambari.server.controller.metrics.MetricsProvider;
+import org.apache.ambari.server.controller.metrics.MetricHostProvider;
+import org.apache.ambari.server.controller.metrics.ThreadPoolEnabledPropertyProvider;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.SystemException;
-import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.controller.utilities.StreamProvider;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
@@ -45,11 +32,22 @@ import org.codehaus.jackson.map.ObjectReader;
import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
/**
* Property provider implementation for JMX sources.
*/
-public class JMXPropertyProvider extends MetricsProvider {
+public class JMXPropertyProvider extends ThreadPoolEnabledPropertyProvider {
private static final String NAME_KEY = "name";
private static final String PORT_KEY = "tag.port";
@@ -110,7 +108,7 @@ public class JMXPropertyProvider extends MetricsProvider {
* @param componentMetrics the map of supported metrics
* @param streamProvider the stream provider
* @param jmxHostProvider the JMX host mapping
- * @param metricsHostProvider the host mapping
+ * @param metricHostProvider the host mapping
* @param clusterNamePropertyId the cluster name property id
* @param hostNamePropertyId the host name property id
* @param componentNamePropertyId the component name property id
@@ -119,13 +117,13 @@ public class JMXPropertyProvider extends MetricsProvider {
public JMXPropertyProvider(Map<String, Map<String, PropertyInfo>> componentMetrics,
StreamProvider streamProvider,
JMXHostProvider jmxHostProvider,
- MetricsHostProvider metricsHostProvider,
+ MetricHostProvider metricHostProvider,
String clusterNamePropertyId,
String hostNamePropertyId,
String componentNamePropertyId,
String statePropertyId) {
- super(componentMetrics, hostNamePropertyId, metricsHostProvider);
+ super(componentMetrics, hostNamePropertyId, metricHostProvider);
this.streamProvider = streamProvider;
this.jmxHostProvider = jmxHostProvider;
@@ -148,7 +146,7 @@ public class JMXPropertyProvider extends MetricsProvider {
* @return the populated resource; null if the resource should NOT be part of the result set for the given predicate
*/
@Override
- protected Resource populateResource(Resource resource, Request request, Predicate predicate, MetricsProvider.Ticket ticket)
+ protected Resource populateResource(Resource resource, Request request, Predicate predicate, Ticket ticket)
throws SystemException {
Set<String> ids = getRequestPropertyIds(request, predicate);
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricHostProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricHostProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricHostProvider.java
new file mode 100644
index 0000000..fec2b57
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricHostProvider.java
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.controller.metrics;
+
+import org.apache.ambari.server.controller.spi.SystemException;
+import static org.apache.ambari.server.controller.metrics.MetricsPropertyProvider.MetricsService;
+
+public interface MetricHostProvider {
+ /**
+ * Get the metrics server host name for the given cluster name.
+ *
+ * @param clusterName the cluster name
+ *
+ * @return the metrics server
+ *
+ * @throws org.apache.ambari.server.controller.spi.SystemException if unable to get the metrics server host name
+ */
+ public String getCollectorHostName(String clusterName, MetricsService service) throws SystemException;
+
+ /**
+ * Get the host name for the given cluster name and component name.
+ *
+ * @param clusterName the cluster name
+ * @param componentName the component name
+ * @return the host name
+ * @throws org.apache.ambari.server.controller.spi.SystemException
+ * if unable to get the JMX host name
+ */
+ public String getHostName(String clusterName, String componentName) throws SystemException;
+
+ /**
+ * Get the metrics server port for the given cluster name.
+ *
+ * @param clusterName the cluster name
+ *
+ * @return the metrics server
+ *
+ * @throws org.apache.ambari.server.controller.spi.SystemException if unable to get the metrics server host name
+ */
+ public String getCollectorPortName(String clusterName, MetricsService service) throws SystemException;
+
+ /**
+ * Get the status of metrics server host for the given cluster name.
+ *
+ * @param clusterName the cluster name
+ *
+ * @return true if heartbeat with metrics server host wasn't lost
+ *
+ * @throws SystemException if unable to get the status of metrics server host
+ */
+ public boolean isCollectorHostLive(String clusterName, MetricsService service) throws SystemException;
+
+ /**
+ * Get the status of metrics server component for the given cluster name.
+ *
+ * @param clusterName the cluster name
+ *
+ * @return true if metrics server component is started
+ *
+ * @throws SystemException if unable to get the status of metrics server component
+ */
+ public boolean isCollectorComponentLive(String clusterName, MetricsService service) throws SystemException;
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsHostProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsHostProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsHostProvider.java
deleted file mode 100644
index 1a96829..0000000
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsHostProvider.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.server.controller.metrics;
-
-
-import org.apache.ambari.server.controller.spi.SystemException;
-
-public interface MetricsHostProvider {
-
- /**
- * Get the host name for the given cluster name and component name.
- *
- * @param clusterName the cluster name
- * @param componentName the component name
- * @return the host name
- * @throws org.apache.ambari.server.controller.spi.SystemException
- * if unable to get the JMX host name
- */
- public String getHostName(String clusterName, String componentName)
- throws SystemException;
-
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java
new file mode 100644
index 0000000..4f1964f
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsPropertyProvider.java
@@ -0,0 +1,204 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.controller.metrics;
+
+import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
+import org.apache.ambari.server.controller.internal.AbstractPropertyProvider;
+import org.apache.ambari.server.controller.internal.PropertyInfo;
+import org.apache.ambari.server.controller.metrics.ganglia.GangliaComponentPropertyProvider;
+import org.apache.ambari.server.controller.metrics.ganglia.GangliaHostComponentPropertyProvider;
+import org.apache.ambari.server.controller.metrics.ganglia.GangliaHostPropertyProvider;
+import org.apache.ambari.server.controller.metrics.timeline.AMSComponentPropertyProvider;
+import org.apache.ambari.server.controller.metrics.timeline.AMSHostComponentPropertyProvider;
+import org.apache.ambari.server.controller.metrics.timeline.AMSHostPropertyProvider;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.spi.SystemException;
+import org.apache.ambari.server.controller.utilities.StreamProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+import static org.apache.ambari.server.controller.metrics.MetricsPropertyProvider.MetricsService.TIMELINE_METRICS;
+
+public abstract class MetricsPropertyProvider extends AbstractPropertyProvider {
+ protected final static Logger LOG =
+ LoggerFactory.getLogger(MetricsPropertyProvider.class);
+
+ protected static final Pattern questionMarkPattern = Pattern.compile("\\?");
+
+ protected final StreamProvider streamProvider;
+
+ protected final MetricHostProvider hostProvider;
+
+ protected final String clusterNamePropertyId;
+
+ protected final String hostNamePropertyId;
+
+ protected final String componentNamePropertyId;
+
+ protected final ComponentSSLConfiguration configuration;
+
+ /**
+ * Enumeration to distinguish metrics service installed for a cluster
+ */
+ public enum MetricsService {
+ GANGLIA,
+ TIMELINE_METRICS
+ }
+
+ public MetricsPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
+ StreamProvider streamProvider,
+ ComponentSSLConfiguration configuration,
+ MetricHostProvider hostProvider,
+ String clusterNamePropertyId,
+ String hostNamePropertyId,
+ String componentNamePropertyId) {
+
+ super(componentPropertyInfoMap);
+
+ this.streamProvider = streamProvider;
+ this.configuration = configuration;
+ this.hostProvider = hostProvider;
+ this.clusterNamePropertyId = clusterNamePropertyId;
+ this.hostNamePropertyId = hostNamePropertyId;
+ this.componentNamePropertyId = componentNamePropertyId;
+ }
+
+ public static MetricsPropertyProvider createInstance(
+ MetricsService metricsService,
+ Resource.Type type,
+ Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
+ StreamProvider streamProvider,
+ ComponentSSLConfiguration configuration,
+ MetricHostProvider hostProvider,
+ String clusterNamePropertyId,
+ String hostNamePropertyId,
+ String componentNamePropertyId) {
+
+ if (type.isInternalType()) {
+ switch (type.getInternalType()) {
+ case Host:
+ if (metricsService.equals(TIMELINE_METRICS)) {
+ return new AMSHostPropertyProvider(componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId,
+ hostNamePropertyId);
+ } else {
+ return new GangliaHostPropertyProvider(
+ componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId,
+ hostNamePropertyId);
+ }
+ case HostComponent:
+ if (metricsService.equals(TIMELINE_METRICS)) {
+ return new AMSHostComponentPropertyProvider(
+ componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId,
+ hostNamePropertyId,
+ componentNamePropertyId);
+ } else {
+ return new GangliaHostComponentPropertyProvider(
+ componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId,
+ hostNamePropertyId,
+ componentNamePropertyId);
+ }
+ case Component:
+ if (metricsService.equals(TIMELINE_METRICS)) {
+ return new AMSComponentPropertyProvider(
+ componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId,
+ componentNamePropertyId);
+ } else {
+ return new GangliaComponentPropertyProvider(
+ componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId,
+ componentNamePropertyId);
+ }
+ default:
+ break;
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * Get the host name for the given resource.
+ *
+ * @param resource the resource
+ *
+ * @return the host name
+ */
+ protected abstract String getHostName(Resource resource);
+
+ /**
+ * Get the component name for the given resource.
+ *
+ * @param resource the resource
+ *
+ * @return the component name
+ */
+ protected abstract String getComponentName(Resource resource);
+
+ /**
+ * Get a comma delimited string from the given set of strings or
+ * an empty string if the size of the given set is greater than
+ * the given limit.
+ *
+ * @param set the set of strings
+ * @param limit the upper size limit for the list
+ *
+ * @return a comma delimited string of strings
+ */
+ protected static String getSetString(Set<String> set, int limit) {
+ StringBuilder sb = new StringBuilder();
+
+ if (limit == -1 || set.size() <= limit) {
+ for (String cluster : set) {
+ if (sb.length() > 0) {
+ sb.append(',');
+ }
+ sb.append(cluster);
+ }
+ }
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsProvider.java
deleted file mode 100644
index 27c55f4..0000000
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsProvider.java
+++ /dev/null
@@ -1,302 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.server.controller.metrics;
-
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CompletionService;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorCompletionService;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-
-import org.apache.ambari.server.controller.internal.AbstractPropertyProvider;
-import org.apache.ambari.server.controller.internal.PropertyInfo;
-import org.apache.ambari.server.controller.spi.Predicate;
-import org.apache.ambari.server.controller.spi.Request;
-import org.apache.ambari.server.controller.spi.Resource;
-import org.apache.ambari.server.controller.spi.SystemException;
-
-/**
- * Unites common functionality for multithreaded metrics providers
- * (JMX and REST as of now). Shares the same pool of executor threads.
- */
-public abstract class MetricsProvider extends AbstractPropertyProvider {
-
- /**
- * Host states that make available metrics collection
- */
- public static final Set<String> healthyStates = Collections.singleton("STARTED");
- protected final String hostNamePropertyId;
- private final MetricsHostProvider metricsHostProvider;
-
- /**
- * Executor service is shared between all childs of current class
- */
- private static final ExecutorService EXECUTOR_SERVICE = initExecutorService();
- private static final int THREAD_POOL_CORE_SIZE = 20;
- private static final int THREAD_POOL_MAX_SIZE = 100;
- private static final long THREAD_POOL_TIMEOUT_MILLIS = 30000L;
-
- private static final long DEFAULT_POPULATE_TIMEOUT_MILLIS = 10000L;
- /**
- * The amount of time that this provider will wait for JMX metric values to be
- * returned from the JMX sources. If no results are returned for this amount of
- * time then the request to populate the resources will fail.
- */
- protected long populateTimeout = DEFAULT_POPULATE_TIMEOUT_MILLIS;
- public static final String TIMED_OUT_MSG = "Timed out waiting for metrics.";
-
- // ----- Constructors ------------------------------------------------------
-
- /**
- * Construct a provider.
- *
- * @param componentMetrics map of metrics for this provider
- */
- public MetricsProvider(Map<String, Map<String, PropertyInfo>> componentMetrics,
- String hostNamePropertyId,
- MetricsHostProvider metricsHostProvider) {
- super(componentMetrics);
- this.hostNamePropertyId = hostNamePropertyId;
- this.metricsHostProvider = metricsHostProvider;
- }
-
- // ----- Thread pool -------------------------------------------------------
-
- /**
- * Generates thread pool with default parameters
- */
-
-
- private static ExecutorService initExecutorService() {
- LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(); // unlimited Queue
-
- ThreadPoolExecutor threadPoolExecutor =
- new ThreadPoolExecutor(
- THREAD_POOL_CORE_SIZE,
- THREAD_POOL_MAX_SIZE,
- THREAD_POOL_TIMEOUT_MILLIS,
- TimeUnit.MILLISECONDS,
- queue);
-
- threadPoolExecutor.allowCoreThreadTimeOut(true);
-
- return threadPoolExecutor;
- }
-
- public static ExecutorService getExecutorService() {
- return EXECUTOR_SERVICE;
- }
-
- // ----- Common PropertyProvider implementation details --------------------
-
- @Override
- public Set<Resource> populateResources(Set<Resource> resources, Request request, Predicate predicate)
- throws SystemException {
-
- // Get a valid ticket for the request.
- Ticket ticket = new Ticket();
-
- CompletionService<Resource> completionService =
- new ExecutorCompletionService<Resource>(EXECUTOR_SERVICE);
-
- // In a large cluster we could have thousands of resources to populate here.
- // Distribute the work across multiple threads.
- for (Resource resource : resources) {
- completionService.submit(getPopulateResourceCallable(resource, request, predicate, ticket));
- }
-
- Set<Resource> keepers = new HashSet<Resource>();
- try {
- for (int i = 0; i < resources.size(); ++ i) {
- Future<Resource> resourceFuture =
- completionService.poll(populateTimeout, TimeUnit.MILLISECONDS);
-
- if (resourceFuture == null) {
- // its been more than the populateTimeout since the last callable completed ...
- // invalidate the ticket to abort the threads and don't wait any longer
- ticket.invalidate();
- LOG.error(TIMED_OUT_MSG);
- break;
- } else {
- // future should already be completed... no need to wait on get
- Resource resource = resourceFuture.get();
- if (resource != null) {
- keepers.add(resource);
- }
- }
- }
- } catch (InterruptedException e) {
- logException(e);
- } catch (ExecutionException e) {
- rethrowSystemException(e.getCause());
- }
- return keepers;
- }
-
- /**
- * Get a callable that can be used to populate the given resource.
- *
- * @param resource the resource to be populated
- * @param request the request
- * @param predicate the predicate
- * @param ticket a valid ticket
- *
- * @return a callable that can be used to populate the given resource
- */
- private Callable<Resource> getPopulateResourceCallable(
- final Resource resource, final Request request, final Predicate predicate, final Ticket ticket) {
- return new Callable<Resource>() {
- public Resource call() throws SystemException {
- return populateResource(resource, request, predicate, ticket);
- }
- };
- }
-
-
- /**
- * Populate a resource by obtaining the requested JMX properties.
- *
- * @param resource the resource to be populated
- * @param request the request
- * @param predicate the predicate
- * @return the populated resource; null if the resource should NOT be part of the result set for the given predicate
- */
-
-
- protected abstract Resource populateResource(Resource resource,
- Request request, Predicate predicate, Ticket ticket)
-
- throws SystemException;
-
- /**
- * Set the populate timeout value for this provider.
- *
- * @param populateTimeout the populate timeout value
- */
-
-
- protected void setPopulateTimeout(long populateTimeout) {
- this.populateTimeout = populateTimeout;
-
- }
-
-
- // ----- helper methods ----------------------------------------------------
-
- /**
- * Determine whether or not the given property id was requested.
- */
- protected static boolean isRequestedPropertyId(String propertyId, String requestedPropertyId, Request request) {
- return request.getPropertyIds().isEmpty() || propertyId.startsWith(requestedPropertyId);
- }
-
- /**
- * Log an error for the given exception.
- *
- * @param throwable the caught exception
- *
- * @return the error message that was logged
- */
- protected static String logException(Throwable throwable) {
- String msg = "Caught exception getting JMX metrics : " + throwable.getLocalizedMessage();
-
- LOG.debug(msg, throwable);
-
- return msg;
- }
-
- /**
- * Rethrow the given exception as a System exception and log the message.
- *
- * @param throwable the caught exception
- *
- * @throws org.apache.ambari.server.controller.spi.SystemException always around the given exception
- */
- protected static void rethrowSystemException(Throwable throwable) throws SystemException {
- String msg = logException(throwable);
-
- if (throwable instanceof SystemException) {
- throw (SystemException) throwable;
- }
- throw new SystemException (msg, throwable);
- }
-
- /**
- * Returns a hostname for component
- */
-
-
- public String getHost(Resource resource, String clusterName, String componentName) throws SystemException {
- return hostNamePropertyId == null ?
- metricsHostProvider.getHostName(clusterName, componentName) :
- (String) resource.getPropertyValue(hostNamePropertyId);
-
- }
-
-
- /**
- * Get complete URL from parts
- */
-
- protected String getSpec(String protocol, String hostName,
- String port, String url) {
- return protocol + "://" + hostName + ":" + port + url;
-
- }
-
- // ----- inner class : Ticket ----------------------------------------------
-
- /**
- * Ticket used to cancel provider threads. The provider threads should
- * monitor the validity of the passed in ticket and bail out if it becomes
- * invalid (as in a timeout).
- */
- protected static class Ticket {
- /**
- * Indicate whether or not the ticket is valid.
- */
- private volatile boolean valid = true;
-
- /**
- * Invalidate the ticket.
- */
- public void invalidate() {
- valid = false;
- }
-
- /**
- * Determine whether or not this ticket is valid.
- *
- * @return true if the ticket is valid
- */
- public boolean isValid() {
- return valid;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java
new file mode 100644
index 0000000..39af1d9
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/MetricsReportPropertyProvider.java
@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.controller.metrics;
+
+import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
+import org.apache.ambari.server.controller.internal.AbstractPropertyProvider;
+import org.apache.ambari.server.controller.internal.PropertyInfo;
+import org.apache.ambari.server.controller.metrics.ganglia.GangliaReportPropertyProvider;
+import org.apache.ambari.server.controller.metrics.timeline.AMSReportPropertyProvider;
+import org.apache.ambari.server.controller.utilities.StreamProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+import static org.apache.ambari.server.controller.metrics.MetricsPropertyProvider.MetricsService;
+
+public abstract class MetricsReportPropertyProvider extends AbstractPropertyProvider {
+
+ protected final StreamProvider streamProvider;
+
+ protected final MetricHostProvider hostProvider;
+
+ protected final String clusterNamePropertyId;
+
+ protected final ComponentSSLConfiguration configuration;
+
+
+ // ----- Constants --------------------------------------------------------
+
+ protected final static Logger LOG =
+ LoggerFactory.getLogger(MetricsReportPropertyProvider.class);
+
+
+ // ----- Constructors ------------------------------------------------------
+
+ public MetricsReportPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
+ StreamProvider streamProvider,
+ ComponentSSLConfiguration configuration,
+ MetricHostProvider hostProvider,
+ String clusterNamePropertyId) {
+ super(componentPropertyInfoMap);
+
+ this.streamProvider = streamProvider;
+ this.hostProvider = hostProvider;
+ this.clusterNamePropertyId = clusterNamePropertyId;
+ this.configuration = configuration;
+ }
+
+ public static MetricsReportPropertyProvider createInstance(
+ MetricsService metricsService,
+ Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
+ StreamProvider streamProvider,
+ ComponentSSLConfiguration configuration,
+ MetricHostProvider hostProvider,
+ String clusterNamePropertyId) {
+ if (metricsService.equals(MetricsService.GANGLIA)) {
+ return new GangliaReportPropertyProvider(
+ componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId);
+ } else {
+ return new AMSReportPropertyProvider(
+ componentPropertyInfoMap,
+ streamProvider,
+ configuration,
+ hostProvider,
+ clusterNamePropertyId);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/RestMetricsPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/RestMetricsPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/RestMetricsPropertyProvider.java
index 61b576c..b92537b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/RestMetricsPropertyProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/RestMetricsPropertyProvider.java
@@ -57,7 +57,7 @@ import java.util.Set;
* For every metric, finds a relevant JSON value and returns is as
* a resource property.
*/
-public class RestMetricsPropertyProvider extends MetricsProvider {
+public class RestMetricsPropertyProvider extends ThreadPoolEnabledPropertyProvider {
protected final static Logger LOG =
LoggerFactory.getLogger(RestMetricsPropertyProvider.class);
@@ -76,7 +76,7 @@ public class RestMetricsPropertyProvider extends MetricsProvider {
private final String clusterNamePropertyId;
private final String componentNamePropertyId;
private final String statePropertyId;
- private MetricsHostProvider metricsHostProvider;
+ private MetricHostProvider metricHostProvider;
private final String componentName;
private static final String DEFAULT_PORT_PROPERTY = "default_port";
@@ -109,31 +109,31 @@ public class RestMetricsPropertyProvider extends MetricsProvider {
* @param metricsProperties the map of per-component metrics properties
* @param componentMetrics the map of supported metrics for component
* @param streamProvider the stream provider
- * @param metricsHostProvider metricsHostProvider instance
+ * @param metricHostProvider metricsHostProvider instance
* @param clusterNamePropertyId the cluster name property id
* @param hostNamePropertyId the host name property id
* @param componentNamePropertyId the component name property id
* @param statePropertyId the state property id
*/
public RestMetricsPropertyProvider(
- Injector injector,
- Map<String, String> metricsProperties,
- Map<String, Map<String, PropertyInfo>> componentMetrics,
- StreamProvider streamProvider,
- MetricsHostProvider metricsHostProvider,
- String clusterNamePropertyId,
- String hostNamePropertyId,
- String componentNamePropertyId,
- String statePropertyId,
- String componentName){
-
- super(componentMetrics, hostNamePropertyId, metricsHostProvider);
+ Injector injector,
+ Map<String, String> metricsProperties,
+ Map<String, Map<String, PropertyInfo>> componentMetrics,
+ StreamProvider streamProvider,
+ MetricHostProvider metricHostProvider,
+ String clusterNamePropertyId,
+ String hostNamePropertyId,
+ String componentNamePropertyId,
+ String statePropertyId,
+ String componentName){
+
+ super(componentMetrics, hostNamePropertyId, metricHostProvider);
this.metricsProperties = metricsProperties;
this.streamProvider = streamProvider;
this.clusterNamePropertyId = clusterNamePropertyId;
this.componentNamePropertyId = componentNamePropertyId;
this.statePropertyId = statePropertyId;
- this.metricsHostProvider = metricsHostProvider;
+ this.metricHostProvider = metricHostProvider;
injector.injectMembers(this);
this.componentName = componentName;
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ThreadPoolEnabledPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ThreadPoolEnabledPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ThreadPoolEnabledPropertyProvider.java
new file mode 100644
index 0000000..8a35636
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ThreadPoolEnabledPropertyProvider.java
@@ -0,0 +1,301 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.controller.metrics;
+
+import org.apache.ambari.server.controller.internal.AbstractPropertyProvider;
+import org.apache.ambari.server.controller.internal.PropertyInfo;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.spi.SystemException;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Unites common functionality for multithreaded metrics providers
+ * (JMX and REST as of now). Shares the same pool of executor threads.
+ */
+public abstract class ThreadPoolEnabledPropertyProvider extends AbstractPropertyProvider {
+
+ /**
+ * Host states that make available metrics collection
+ */
+ public static final Set<String> healthyStates = Collections.singleton("STARTED");
+ protected final String hostNamePropertyId;
+ private final MetricHostProvider metricHostProvider;
+
+ /**
+ * Executor service is shared between all childs of current class
+ */
+ private static final ExecutorService EXECUTOR_SERVICE = initExecutorService();
+ private static final int THREAD_POOL_CORE_SIZE = 20;
+ private static final int THREAD_POOL_MAX_SIZE = 100;
+ private static final long THREAD_POOL_TIMEOUT_MILLIS = 30000L;
+
+ private static final long DEFAULT_POPULATE_TIMEOUT_MILLIS = 10000L;
+ /**
+ * The amount of time that this provider will wait for JMX metric values to be
+ * returned from the JMX sources. If no results are returned for this amount of
+ * time then the request to populate the resources will fail.
+ */
+ protected long populateTimeout = DEFAULT_POPULATE_TIMEOUT_MILLIS;
+ public static final String TIMED_OUT_MSG = "Timed out waiting for metrics.";
+
+ // ----- Constructors ------------------------------------------------------
+
+ /**
+ * Construct a provider.
+ *
+ * @param componentMetrics map of metrics for this provider
+ */
+ public ThreadPoolEnabledPropertyProvider(Map<String, Map<String, PropertyInfo>> componentMetrics,
+ String hostNamePropertyId,
+ MetricHostProvider metricHostProvider) {
+ super(componentMetrics);
+ this.hostNamePropertyId = hostNamePropertyId;
+ this.metricHostProvider = metricHostProvider;
+ }
+
+ // ----- Thread pool -------------------------------------------------------
+
+ /**
+ * Generates thread pool with default parameters
+ */
+
+
+ private static ExecutorService initExecutorService() {
+ LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(); // unlimited Queue
+
+ ThreadPoolExecutor threadPoolExecutor =
+ new ThreadPoolExecutor(
+ THREAD_POOL_CORE_SIZE,
+ THREAD_POOL_MAX_SIZE,
+ THREAD_POOL_TIMEOUT_MILLIS,
+ TimeUnit.MILLISECONDS,
+ queue);
+
+ threadPoolExecutor.allowCoreThreadTimeOut(true);
+
+ return threadPoolExecutor;
+ }
+
+ public static ExecutorService getExecutorService() {
+ return EXECUTOR_SERVICE;
+ }
+
+ // ----- Common PropertyProvider implementation details --------------------
+
+ @Override
+ public Set<Resource> populateResources(Set<Resource> resources, Request request, Predicate predicate)
+ throws SystemException {
+
+ // Get a valid ticket for the request.
+ Ticket ticket = new Ticket();
+
+ CompletionService<Resource> completionService =
+ new ExecutorCompletionService<Resource>(EXECUTOR_SERVICE);
+
+ // In a large cluster we could have thousands of resources to populate here.
+ // Distribute the work across multiple threads.
+ for (Resource resource : resources) {
+ completionService.submit(getPopulateResourceCallable(resource, request, predicate, ticket));
+ }
+
+ Set<Resource> keepers = new HashSet<Resource>();
+ try {
+ for (int i = 0; i < resources.size(); ++ i) {
+ Future<Resource> resourceFuture =
+ completionService.poll(populateTimeout, TimeUnit.MILLISECONDS);
+
+ if (resourceFuture == null) {
+ // its been more than the populateTimeout since the last callable completed ...
+ // invalidate the ticket to abort the threads and don't wait any longer
+ ticket.invalidate();
+ LOG.error(TIMED_OUT_MSG);
+ break;
+ } else {
+ // future should already be completed... no need to wait on get
+ Resource resource = resourceFuture.get();
+ if (resource != null) {
+ keepers.add(resource);
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ logException(e);
+ } catch (ExecutionException e) {
+ rethrowSystemException(e.getCause());
+ }
+ return keepers;
+ }
+
+ /**
+ * Get a callable that can be used to populate the given resource.
+ *
+ * @param resource the resource to be populated
+ * @param request the request
+ * @param predicate the predicate
+ * @param ticket a valid ticket
+ *
+ * @return a callable that can be used to populate the given resource
+ */
+ private Callable<Resource> getPopulateResourceCallable(
+ final Resource resource, final Request request, final Predicate predicate, final Ticket ticket) {
+ return new Callable<Resource>() {
+ public Resource call() throws SystemException {
+ return populateResource(resource, request, predicate, ticket);
+ }
+ };
+ }
+
+
+ /**
+ * Populate a resource by obtaining the requested JMX properties.
+ *
+ * @param resource the resource to be populated
+ * @param request the request
+ * @param predicate the predicate
+ * @return the populated resource; null if the resource should NOT be part of the result set for the given predicate
+ */
+
+
+ protected abstract Resource populateResource(Resource resource,
+ Request request, Predicate predicate, Ticket ticket)
+
+ throws SystemException;
+
+ /**
+ * Set the populate timeout value for this provider.
+ *
+ * @param populateTimeout the populate timeout value
+ */
+
+
+ protected void setPopulateTimeout(long populateTimeout) {
+ this.populateTimeout = populateTimeout;
+
+ }
+
+
+ // ----- helper methods ----------------------------------------------------
+
+ /**
+ * Determine whether or not the given property id was requested.
+ */
+ protected static boolean isRequestedPropertyId(String propertyId, String requestedPropertyId, Request request) {
+ return request.getPropertyIds().isEmpty() || propertyId.startsWith(requestedPropertyId);
+ }
+
+ /**
+ * Log an error for the given exception.
+ *
+ * @param throwable the caught exception
+ *
+ * @return the error message that was logged
+ */
+ protected static String logException(Throwable throwable) {
+ String msg = "Caught exception getting JMX metrics : " + throwable.getLocalizedMessage();
+
+ LOG.debug(msg, throwable);
+
+ return msg;
+ }
+
+ /**
+ * Rethrow the given exception as a System exception and log the message.
+ *
+ * @param throwable the caught exception
+ *
+ * @throws org.apache.ambari.server.controller.spi.SystemException always around the given exception
+ */
+ protected static void rethrowSystemException(Throwable throwable) throws SystemException {
+ String msg = logException(throwable);
+
+ if (throwable instanceof SystemException) {
+ throw (SystemException) throwable;
+ }
+ throw new SystemException (msg, throwable);
+ }
+
+ /**
+ * Returns a hostname for component
+ */
+
+
+ public String getHost(Resource resource, String clusterName, String componentName) throws SystemException {
+ return hostNamePropertyId == null ?
+ metricHostProvider.getHostName(clusterName, componentName) :
+ (String) resource.getPropertyValue(hostNamePropertyId);
+
+ }
+
+
+ /**
+ * Get complete URL from parts
+ */
+
+ protected String getSpec(String protocol, String hostName,
+ String port, String url) {
+ return protocol + "://" + hostName + ":" + port + url;
+
+ }
+
+ // ----- inner class : Ticket ----------------------------------------------
+
+ /**
+ * Ticket used to cancel provider threads. The provider threads should
+ * monitor the validity of the passed in ticket and bail out if it becomes
+ * invalid (as in a timeout).
+ */
+ protected static class Ticket {
+ /**
+ * Indicate whether or not the ticket is valid.
+ */
+ private volatile boolean valid = true;
+
+ /**
+ * Invalidate the ticket.
+ */
+ public void invalidate() {
+ valid = false;
+ }
+
+ /**
+ * Determine whether or not this ticket is valid.
+ *
+ * @return true if the ticket is valid
+ */
+ public boolean isValid() {
+ return valid;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaComponentPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaComponentPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaComponentPropertyProvider.java
new file mode 100644
index 0000000..e531dff
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaComponentPropertyProvider.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.controller.metrics.ganglia;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
+import org.apache.ambari.server.controller.internal.PropertyInfo;
+import org.apache.ambari.server.controller.metrics.MetricHostProvider;
+import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.utilities.StreamProvider;
+
+/**
+ * Ganglia property provider implementation for component resources.
+ */
+public class GangliaComponentPropertyProvider extends GangliaPropertyProvider {
+
+
+ // ----- Constructors ------------------------------------------------------
+
+ public GangliaComponentPropertyProvider(Map<String, Map<String, PropertyInfo>> componentMetrics,
+ StreamProvider streamProvider,
+ ComponentSSLConfiguration configuration,
+ MetricHostProvider hostProvider,
+ String clusterNamePropertyId,
+ String componentNamePropertyId) {
+
+ super(componentMetrics, streamProvider, configuration, hostProvider,
+ clusterNamePropertyId, null, componentNamePropertyId);
+ }
+
+
+ // ----- GangliaPropertyProvider -------------------------------------------
+
+ @Override
+ protected String getHostName(Resource resource) {
+ return "__SummaryInfo__";
+ }
+
+ @Override
+ protected String getComponentName(Resource resource) {
+ return (String) resource.getPropertyValue(getComponentNamePropertyId());
+ }
+
+ @Override
+ protected Set<String> getGangliaClusterNames(Resource resource, String clusterName) {
+ String component = getComponentName(resource);
+
+ return new HashSet<String>(GANGLIA_CLUSTER_NAME_MAP.containsKey(component) ?
+ GANGLIA_CLUSTER_NAME_MAP.get(component) :
+ Collections.<String>emptyList());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostComponentPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostComponentPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostComponentPropertyProvider.java
new file mode 100644
index 0000000..5508cbe
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostComponentPropertyProvider.java
@@ -0,0 +1,71 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.controller.metrics.ganglia;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
+import org.apache.ambari.server.controller.internal.PropertyInfo;
+import org.apache.ambari.server.controller.metrics.MetricHostProvider;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.utilities.StreamProvider;
+
+/**
+ * Ganglia property provider implementation for host component resources.
+ */
+public class GangliaHostComponentPropertyProvider extends GangliaPropertyProvider {
+ // ----- Constructors ------------------------------------------------------
+
+ public GangliaHostComponentPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
+ StreamProvider streamProvider,
+ ComponentSSLConfiguration configuration,
+ MetricHostProvider hostProvider,
+ String clusterNamePropertyId,
+ String hostNamePropertyId,
+ String componentNamePropertyId) {
+
+ super(componentPropertyInfoMap, streamProvider, configuration, hostProvider,
+ clusterNamePropertyId, hostNamePropertyId, componentNamePropertyId);
+ }
+
+
+ // ----- GangliaPropertyProvider -------------------------------------------
+
+ @Override
+ protected String getHostName(Resource resource) {
+ return (String) resource.getPropertyValue(getHostNamePropertyId());
+ }
+
+ @Override
+ protected String getComponentName(Resource resource) {
+ return (String) resource.getPropertyValue(getComponentNamePropertyId());
+ }
+
+ @Override
+ protected Set<String> getGangliaClusterNames(Resource resource, String clusterName) {
+ String component = getComponentName(resource);
+
+ return new HashSet<String>(GANGLIA_CLUSTER_NAME_MAP.containsKey(component) ?
+ GANGLIA_CLUSTER_NAME_MAP.get(component) :
+ Collections.<String>emptyList());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostPropertyProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostPropertyProvider.java
new file mode 100644
index 0000000..df58ee5
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaHostPropertyProvider.java
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.controller.metrics.ganglia;
+
+import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
+import org.apache.ambari.server.controller.internal.PropertyInfo;
+import org.apache.ambari.server.controller.metrics.MetricHostProvider;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.utilities.StreamProvider;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Ganglia property provider implementation for host resources.
+ */
+public class GangliaHostPropertyProvider extends GangliaPropertyProvider {
+
+ /**
+ * Set of Ganglia cluster names.
+ */
+ private static final Set<String> GANGLIA_CLUSTER_NAMES = new HashSet<String>();
+
+ static {
+ GANGLIA_CLUSTER_NAMES.add("HDPNameNode");
+ GANGLIA_CLUSTER_NAMES.add("HDPSlaves");
+ GANGLIA_CLUSTER_NAMES.add("HDPJobTracker");
+ GANGLIA_CLUSTER_NAMES.add("HDPResourceManager");
+ GANGLIA_CLUSTER_NAMES.add("HDPHBaseMaster");
+ GANGLIA_CLUSTER_NAMES.add("HDPHistoryServer");
+ GANGLIA_CLUSTER_NAMES.add("HDPNameNode");
+ GANGLIA_CLUSTER_NAMES.add("HDPTaskTracker");
+ GANGLIA_CLUSTER_NAMES.add("HDPHBaseRegionServer");
+ GANGLIA_CLUSTER_NAMES.add("HDPFlumeServer");
+ GANGLIA_CLUSTER_NAMES.add("HDPJournalNode");
+ }
+
+ // ----- Constructors ------------------------------------------------------
+
+ public GangliaHostPropertyProvider(Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
+ StreamProvider streamProvider,
+ ComponentSSLConfiguration configuration,
+ MetricHostProvider hostProvider,
+ String clusterNamePropertyId,
+ String hostNamePropertyId) {
+
+ super(componentPropertyInfoMap, streamProvider, configuration, hostProvider,
+ clusterNamePropertyId, hostNamePropertyId, null);
+ }
+
+
+ // ----- GangliaPropertyProvider -------------------------------------------
+
+ @Override
+ protected String getHostName(Resource resource) {
+ return (String) resource.getPropertyValue(getHostNamePropertyId());
+ }
+
+ @Override
+ protected String getComponentName(Resource resource) {
+ return "*";
+ }
+
+ @Override
+ protected Set<String> getGangliaClusterNames(Resource resource, String clusterName) {
+ return GANGLIA_CLUSTER_NAMES;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/a52f8a55/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaMetric.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaMetric.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaMetric.java
new file mode 100644
index 0000000..cca685c
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/ganglia/GangliaMetric.java
@@ -0,0 +1,257 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ambari.server.controller.metrics.ganglia;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+@JsonIgnoreProperties(ignoreUnknown = true)
+
+
+/**
+ * Data structure for temporal data returned from Ganglia Web.
+ */
+public class GangliaMetric {
+
+ // Note that the member names correspond to the names in the JSON returned from Ganglia Web.
+
+ /**
+ * The name.
+ */
+ private String ds_name;
+
+ /**
+ * The ganglia cluster name.
+ */
+ private String cluster_name;
+
+ /**
+ * The graph type.
+ */
+ private String graph_type;
+
+ /**
+ * The host name.
+ */
+ private String host_name;
+
+ /**
+ * The metric name.
+ */
+ private String metric_name;
+
+ /**
+ * The temporal data points.
+ */
+ private Number[][] datapoints;
+
+
+ private static final Set<String> PERCENTAGE_METRIC;
+
+ //BUG-3386 Cluster CPU Chart is off the charts
+ // Here can be added other percentage metrics
+ static {
+ Set<String> temp = new HashSet<String>();
+ temp.add("cpu_wio");
+ temp.add("cpu_idle");
+ temp.add("cpu_nice");
+ temp.add("cpu_aidle");
+ temp.add("cpu_system");
+ temp.add("cpu_user");
+ PERCENTAGE_METRIC = Collections.unmodifiableSet(temp);
+ }
+
+
+ // ----- GangliaMetric -----------------------------------------------------
+
+ public String getDs_name() {
+ return ds_name;
+ }
+
+ public void setDs_name(String ds_name) {
+ this.ds_name = ds_name;
+ }
+
+ public String getCluster_name() {
+ return cluster_name;
+ }
+
+ public void setCluster_name(String cluster_name) {
+ this.cluster_name = cluster_name;
+ }
+
+ public String getGraph_type() {
+ return graph_type;
+ }
+
+ public void setGraph_type(String graph_type) {
+ this.graph_type = graph_type;
+ }
+
+ public String getHost_name() {
+ return host_name;
+ }
+
+ public void setHost_name(String host_name) {
+ this.host_name = host_name;
+ }
+
+ public String getMetric_name() {
+ return metric_name;
+ }
+
+ public void setMetric_name(String metric_name) {
+ this.metric_name = metric_name;
+ }
+
+ public Number[][] getDatapoints() {
+ return datapoints;
+ }
+
+
+ public void setDatapoints(Number[][] datapoints) {
+ this.datapoints = datapoints;
+ }
+
+ public void setDatapointsFromList(List<GangliaMetric.TemporalMetric> listTemporalMetrics) {
+ //this.datapoints = datapoints;
+ Number[][] datapointsArray = new Number[listTemporalMetrics.size()][2];
+ int cnt = 0;
+ if (PERCENTAGE_METRIC.contains(metric_name)) {
+ int firstIndex = 0;
+ int lastIndex = listTemporalMetrics.size() - 1;
+ for (int i = firstIndex; i <= lastIndex; ++i) {
+ GangliaMetric.TemporalMetric m = listTemporalMetrics.get(i);
+ Number val = m.getValue();
+ if (100.0 >= val.doubleValue()) {
+ datapointsArray[cnt][0] = val;
+ datapointsArray[cnt][1] = m.getTime();
+ cnt++;
+ }
+ }
+ } else {
+ int firstIndex = 0;
+ int lastIndex = listTemporalMetrics.size() - 1;
+ for (int i = firstIndex; i <= lastIndex; ++i) {
+ GangliaMetric.TemporalMetric m = listTemporalMetrics.get(i);
+ datapointsArray[i][0] = m.getValue();
+ datapointsArray[i][1] = m.getTime();
+ cnt++;
+ }
+ }
+
+ this.datapoints = new Number[cnt][2];
+ for (int i = 0; i < this.datapoints.length; i++) {
+ this.datapoints[i][0] = datapointsArray[i][0];
+ this.datapoints[i][1] = datapointsArray[i][1];
+ }
+
+ }
+
+ // ----- Object overrides --------------------------------------------------
+
+ @Override
+ public String toString() {
+ StringBuilder stringBuilder = new StringBuilder();
+
+ stringBuilder.append("\n");
+ stringBuilder.append("name=");
+ stringBuilder.append(ds_name);
+ stringBuilder.append("\n");
+ stringBuilder.append("cluster name=");
+ stringBuilder.append(cluster_name);
+ stringBuilder.append("\n");
+ stringBuilder.append("graph type=");
+ stringBuilder.append(graph_type);
+ stringBuilder.append("\n");
+ stringBuilder.append("host name=");
+ stringBuilder.append(host_name);
+ stringBuilder.append("\n");
+ stringBuilder.append("api name=");
+ stringBuilder.append(metric_name);
+ stringBuilder.append("\n");
+
+ stringBuilder.append("datapoints (value/timestamp):");
+ stringBuilder.append("\n");
+
+
+ boolean first = true;
+ stringBuilder.append("[");
+ for (Number[] m : datapoints) {
+ if (!first) {
+ stringBuilder.append(",");
+ }
+ stringBuilder.append("[");
+ stringBuilder.append(m[0]);
+ stringBuilder.append(",");
+ stringBuilder.append(m[1].longValue());
+ stringBuilder.append("]");
+ first = false;
+ }
+ stringBuilder.append("]");
+
+ return stringBuilder.toString();
+ }
+
+ public static class TemporalMetric {
+ private Number m_value;
+ private Number m_time;
+ private boolean valid;
+
+ public boolean isValid() {
+ return valid;
+ }
+
+ public TemporalMetric(String value, Number time) {
+ valid = true;
+ try{
+ m_value = convertToNumber(value);
+ } catch (NumberFormatException e) {
+ valid = false;
+ }
+ m_time = time;
+ }
+
+ public Number getValue() {
+ return m_value;
+ }
+
+ public Number getTime() {
+ return m_time;
+ }
+
+ private Number convertToNumber(String s) throws NumberFormatException {
+ Number res;
+ if(s.contains(".")){
+ Double d = Double.parseDouble(s);
+ if(d.isNaN() || d.isInfinite()){
+ throw new NumberFormatException(s);
+ } else {
+ res = d;
+ }
+ } else {
+ res = Long.parseLong(s);
+ }
+ return res;
+ }
+
+ }
+}