/*
    * AbstractMetricsContext.java
    *
    * 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.util.ArrayList;
  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 java.util.Timer;
  import java.util.TimerTask;
  import java.util.TreeMap;
  import java.util.Map.Entry;
  
  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. 
   *
   * 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.
   */
  public abstract class AbstractMetricsContext implements MetricsContext {
      
    private int period = MetricsContext.DEFAULT_PERIOD;
    private Timer timer = null;
    private boolean computeRate = true;    
    private Set<Updater> updaters = new HashSet<Updater>(1);
    private volatile boolean isMonitoring = false;
      
    private ContextFactory factory = null;
    private String contextName = null;
      
    static class TagMap extends TreeMap<String,Object> {
      private static final long serialVersionUID = 3546309335061952993L;
      TagMap() {
        super();
      }
      TagMap(TagMap orig) {
        super(orig);
      }
      /**
       * Returns true if this tagmap contains every tag in other.
       * @param other 
       * @return 
       */
      public boolean containsAll(TagMap other) {
        for (Map.Entry<String,Object> entry : other.entrySet()) {
          Object value = get(entry.getKey());
          if (value == null || !value.equals(entry.getValue())) {
            // either key does not exist here, or the value is different
            return false;
          }
        }
        return true;
      }
    }
    
    static class MetricMap extends TreeMap<String,Number> {
      private static final long serialVersionUID = -7495051861141631609L;
    }
              
    static class RecordMap extends HashMap<TagMap,MetricMap> {
      private static final long serialVersionUID = 259835619700264611L;
    }
      
    private Map<String,RecordMap> bufferedData = new HashMap<String,RecordMap>();
      
  
    /**
    * 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) {
     String prefix = contextName + "." + tableName + ".";
     Map<String,String> result = new HashMap<String,String>();
     for (String attributeName : factory.getAttributeNames()) {
       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.
    * @return the factory by which the 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 final synchronized MetricsRecord createRecord(String recordName) {
     if (bufferedData.get(recordName) == null) {
       bufferedData.put(recordName, new RecordMap());
     }
     return newRecord(recordName);
   }
     
   /**
    * Subclasses should override this if they subclass MetricsRecordImpl.
    * @param recordName the name of the record
    * @return newly created instance of MetricsRecordImpl or subclass
    */
   protected MetricsRecord newRecord(String recordName) {
     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("Timer thread for monitoring " + getContextName(), 
                         true);
       TimerTask task = new TimerTask() {
           public void run() {
             try {
               timerEvent();
             }
             catch (IOException ioe) {
               ioe.printStackTrace();
             }
           }
         };
       long millis = period * 1000;
       timer.scheduleAtFixedRate(task, millis, millis);
     }
   }
     
   /**
    * Stops timer if it is running
    */
   private synchronized void stopTimer() {
     if (timer != null) {
       timer.cancel();
       timer = null;
     }
   }
     
   /**
    * Timer callback.
    */
   private void timerEvent() throws IOException {
     if (isMonitoring) {
       Collection<Updater> myUpdaters;
       synchronized (this) {
         myUpdaters = new ArrayList<Updater>(updaters);
       }     
       // Run all the registered updates without holding a lock
       // on this context
       for (Updater updater : myUpdaters) {
         try {
           updater.doUpdates(this);
         }
         catch (Throwable throwable) {
           throwable.printStackTrace();
         }
       }
       emitRecords();
     }
   }
     
   /**
    *  Emits the records.
    */
   private synchronized void emitRecords() throws IOException {
     for (Entry<String, RecordMap> record : bufferedData.entrySet()) {
       String recordName = record.getKey();
       RecordMap recordMap = record.getValue();
       for (Entry<TagMap, MetricMap> entry : record.getValue().entrySet()) {
         OutputRecord outRec = new OutputRecord(entry.getKey(), entry.getValue());
         emitRecord(contextName, recordName, outRec);
       }
     }
     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() throws IOException {
   }
     
   /**
    * 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();
         
     RecordMap recordMap = getRecordMap(recordName);
     synchronized (recordMap) {
       MetricMap metricMap = recordMap.get(tagTable);
       if (metricMap == null) {
         metricMap = new MetricMap();
         TagMap tagMap = new TagMap(tagTable); // clone tags
         recordMap.put(tagMap, metricMap);
       }
 
       Set<Entry<String, MetricValue>> entrySet = metricUpdates.entrySet();
       for (Entry<String, MetricValue> entry : entrySet) {
         String metricName = entry.getKey ();
         MetricValue updateValue = entry.getValue ();
         Number updateNumber = updateValue.getNumber();
         Number currentNumber = metricMap.get(metricName);
         if (currentNumber == null || updateValue.isAbsolute()) {
           metricMap.put(metricName, updateNumber);
         }
         else {
           Number newNumber = sum(updateNumber, currentNumber);
           metricMap.put(metricName, newNumber);
           metricMap.put(metricName+"_raw", updateNumber);
           if (computeRate ) {
               double rate = updateNumber.doubleValue() * 60.0 / period;
               metricMap.put(metricName+"_rate", rate);
             }
           computeRate = true;
         }
       }
     }
   }
     
   private synchronized RecordMap getRecordMap(String recordName) {
     return bufferedData.get(recordName);
   }
     
   /**
    * 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 new Float(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 if (a instanceof Long) {
       return Long.valueOf((a.longValue() + b.longValue()));
     }
     else {
       // should never happen
       throw new MetricsException("Invalid number type");
     }
             
   }
     
   /**
    * Called by MetricsRecordImpl.remove().  Removes all matching rows in
    * the internal table of metric data.  A row matches if it has the same
    * tag names and values as record, but it may also have additional
    * tags.
    */    
   protected void remove(MetricsRecordImpl record) {
     String recordName = record.getRecordName();
     TagMap tagTable = record.getTagTable();
         
     RecordMap recordMap = getRecordMap(recordName);
     synchronized (recordMap) {
       Iterator<TagMap> it = recordMap.keySet().iterator();
       while (it.hasNext()) {
         TagMap rowTags = it.next();
         if (rowTags.containsAll(tagTable)) {
           it.remove();
         }
       }
     }
   }
     
   /**
    * Returns the timer period.
    */
   public int getPeriod() {
     return period;
   }
     
   /**
    * Sets the timer period
    */
   protected void setPeriod(int period) {
     this.period = period;
   }
 }