HADOOP-4758. Add a splitter for metrics contexts to support more than one type of collector.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@723181 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -158,6 +158,9 @@ Trunk (unreleased changes)
     HADOOP-4708. Add support for dfsadmin commands in TestCLI. (Boris Shkolnik
     via cdouglas)
 
+    HADOOP-4758. Add a splitter for metrics contexts to support more than one
+    type of collector. (cdouglas)
+
   OPTIMIZATIONS
 
     HADOOP-3293. Fixes FileInputFormat to do provide locations for splits

src/core/org/apache/hadoop/metrics/ContextFactory.java

@@ -45,8 +45,8 @@ public class ContextFactory {
   private static ContextFactory theFactory = null;
     
   private Map<String,Object> attributeMap = new HashMap<String,Object>();
-  private Map<String,AbstractMetricsContext> contextMap = 
-    new HashMap<String,AbstractMetricsContext>();
+  private Map<String,MetricsContext> contextMap = 
+    new HashMap<String,MetricsContext>();
     
   // Used only when contexts, or the ContextFactory itself, cannot be
   // created.
@@ -119,23 +119,29 @@ public class ContextFactory {
    * @param contextName the name of the context
    * @return the named MetricsContext
    */
-  public synchronized MetricsContext getContext(String contextName) 
-    throws IOException, ClassNotFoundException, InstantiationException, IllegalAccessException
-  {
-    AbstractMetricsContext metricsContext = contextMap.get(contextName);
+  public synchronized MetricsContext getContext(String refName, String contextName)
+      throws IOException, ClassNotFoundException,
+             InstantiationException, IllegalAccessException {
+    MetricsContext metricsContext = contextMap.get(refName);
     if (metricsContext == null) {
-      String classNameAttribute = contextName + CONTEXT_CLASS_SUFFIX;
+      String classNameAttribute = refName + CONTEXT_CLASS_SUFFIX;
       String className = (String) getAttribute(classNameAttribute);
       if (className == null) {
         className = DEFAULT_CONTEXT_CLASSNAME;
       }
       Class contextClass = Class.forName(className);
-      metricsContext = (AbstractMetricsContext) contextClass.newInstance();
+      metricsContext = (MetricsContext) contextClass.newInstance();
       metricsContext.init(contextName, this);
       contextMap.put(contextName, metricsContext);
     }
     return metricsContext;
   }
+
+  public synchronized MetricsContext getContext(String contextName)
+    throws IOException, ClassNotFoundException, InstantiationException,
+           IllegalAccessException {
+    return getContext(contextName, contextName);
+  }
     
   /**
    * Returns a "null" context - one which does nothing.

src/core/org/apache/hadoop/metrics/MetricsContext.java

@@ -31,7 +31,14 @@ public interface MetricsContext {
    * Default period in seconds at which data is sent to the metrics system.
    */
   public static final int DEFAULT_PERIOD = 5;
-    
+
+  /**
+   * Initialize this context.
+   * @param contextName The given name for this context
+   * @param factory The creator of this context
+   */
+  public void init(String contextName, ContextFactory factory);
+
   /**
    * Returns the context name.
    *

src/core/org/apache/hadoop/metrics/MetricsUtil.java

@@ -41,15 +41,20 @@ public class MetricsUtil {
    */
   private MetricsUtil() {}
     
+  public static MetricsContext getContext(String contextName) {
+    return getContext(contextName, contextName);
+  }
+
   /**
    * Utility method to return the named context.
    * If the desired context cannot be created for any reason, the exception
    * is logged, and a null context is returned.
    */
-  public static MetricsContext getContext(String contextName) {
+  public static MetricsContext getContext(String refName, String contextName) {
     MetricsContext metricsContext;
     try {
-      metricsContext = ContextFactory.getFactory().getContext(contextName);
+      metricsContext =
+        ContextFactory.getFactory().getContext(refName, contextName);
       if (!metricsContext.isMonitoring()) {
         metricsContext.startMonitoring();
       }

src/core/org/apache/hadoop/metrics/spi/AbstractMetricsContext.java

@@ -208,7 +208,7 @@ public abstract class AbstractMetricsContext implements MetricsContext {
    * @param recordName the name of the record
    * @return newly created instance of MetricsRecordImpl or subclass
    */
-  protected MetricsRecordImpl newRecord(String recordName) {
+  protected MetricsRecord newRecord(String recordName) {
     return new MetricsRecordImpl(recordName, this);
   }
     

src/core/org/apache/hadoop/metrics/spi/CompositeContext.java

@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.metrics.spi;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.metrics.ContextFactory;
+import org.apache.hadoop.metrics.MetricsContext;
+import org.apache.hadoop.metrics.MetricsException;
+import org.apache.hadoop.metrics.MetricsRecord;
+import org.apache.hadoop.metrics.MetricsUtil;
+import org.apache.hadoop.metrics.Updater;
+
+public class CompositeContext extends AbstractMetricsContext {
+
+  private static final Log LOG = LogFactory.getLog(CompositeContext.class);
+  private static final String ARITY_LABEL = "arity";
+  private static final String SUB_FMT = "%s.sub%d";
+  private final ArrayList<MetricsContext> subctxt =
+    new ArrayList<MetricsContext>();
+
+  public CompositeContext() {
+  }
+
+  public void init(String contextName, ContextFactory factory) {
+    super.init(contextName, factory);
+    int nKids;
+    try {
+      String sKids = getAttribute(ARITY_LABEL);
+      nKids = Integer.valueOf(sKids);
+    } catch (Exception e) {
+      LOG.error("Unable to initialize composite metric " + contextName +
+                ": could not init arity", e);
+      return;
+    }
+    for (int i = 0; i < nKids; ++i) {
+      MetricsContext ctxt = MetricsUtil.getContext(
+          String.format(SUB_FMT, contextName, i), contextName);
+      if (null != ctxt) {
+        subctxt.add(ctxt);
+      }
+    }
+  }
+
+  @Override
+  public MetricsRecord newRecord(String recordName) {
+    return (MetricsRecord) Proxy.newProxyInstance(
+        MetricsRecord.class.getClassLoader(),
+        new Class[] { MetricsRecord.class },
+        new MetricsRecordDelegator(recordName, subctxt));
+  }
+
+  @Override
+  protected void emitRecord(String contextName, String recordName,
+      OutputRecord outRec) throws IOException {
+    for (MetricsContext ctxt : subctxt) {
+      try {
+        ((AbstractMetricsContext)ctxt).emitRecord(
+          contextName, recordName, outRec);
+        if (contextName == null || recordName == null || outRec == null) {
+          throw new IOException(contextName + ":" + recordName + ":" + outRec);
+        }
+      } catch (IOException e) {
+        LOG.warn("emitRecord failed: " + ctxt.getContextName(), e);
+      }
+    }
+  }
+
+  @Override
+  protected void flush() throws IOException {
+    for (MetricsContext ctxt : subctxt) {
+      try {
+        ((AbstractMetricsContext)ctxt).flush();
+      } catch (IOException e) {
+        LOG.warn("flush failed: " + ctxt.getContextName(), e);
+      }
+    }
+  }
+
+  @Override
+  public void startMonitoring() throws IOException {
+    for (MetricsContext ctxt : subctxt) {
+      try {
+        ctxt.startMonitoring();
+      } catch (IOException e) {
+        LOG.warn("startMonitoring failed: " + ctxt.getContextName(), e);
+      }
+    }
+  }
+
+  @Override
+  public void stopMonitoring() {
+    for (MetricsContext ctxt : subctxt) {
+      ctxt.stopMonitoring();
+    }
+  }
+
+  /**
+   * Return true if all subcontexts are monitoring.
+   */
+  @Override
+  public boolean isMonitoring() {
+    boolean ret = true;
+    for (MetricsContext ctxt : subctxt) {
+      ret &= ctxt.isMonitoring();
+    }
+    return ret;
+  }
+
+  @Override
+  public void close() {
+    for (MetricsContext ctxt : subctxt) {
+      ctxt.close();
+    }
+  }
+
+  @Override
+  public void registerUpdater(Updater updater) {
+    for (MetricsContext ctxt : subctxt) {
+      ctxt.registerUpdater(updater);
+    }
+  }
+
+  @Override
+  public void unregisterUpdater(Updater updater) {
+    for (MetricsContext ctxt : subctxt) {
+      ctxt.unregisterUpdater(updater);
+    }
+  }
+
+  private static class MetricsRecordDelegator implements InvocationHandler {
+    private static final Method m_getRecordName = initMethod();
+    private static Method initMethod() {
+      try {
+        return MetricsRecord.class.getMethod("getRecordName", new Class[0]);
+      } catch (Exception e) {
+        throw new RuntimeException("Internal error", e);
+      }
+    }
+
+    private final String recordName;
+    private final ArrayList<MetricsRecord> subrecs;
+
+    MetricsRecordDelegator(String recordName, ArrayList<MetricsContext> ctxts) {
+      this.recordName = recordName;
+      this.subrecs = new ArrayList<MetricsRecord>(ctxts.size());
+      for (MetricsContext ctxt : ctxts) {
+        subrecs.add(ctxt.createRecord(recordName));
+      }
+    }
+
+    public Object invoke(Object p, Method m, Object[] args) throws Throwable {
+      if (m_getRecordName.equals(m)) {
+        return recordName;
+      }
+      assert Void.TYPE.equals(m.getReturnType());
+      for (MetricsRecord rec : subrecs) {
+        m.invoke(rec, args);
+      }
+      return null;
+    }
+  }
+
+}