Fix HADOOP-132. Add an API for reporting metrics (as yet unused). Contributed by David Bowen.

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

CHANGES.txt

@@ -86,6 +86,10 @@ Trunk (unreleased)
 23. Fix HADOOP-162.  Fix ConcurrentModificationException when
     releasing file locks. (omalley via cutting)
 
+24. Fix HADOOP-132.  Initial check-in of new Metrics API, including 
+    implementations for writing metric data to a file and for sending
+    it to Ganglia.
+
 
 Release 0.1.1 - 2006-04-08
 

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

@@ -0,0 +1,163 @@
+/*
+ * ContextFactory.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.hadoop.metrics.spi.AbstractMetricsContext;
+
+/**
+ * Factory class for creating MetricsContext objects.  To obtain an instance
+ * of this class, use the static <code>getFactory()</code> method.
+ */
+public class ContextFactory {
+    
+    private static final String PROPERTIES_FILE = 
+            "/hadoop-metrics.properties";
+    private static final String CONTEXT_CLASS_SUFFIX =
+            ".class";
+    private static final String DEFAULT_CONTEXT_CLASSNAME =
+            "org.apache.hadoop.metrics.spi.NullContext";
+    
+    private static ContextFactory theFactory = null;
+    
+    // private Map<String,Object> attributeMap = new HashMap<String,Object>();
+    private Map attributeMap = new HashMap();
+    
+    /** Creates a new instance of ContextFactory */
+    protected ContextFactory() {
+    }
+    
+    /**
+     * Returns the value of the named attribute, or null if there is no 
+     * attribute of that name.
+     *
+     * @param attributeName the attribute name
+     * @return the attribute value
+     */
+    public Object getAttribute(String attributeName) {
+        return attributeMap.get(attributeName);
+    }
+    
+    /**
+     * Returns the names of all the factory's attributes.
+     * 
+     * @return the attribute names
+     */
+    public String[] getAttributeNames() {
+        String[] result = new String[attributeMap.size()];
+        int i = 0;
+        // for (String attributeName : attributeMap.keySet()) {
+        Iterator it = attributeMap.keySet().iterator();
+        while (it.hasNext()) {
+            result[i++] = (String) it.next();
+        }
+        return result;
+    }
+    
+    /**
+     * Sets the named factory attribute to the specified value, creating it
+     * if it did not already exist.  If the value is null, this is the same as
+     * calling removeAttribute.
+     *
+     * @param attributeName the attribute name
+     * @param value the new attribute value
+     */
+    public void setAttribute(String attributeName, Object value) {
+        attributeMap.put(attributeName, value);
+    }
+
+    /**
+     * Removes the named attribute if it exists.
+     *
+     * @param attributeName the attribute name
+     */
+    public void removeAttribute(String attributeName) {
+        attributeMap.remove(attributeName);
+    }
+    
+    /**
+     * Returns the named MetricsContext instance, constructing it if necessary 
+     * using the factory's current configuration attributes. <p/>
+     * 
+     * When constructing the instance, if the factory property 
+     * <i>contextName</i>.class</code> exists, 
+     * its value is taken to be the name of the class to instantiate.  Otherwise,
+     * the default is to create an instance of 
+     * <code>org.apache.hadoop.metrics.spi.NullContext</code>, which is a 
+     * dummy "no-op" context which will cause all metric data to be discarded.
+     * 
+     * @param contextName the name of the context
+     * @return the named MetricsContext
+     */
+    public MetricsContext getContext(String contextName) 
+        throws IOException, ClassNotFoundException, InstantiationException, IllegalAccessException
+    {
+        String classNameAttribute = contextName + CONTEXT_CLASS_SUFFIX;
+        String className = (String) getAttribute(classNameAttribute);
+        if (className == null) {
+            className = DEFAULT_CONTEXT_CLASSNAME;
+        }
+        Class contextClass = Class.forName(className);
+        AbstractMetricsContext metricsContext = 
+                (AbstractMetricsContext) contextClass.newInstance();
+        metricsContext.init(contextName, this);
+        return metricsContext;
+    }
+    
+    /**
+     * Returns the singleton ContextFactory instance, constructing it if 
+     * necessary. <p/>
+     * 
+     * When the instance is constructed, this method checks if the file 
+     * <code>hadoop-metrics.properties</code> exists on the class path.  If it 
+     * exists, it must be in the format defined by java.util.Properties, and all 
+     * the properties in the file are set as attributes on the newly created
+     * ContextFactory instance.
+     *
+     * @return the singleton ContextFactory instance
+     */
+    public static synchronized ContextFactory getFactory() throws IOException {
+        if (theFactory == null) {
+            theFactory = new ContextFactory();
+            theFactory.setAttributes();
+        }
+        return theFactory;
+    }
+    
+    private void setAttributes() throws IOException {
+        InputStream is = getClass().getResourceAsStream(PROPERTIES_FILE);
+        if (is != null) {
+            Properties properties = new Properties();
+            properties.load(is);
+            //for (Object propertyNameObj : properties.keySet()) {
+            Iterator it = properties.keySet().iterator();
+            while (it.hasNext()) {
+                String propertyName = (String) it.next();
+                String propertyValue = properties.getProperty(propertyName);
+                setAttribute(propertyName, propertyValue);
+            }
+        }
+    }
+    
+}

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

@@ -0,0 +1,95 @@
+/*
+ * MetricsContext.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The main interface to the metrics package. 
+ */
+public interface MetricsContext {
+    
+    /**
+     * Default period in seconds at which data is sent to the metrics system.
+     */
+    public static final int DEFAULT_PERIOD = 5;
+    
+    /**
+     * Returns the context name.
+     *
+     * @return the context name
+     */
+    public abstract String getContextName();
+    
+    /**
+     * Starts or restarts monitoring, the emitting of metrics records as they are 
+     * updated. 
+     */
+    public abstract void startMonitoring()
+        throws IOException;
+
+    /**
+     * Stops monitoring.  This does not free any data that the implementation
+     * may have buffered for sending at the next timer event. It
+     * is OK to call <code>startMonitoring()</code> again after calling 
+     * this.
+     * @see #close()
+     */
+    public abstract void stopMonitoring();
+    
+    /**
+     * Returns true if monitoring is currently in progress.
+     */
+    public abstract boolean isMonitoring();
+    
+    /**
+     * Stops monitoring and also frees any buffered data, returning this 
+     * object to its initial state.  
+     */
+    public abstract void close();
+    
+    /**
+     * Creates a new MetricsRecord instance with the given <code>recordName</code>.
+     * Throws an exception if the metrics implementation is configured with a fixed
+     * set of record names and <code>recordName</code> is not in that set.
+     *
+     * @param recordName the name of the record
+     * @throws MetricsException if recordName conflicts with configuration data
+     */
+    public abstract MetricsRecord createRecord(String recordName);
+    
+    /**
+     * Registers a callback to be called at regular time intervals, as 
+     * determined by the implementation-class specific configuration.
+     *
+     * @param updater object to be run periodically; it should updated
+     * some metrics records and then return
+     */
+    public abstract void registerUpdater(Updater updater);
+
+    /**
+     * Removes a callback, if it exists.
+     * 
+     * @param updater object to be removed from the callback list
+     */
+    public abstract void unregisterUpdater(Updater updater);
+    
+}

