content: /*
   * 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);
      }
    }
  }
  