/*
 * 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.tools.backfilling;

import java.io.File;

import org.apache.hadoop.chukwa.ChunkImpl;
import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
import org.apache.hadoop.chukwa.datacollection.ChunkQueue;
import org.apache.hadoop.chukwa.datacollection.DataFactory;
import org.apache.hadoop.chukwa.datacollection.adaptor.*;
import org.apache.hadoop.chukwa.datacollection.agent.AdaptorFactory;
import org.apache.hadoop.chukwa.datacollection.agent.AdaptorManager;
import org.apache.hadoop.chukwa.datacollection.connector.Connector;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

public class BackfillingLoader {
  static Logger log = Logger.getLogger(BackfillingLoader.class);
  
  protected Configuration conf = null;
  protected ChunkQueue queue = null;
  protected Connector connector = null;
  
  private String cluster =  null;
  private String machine =  null;
  private String adaptorName =  null;
  private String recordType =  null;
  private String logFile =  null;
  
  public BackfillingLoader(Configuration conf, String cluster, String machine, 
      String adaptorName, String recordType, String logFile) {
    
    this.conf = conf;
    this.cluster = cluster.trim();
    this.machine = machine.trim();
    this.adaptorName = adaptorName;
    this.recordType = recordType;
    this.logFile = logFile;
    
    log.info("cluster >>>" + cluster) ;
    log.info("machine >>>" + machine) ;
    log.info("adaptorName >>>" + adaptorName) ;
    log.info("recordType >>>" + recordType) ;
    log.info("logFile >>>" + logFile) ;
    
    // Set the right cluster and machine information
    DataFactory.getInstance().addDefaultTag("cluster=\"" + this.cluster + "\"");
    ChunkImpl.setHostAddress(this.machine);
    
    queue = DataFactory.getInstance().getEventQueue();
    connector = new QueueToWriterConnector(conf,true);
  }
  
  public void process() throws AdaptorException {
    File file = new File(logFile);
    connector.start();
    Adaptor adaptor = AdaptorFactory.createAdaptor(adaptorName);
    adaptor.parseArgs(recordType, "0 " +file.getAbsolutePath(),AdaptorManager.NULL);
    adaptor.start("", recordType,  0l,queue);
    adaptor.shutdown(AdaptorShutdownPolicy.WAIT_TILL_FINISHED);
    connector.shutdown();
    if(!file.renameTo(new File(logFile + ".sav"))) {
      System.err.println("Error in renaming "+logFile+" to "+logFile+".sav");
    }
  }
  
  public static void usage() {
    System.out.println("java org.apache.hadoop.chukwa.tools.backfilling.BackfillingLoader <cluster> <machine> <adaptorName> <recordType> <logFile>");
  }
  
  /**
   * @param args is command line parameters
   * @throws Exception if problem loading data to HDFS
   */
  public static void main(String[] args) throws Exception {

    if (args.length != 5) {
	usage();
	return;
    }
    

    String cluster = args[0];
    String machine = args[1];
    String adaptorName = args[2];
    String recordType = args[3];
    String logFile = args[4];

    BackfillingLoader loader = new BackfillingLoader(new ChukwaConfiguration(),cluster,machine,adaptorName,recordType,logFile);
    loader.process();
  }

}