/*
    * 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.chukwa.database;
  
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import java.util.*;
  import java.io.File;
  import java.io.FilenameFilter;
  
  public class DatabaseConfig {
    private Configuration config = null;
    public final static long CENTURY = 36500 * 24 * 60 * 60 * 1000L;
    public final static long DECADE = 3650 * 24 * 60 * 60 * 1000L;
    public final static long YEAR = 365 * 24 * 60 * 60 * 1000L;
    public final static long QUARTER = 91250 * 24 * 60 * 60L;
    public final static long MONTH = 30 * 24 * 60 * 60 * 1000L;
    public final static long WEEK = 7 * 24 * 60 * 60 * 1000L;
    public final static long DAY = 24 * 60 * 60 * 1000L;
    public final static String MDL_XML = "mdl.xml";
  
    public DatabaseConfig(String path) {
      Path fileResource = new Path(path);
      config = new Configuration();
      config.addResource(fileResource);
    }
  
    public DatabaseConfig() {
      String dataConfig = System.getenv("CHUKWA_CONF_DIR");
      if (dataConfig == null) {
        dataConfig = MDL_XML;
      } else {
        dataConfig += File.separator + MDL_XML;
      }
      Path fileResource = new Path(dataConfig);
      config = new Configuration();
      config.addResource(fileResource);
      
      if (System.getenv("CHUKWA_CONF_DIR") != null) {
        // Allow site-specific MDL files to be included in the 
        // configuration so as to keep the "main" mdl.xml pure.
        File confDir = new File(System.getenv("CHUKWA_CONF_DIR"));
        File[] confFiles = confDir.listFiles(new FilenameFilter() {
  
          @Override
          public boolean accept(File dir, String name) {
            // Implements a naming convention of ending with "mdl.xml"
            // but is careful not to pick up mdl.xml itself again.
            return name.endsWith(MDL_XML) && !name.equals(MDL_XML);
          }
  
        });
  
        if (confFiles != null) {
          for (File confFile : confFiles) 
            config.addResource(new Path(confFile.getAbsolutePath()));
        }
      }
    }
  
    public String get(String key) {
      return config.get(key);
    }
  
    public void put(String key, String value) {
      this.config.set(key, value);
    }
  
    public Iterator<?> iterator() {
      return this.config.iterator();
    }
  
    public HashMap<String, String> startWith(String key) {
      HashMap<String, String> transformer = new HashMap<String, String>();
      Iterator<?> entries = config.iterator();
      while (entries.hasNext()) {
        String entry = entries.next().toString();
        if (entry.startsWith(key)) {
          String[] metrics = entry.split("=");
          transformer.put(metrics[0], metrics[1]);
        }
      }
     return transformer;
   }
 
   public String[] findTableName(String tableName, long start, long end) {
     String[] tableNames = null;
     String tableType = "_week";
     long now = (new Date()).getTime();
     long timeWindow = end - start;
     long partitionSize = WEEK;
     boolean fallback = true;
 
     if (config.get("consolidator.table." + tableName) == null) {
       tableNames = new String[1];
       tableNames[0] = tableName;
       return tableNames;
     }
     if (timeWindow <= 0) {
       timeWindow = 1;
     }
     if (timeWindow > DECADE) {
       tableType = "_century";
       partitionSize = CENTURY;
     } else if (timeWindow > YEAR) {
       tableType = "_decade";
       partitionSize = DECADE;
     } else if (timeWindow > QUARTER) {
       tableType = "_year";
       partitionSize = YEAR;
     } else if (timeWindow > MONTH) {
       tableType = "_quarter";
       partitionSize = QUARTER;
     } else if (timeWindow > WEEK) {
       tableType = "_month";
       partitionSize = MONTH;
     } else {
       tableType = "_week";
       partitionSize = WEEK;
     }
 
     long currentPartition = now / partitionSize;
     long startPartition = start / partitionSize;
     long endPartition = end / partitionSize;
     while (fallback && partitionSize != CENTURY * 100) {
       // Check if the starting date is in the far distance from current time. If
       // it is, use down sampled data.
       if (startPartition + 2 < currentPartition) {
         fallback = true;
         if (partitionSize == DAY) {
           tableType = "_week";
           partitionSize = WEEK;
         } else if (partitionSize == WEEK) {
           tableType = "_month";
           partitionSize = MONTH;
         } else if (partitionSize == MONTH) {
           tableType = "_year";
           partitionSize = YEAR;
         } else if (partitionSize == YEAR) {
           tableType = "_decade";
           partitionSize = DECADE;
         } else if (partitionSize == DECADE) {
           tableType = "_century";
           partitionSize = CENTURY;
         } else {
           partitionSize = 100 * CENTURY;
         }
         currentPartition = now / partitionSize;
         startPartition = start / partitionSize;
         endPartition = end / partitionSize;
       } else {
         fallback = false;
       }
     }
 
     if (startPartition != endPartition) {
       int delta = (int) (endPartition - startPartition);
       tableNames = new String[delta + 1];
       for (int i = 0; i <= delta; i++) {
         long partition = startPartition + (long) i;
         tableNames[i] = tableName + "_" + partition + tableType;
       }
     } else {
       tableNames = new String[1];
       tableNames[0] = tableName + "_" + startPartition + tableType;
     }
     return tableNames;
   }
 
   public String[] findTableNameForCharts(String tableName, long start, long end) {
     String[] tableNames = null;
     String tableType = "_week";
     long now = (new Date()).getTime();
     long timeWindow = end - start;
     if (timeWindow > 60 * 60 * 1000) {
       timeWindow = timeWindow + 1;
     }
     long partitionSize = WEEK;
     boolean fallback = true;
 
     if (config.get("consolidator.table." + tableName) == null) {
       tableNames = new String[1];
       tableNames[0] = tableName;
       return tableNames;
     }
 
     if (timeWindow <= 0) {
       timeWindow = 1;
     }
     if (timeWindow > DECADE) {
       tableType = "_decade";
       partitionSize = CENTURY;
     } else if (timeWindow > YEAR) {
       tableType = "_decade";
       partitionSize = CENTURY;
     } else if (timeWindow > QUARTER) {
       tableType = "_decade";
       partitionSize = DECADE;
     } else if (timeWindow > MONTH) {
       tableType = "_year";
       partitionSize = YEAR;
     } else if (timeWindow > WEEK) {
       tableType = "_quarter";
       partitionSize = QUARTER;
     } else if (timeWindow > DAY) {
       tableType = "_month";
       partitionSize = MONTH;
     } else {
       tableType = "_week";
       partitionSize = WEEK;
     }
 
 
     long currentPartition = now / partitionSize;
     long startPartition = start / partitionSize;
     long endPartition = end / partitionSize;
     while (fallback && partitionSize != DECADE * 100) {
       // Check if the starting date is in the far distance from current time. If
       // it is, use down sampled data.
       if (startPartition + 2 < currentPartition) {
         fallback = true;
         if (partitionSize == DAY) {
           tableType = "_month";
           partitionSize = MONTH;
         } else if (partitionSize == WEEK) {
           tableType = "_quarter";
           partitionSize = QUARTER;
         } else if (partitionSize == MONTH) {
           tableType = "_year";
           partitionSize = YEAR;
         } else if (partitionSize == YEAR) {
           tableType = "_decade";
           partitionSize = DECADE;
         } else {
           partitionSize = CENTURY;
         }
         currentPartition = now / partitionSize;
         startPartition = start / partitionSize;
         endPartition = end / partitionSize;
       } else {
         fallback = false;
       }
     }
 
     if (startPartition != endPartition) {
       int delta = (int) (endPartition - startPartition);
       tableNames = new String[delta + 1];
       for (int i = 0; i <= delta; i++) {
         long partition = startPartition + (long) i;
         tableNames[i] = tableName + "_" + partition + tableType;
       }
     } else {
       tableNames = new String[1];
       tableNames[0] = tableName + "_" + startPartition + tableType;
     }
 
     return tableNames;
   }
 
   public static void main(String[] args) {
     DatabaseConfig dbc = new DatabaseConfig();
     String[] names = dbc.findTableName("system_metrics", 1216140020000L,
         1218645620000L);
     for (String n : names) {
       System.out.println("name:" + n);
     }
   }
 }