src/java/org/apache/hadoop/metrics/MetricsException.java

@@ -0,0 +1,38 @@
+/*
+ * MetricsException.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+/**
+ * General-purpose, unchecked metrics exception.
+ */
+public class MetricsException extends RuntimeException {
+    
+    /** Creates a new instance of MetricsException */
+    public MetricsException() {
+    }
+    
+    /** Creates a new instance of MetricsException 
+     *
+     * @param message an error message
+     */
+    public MetricsException(String message) {
+        super(message);
+    }
+    
+}

src/java/org/apache/hadoop/metrics/MetricsRecord.java

@@ -0,0 +1,204 @@
+/*
+ * MetricsRecord.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+/**
+ * A named and optionally tagged set of records to be sent to the metrics
+ * system. <p/>
+ *
+ * A record name identifies the kind of data to be reported. For example, a
+ * program reporting statistics relating to the disks on a computer might use
+ * a record name "diskStats".<p/>
+ *
+ * A record has zero or more <i>tags</i>. A tag has a name and a value. To
+ * continue the example, the "diskStats" record might use a tag named
+ * "diskName" to identify a particular disk.  Sometimes it is useful to have
+ * more than one tag, so there might also be a "diskType" with value "ide" or
+ * "scsi" or whatever.<p/>
+ *
+ * A record also has zero or more <i>metrics</i>.  These are the named
+ * values that are to be reported to the metrics system.  In the "diskStats"
+ * example, possible metric names would be "diskPercentFull", "diskPercentBusy", 
+ * "kbReadPerSecond", etc.<p/>
+ * 
+ * The general procedure for using a MetricsRecord is to fill in its tag and
+ * metric values, and then call <code>update()</code> to pass the record to the
+ * client library.
+ * Metric data is not immediately sent to the metrics system
+ * each time that <code>update()</code> is called. 
+ * An internal table is maintained, identified by the record name. This
+ * table has columns 
+ * corresponding to the tag and the metric names, and rows 
+ * corresponding to each unique set of tag values. An update
+ * either modifies an existing row in the table, or adds a new row with a set of
+ * tag values that are different from all the other rows.  Note that if there
+ * are no tags, then there can be at most one row in the table. <p/>
+ * 
+ * Once a row is added to the table, its data will be sent to the metrics system 
+ * on every timer period, whether or not it has been updated since the previous
+ * timer period.  If this is inappropriate, for example if metrics were being
+ * reported by some transient object in an application, the <code>remove()</code>
+ * method can be used to remove the row and thus stop the data from being
+ * sent.<p/>
+ *
+ * Note that the <code>update()</code> method is atomic.  This means that it is
+ * safe for different threads to be updating the same metric.  More precisely,
+ * it is OK for different threads to call <code>update()</code> on MetricsRecord instances 
+ * with the same set of tag names and tag values.  Different threads should 
+ * <b>not</b> use the same MetricsRecord instance at the same time.
+ */
+public interface MetricsRecord {
+    
+    /**
+     * Returns the record name. 
+     *
+     * @return the record name
+     */
+    public abstract String getRecordName();
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public abstract void setTag(String tagName, String tagValue);
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public abstract void setTag(String tagName, int tagValue);
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public abstract void setTag(String tagName, short tagValue);
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public abstract void setTag(String tagName, byte tagValue);
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void setMetric(String metricName, int metricValue);
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void setMetric(String metricName, short metricValue);
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void setMetric(String metricName, byte metricValue);
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void setMetric(String metricName, float metricValue);
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void incrMetric(String metricName, int metricValue);
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void incrMetric(String metricName, short metricValue);
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void incrMetric(String metricName, byte metricValue);
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public abstract void incrMetric(String metricName, float metricValue);
+    
+    /**
+     * Updates the table of buffered data which is to be sent periodically.
+     * If the tag values match an existing row, that row is updated; 
+     * otherwise, a new row is added.
+     */
+    public abstract void update();
+    
+    /**
+     * Removes, from the buffered data table, the row (if it exists) having tags 
+     * that equal the tags that have been set on this record. 
+     */
+    public abstract void remove();
+    
+}

src/java/org/apache/hadoop/metrics/Updater.java

@@ -0,0 +1,31 @@
+/*
+ * Updater.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+/**
+ * Call-back interface.  See <code>MetricsContext.registerUpdater()</code>.
+ */
+public interface Updater {
+    
+    /**
+     * Timer-based call-back from the metric library. 
+     */
+    public abstract void doUpdates(MetricsContext context);
+
+}

src/java/org/apache/hadoop/metrics/file/FileContext.java

