core/src/main/java/org/apache/hadoop/chukwa/extraction/engine/datasource/record/RecordDS.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.chukwa.extraction.engine.datasource.record;


import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.LinkedList;
import java.util.List;
import java.util.TreeMap;
import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
import org.apache.hadoop.chukwa.extraction.engine.Record;
import org.apache.hadoop.chukwa.extraction.engine.SearchResult;
import org.apache.hadoop.chukwa.extraction.engine.Token;
import org.apache.hadoop.chukwa.extraction.engine.datasource.DataSource;
import org.apache.hadoop.chukwa.extraction.engine.datasource.DataSourceException;
import org.apache.hadoop.chukwa.inputtools.mdl.DataConfig;
import org.apache.hadoop.fs.FileSystem;

public class RecordDS implements DataSource {

  private static FileSystem fs = null;
  private static ChukwaConfiguration conf = null;

  private static String rootFolder = null;
  private static DataConfig dataConfig = null;

  static {
    dataConfig = new DataConfig();
    rootFolder = dataConfig.get("chukwa.engine.dsDirectory.rootFolder");
    conf = new ChukwaConfiguration();
    try {
      fs = FileSystem.get(conf);
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  public SearchResult search(SearchResult result, String cluster,
      String dataSource, long t0, long t1, String filter, Token token)
      throws DataSourceException {

    String filePath = rootFolder + "/" + cluster + "/" + dataSource;

    System.out.println("filePath [" + filePath + "]");
    Calendar calendar = Calendar.getInstance();
    calendar.setTimeInMillis(t1);

    TreeMap<Long, List<Record>> records = result.getRecords();
    int maxCount = 200;
    SimpleDateFormat sdf = new java.text.SimpleDateFormat("_yyyyMMdd_HH_");
    do {
      System.out.println("start Date [" + calendar.getTime() + "]");
      String fileName = sdf.format(calendar.getTime());
      int minutes = calendar.get(Calendar.MINUTE);
      int dec = minutes / 10;
      fileName += dec;

      int m = minutes - (dec * 10);
      if (m < 5) {
        fileName += "0.1.evt";
      } else {
        fileName += "5.1.evt";
      }

      fileName = filePath + "/" + dataSource + fileName;

      // System.out.println("JB fileName  [" +fileName + "]");

      try {
        System.out.println("BEFORE fileName  [" + fileName + "]");

        // List<Record> evts =
        // ChukwaFileParser.readData(cluster,dataSource,maxCount, t1, t0,
        // Long.MAX_VALUE, filter, fileName, fs);
        List<Record> evts = ChukwaSequenceFileParser.readData(cluster,
            dataSource, maxCount, t1, t0, Long.MAX_VALUE, filter, fileName, fs,
            conf);

        maxCount = maxCount - evts.size();
        System.out.println("AFTER fileName  [" + fileName + "] count="
            + evts.size() + " maxCount=" + maxCount);
        for (Record evt : evts) {
          System.out.println("AFTER Loop  [" + evt.toString() + "]");
          long timestamp = evt.getTime();
          if (records.containsKey(timestamp)) {
            records.get(timestamp).add(evt);
          } else {
            List<Record> list = new LinkedList<Record>();
            list.add(evt);
            records.put(timestamp, list);
          }
        }
      } catch (Exception e) {
        e.printStackTrace();
      }

      if (maxCount <= 0) {
        System.out.println("BREAKING LOOP AFTER [" + fileName + "] maxCount="
            + maxCount);
        break;
      }

      calendar.add(Calendar.MINUTE, -5);

      System.out.println("calendar  [" + calendar.getTimeInMillis() + "] ");
      System.out.println("end       [" + (t0 - 1000 * 60 * 5) + "] ");
    } while (calendar.getTimeInMillis() > (t0 - 1000 * 60 * 5)); // <= need some
                                                                 // code here
    // Need more than this to compute the end

    return result;
  }

  public boolean isThreadSafe() {
    return true;
  }
}