HADOOP-7507. Allow ganglia metrics to include the metrics system tags in the gmetric names. Contributed by Alejandro Abdelnur.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.23@1166459 13f79535-47bb-0310-9956-ffa450edef68

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -361,6 +361,9 @@ Release 0.23.0 - Unreleased
     HADOOP-7580. Add a version of getLocalPathForWrite to LocalDirAllocator
     which doesn't create dirs. (Chris Douglas & Siddharth Seth via acmurthy) 
 
+    HADOOP-7507. Allow ganglia metrics to include the metrics system tags
+                 in the gmetric names. (Alejandro Abdelnur via todd)
+
   OPTIMIZATIONS
   
     HADOOP-7333. Performance improvement in PureJavaCrc32. (Eric Caspole

hadoop-common-project/hadoop-common/src/main/conf/hadoop-metrics2.properties

@@ -43,6 +43,16 @@
 #*.sink.ganglia.slope=jvm.metrics.gcCount=zero,jvm.metrics.memHeapUsedM=both
 #*.sink.ganglia.dmax=jvm.metrics.threadsBlocked=70,jvm.metrics.memHeapUsedM=40
 
+# Tag values to use for the ganglia prefix. If not defined no tags are used.
+# If '*' all tags are used. If specifiying multiple tags separate them with 
+# commas. Note that the last segment of the property name is the context name.
+#
+#*.sink.ganglia.tagsForPrefix.jvm=ProcesName
+#*.sink.ganglia.tagsForPrefix.dfs=
+#*.sink.ganglia.tagsForPrefix.rpc=
+#*.sink.ganglia.tagsForPrefix.mapred=
+#*.sink.ganglia.tagsForPrefix.fairscheduler=
+
 #namenode.sink.ganglia.servers=yourgangliahost_1:8649,yourgangliahost_2:8649
 
 #datanode.sink.ganglia.servers=yourgangliahost_1:8649,yourgangliahost_2:8649

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics/ganglia/GangliaContext.java

@@ -132,6 +132,12 @@ public class GangliaContext extends AbstractMetricsContext {
     StringBuilder sb = new StringBuilder();
     sb.append(contextName);
     sb.append('.');
+
+    if (contextName.equals("jvm") && outRec.getTag("processName") != null) {
+      sb.append(outRec.getTag("processName"));
+      sb.append('.');
+    }
+
     sb.append(recordName);
     sb.append('.');
     int sbBaseLen = sb.length();

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/sink/ganglia/GangliaSink30.java

@@ -20,13 +20,21 @@ package org.apache.hadoop.metrics2.sink.ganglia;
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
 
+import org.apache.commons.configuration.SubsetConfiguration;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.metrics2.AbstractMetric;
 import org.apache.hadoop.metrics2.MetricsException;
 import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.hadoop.metrics2.impl.MsInfo;
 import org.apache.hadoop.metrics2.util.MetricsCache;
 import org.apache.hadoop.metrics2.util.MetricsCache.Record;
 
@@ -38,8 +46,67 @@ public class GangliaSink30 extends AbstractGangliaSink {
 
   public final Log LOG = LogFactory.getLog(this.getClass());
 
+  private static final String TAGS_FOR_PREFIX_PROPERTY_PREFIX = "tagsForPrefix.";
+  
   private MetricsCache metricsCache = new MetricsCache();
 
+  // a key with a NULL value means ALL
+  private Map<String,Set<String>> useTagsMap = new HashMap<String,Set<String>>();
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public void init(SubsetConfiguration conf) {
+    super.init(conf);
+
+    conf.setListDelimiter(',');
+    Iterator<String> it = (Iterator<String>) conf.getKeys();
+    while (it.hasNext()) {
+      String propertyName = it.next();
+      if (propertyName.startsWith(TAGS_FOR_PREFIX_PROPERTY_PREFIX)) {
+        String contextName = propertyName.substring(TAGS_FOR_PREFIX_PROPERTY_PREFIX.length());
+        String[] tags = conf.getStringArray(propertyName);
+        boolean useAllTags = false;
+        Set<String> set = null;
+        if (tags.length > 0) {
+          set = new HashSet<String>();
+          for (String tag : tags) {
+            tag = tag.trim();
+            useAllTags |= tag.equals("*");
+            if (tag.length() > 0) {
+              set.add(tag);
+            }
+          }
+          if (useAllTags) {
+            set = null;
+          }
+        }
+        useTagsMap.put(contextName, set);
+      }
+    }
+  }
+
+  @InterfaceAudience.Private
+  public void appendPrefix(MetricsRecord record, StringBuilder sb) {
+    String contextName = record.context();
+    Collection<MetricsTag> tags = record.tags();
+    if (useTagsMap.containsKey(contextName)) {
+      Set<String> useTags = useTagsMap.get(contextName);
+      for (MetricsTag t : tags) {
+        if (useTags == null || useTags.contains(t.name())) {
+
+          // the context is always skipped here because it is always added
+          
+          // the hostname is always skipped to avoid case-mismatches 
+          // from different DNSes.
+
+          if (t.info() != MsInfo.Context && t.info() != MsInfo.Hostname && t.value() != null) {
+            sb.append('.').append(t.name()).append('=').append(t.value());
+          }
+        }
+      }
+    }          
+  }
+  
   @Override
   public void putMetrics(MetricsRecord record) {
     // The method handles both cases whether Ganglia support dense publish
@@ -53,6 +120,8 @@ public class GangliaSink30 extends AbstractGangliaSink {
       sb.append('.');
       sb.append(recordName);
 
+      appendPrefix(record, sb);
+      
       String groupName = sb.toString();
       sb.append('.');
       int sbBaseLen = sb.length();

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/impl/TestGangliaMetrics.java

@@ -26,12 +26,17 @@ import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.SocketException;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsTag;
 import org.apache.hadoop.metrics2.annotation.Metric;
 import org.apache.hadoop.metrics2.annotation.Metrics;
 import org.apache.hadoop.metrics2.lib.MetricsRegistry;
@@ -54,6 +59,44 @@ public class TestGangliaMetrics {
       "test.s1rec.S1NumOps",
       "test.s1rec.S1AvgTime" };
 
+  @Test
+  public void testTagsForPrefix() throws Exception {
+    ConfigBuilder cb = new ConfigBuilder()
+      .add("test.sink.ganglia.tagsForPrefix.all", "*")
+      .add("test.sink.ganglia.tagsForPrefix.some", "NumActiveSinks, NumActiveSources")
+      .add("test.sink.ganglia.tagsForPrefix.none", "");
+    GangliaSink30 sink = new GangliaSink30();
+    sink.init(cb.subset("test.sink.ganglia"));
+
+    List<MetricsTag> tags = new ArrayList<MetricsTag>();
+    tags.add(new MetricsTag(MsInfo.Context, "all"));
+    tags.add(new MetricsTag(MsInfo.NumActiveSources, "foo"));
+    tags.add(new MetricsTag(MsInfo.NumActiveSinks, "bar"));
+    tags.add(new MetricsTag(MsInfo.NumAllSinks, "haa"));
+    tags.add(new MetricsTag(MsInfo.Hostname, "host"));
+    Set<AbstractMetric> metrics = new HashSet<AbstractMetric>();
+    MetricsRecord record = new MetricsRecordImpl(MsInfo.Context, (long) 1, tags, metrics);
+
+    StringBuilder sb = new StringBuilder();
+    sink.appendPrefix(record, sb);
+    assertEquals(".NumActiveSources=foo.NumActiveSinks=bar.NumAllSinks=haa", sb.toString());
+
+    tags.set(0, new MetricsTag(MsInfo.Context, "some"));
+    sb = new StringBuilder();
+    sink.appendPrefix(record, sb);
+    assertEquals(".NumActiveSources=foo.NumActiveSinks=bar", sb.toString());
+
+    tags.set(0, new MetricsTag(MsInfo.Context, "none"));
+    sb = new StringBuilder();
+    sink.appendPrefix(record, sb);
+    assertEquals("", sb.toString());
+
+    tags.set(0, new MetricsTag(MsInfo.Context, "nada"));
+    sb = new StringBuilder();
+    sink.appendPrefix(record, sb);
+    assertEquals("", sb.toString());
+  }
+  
   @Test public void testGangliaMetrics2() throws Exception {
     ConfigBuilder cb = new ConfigBuilder().add("default.period", 10)
         .add("test.sink.gsink30.context", "test") // filter out only "test"