@@ -0,0 +1,156 @@
+/*
+ * FileContext.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.file;
+
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.Iterator;
+import org.apache.hadoop.metrics.ContextFactory;
+import org.apache.hadoop.metrics.MetricsException;
+import org.apache.hadoop.metrics.spi.AbstractMetricsContext;
+import org.apache.hadoop.metrics.spi.OutputRecord;
+
+/**
+ * Metrics context for writing metrics to a file.<p/>
+ *
+ * This class is configured by setting ContextFactory attributes which in turn
+ * are usually configured through a properties file.  All the attributes are
+ * prefixed by the contextName. For example, the properties file might contain:
+ * <pre>
+ * myContextName.fileName=/tmp/metrics.log
+ * myContextName.period=5
+ * </pre>
+ *
+ */
+public class FileContext extends AbstractMetricsContext {
+    
+    /* Configuration attribute names */
+    protected static final String FILE_NAME_PROPERTY = "fileName";
+    protected static final String PERIOD_PROPERTY = "period";
+    
+    private File file = null;              // file for metrics to be written to
+    private PrintWriter writer = null;
+    
+    /** Creates a new instance of FileContext */
+    public FileContext() {}
+    
+    public void init(String contextName, ContextFactory factory) {
+        super.init(contextName, factory);
+        
+        String fileName = getAttribute(FILE_NAME_PROPERTY);
+        if (fileName != null) {
+            file = new File(fileName);
+        }
+        
+        String periodStr = getAttribute(PERIOD_PROPERTY);
+        if (periodStr != null) {
+            int period = 0;
+            try {
+                period = Integer.parseInt(periodStr);
+            } catch (NumberFormatException nfe) {
+            }
+            if (period <= 0) {
+                throw new MetricsException("Invalid period: " + periodStr);
+            }
+            setPeriod(period);
+        }
+    }
+
+    /**
+     * Returns the configured file name, or null.
+     */
+    public String getFileName() {
+        if (file == null) {
+            return null;
+        } else {
+            return file.getName();
+        }
+    }
+    
+    /**
+     * Starts or restarts monitoring, by opening in append-mode, the
+     * file specified by the <code>fileName</code> attribute,
+     * if specified. Otherwise the data will be written to standard
+     * output.
+     */
+    public void startMonitoring()
+        throws IOException 
+    {
+        if (file == null) {
+            writer = new PrintWriter(new BufferedOutputStream(System.out));
+        } else {
+            writer = new PrintWriter(new FileWriter(file, true));
+        }
+        super.startMonitoring();
+    }
+    
+    /**
+     * Stops monitoring, closing the file.
+     * @see #close()
+     */
+    public void stopMonitoring() {
+        super.stopMonitoring();
+        
+        if (writer != null) {
+            writer.close();
+            writer = null;
+        }
+    }
+    
+    /**
+     * Emits a metrics record to a file.
+     */
+    public void emitRecord(String contextName, String recordName, OutputRecord outRec) {
+        writer.print(contextName);
+        writer.print(".");
+        writer.print(recordName);
+        String separator = ": ";
+        // for (String tagName : outRec.getTagNames()) {
+        Iterator tagIt = outRec.getTagNames().iterator();
+        while (tagIt.hasNext()) {
+            String tagName = (String) tagIt.next();
+            writer.print(separator);
+            separator = ", ";
+            writer.print(tagName);
+            writer.print("=");
+            writer.print(outRec.getTag(tagName));
+        }
+        // for (String metricName : outRec.getMetricNames()) {
+        Iterator metricIt = outRec.getMetricNames().iterator();
+        while (metricIt.hasNext()) {
+            String metricName = (String) metricIt.next();
+            writer.print(separator);
+            separator = ", ";
+            writer.print(metricName);
+            writer.print("=");
+            writer.print(outRec.getMetric(metricName));
+        }
+        writer.println();
+    }
+    
+    /**
+     * Flushes the output writer, forcing updates to disk.
+     */
+    public void flush() {
+        writer.flush();
+    }
+}

src/java/org/apache/hadoop/metrics/file/package.html

@@ -0,0 +1,41 @@
+<html>
+<body>
+<!--
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.
+-->
+
+Implementation of the metrics package that writes the metrics to a file.
+Programmers should not normally need to use this package directly. Instead
+they should use org.hadoop.metrics.
+
+<p/>
+These are the implementation specific factory attributes 
+(See ContextFactory.getFactory()):
+
+<dl>
+    <dt><i>contextName</i>.fileName</dt>
+    <dd>The path of the file to which metrics in context <i>contextName</i>
+    are to be appended.  If this attribute is not specified, the metrics
+    are written to standard output by default.</dd>
+    
+    <dt><i>contextName</i>.period</dt>
+    <dd>The period in seconds on which the metric data is written to the
+    file.</dd>
+    
+</dl>
+
+
+</body>
+</html>

src/java/org/apache/hadoop/metrics/ganglia/GangliaContext.java

