Repository: ambari Updated Branches: refs/heads/trunk d6fbf4694 -> e74fc95b8
AMBARI-5974 Customize rrdcached configuration to improve io operations (dsen) Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/e74fc95b Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/e74fc95b Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/e74fc95b Branch: refs/heads/trunk Commit: e74fc95b838c0a9e024a32cab9495c1dc6c90e4e Parents: d6fbf46 Author: Dmitry Sen <[email protected]> Authored: Fri May 30 22:02:34 2014 +0300 Committer: Dmitry Sen <[email protected]> Committed: Fri May 30 22:25:45 2014 +0300 ---------------------------------------------------------------------- .../ganglia/GangliaPropertyProvider.java | 29 ++- .../controller/jmx/JMXPropertyProvider.java | 24 +-- .../nagios/NagiosPropertyProvider.java | 13 +- .../services/GANGLIA/configuration/global.xml | 15 ++ .../1.3.2/services/GANGLIA/package/files/rrd.py | 215 ------------------- .../GANGLIA/package/files/startRrdcached.sh | 8 +- .../GANGLIA/package/scripts/ganglia_server.py | 7 +- .../services/GANGLIA/package/scripts/params.py | 3 + .../GANGLIA/package/templates/gangliaLib.sh.j2 | 3 + .../GANGLIA/package/templates/rrd.py.j2 | 215 +++++++++++++++++++ .../services/GANGLIA/configuration/global.xml | 15 ++ .../2.0.6/services/GANGLIA/package/files/rrd.py | 215 ------------------- .../GANGLIA/package/files/startRrdcached.sh | 9 +- .../GANGLIA/package/scripts/ganglia_server.py | 7 +- .../services/GANGLIA/package/scripts/params.py | 3 + .../GANGLIA/package/templates/gangliaLib.sh.j2 | 3 + .../GANGLIA/package/templates/rrd.py.j2 | 215 +++++++++++++++++++ .../stacks/1.3.2/GANGLIA/test_ganglia_server.py | 5 +- .../stacks/2.0.6/GANGLIA/test_ganglia_server.py | 5 +- 19 files changed, 521 insertions(+), 488 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/java/org/apache/ambari/server/controller/ganglia/GangliaPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/ganglia/GangliaPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/ganglia/GangliaPropertyProvider.java index a0c9b3c..4d623c1 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/ganglia/GangliaPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/ganglia/GangliaPropertyProvider.java @@ -39,6 +39,8 @@ import java.util.regex.Pattern; */ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { + private static final Pattern questionMarkPattern = Pattern.compile("\\?"); + private final StreamProvider streamProvider; private final GangliaHostProvider hostProvider; @@ -58,19 +60,19 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { static { - GANGLIA_CLUSTER_NAME_MAP.put("NAMENODE", Arrays.asList("HDPNameNode")); + GANGLIA_CLUSTER_NAME_MAP.put("NAMENODE", Collections.singletonList("HDPNameNode")); GANGLIA_CLUSTER_NAME_MAP.put("DATANODE", Arrays.asList("HDPDataNode", "HDPSlaves")); - GANGLIA_CLUSTER_NAME_MAP.put("JOBTRACKER", Arrays.asList("HDPJobTracker")); + GANGLIA_CLUSTER_NAME_MAP.put("JOBTRACKER", Collections.singletonList("HDPJobTracker")); GANGLIA_CLUSTER_NAME_MAP.put("TASKTRACKER", Arrays.asList("HDPTaskTracker", "HDPSlaves")); - GANGLIA_CLUSTER_NAME_MAP.put("RESOURCEMANAGER", Arrays.asList("HDPResourceManager")); + GANGLIA_CLUSTER_NAME_MAP.put("RESOURCEMANAGER", Collections.singletonList("HDPResourceManager")); GANGLIA_CLUSTER_NAME_MAP.put("NODEMANAGER", Arrays.asList("HDPNodeManager", "HDPSlaves")); - GANGLIA_CLUSTER_NAME_MAP.put("HISTORYSERVER", Arrays.asList("HDPHistoryServer")); - GANGLIA_CLUSTER_NAME_MAP.put("HBASE_MASTER", Arrays.asList("HDPHBaseMaster")); + GANGLIA_CLUSTER_NAME_MAP.put("HISTORYSERVER", Collections.singletonList("HDPHistoryServer")); + GANGLIA_CLUSTER_NAME_MAP.put("HBASE_MASTER", Collections.singletonList("HDPHBaseMaster")); GANGLIA_CLUSTER_NAME_MAP.put("HBASE_REGIONSERVER", Arrays.asList("HDPHBaseRegionServer", "HDPSlaves")); GANGLIA_CLUSTER_NAME_MAP.put("FLUME_HANDLER", Arrays.asList("HDPFlumeServer", "HDPSlaves")); GANGLIA_CLUSTER_NAME_MAP.put("JOURNALNODE", Arrays.asList("HDPJournalNode", "HDPSlaves")); - GANGLIA_CLUSTER_NAME_MAP.put("NIMBUS", Arrays.asList("HDPNimbus")); - GANGLIA_CLUSTER_NAME_MAP.put("SUPERVISOR", Arrays.asList("HDPSupervisor")); + GANGLIA_CLUSTER_NAME_MAP.put("NIMBUS", Collections.singletonList("HDPNimbus")); + GANGLIA_CLUSTER_NAME_MAP.put("SUPERVISOR", Collections.singletonList("HDPSupervisor")); } protected final static Logger LOG = @@ -108,8 +110,6 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { return resources; } - Set<Resource> keepers = new HashSet<Resource>(); - Map<String, Map<TemporalInfo, RRDRequest>> requestMap = getRRDRequests(resources, request, ids); // For each cluster... @@ -117,10 +117,9 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { // For each request ... for (RRDRequest rrdRequest : clusterEntry.getValue().values() ) { //todo: property provider can reduce set of resources - keepers.addAll(rrdRequest.populateResources()); + rrdRequest.populateResources(); } } - //todo: ignoring keepers returned by the provider return resources; } @@ -267,7 +266,7 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { * * @return the spec, like http://example.com/path?param1=val1¶mn=valn * - * @throws SystemException if unable to get the Ganglia Collector host name + * @throws org.apache.ambari.server.controller.spi.SystemException if unable to get the Ganglia Collector host name */ private String getSpec(String clusterName, Set<String> clusterSet, @@ -364,7 +363,7 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { if (limit == -1 || set.size() <= limit) { for (String cluster : set) { if (sb.length() > 0) { - sb.append(","); + sb.append(','); } sb.append(cluster); } @@ -422,7 +421,7 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { * * @return a collection of populated resources * - * @throws SystemException if unable to populate the resources + * @throws org.apache.ambari.server.controller.spi.SystemException if unable to populate the resources */ public Collection<Resource> populateResources() throws SystemException { @@ -434,7 +433,7 @@ public abstract class GangliaPropertyProvider extends AbstractPropertyProvider { //Parameters String params = null; - String[] tokens = specWithParams.split("\\?", 2); + String[] tokens = questionMarkPattern.split(specWithParams, 2); try { spec = tokens[0]; http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/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 927eb00..a4dc2b2 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 @@ -46,7 +46,6 @@ 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.JsonFactory; import org.codehaus.jackson.map.DeserializationConfig; import org.codehaus.jackson.map.ObjectMapper; import org.codehaus.jackson.map.ObjectReader; @@ -113,8 +112,6 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { jmxObjectMapper.configure(DeserializationConfig.Feature.USE_ANNOTATIONS, false); jmxObjectReader = jmxObjectMapper.reader(JMXMetricHolder.class); - JsonFactory factory = new JsonFactory(); - ObjectMapper stormObjectMapper = new ObjectMapper(factory); TypeReference<HashMap<String,Object>> typeRef = new TypeReference< HashMap<String,Object> @@ -125,6 +122,9 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { protected final static Logger LOG = LoggerFactory.getLogger(JMXPropertyProvider.class); + private static final Pattern dotReplacementCharPattern = + Pattern.compile(DOT_REPLACEMENT_CHAR); + private final StreamProvider streamProvider; private final JMXHostProvider jmxHostProvider; @@ -347,7 +347,7 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { if (null == componentName || !componentName.equals(STORM_REST_API)) { getHadoopMetricValue(in, ids, resource, request); } else { - getStormMetricValue(in, ids, resource, request); + getStormMetricValue(in, ids, resource); } } finally { @@ -391,7 +391,6 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { String property = propertyInfo.getPropertyId(); String category = ""; - List<String> keyList = new LinkedList<String>(); int keyStartIndex = property.indexOf('['); @@ -450,13 +449,13 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { * Storm-specific metrics fetching */ private void getStormMetricValue(InputStream in, Set<String> ids, - Resource resource, Request request) throws IOException { + Resource resource) throws IOException { HashMap<String, Object> metricHolder = stormObjectReader.readValue(in); for (String category : ids) { Map<String, PropertyInfo> defProps = getComponentMetrics().get(STORM_REST_API); - for (String depProp : defProps.keySet()) { - if (depProp.startsWith(category)) { - PropertyInfo propInfo = defProps.get(depProp); + for (Map.Entry<String, PropertyInfo> depEntry : defProps.entrySet()) { + if (depEntry.getKey().startsWith(category)) { + PropertyInfo propInfo = depEntry.getValue(); String propName = propInfo.getPropertyId(); Object propertyValue = metricHolder.get(propName); String absId = PropertyHelper.getPropertyId(category, propName); @@ -471,7 +470,7 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { String category, String property, List<String> keyList) { Map<String, Object> properties = categories.get(category); if (property.contains(DOT_REPLACEMENT_CHAR)) { - property = property.replaceAll(DOT_REPLACEMENT_CHAR, "."); + property = dotReplacementCharPattern.matcher(property).replaceAll("."); } if (properties != null && properties.containsKey(property)) { Object value = properties.get(property); @@ -497,8 +496,7 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { } private String getJMXProtocol(String clusterName, String componentName) { - String protocol = jmxHostProvider.getJMXProtocol(clusterName, componentName); - return protocol; + return jmxHostProvider.getJMXProtocol(clusterName, componentName); } private String getHost(Resource resource, String clusterName, String componentName) throws SystemException { @@ -548,7 +546,7 @@ public class JMXPropertyProvider extends AbstractPropertyProvider { * * @param throwable the caught exception * - * @throws SystemException always around the given exception + * @throws org.apache.ambari.server.controller.spi.SystemException always around the given exception */ private static void rethrowSystemException(Throwable throwable) throws SystemException { String msg = logException(throwable); http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/java/org/apache/ambari/server/controller/nagios/NagiosPropertyProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/nagios/NagiosPropertyProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/nagios/NagiosPropertyProvider.java index 52e84e0..ba7309a 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/nagios/NagiosPropertyProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/nagios/NagiosPropertyProvider.java @@ -23,11 +23,9 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; -import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashMap; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; @@ -39,6 +37,7 @@ import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; import org.apache.ambari.server.AmbariException; import org.apache.ambari.server.configuration.ComponentSSLConfiguration; @@ -84,7 +83,9 @@ public class NagiosPropertyProvider extends BaseProvider implements PropertyProv "ZooKeeper Server process", "Supervisors process")); private static final List<String> IGNORABLE_FOR_HOSTS = new ArrayList<String>( - Arrays.asList("percent")); + Collections.singletonList("percent")); + + private static final Pattern COMMA_PATTERN = Pattern.compile(","); // holds alerts for clusters. clusterName is the key private static final Map<String, List<NagiosAlert>> CLUSTER_ALERTS = @@ -120,14 +121,12 @@ public class NagiosPropertyProvider extends BaseProvider implements PropertyProv String ignores = config.getProperty(Configuration.NAGIOS_IGNORE_FOR_SERVICES_KEY); if (null != ignores) { - for (String str : ignores.split(",")) - IGNORABLE_FOR_SERVICES.add(str); + Collections.addAll(IGNORABLE_FOR_SERVICES, COMMA_PATTERN.split(ignores)); } ignores = config.getProperty(Configuration.NAGIOS_IGNORE_FOR_HOSTS_KEY); if (null != ignores) { - for (String str : ignores.split(",")) - IGNORABLE_FOR_HOSTS.add(str); + Collections.addAll(IGNORABLE_FOR_HOSTS, COMMA_PATTERN.split(ignores)); } } http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/configuration/global.xml ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/configuration/global.xml b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/configuration/global.xml index 49d38ae..a1dd4bb 100644 --- a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/configuration/global.xml +++ b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/configuration/global.xml @@ -46,5 +46,20 @@ <value>/var/lib/ganglia/rrds</value> <description>Default directory for saving the rrd files on ganglia server</description> </property> + <property> + <name>rrdcached_timeout</name> + <value>3600</value> + <description>(-w) Data is written to disk every timeout seconds. If this option is not specified the default interval of 300 seconds will be used.</description> + </property> + <property> + <name>rrdcached_delay</name> + <value>1800</value> + <description>(-z) If specified, rrdcached will delay writing of each RRD for a random number of seconds in the range [0,delay). This will avoid too many writes being queued simultaneously. This value should be no greater than the value specified in -w. By default, there is no delay.</description> + </property> + <property> + <name>rrdcached_write_threads</name> + <value>10</value> + <description>(-t) Specifies the number of threads used for writing RRD files. The default is 4. Increasing this number will allow rrdcached to have more simultaneous I/O requests into the kernel. This may allow the kernel to re-order disk writes, resulting in better disk throughput.</description> + </property> </configuration> http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/rrd.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/rrd.py b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/rrd.py deleted file mode 100644 index 2d32af9..0000000 --- a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/rrd.py +++ /dev/null @@ -1,215 +0,0 @@ -#!/usr/bin/env python - -''' -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. -''' - -import cgi -import os -import rrdtool -import sys -import time -import re -import urlparse - -# place this script in /var/www/cgi-bin of the Ganglia collector -# requires 'yum install rrdtool-python' on the Ganglia collector - - -def printMetric(clusterName, hostName, metricName, file, cf, start, end, - resolution, pointInTime): - if clusterName.endswith("rrds"): - clusterName = "" - - args = [file, cf] - - if start is not None: - args.extend(["-s", start]) - else: - args.extend(["-s", "now-10m"]) - - if end is not None: - args.extend(["-e", end]) - - if resolution is not None: - args.extend(["-r", resolution]) - - rrdMetric = rrdtool.fetch(args) - # ds_name - sys.stdout.write(rrdMetric[1][0]) - sys.stdout.write("\n") - - sys.stdout.write(clusterName) - sys.stdout.write("\n") - sys.stdout.write(hostName) - sys.stdout.write("\n") - sys.stdout.write(metricName) - sys.stdout.write("\n") - - # write time - sys.stdout.write(str(rrdMetric[0][0])) - sys.stdout.write("\n") - # write step - sys.stdout.write(str(rrdMetric[0][2])) - sys.stdout.write("\n") - - if not pointInTime: - valueCount = 0 - lastValue = None - - for tuple in rrdMetric[2]: - - thisValue = tuple[0] - - if valueCount > 0 and thisValue == lastValue: - valueCount += 1 - else: - if valueCount > 1: - sys.stdout.write("[~r]") - sys.stdout.write(str(valueCount)) - sys.stdout.write("\n") - - if thisValue is None: - sys.stdout.write("[~n]\n") - else: - sys.stdout.write(str(thisValue)) - sys.stdout.write("\n") - - valueCount = 1 - lastValue = thisValue - else: - value = None - idx = -1 - tuple = rrdMetric[2] - tupleLastIdx = len(tuple) * -1 - - while value is None and idx >= tupleLastIdx: - value = tuple[idx][0] - idx -= 1 - - if value is not None: - sys.stdout.write(str(value)) - sys.stdout.write("\n") - - sys.stdout.write("[~EOM]\n") - return - - -def stripList(l): - return ([x.strip() for x in l]) - - -sys.stdout.write("Content-type: text/plain\n\n") - -# write start time -sys.stdout.write(str(time.mktime(time.gmtime()))) -sys.stdout.write("\n") - -requestMethod = os.environ['REQUEST_METHOD'] - -if requestMethod == 'POST': - postData = sys.stdin.readline() - queryString = cgi.parse_qs(postData) - queryString = dict((k, v[0]) for k, v in queryString.items()) -elif requestMethod == 'GET': - queryString = dict(cgi.parse_qsl(os.environ['QUERY_STRING'])); - -if "m" in queryString: - metricParts = queryString["m"].split(",") -else: - metricParts = [""] -metricParts = stripList(metricParts) - -hostParts = [] -if "h" in queryString: - hostParts = queryString["h"].split(",") -hostParts = stripList(hostParts) - -if "c" in queryString: - clusterParts = queryString["c"].split(",") -else: - clusterParts = [""] -clusterParts = stripList(clusterParts) - -if "p" in queryString: - rrdPath = queryString["p"] -else: - rrdPath = "/var/lib/ganglia/rrds/" - -start = None -if "s" in queryString: - start = queryString["s"] - -end = None -if "e" in queryString: - end = queryString["e"] - -resolution = None -if "r" in queryString: - resolution = queryString["r"] - -if "cf" in queryString: - cf = queryString["cf"] -else: - cf = "AVERAGE" - -if "pt" in queryString: - pointInTime = True -else: - pointInTime = False - - -def _walk(*args, **kwargs): - for root, dirs, files in os.walk(*args, **kwargs): - for dir in dirs: - qualified_dir = os.path.join(root, dir) - if os.path.islink(qualified_dir): - for x in os.walk(qualified_dir, **kwargs): - yield x - yield (root, dirs, files) - - -for cluster in clusterParts: - for path, dirs, files in _walk(rrdPath + cluster): - pathParts = path.split("/") - #Process only path which contains files. If no host parameter passed - process all hosts folders and summary info - #If host parameter passed - process only this host folder - if len(files) > 0 and (len(hostParts) == 0 or pathParts[-1] in hostParts): - for metric in metricParts: - file = metric + ".rrd" - fileFullPath = os.path.join(path, file) - if os.path.exists(fileFullPath): - #Exact name of metric - printMetric(pathParts[-2], pathParts[-1], file[:-4], - os.path.join(path, file), cf, start, end, resolution, - pointInTime) - else: - #Regex as metric name - metricRegex = metric + '\.rrd$' - p = re.compile(metricRegex) - matchedFiles = filter(p.match, files) - for matchedFile in matchedFiles: - printMetric(pathParts[-2], pathParts[-1], matchedFile[:-4], - os.path.join(path, matchedFile), cf, start, end, - resolution, pointInTime) - -sys.stdout.write("[~EOF]\n") -# write end time -sys.stdout.write(str(time.mktime(time.gmtime()))) -sys.stdout.write("\n") - -sys.stdout.flush http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/startRrdcached.sh ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/startRrdcached.sh b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/startRrdcached.sh index e79472b..e44af17 100644 --- a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/startRrdcached.sh +++ b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/files/startRrdcached.sh @@ -31,15 +31,11 @@ rrdcachedRunningPid=`getRrdcachedRunningPid`; # Only attempt to start rrdcached if there's not already one running. if [ -z "${rrdcachedRunningPid}" ] then - #changed because problem puppet had with nobody user - #sudo -u ${GMETAD_USER} ${RRDCACHED_BIN} -p ${RRDCACHED_PID_FILE} \ - # -m 664 -l unix:${RRDCACHED_ALL_ACCESS_UNIX_SOCKET} \ - # -m 777 -P FLUSH,STATS,HELP -l unix:${RRDCACHED_LIMITED_ACCESS_UNIX_SOCKET} \ - # -b /var/lib/ganglia/rrds -B su - ${GMETAD_USER} -c "${RRDCACHED_BIN} -p ${RRDCACHED_PID_FILE} \ -m 664 -l unix:${RRDCACHED_ALL_ACCESS_UNIX_SOCKET} \ -m 777 -P FLUSH,STATS,HELP -l unix:${RRDCACHED_LIMITED_ACCESS_UNIX_SOCKET} \ - -b ${RRDCACHED_BASE_DIR} -B" + -b ${RRDCACHED_BASE_DIR} -B -t ${RRDCACHED_WRITE_THREADS} \ + -w ${RRDCACHED_TIMEOUT} -z ${RRDCACHED_DELAY} -F" # Ideally, we'd use ${RRDCACHED_BIN}'s -s ${WEBSERVER_GROUP} option for # this, but it doesn't take sometimes due to a lack of permissions, http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/ganglia_server.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/ganglia_server.py b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/ganglia_server.py index 54541b5..c2b647c 100644 --- a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/ganglia_server.py +++ b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/ganglia_server.py @@ -96,9 +96,10 @@ def server_files(): recursive=True ) rrd_py_file_path = path.join(rrd_py_path, "rrd.py") - File(rrd_py_file_path, - content=StaticFile("rrd.py"), - mode=0755 + TemplateConfig(rrd_py_file_path, + owner="root", + group="root", + mode=0755 ) rrd_file_owner = params.gmetad_user http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/params.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/params.py b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/params.py index ef13dc8..030181f 100644 --- a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/params.py +++ b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/scripts/params.py @@ -33,6 +33,9 @@ gmond_user = config['configurations']['global']["gmond_user"] webserver_group = "apache" rrdcached_default_base_dir = "/var/lib/ganglia/rrds" rrdcached_base_dir = config['configurations']['global']["rrdcached_base_dir"] +rrdcached_timeout = default("/configurations/global/rrdcached_timeout", 3600) +rrdcached_delay = default("/configurations/global/rrdcached_delay", 1800) +rrdcached_write_threads = default("/configurations/global/rrdcached_write_threads", 10) ganglia_server_host = config["clusterHostInfo"]["ganglia_server_host"][0] http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/gangliaLib.sh.j2 ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/gangliaLib.sh.j2 b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/gangliaLib.sh.j2 index c0cd66a..f066b57 100644 --- a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/gangliaLib.sh.j2 +++ b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/gangliaLib.sh.j2 @@ -41,6 +41,9 @@ cd `dirname ${0}`; GANGLIA_CONF_DIR={{ganglia_conf_dir}}; GANGLIA_RUNTIME_DIR={{ganglia_runtime_dir}}; RRDCACHED_BASE_DIR={{rrdcached_base_dir}}; +RRDCACHED_WRITE_THREADS={{rrdcached_write_threads}} +RRDCACHED_TIMEOUT={{rrdcached_timeout}} +RRDCACHED_DELAY={{rrdcached_delay}} # This file contains all the info about each Ganglia Cluster in our Grid. GANGLIA_CLUSTERS_CONF_FILE=./gangliaClusters.conf; http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/rrd.py.j2 ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/rrd.py.j2 b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/rrd.py.j2 new file mode 100644 index 0000000..f6d6bd0 --- /dev/null +++ b/ambari-server/src/main/resources/stacks/HDP/1.3.2/services/GANGLIA/package/templates/rrd.py.j2 @@ -0,0 +1,215 @@ +#!/usr/bin/env python + +''' +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. +''' + +import cgi +import os +import rrdtool +import sys +import time +import re +import urlparse + +# place this script in /var/www/cgi-bin of the Ganglia collector +# requires 'yum install rrdtool-python' on the Ganglia collector + + +def printMetric(clusterName, hostName, metricName, file, cf, start, end, + resolution, pointInTime): + if clusterName.endswith("rrds"): + clusterName = "" + + args = [file, cf, "--daemon", "unix:{{ganglia_runtime_dir}}/rrdcached.sock"] + + if start is not None: + args.extend(["-s", start]) + else: + args.extend(["-s", "now-10m"]) + + if end is not None: + args.extend(["-e", end]) + + if resolution is not None: + args.extend(["-r", resolution]) + + rrdMetric = rrdtool.fetch(args) + # ds_name + sys.stdout.write(rrdMetric[1][0]) + sys.stdout.write("\n") + + sys.stdout.write(clusterName) + sys.stdout.write("\n") + sys.stdout.write(hostName) + sys.stdout.write("\n") + sys.stdout.write(metricName) + sys.stdout.write("\n") + + # write time + sys.stdout.write(str(rrdMetric[0][0])) + sys.stdout.write("\n") + # write step + sys.stdout.write(str(rrdMetric[0][2])) + sys.stdout.write("\n") + + if not pointInTime: + valueCount = 0 + lastValue = None + + for tuple in rrdMetric[2]: + + thisValue = tuple[0] + + if valueCount > 0 and thisValue == lastValue: + valueCount += 1 + else: + if valueCount > 1: + sys.stdout.write("[~r]") + sys.stdout.write(str(valueCount)) + sys.stdout.write("\n") + + if thisValue is None: + sys.stdout.write("[~n]\n") + else: + sys.stdout.write(str(thisValue)) + sys.stdout.write("\n") + + valueCount = 1 + lastValue = thisValue + else: + value = None + idx = -1 + tuple = rrdMetric[2] + tupleLastIdx = len(tuple) * -1 + + while value is None and idx >= tupleLastIdx: + value = tuple[idx][0] + idx -= 1 + + if value is not None: + sys.stdout.write(str(value)) + sys.stdout.write("\n") + + sys.stdout.write("[~EOM]\n") + return + + +def stripList(l): + return ([x.strip() for x in l]) + + +sys.stdout.write("Content-type: text/plain\n\n") + +# write start time +sys.stdout.write(str(time.mktime(time.gmtime()))) +sys.stdout.write("\n") + +requestMethod = os.environ['REQUEST_METHOD'] + +if requestMethod == 'POST': + postData = sys.stdin.readline() + queryString = cgi.parse_qs(postData) + queryString = dict((k, v[0]) for k, v in queryString.items()) +elif requestMethod == 'GET': + queryString = dict(cgi.parse_qsl(os.environ['QUERY_STRING'])); + +if "m" in queryString: + metricParts = queryString["m"].split(",") +else: + metricParts = [""] +metricParts = stripList(metricParts) + +hostParts = [] +if "h" in queryString: + hostParts = queryString["h"].split(",") +hostParts = stripList(hostParts) + +if "c" in queryString: + clusterParts = queryString["c"].split(",") +else: + clusterParts = [""] +clusterParts = stripList(clusterParts) + +if "p" in queryString: + rrdPath = queryString["p"] +else: + rrdPath = "/var/lib/ganglia/rrds/" + +start = None +if "s" in queryString: + start = queryString["s"] + +end = None +if "e" in queryString: + end = queryString["e"] + +resolution = None +if "r" in queryString: + resolution = queryString["r"] + +if "cf" in queryString: + cf = queryString["cf"] +else: + cf = "AVERAGE" + +if "pt" in queryString: + pointInTime = True +else: + pointInTime = False + + +def _walk(*args, **kwargs): + for root, dirs, files in os.walk(*args, **kwargs): + for dir in dirs: + qualified_dir = os.path.join(root, dir) + if os.path.islink(qualified_dir): + for x in os.walk(qualified_dir, **kwargs): + yield x + yield (root, dirs, files) + + +for cluster in clusterParts: + for path, dirs, files in _walk(rrdPath + cluster): + pathParts = path.split("/") + #Process only path which contains files. If no host parameter passed - process all hosts folders and summary info + #If host parameter passed - process only this host folder + if len(files) > 0 and (len(hostParts) == 0 or pathParts[-1] in hostParts): + for metric in metricParts: + file = metric + ".rrd" + fileFullPath = os.path.join(path, file) + if os.path.exists(fileFullPath): + #Exact name of metric + printMetric(pathParts[-2], pathParts[-1], file[:-4], + os.path.join(path, file), cf, start, end, resolution, + pointInTime) + else: + #Regex as metric name + metricRegex = metric + '\.rrd$' + p = re.compile(metricRegex) + matchedFiles = filter(p.match, files) + for matchedFile in matchedFiles: + printMetric(pathParts[-2], pathParts[-1], matchedFile[:-4], + os.path.join(path, matchedFile), cf, start, end, + resolution, pointInTime) + +sys.stdout.write("[~EOF]\n") +# write end time +sys.stdout.write(str(time.mktime(time.gmtime()))) +sys.stdout.write("\n") + +sys.stdout.flush http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/configuration/global.xml ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/configuration/global.xml b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/configuration/global.xml index 49d38ae..a1dd4bb 100644 --- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/configuration/global.xml +++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/configuration/global.xml @@ -46,5 +46,20 @@ <value>/var/lib/ganglia/rrds</value> <description>Default directory for saving the rrd files on ganglia server</description> </property> + <property> + <name>rrdcached_timeout</name> + <value>3600</value> + <description>(-w) Data is written to disk every timeout seconds. If this option is not specified the default interval of 300 seconds will be used.</description> + </property> + <property> + <name>rrdcached_delay</name> + <value>1800</value> + <description>(-z) If specified, rrdcached will delay writing of each RRD for a random number of seconds in the range [0,delay). This will avoid too many writes being queued simultaneously. This value should be no greater than the value specified in -w. By default, there is no delay.</description> + </property> + <property> + <name>rrdcached_write_threads</name> + <value>10</value> + <description>(-t) Specifies the number of threads used for writing RRD files. The default is 4. Increasing this number will allow rrdcached to have more simultaneous I/O requests into the kernel. This may allow the kernel to re-order disk writes, resulting in better disk throughput.</description> + </property> </configuration> http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/rrd.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/rrd.py b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/rrd.py deleted file mode 100644 index 2d32af9..0000000 --- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/rrd.py +++ /dev/null @@ -1,215 +0,0 @@ -#!/usr/bin/env python - -''' -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. -''' - -import cgi -import os -import rrdtool -import sys -import time -import re -import urlparse - -# place this script in /var/www/cgi-bin of the Ganglia collector -# requires 'yum install rrdtool-python' on the Ganglia collector - - -def printMetric(clusterName, hostName, metricName, file, cf, start, end, - resolution, pointInTime): - if clusterName.endswith("rrds"): - clusterName = "" - - args = [file, cf] - - if start is not None: - args.extend(["-s", start]) - else: - args.extend(["-s", "now-10m"]) - - if end is not None: - args.extend(["-e", end]) - - if resolution is not None: - args.extend(["-r", resolution]) - - rrdMetric = rrdtool.fetch(args) - # ds_name - sys.stdout.write(rrdMetric[1][0]) - sys.stdout.write("\n") - - sys.stdout.write(clusterName) - sys.stdout.write("\n") - sys.stdout.write(hostName) - sys.stdout.write("\n") - sys.stdout.write(metricName) - sys.stdout.write("\n") - - # write time - sys.stdout.write(str(rrdMetric[0][0])) - sys.stdout.write("\n") - # write step - sys.stdout.write(str(rrdMetric[0][2])) - sys.stdout.write("\n") - - if not pointInTime: - valueCount = 0 - lastValue = None - - for tuple in rrdMetric[2]: - - thisValue = tuple[0] - - if valueCount > 0 and thisValue == lastValue: - valueCount += 1 - else: - if valueCount > 1: - sys.stdout.write("[~r]") - sys.stdout.write(str(valueCount)) - sys.stdout.write("\n") - - if thisValue is None: - sys.stdout.write("[~n]\n") - else: - sys.stdout.write(str(thisValue)) - sys.stdout.write("\n") - - valueCount = 1 - lastValue = thisValue - else: - value = None - idx = -1 - tuple = rrdMetric[2] - tupleLastIdx = len(tuple) * -1 - - while value is None and idx >= tupleLastIdx: - value = tuple[idx][0] - idx -= 1 - - if value is not None: - sys.stdout.write(str(value)) - sys.stdout.write("\n") - - sys.stdout.write("[~EOM]\n") - return - - -def stripList(l): - return ([x.strip() for x in l]) - - -sys.stdout.write("Content-type: text/plain\n\n") - -# write start time -sys.stdout.write(str(time.mktime(time.gmtime()))) -sys.stdout.write("\n") - -requestMethod = os.environ['REQUEST_METHOD'] - -if requestMethod == 'POST': - postData = sys.stdin.readline() - queryString = cgi.parse_qs(postData) - queryString = dict((k, v[0]) for k, v in queryString.items()) -elif requestMethod == 'GET': - queryString = dict(cgi.parse_qsl(os.environ['QUERY_STRING'])); - -if "m" in queryString: - metricParts = queryString["m"].split(",") -else: - metricParts = [""] -metricParts = stripList(metricParts) - -hostParts = [] -if "h" in queryString: - hostParts = queryString["h"].split(",") -hostParts = stripList(hostParts) - -if "c" in queryString: - clusterParts = queryString["c"].split(",") -else: - clusterParts = [""] -clusterParts = stripList(clusterParts) - -if "p" in queryString: - rrdPath = queryString["p"] -else: - rrdPath = "/var/lib/ganglia/rrds/" - -start = None -if "s" in queryString: - start = queryString["s"] - -end = None -if "e" in queryString: - end = queryString["e"] - -resolution = None -if "r" in queryString: - resolution = queryString["r"] - -if "cf" in queryString: - cf = queryString["cf"] -else: - cf = "AVERAGE" - -if "pt" in queryString: - pointInTime = True -else: - pointInTime = False - - -def _walk(*args, **kwargs): - for root, dirs, files in os.walk(*args, **kwargs): - for dir in dirs: - qualified_dir = os.path.join(root, dir) - if os.path.islink(qualified_dir): - for x in os.walk(qualified_dir, **kwargs): - yield x - yield (root, dirs, files) - - -for cluster in clusterParts: - for path, dirs, files in _walk(rrdPath + cluster): - pathParts = path.split("/") - #Process only path which contains files. If no host parameter passed - process all hosts folders and summary info - #If host parameter passed - process only this host folder - if len(files) > 0 and (len(hostParts) == 0 or pathParts[-1] in hostParts): - for metric in metricParts: - file = metric + ".rrd" - fileFullPath = os.path.join(path, file) - if os.path.exists(fileFullPath): - #Exact name of metric - printMetric(pathParts[-2], pathParts[-1], file[:-4], - os.path.join(path, file), cf, start, end, resolution, - pointInTime) - else: - #Regex as metric name - metricRegex = metric + '\.rrd$' - p = re.compile(metricRegex) - matchedFiles = filter(p.match, files) - for matchedFile in matchedFiles: - printMetric(pathParts[-2], pathParts[-1], matchedFile[:-4], - os.path.join(path, matchedFile), cf, start, end, - resolution, pointInTime) - -sys.stdout.write("[~EOF]\n") -# write end time -sys.stdout.write(str(time.mktime(time.gmtime()))) -sys.stdout.write("\n") - -sys.stdout.flush http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/startRrdcached.sh ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/startRrdcached.sh b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/startRrdcached.sh index 1ff0201..807c203 100644 --- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/startRrdcached.sh +++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/files/startRrdcached.sh @@ -31,15 +31,10 @@ rrdcachedRunningPid=`getRrdcachedRunningPid`; # Only attempt to start rrdcached if there's not already one running. if [ -z "${rrdcachedRunningPid}" ] then - #changed because problem puppet had with nobody user - #sudo -u ${GMETAD_USER} ${RRDCACHED_BIN} -p ${RRDCACHED_PID_FILE} \ - # -m 664 -l unix:${RRDCACHED_ALL_ACCESS_UNIX_SOCKET} \ - # -m 777 -P FLUSH,STATS,HELP -l unix:${RRDCACHED_LIMITED_ACCESS_UNIX_SOCKET} \ - # -b /var/lib/ganglia/rrds -B su - ${GMETAD_USER} -c "${RRDCACHED_BIN} -p ${RRDCACHED_PID_FILE} \ -m 664 -l unix:${RRDCACHED_ALL_ACCESS_UNIX_SOCKET} \ - -m 777 -P FLUSH,STATS,HELP -l unix:${RRDCACHED_LIMITED_ACCESS_UNIX_SOCKET} \ - -b ${RRDCACHED_BASE_DIR} -B" + -b ${RRDCACHED_BASE_DIR} -B -t ${RRDCACHED_WRITE_THREADS} \ + -w ${RRDCACHED_TIMEOUT} -z ${RRDCACHED_DELAY} -F" # Ideally, we'd use ${RRDCACHED_BIN}'s -s ${WEBSERVER_GROUP} option for # this, but it doesn't take sometimes due to a lack of permissions, http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/ganglia_server.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/ganglia_server.py b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/ganglia_server.py index 44eb1d3..5e86dbf 100644 --- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/ganglia_server.py +++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/ganglia_server.py @@ -96,9 +96,10 @@ def server_files(): recursive=True ) rrd_py_file_path = path.join(rrd_py_path, "rrd.py") - File(rrd_py_file_path, - content=StaticFile("rrd.py"), - mode=0755 + TemplateConfig(rrd_py_file_path, + owner="root", + group="root", + mode=0755 ) rrd_file_owner = params.gmetad_user http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/params.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/params.py b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/params.py index 3a5bcec..aae06e9 100644 --- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/params.py +++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/scripts/params.py @@ -39,6 +39,9 @@ gmond_apps = set(gmond_apps) & set(gmond_allowed_apps) webserver_group = "apache" rrdcached_default_base_dir = "/var/lib/ganglia/rrds" rrdcached_base_dir = config['configurations']['global']["rrdcached_base_dir"] +rrdcached_timeout = default("/configurations/global/rrdcached_timeout", 3600) +rrdcached_delay = default("/configurations/global/rrdcached_delay", 1800) +rrdcached_write_threads = default("/configurations/global/rrdcached_write_threads", 10) ganglia_server_host = config["clusterHostInfo"]["ganglia_server_host"][0] http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/gangliaLib.sh.j2 ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/gangliaLib.sh.j2 b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/gangliaLib.sh.j2 index 1da4f35..7d6bb98 100644 --- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/gangliaLib.sh.j2 +++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/gangliaLib.sh.j2 @@ -42,6 +42,9 @@ cd `dirname ${0}`; GANGLIA_CONF_DIR={{ganglia_conf_dir}}; GANGLIA_RUNTIME_DIR={{ganglia_runtime_dir}}; RRDCACHED_BASE_DIR={{rrdcached_base_dir}}; +RRDCACHED_WRITE_THREADS={{rrdcached_write_threads}} +RRDCACHED_TIMEOUT={{rrdcached_timeout}} +RRDCACHED_DELAY={{rrdcached_delay}} # This file contains all the info about each Ganglia Cluster in our Grid. GANGLIA_CLUSTERS_CONF_FILE=./gangliaClusters.conf; http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/rrd.py.j2 ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/rrd.py.j2 b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/rrd.py.j2 new file mode 100644 index 0000000..f6d6bd0 --- /dev/null +++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/GANGLIA/package/templates/rrd.py.j2 @@ -0,0 +1,215 @@ +#!/usr/bin/env python + +''' +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. +''' + +import cgi +import os +import rrdtool +import sys +import time +import re +import urlparse + +# place this script in /var/www/cgi-bin of the Ganglia collector +# requires 'yum install rrdtool-python' on the Ganglia collector + + +def printMetric(clusterName, hostName, metricName, file, cf, start, end, + resolution, pointInTime): + if clusterName.endswith("rrds"): + clusterName = "" + + args = [file, cf, "--daemon", "unix:{{ganglia_runtime_dir}}/rrdcached.sock"] + + if start is not None: + args.extend(["-s", start]) + else: + args.extend(["-s", "now-10m"]) + + if end is not None: + args.extend(["-e", end]) + + if resolution is not None: + args.extend(["-r", resolution]) + + rrdMetric = rrdtool.fetch(args) + # ds_name + sys.stdout.write(rrdMetric[1][0]) + sys.stdout.write("\n") + + sys.stdout.write(clusterName) + sys.stdout.write("\n") + sys.stdout.write(hostName) + sys.stdout.write("\n") + sys.stdout.write(metricName) + sys.stdout.write("\n") + + # write time + sys.stdout.write(str(rrdMetric[0][0])) + sys.stdout.write("\n") + # write step + sys.stdout.write(str(rrdMetric[0][2])) + sys.stdout.write("\n") + + if not pointInTime: + valueCount = 0 + lastValue = None + + for tuple in rrdMetric[2]: + + thisValue = tuple[0] + + if valueCount > 0 and thisValue == lastValue: + valueCount += 1 + else: + if valueCount > 1: + sys.stdout.write("[~r]") + sys.stdout.write(str(valueCount)) + sys.stdout.write("\n") + + if thisValue is None: + sys.stdout.write("[~n]\n") + else: + sys.stdout.write(str(thisValue)) + sys.stdout.write("\n") + + valueCount = 1 + lastValue = thisValue + else: + value = None + idx = -1 + tuple = rrdMetric[2] + tupleLastIdx = len(tuple) * -1 + + while value is None and idx >= tupleLastIdx: + value = tuple[idx][0] + idx -= 1 + + if value is not None: + sys.stdout.write(str(value)) + sys.stdout.write("\n") + + sys.stdout.write("[~EOM]\n") + return + + +def stripList(l): + return ([x.strip() for x in l]) + + +sys.stdout.write("Content-type: text/plain\n\n") + +# write start time +sys.stdout.write(str(time.mktime(time.gmtime()))) +sys.stdout.write("\n") + +requestMethod = os.environ['REQUEST_METHOD'] + +if requestMethod == 'POST': + postData = sys.stdin.readline() + queryString = cgi.parse_qs(postData) + queryString = dict((k, v[0]) for k, v in queryString.items()) +elif requestMethod == 'GET': + queryString = dict(cgi.parse_qsl(os.environ['QUERY_STRING'])); + +if "m" in queryString: + metricParts = queryString["m"].split(",") +else: + metricParts = [""] +metricParts = stripList(metricParts) + +hostParts = [] +if "h" in queryString: + hostParts = queryString["h"].split(",") +hostParts = stripList(hostParts) + +if "c" in queryString: + clusterParts = queryString["c"].split(",") +else: + clusterParts = [""] +clusterParts = stripList(clusterParts) + +if "p" in queryString: + rrdPath = queryString["p"] +else: + rrdPath = "/var/lib/ganglia/rrds/" + +start = None +if "s" in queryString: + start = queryString["s"] + +end = None +if "e" in queryString: + end = queryString["e"] + +resolution = None +if "r" in queryString: + resolution = queryString["r"] + +if "cf" in queryString: + cf = queryString["cf"] +else: + cf = "AVERAGE" + +if "pt" in queryString: + pointInTime = True +else: + pointInTime = False + + +def _walk(*args, **kwargs): + for root, dirs, files in os.walk(*args, **kwargs): + for dir in dirs: + qualified_dir = os.path.join(root, dir) + if os.path.islink(qualified_dir): + for x in os.walk(qualified_dir, **kwargs): + yield x + yield (root, dirs, files) + + +for cluster in clusterParts: + for path, dirs, files in _walk(rrdPath + cluster): + pathParts = path.split("/") + #Process only path which contains files. If no host parameter passed - process all hosts folders and summary info + #If host parameter passed - process only this host folder + if len(files) > 0 and (len(hostParts) == 0 or pathParts[-1] in hostParts): + for metric in metricParts: + file = metric + ".rrd" + fileFullPath = os.path.join(path, file) + if os.path.exists(fileFullPath): + #Exact name of metric + printMetric(pathParts[-2], pathParts[-1], file[:-4], + os.path.join(path, file), cf, start, end, resolution, + pointInTime) + else: + #Regex as metric name + metricRegex = metric + '\.rrd$' + p = re.compile(metricRegex) + matchedFiles = filter(p.match, files) + for matchedFile in matchedFiles: + printMetric(pathParts[-2], pathParts[-1], matchedFile[:-4], + os.path.join(path, matchedFile), cf, start, end, + resolution, pointInTime) + +sys.stdout.write("[~EOF]\n") +# write end time +sys.stdout.write(str(time.mktime(time.gmtime()))) +sys.stdout.write("\n") + +sys.stdout.flush http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/test/python/stacks/1.3.2/GANGLIA/test_ganglia_server.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/python/stacks/1.3.2/GANGLIA/test_ganglia_server.py b/ambari-server/src/test/python/stacks/1.3.2/GANGLIA/test_ganglia_server.py index bc5170c..e4a82ac 100644 --- a/ambari-server/src/test/python/stacks/1.3.2/GANGLIA/test_ganglia_server.py +++ b/ambari-server/src/test/python/stacks/1.3.2/GANGLIA/test_ganglia_server.py @@ -178,8 +178,9 @@ class TestGangliaServer(RMFTestCase): self.assertResourceCalled('Directory', '/srv/www/cgi-bin', recursive = True, ) - self.assertResourceCalled('File', '/srv/www/cgi-bin/rrd.py', - content = StaticFile('rrd.py'), + self.assertResourceCalled('TemplateConfig', '/srv/www/cgi-bin/rrd.py', + owner = "root", + group = "root", mode = 0755, ) self.assertResourceCalled('Directory', '/var/lib/ganglia/rrds', http://git-wip-us.apache.org/repos/asf/ambari/blob/e74fc95b/ambari-server/src/test/python/stacks/2.0.6/GANGLIA/test_ganglia_server.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/python/stacks/2.0.6/GANGLIA/test_ganglia_server.py b/ambari-server/src/test/python/stacks/2.0.6/GANGLIA/test_ganglia_server.py index f76ce52..afd5f51 100644 --- a/ambari-server/src/test/python/stacks/2.0.6/GANGLIA/test_ganglia_server.py +++ b/ambari-server/src/test/python/stacks/2.0.6/GANGLIA/test_ganglia_server.py @@ -168,8 +168,9 @@ class TestGangliaServer(RMFTestCase): self.assertResourceCalled('Directory', '/srv/www/cgi-bin', recursive = True, ) - self.assertResourceCalled('File', '/srv/www/cgi-bin/rrd.py', - content = StaticFile('rrd.py'), + self.assertResourceCalled('TemplateConfig', '/srv/www/cgi-bin/rrd.py', + owner = "root", + group = "root", mode = 0755, ) self.assertResourceCalled('Directory', '/var/lib/ganglia/rrds',