@@ -0,0 +1,246 @@
+/*
+ * GangliaContext.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.ganglia;
+
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.metrics.ContextFactory;
+import org.apache.hadoop.metrics.MetricsException;
+import org.apache.hadoop.metrics.spi.AbstractMetricsContext;
+import org.apache.hadoop.metrics.spi.OutputRecord;
+import org.apache.hadoop.metrics.spi.Util;
+
+/**
+ * Context for sending metrics to Ganglia.
+ * 
+ */
+public class GangliaContext extends AbstractMetricsContext {
+    
+    private static final String PERIOD_PROPERTY = "period";
+    private static final String SERVERS_PROPERTY = "servers";
+    private static final String UNITS_PROPERTY = "units";
+    private static final String SLOPE_PROPERTY = "slope";
+    private static final String TMAX_PROPERTY = "tmax";
+    private static final String DMAX_PROPERTY = "dmax";
+    
+    private static final String DEFAULT_UNITS = "";
+    private static final String DEFAULT_SLOPE = "both";
+    private static final int DEFAULT_TMAX = 60;
+    private static final int DEFAULT_DMAX = 0;
+    private static final int DEFAULT_PORT = 8649;
+    private static final int BUFFER_SIZE = 1500;       // as per libgmond.c
+    
+    //private static final Map<Class,String> typeTable = new HashMap<Class,String>(5);
+    private static final Map typeTable = new HashMap(5);
+    
+    static {
+        typeTable.put(String.class, "string");
+        typeTable.put(Byte.class, "int8");
+        typeTable.put(Short.class, "int16");
+        typeTable.put(Integer.class, "int32");
+        typeTable.put(Float.class, "float");
+    }
+    
+    private byte[] buffer = new byte[BUFFER_SIZE];
+    private int offset;
+    
+    //private List<SocketAddress> metricsServers;
+    private List metricsServers;
+    //private Map<String,String> unitsTable;
+    private Map unitsTable;
+    //private Map<String,String> slopeTable;
+    private Map slopeTable;
+    //private Map<String,String> tmaxTable;
+    private Map tmaxTable;
+    //private Map<String,String> dmaxTable;
+    private Map dmaxTable;
+    
+    private DatagramSocket datagramSocket;
+    
+    /** Creates a new instance of GangliaContext */
+    public GangliaContext() {
+    }
+    
+    public void init(String contextName, ContextFactory factory) 
+    {
+        super.init(contextName, factory);
+        
+        String periodStr = getAttribute(PERIOD_PROPERTY);
+        if (periodStr != null) {
+            int period = 0;
+            try {
+                period = Integer.parseInt(periodStr);
+            } catch (NumberFormatException nfe) {
+            }
+            if (period <= 0) {
+                throw new MetricsException("Invalid period: " + periodStr);
+            }
+            setPeriod(period);
+        }
+        
+        metricsServers = 
+                Util.parse(getAttribute(SERVERS_PROPERTY), DEFAULT_PORT); 
+        
+        unitsTable = getAttributeTable(UNITS_PROPERTY);
+        slopeTable = getAttributeTable(SLOPE_PROPERTY);
+        tmaxTable  = getAttributeTable(TMAX_PROPERTY);
+        dmaxTable  = getAttributeTable(DMAX_PROPERTY);
+        
+        try {
+            datagramSocket = new DatagramSocket();
+        }
+        catch (SocketException se) {
+            se.printStackTrace();
+        }
+    }
+        
+    public void emitRecord(String contextName, String recordName, OutputRecord outRec) 
+        throws IOException
+    {
+        
+        // metric name formed from record name and tag values
+        StringBuffer nameBuf = new StringBuffer(recordName);
+        // for (String tagName : outRec.getTagNames()) {
+        Iterator tagIt = outRec.getTagNames().iterator();
+        while (tagIt.hasNext()) {
+            String tagName = (String) tagIt.next();
+            nameBuf.append('.');
+            nameBuf.append(outRec.getTag(tagName));
+        }
+        nameBuf.append('.');
+        String namePrefix = new String(nameBuf);
+        
+        // emit each metric in turn
+        // for (String metricName : outRec.getMetricNames()) {
+        Iterator metricIt = outRec.getMetricNames().iterator();
+        while (metricIt.hasNext()) {
+            String metricName = (String) metricIt.next();
+            Object metric = outRec.getMetric(metricName);
+            String type = (String) typeTable.get(metric.getClass());
+            emitMetric(namePrefix + metricName, type, metric.toString());
+        }
+        
+    }
+    
+    private void emitMetric(String name, String type,  String value) 
+        throws IOException
+    {
+        String units = getUnits(name);
+        int slope = getSlope(name);
+        int tmax = getTmax(name);
+        int dmax = getDmax(name);
+        
+        offset = 0;
+        xdr_int(0);             // metric_user_defined
+        xdr_string(type);
+        xdr_string(name);
+        xdr_string(value);
+        xdr_string(units);
+        xdr_int(slope);
+        xdr_int(tmax);
+        xdr_int(dmax);
+        
+        // for (SocketAddress socketAddress : metricsServers) {
+        Iterator socketIt = metricsServers.iterator();
+        while (socketIt.hasNext()) {
+            SocketAddress socketAddress = (SocketAddress) socketIt.next();
+            DatagramPacket packet = 
+                new DatagramPacket(buffer, offset, socketAddress);
+            datagramSocket.send(packet);
+        }
+    }
+    
+    private String getUnits(String metricName) {
+        String result = (String) unitsTable.get(metricName);
+        if (result == null) {
+            result = DEFAULT_UNITS;
+        }
+        return result;
+    }
+    
+    private int getSlope(String metricName) {
+        String slopeString = (String) slopeTable.get(metricName);
+        if (slopeString == null) {
+            slopeString = DEFAULT_SLOPE; 
+        }
+        return ("zero".equals(slopeString) ? 0 : 3); // see gmetric.c
+    }
+    
+    private int getTmax(String metricName) {
+        String tmaxString = (String) tmaxTable.get(metricName);
+        if (tmaxString == null) {
+            return DEFAULT_TMAX;
+        }
+        else {
+            return Integer.parseInt(tmaxString);
+        }
+    }
+    
+    private int getDmax(String metricName) {
+        String dmaxString = (String) dmaxTable.get(metricName);
+        if (dmaxString == null) {
+            return DEFAULT_DMAX;
+        }
+        else {
+            return Integer.parseInt(dmaxString);
+        }
+    }
+    
+    /**
+     * Puts a string into the buffer by first writing the size of the string
+     * as an int, followed by the bytes of the string, padded if necessary to
+     * a multiple of 4.
+     */
+    private void xdr_string(String s) {
+	byte[] bytes = s.getBytes();
+        int len = bytes.length;
+        xdr_int(len);
+        System.arraycopy(bytes, 0, buffer, offset, len);
+        offset += len;
+        pad();
+    }
+
+    /**
+     * Pads the buffer with zero bytes up to the nearest multiple of 4.
+     */
+    private void pad() {
+	int newOffset = ((offset + 3) / 4) * 4;
+	while (offset < newOffset) {
+            buffer[offset++] = 0;
+        }
+    }
+        
+    /**
+     * Puts an integer into the buffer as 4 bytes, big-endian.
+     */
+    private void xdr_int(int i) {
+	buffer[offset++] = (byte)((i >> 24) & 0xff);
+	buffer[offset++] = (byte)((i >> 16) & 0xff);
+	buffer[offset++] = (byte)((i >> 8) & 0xff);
+	buffer[offset++] = (byte)(i & 0xff);
+    }
+    
+}

src/java/org/apache/hadoop/metrics/ganglia/package.html

@@ -0,0 +1,55 @@
+<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
+
+<html>
+<body>
+<!--
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.
+-->
+
+Implementation of the metrics package that sends metric data to 
+<a href="http://ganglia.sourceforge.net/">Ganglia</a>.
+Programmers should not normally need to use this package directly. Instead
+they should use org.hadoop.metrics.
+
+<p/>
+These are the implementation specific factory attributes 
+(See ContextFactory.getFactory()):
+
+<dl>
+    <dt><i>contextName</i>.servers</dt>
+    <dd>Space and/or comma separated sequence of servers to which UDP
+    messages should be sent.</dd>
+    
+    <dt><i>contextName</i>.period</dt>
+    <dd>The period in seconds on which the metric data is sent to the
+    server(s).</dd>
+    
+    <dt><i>contextName</i>.units.<i>recordName</i>.<i>metricName</i></dt>
+    <dd>The units for the specified metric in the specified record.</dd>
+    
+    <dt><i>contextName</i>.slope.<i>recordName</i>.<i>metricName</i></dt>
+    <dd>The slope for the specified metric in the specified record.</dd>
+    
+    <dt><i>contextName</i>.tmax.<i>recordName</i>.<i>metricName</i></dt>
+    <dd>The tmax for the specified metric in the specified record.</dd>
+    
+    <dt><i>contextName</i>.dmax.<i>recordName</i>.<i>metricName</i></dt>
+    <dd>The dmax for the specified metric in the specified record.</dd>
+    
+</dl>
+
+
+</body>
+</html>

src/java/org/apache/hadoop/metrics/package.html

@@ -0,0 +1,157 @@
+<html>
+  <head>
+    <title>org.apache.hadoop.metrics</title>
+  </head>
+<body>
+<!--
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.
+-->
+
+This package defines an API for reporting performance metric information.
+<p/>
+The API is abstract so that it can be implemented on top of
+a variety of metrics client libraries.  The choice of 
+client library is a configuration option, and different 
+modules within the same application can use
+different metrics implementation libraries.
+<p/>
+Sub-packages:
+<dl>
+    <dt><code>org.apache.hadoop.metrics.spi</code></dt>
+    <dd>The abstract Server Provider Interface package. Those wishing to
+    integrate the metrics API with a particular metrics client library should 
+    extend this package.</dd>
+    
+    <dt><code>org.apache.hadoop.metrics.file</code></dt>
+    <dd>An implementation package which writes the metric data to 
+    a file, or sends it to the standard output stream.</dd>
+ 
+    <dt> <code>org.apache.hadoop.metrics.ganglia</code></dt>
+    <dd>An implementation package which sends metric data to 
+    <a href="http://ganglia.sourceforge.net/">Ganglia</a>.</dd>
+</dl>
+
+<h3>Introduction to the Metrics API</h3>
+
+Here is a simple example of how to use this package to report a single
+metric value:
+<pre>
+    private ContextFactory contextFactory = ContextFactory.getFactory();
+    
+    void reportMyMetric(float myMetric) {
+        MetricsContext myContext = contextFactory.getContext("myContext");
+        MetricsRecord myRecord = myContext.getRecord("myRecord");
+        myRecord.setMetric("myMetric", myMetric);
+        myRecord.update();
+    }
+</pre>
+  
+In this example there are three names:
+<dl>
+  <dt><i>myContext</i></dt>
+  <dd>The context name will typically identify either the application, or else a
+  module within an application or library.</dd>
+  
+  <dt><i>myRecord</i></dt>
+  <dd>The record name generally identifies some entity for which a set of
+  metrics are to be reported.  For example, you could have a record named 
+  "cacheStats" for reporting a number of statistics relating to the usage of
+  some cache in your application.</dd>
+  
+  <dt><i>myMetric</i></dt>
+  <dd>This identifies a particular metric.  For example, you might have metrics
+  named "cache_hits" and "cache_misses".
+  </dd>
+</dl>
+
+<h3>Tags</h3>
+
+In some cases it is useful to have multiple records with the same name. For 
+example, suppose that you want to report statistics about each disk on a computer. 
+In this case, the record name would be something like "diskStats", but you also
+need to identify the disk which is done by adding a <i>tag</i> to the record.
+The code could look something like this:
+<pre>
+    private MetricsRecord diskStats =
+            contextFactory.getContext("myContext").getRecord("diskStats");
+            
+    void reportDiskMetrics(String diskName, float diskBusy, float diskUsed) {
+        diskStats.setTag("diskName", diskName);
+        diskStats.setMetric("diskBusy", diskBusy);
+        diskStats.setMetric("diskUsed", diskUsed);
+        diskStats.update();
+    }
+</pre>
+
+<h3>Buffering and Callbacks</h3>
+
+Data is not sent immediately to the metrics system when 
+<code>MetricsRecord.update()</code> is called. Instead it is stored in an
+internal table, and the contents of the table are sent periodically.
+This can be important for two reasons:
+<ol>
+    <li>It means that a programmer is free to put calls to this API in an 
+    inner loop, since updates can be very frequent without slowing down
+    the application significantly.</li>
+    <li>Some implementations can gain efficiency by combining many metrics 
+    into a single UDP message.</li>
+</ol>
+
+The API provides a timer-based callback via the 
+<code>registerUpdater()</code> method.  The benefit of this
+versus using <code>java.util.Timer</code> is that the callbacks will be done 
+immediately before sending the data, making the data as current as possible.
+
+<h3>Configuration</h3>
+
+It is possible to programmatically examine and modify configuration data
+before creating a context, like this:
+<pre>
+    ContextFactory factory = ContextFactory.getFactory();
+    ... examine and/or modify factory attributes ...
+    MetricsContext context = factory.getContext("myContext");
+</pre>
+The factory attributes can be examined and modified using the following
+<code>ContextFactory</code>methods:
+<ul>
+    <li><code>Object getAttribute(String attributeName)</code></li>
+    <li><code>String[] getAttributeNames()</code></li>
+    <li><code>void setAttribute(String name, Object value)</code></li>
+    <li><code>void removeAttribute(attributeName)</code></li>
+</ul>
+
+<p/>
+<code>ContextFactory.getFactory()</code> initializes the factory attributes by
+reading the properties file <code>hadoop-metrics.properties</code> if it exists 
+on the class path.
+
+<p/>
+A factory attribute named:
+<pre>
+<i>contextName</i>.class
+</pre>
+should have as its value the fully qualified name of the class to be 
+instantiated by a call of the <code>CodeFactory</code> method
+<code>getContext(<i>contextName</i>)</code>.  If this factory attribute is not 
+specified, the default is to instantiate 
+<code>org.apache.hadoop.metrics.file.FileContext</code>.
+
+<p/>
+Other factory attributes are specific to a particular implementation of this 
+API and are documented elsewhere.  For example, configuration attributes for
+the file and Ganglia implementations can be found in the javadoc for 
+their respective packages.
+</body>
+</html>

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

@@ -0,0 +1,388 @@
+/*
+ * AbstractMetricsContext.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.TreeMap;
+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.Updater;
+
+/**
+ * The main class of the Service Provider Interface.  This class should be
+ * extended in order to integrate the Metrics API with a specific metrics
+ * client library. <p/>
+ *
+ * This class implements the internal table of metric data, and the timer
+ * on which data is to be sent to the metrics system.  Subclasses must
+ * override the abstract <code>emitRecord</code> method in order to transmit
+ * the data. <p/>
+ */
+public abstract class AbstractMetricsContext implements MetricsContext {
+    
+    private int period = MetricsContext.DEFAULT_PERIOD;
+    private Timer timer = null;
+    
+    //private Set<Updater> updaters = new HashSet<Updater>(1);
+    private Set updaters = new HashSet(1);
+    private boolean isMonitoring = false;
+    
+    private ContextFactory factory = null;
+    private String contextName = null;
+    
+    //static class TagMap extends TreeMap<String,Object> {
+    static class TagMap extends TreeMap {
+        TagMap() {
+            super();
+        }
+        TagMap(TagMap orig) {
+            super(orig);
+        }
+    }
+    //static class MetricMap extends TreeMap<String,Number> {}
+    static class MetricMap extends TreeMap {}
+            
+    //static class RecordMap extends HashMap<TagMap,MetricMap> {}
+    static class RecordMap extends HashMap {}
+    
+    //private Map<String,RecordMap> bufferedData = new HashMap<String,RecordMap>();
+    private Map bufferedData = new HashMap();
+    
+
+    /**
+     * Creates a new instance of AbstractMetricsContext
+     */
+    protected AbstractMetricsContext() {
+    }
+    
+    /**
+     * Initializes the context.
+     */
+    public void init(String contextName, ContextFactory factory) 
+    {
+        this.contextName = contextName;
+        this.factory = factory;
+    }
+    
+    /**
+     * Convenience method for subclasses to access factory attributes.
+     */
+    protected String getAttribute(String attributeName) {
+        String factoryAttribute = contextName + "." + attributeName;
+        return (String) factory.getAttribute(factoryAttribute);  
+    }
+    
+    /**
+     * Returns an attribute-value map derived from the factory attributes
+     * by finding all factory attributes that begin with 
+     * <i>contextName</i>.<i>tableName</i>.  The returned map consists of
+     * those attributes with the contextName and tableName stripped off.
+     */
+    //protected Map<String,String> getAttributeTable(String tableName) {
+    protected Map getAttributeTable(String tableName) {
+        String prefix = contextName + "." + tableName + ".";
+        //Map<String,String> result = new HashMap<String,String>();
+        //for (String attributeName : factory.getAttributeNames()) {
+        Map result = new HashMap();
+        String[] attributeNames = factory.getAttributeNames();
+        for (int i = 0; i < attributeNames.length; i++) {
+            String attributeName = attributeNames[i];
+            if (attributeName.startsWith(prefix)) {
+                String name = attributeName.substring(prefix.length());
+                String value = (String) factory.getAttribute(attributeName);
+                result.put(name, value);
+            }
+        }
+        return result;
+    }
+    
+    /**
+     * Returns the context name.
+     */
+    public String getContextName() {
+        return contextName;
+    }
+    
+    /**
+     * Returns the factory by which this context was created.
+     */
+    public ContextFactory getContextFactory() {
+        return factory;
+    }
+    
+    /**
+     * Starts or restarts monitoring, the emitting of metrics records.
+     */
+    public synchronized void startMonitoring()
+    throws IOException {
+        if (!isMonitoring) {
+            startTimer();
+            isMonitoring = true;
+        }
+    }
+    
+    /**
+     * Stops monitoring.  This does not free buffered data. 
+     * @see #close()
+     */
+    public synchronized void stopMonitoring() {
+        if (isMonitoring) {
+            stopTimer();
+            isMonitoring = false;
+        }
+    }
+    
+    /**
+     * Returns true if monitoring is currently in progress.
+     */
+    public boolean isMonitoring() {
+        return isMonitoring;
+    }
+    
+    /**
+     * Stops monitoring and frees buffered data, returning this
+     * object to its initial state.  
+     */
+    public synchronized void close() {
+        stopMonitoring();
+        clearUpdaters();
+    } 
+    
+    /**
+     * Creates a new AbstractMetricsRecord instance with the given <code>recordName</code>.
+     * Throws an exception if the metrics implementation is configured with a fixed
+     * set of record names and <code>recordName</code> is not in that set.
+     * 
+     * @param recordName the name of the record
+     * @throws MetricsException if recordName conflicts with configuration data
+     */
+    public synchronized MetricsRecord createRecord(String recordName) {
+        if (bufferedData.get(recordName) == null) {
+            bufferedData.put(recordName, new RecordMap());
+        }
+        return new MetricsRecordImpl(recordName, this);
+    }
+    
+    /**
+     * Registers a callback to be called at time intervals determined by
+     * the configuration.
+     *
+     * @param updater object to be run periodically; it should update
+     * some metrics records 
+     */
+    public synchronized void registerUpdater(final Updater updater) {
+        if (!updaters.contains(updater)) {
+            updaters.add(updater);
+        }
+    }
+    
+    /**
+     * Removes a callback, if it exists.
+     *
+     * @param updater object to be removed from the callback list
+     */
+    public synchronized void unregisterUpdater(Updater updater) {
+        updaters.remove(updater);
+    }
+    
+    private synchronized void clearUpdaters() {
+        updaters.clear();
+    }
+    
+    /**
+     * Starts timer if it is not already started
+     */
+    private synchronized void startTimer() {
+         if (timer == null) {
+            timer = new Timer();
+            TimerTask task = new TimerTask() {
+                public void run() {
+                    timerEvent();
+                }
+            };
+            long millis = period * 1000;
+            timer.schedule(task, millis, millis);
+         }
+    }
+    
+    /**
+     * Stops timer if it is running
+     */
+    private synchronized void stopTimer() {
+        if (timer != null) {
+            timer.cancel();
+            timer = null;
+        }
+    }
+    
+    /**
+     * Timer callback.
+     */
+    private synchronized void timerEvent() {
+        if (isMonitoring) {
+            // Run all the registered updates
+            // for (Updater updater : updaters) {
+            Iterator it = updaters.iterator();
+            while (it.hasNext()) {
+                Updater updater = (Updater) it.next();
+                try {
+                    updater.doUpdates(this);
+                }
+                catch (Throwable throwable) {
+                    throwable.printStackTrace();
+                }
+            }
+            
+            // Emit the records
+            //for (String recordName : bufferedData.keySet()) {
+            Iterator recordIt = bufferedData.keySet().iterator();
+            while (recordIt.hasNext()) {
+                String recordName = (String) recordIt.next();
+                RecordMap recordMap = (RecordMap) bufferedData.get(recordName);
+                synchronized (recordMap) {
+                    try {
+                        //for (TagMap tagMap : recordMap.keySet()) {
+                        Iterator tagIt = recordMap.keySet().iterator();
+                        while (tagIt.hasNext()) {
+                            TagMap tagMap = (TagMap) tagIt.next();
+                            MetricMap metricMap = (MetricMap) recordMap.get(tagMap);
+                            OutputRecord outRec = new OutputRecord(tagMap, metricMap);
+                            emitRecord(contextName, recordName, outRec);
+                        }
+                    }
+                    catch (IOException ioe) {
+                        ioe.printStackTrace();
+                    }
+                }
+            }
+            flush();
+        }
+    }
+    
+    /**
+     * Sends a record to the metrics system.
+     */
+    protected abstract void emitRecord(String contextName, String recordName, 
+                                       OutputRecord outRec) throws IOException;
+    
+    /**
+     * Called each period after all records have been emitted, this method does nothing.
+     * Subclasses may override it in order to perform some kind of flush.
+     */
+    protected void flush() {
+    }
+    
+    /**
+     * Called by MetricsRecordImpl.update().  Creates or updates a row in
+     * the internal table of metric data.
+     */
+    protected void update(MetricsRecordImpl record) {
+        String recordName = record.getRecordName();
+        TagMap tagTable = record.getTagTable();
+        //Map<String,MetricValue> metricUpdates = record.getMetricTable();
+        Map metricUpdates = record.getMetricTable();
+        
+        RecordMap recordMap = (RecordMap) bufferedData.get(recordName);
+        synchronized (recordMap) {
+            MetricMap metricMap = (MetricMap) recordMap.get(tagTable);
+            if (metricMap == null) {
+                metricMap = new MetricMap();
+                TagMap tagMap = new TagMap(tagTable); // clone tags
+                recordMap.put(tagMap, metricMap);
+            }
+            //for (String metricName : metricUpdates.keySet()) {
+            Iterator metricIt = metricUpdates.keySet().iterator();
+            while (metricIt.hasNext()) {
+                String metricName = (String) metricIt.next();
+                MetricValue updateValue = (MetricValue) metricUpdates.get(metricName);
+                Number updateNumber = updateValue.getNumber();
+                Number currentNumber = (Number) metricMap.get(metricName);
+                if (currentNumber == null || updateValue.isAbsolute()) {
+                    metricMap.put(metricName, updateNumber);
+                }
+                else {
+                    Number newNumber = sum(updateNumber, currentNumber);
+                    metricMap.put(metricName, newNumber);
+                }
+            }
+        }
+    }
+    
+    /**
+     * Adds two numbers, coercing the second to the type of the first.
+     *
+     */
+    private Number sum(Number a, Number b) {
+        if (a instanceof Integer) {
+            return Integer.valueOf(a.intValue() + b.intValue());
+        }
+        else if (a instanceof Float) {
+            return Float.valueOf(a.floatValue() + b.floatValue());
+        }
+        else if (a instanceof Short) {
+            return Short.valueOf((short)(a.shortValue() + b.shortValue()));
+        }
+        else if (a instanceof Byte) {
+            return Byte.valueOf((byte)(a.byteValue() + b.byteValue()));
+        }
+        else {
+            // should never happen
+            throw new MetricsException("Invalid number type");
+        }
+            
+    }
+    
+    /**
+     * Called by MetricsRecordImpl.remove().  Removes any matching row in
+     * the internal table of metric data.  A row matches if it has the same
+     * tag names and tag values.
+     */    
+    protected void remove(MetricsRecordImpl record) {
+        String recordName = record.getRecordName();
+        TagMap tagTable = record.getTagTable();
+        
+        RecordMap recordMap = (RecordMap) bufferedData.get(recordName);
+        synchronized (recordMap) {
+            recordMap.remove(tagTable);
+        }
+    }
+    
+    /**
+     * Returns the timer period.
+     */
+    public int getPeriod() {
+        return period;
+    }
+    
+    /**
+     * Sets the timer period
+     */
+    protected void setPeriod(int period) {
+        this.period = period;
+    }
+}

src/java/org/apache/hadoop/metrics/spi/MetricValue.java

@@ -0,0 +1,50 @@
+/*
+ * MetricValue.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+/**
+ * A Number that is either an absolute or an incremental amount.
+ */
+public class MetricValue {
+    
+    public static final boolean ABSOLUTE = false;
+    public static final boolean INCREMENT = true;
+    
+    private boolean isIncrement;
+    private Number number;
+    
+    /** Creates a new instance of MetricValue */
+    public MetricValue(Number number, boolean isIncrement) {
+        this.number = number;
+        this.isIncrement = isIncrement;
+    }
+
+    public boolean isIncrement() {
+        return isIncrement;
+    }
+    
+    public boolean isAbsolute() {
+        return !isIncrement;
+    }
+
+    public Number getNumber() {
+        return number;
+    }
+    
+}

src/java/org/apache/hadoop/metrics/spi/MetricsRecordImpl.java

@@ -0,0 +1,230 @@
+/*
+ * MetricsRecordImpl.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.hadoop.metrics.MetricsRecord;
+import org.apache.hadoop.metrics.spi.AbstractMetricsContext.TagMap;
+
+/**
+ * An implementation of MetricsRecord.  Keeps a back-pointer to the context
+ * from which it was created, and delegates back to it on <code>update</code>
+ * and <code>remove()</code>.
+ */
+public class MetricsRecordImpl implements MetricsRecord {
+    
+    private TagMap tagTable = new TagMap();
+    //private Map<String,MetricValue> metricTable = new LinkedHashMap<String,MetricValue>();
+    private Map metricTable = new LinkedHashMap();
+    
+    private String recordName;
+    private AbstractMetricsContext context;
+    
+    
+    /** Creates a new instance of FileRecord */
+    MetricsRecordImpl(String recordName, AbstractMetricsContext context)
+    {
+        this.recordName = recordName;
+        this.context = context;
+    }
+    
+    /**
+     * Returns the record name. 
+     *
+     * @return the record name
+     */
+    public String getRecordName() {
+        return recordName;
+    }
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public void setTag(String tagName, String tagValue) {
+        tagTable.put(tagName, tagValue);
+    }
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public void setTag(String tagName, int tagValue) {
+        tagTable.put(tagName, Integer.valueOf(tagValue));
+    }
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public void setTag(String tagName, short tagValue) {
+        tagTable.put(tagName, Short.valueOf(tagValue));
+    }
+    
+    /**
+     * Sets the named tag to the specified value.
+     *
+     * @param tagName name of the tag
+     * @param tagValue new value of the tag
+     * @throws MetricsException if the tagName conflicts with the configuration
+     */
+    public void setTag(String tagName, byte tagValue) {
+        tagTable.put(tagName, Byte.valueOf(tagValue));
+    }
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void setMetric(String metricName, int metricValue) {
+        setAbsolute(metricName, Integer.valueOf(metricValue));
+    }
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void setMetric(String metricName, short metricValue) {
+        setAbsolute(metricName, Short.valueOf(metricValue));
+    }
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void setMetric(String metricName, byte metricValue) {
+        setAbsolute(metricName, Byte.valueOf(metricValue));
+    }
+    
+    /**
+     * Sets the named metric to the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue new value of the metric
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void setMetric(String metricName, float metricValue) {
+        setAbsolute(metricName, Float.valueOf(metricValue));
+    }
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void incrMetric(String metricName, int metricValue) {
+        setIncrement(metricName, Integer.valueOf(metricValue));
+    }
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void incrMetric(String metricName, short metricValue) {
+        setIncrement(metricName, Short.valueOf(metricValue));
+    }
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void incrMetric(String metricName, byte metricValue) {
+        setIncrement(metricName, Byte.valueOf(metricValue));
+    }
+    
+    /**
+     * Increments the named metric by the specified value.
+     *
+     * @param metricName name of the metric
+     * @param metricValue incremental value
+     * @throws MetricsException if the metricName or the type of the metricValue 
+     * conflicts with the configuration
+     */
+    public void incrMetric(String metricName, float metricValue) {
+        setIncrement(metricName, Float.valueOf(metricValue));
+    }
+    
+    private void setAbsolute(String metricName, Number metricValue) {
+        metricTable.put(metricName, new MetricValue(metricValue, MetricValue.ABSOLUTE));
+    }
+    
+    private void setIncrement(String metricName, Number metricValue) {
+        metricTable.put(metricName, new MetricValue(metricValue, MetricValue.INCREMENT));
+    }
+    
+    /**
+     * Updates the table of buffered data which is to be sent periodically.
+     * If the tag values match an existing row, that row is updated; 
+     * otherwise, a new row is added.
+     */
+    public void update() {
+        context.update(this);
+    }
+    
+    /**
+     * Removes the row, if it exists, in the buffered data table having tags 
+     * that equal the tags that have been set on this record. 
+     */
+    public void remove() {
+        context.remove(this);
+    }
+
+    TagMap getTagTable() {
+        return tagTable;
+    }
+
+    //Map<String, MetricValue> getMetricTable() {
+    Map getMetricTable() {
+        return metricTable;
+    }
+}

src/java/org/apache/hadoop/metrics/spi/NullContext.java

@@ -0,0 +1,56 @@
+/*
+ * NullContext.java
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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;
+
+/**
+ * Null metrics context: a metrics context which does nothing.  Used as the
+ * default context, so that no performance data is emitted if no configuration
+ * data is found.
+ * 
+ */
+public class NullContext extends AbstractMetricsContext {
+    
+    /** Creates a new instance of NullContext */
+    public NullContext() {
+    }
+    
+    /**
+     * Do-nothing version of startMonitoring
+     */
+    public void startMonitoring() {
+    }
+    
+    /**
+     * Do-nothing version of emitRecord
+     */
+    protected void emitRecord(String contextName, String recordName,
+                              OutputRecord outRec) 
+    {}
+    
+    /**
+     * Do-nothing version of update
+     */
+    protected void update(MetricsRecordImpl record) {
+    }
+    
+    /**
+     * Do-nothing version of remove
+     */
+    protected void remove(MetricsRecordImpl record) {
+    }
+}

src/java/org/apache/hadoop/metrics/spi/OutputRecord.java

@@ -0,0 +1,72 @@
+/*
+ * OutputRecord.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.util.Collections;
+import java.util.Set;
+import org.apache.hadoop.metrics.spi.AbstractMetricsContext.MetricMap;
+import org.apache.hadoop.metrics.spi.AbstractMetricsContext.TagMap;
+
+/**
+ * Represents a record of metric data to be sent to a metrics system.
+ */
+public class OutputRecord {
+    
+    private TagMap tagMap;
+    private MetricMap metricMap;
+    
+    /** Creates a new instance of OutputRecord */
+    OutputRecord(TagMap tagMap, MetricMap metricMap) {
+        this.tagMap = tagMap;
+        this.metricMap = metricMap;
+    }
+    
+    /**
+     * Returns the set of tag names
+     */
+    //public Set<String> getTagNames() {
+    public Set getTagNames() {
+        return Collections.unmodifiableSet(tagMap.keySet());
+    }
+    
+    /**
+     * Returns a tag object which is can be a String, Integer, Short or Byte.
+     *
+     * @return the tag value, or null if there is no such tag
+     */
+    public Object getTag(String name) {
+        return tagMap.get(name);
+    }
+    
+    /**
+     * Returns the set of metric names.
+     */
+    //public Set<String> getMetricNames() {
+    public Set getMetricNames() {
+        return Collections.unmodifiableSet(metricMap.keySet());
+    }
+    
+    /**
+     * Returns the metric object which can be a Float, Integer, Short or Byte.
+     */
+    public Number getMetric(String name) {
+        return (Number) metricMap.get(name);
+    }
+    
+}

src/java/org/apache/hadoop/metrics/spi/Util.java

@@ -0,0 +1,64 @@
+/*
+ * Util.java
+ *
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Static utility methods
+ */
+public class Util {
+    
+    /**
+     * Not intended to be instantiated
+     */
+    private Util() {}
+    
+    /**
+     * Parses a space and/or comma separated sequence of server specifications
+     * of the form <i>hostname</i> or <i>hostname:port</i>.  
+     * 
+     * @return a list of SocketAddress objects.
+     */
+    //public static List<SocketAddress> parse(String specs, int defaultPort) {
+    public static List parse(String specs, int defaultPort) {
+        String[] specStrings = specs.split("[ ,]+");
+        //List<SocketAddress> result = new ArrayList<SocketAddress>();
+        List result = new ArrayList();
+        
+        //for (String specString : specStrings) {
+        for (int i = 0; i < specStrings.length; i++) {
+            String specString = specStrings[i];
+            int colon = specString.indexOf(':');
+            if (colon < 0 || colon == specString.length() - 1) {
+                result.add(new InetSocketAddress(specString, defaultPort));
+            } else {
+                String hostname = specString.substring(0, colon);
+                int port = Integer.parseInt(specString.substring(colon+1));
+                result.add(new InetSocketAddress(hostname, port));
+            }
+        }
+        return result;
+    }
+    
+}

src/java/org/apache/hadoop/metrics/spi/package.html

@@ -0,0 +1,35 @@
+<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
+
+<html>
+  <head>
+    <title>org.apache.hadoop.metrics.spi</title>
+  </head>
+  <body>
+ <!--
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.
+-->
+ 
+The Service Provider Interface for the Metrics API.  This package provides
+an interface allowing a variety of metrics reporting implementations to be
+plugged in to the Metrics API.  Examples of such implementations can be found 
+in the packages <code>org.apache.hadoop.metrics.file</code> and
+<code>org.apache.hadoop.metrics.ganglia</code>.<p/>
+
+Plugging in an implementation involves writing a concrete subclass of 
+<code>AbstractMetricsContext</code>.  The subclass should get its
+ configuration information using the <code>getAttribute(<i>attributeName</i>)</code>
+ method.
+  </body>
+</html>