src/test/java/org/apache/hadoop/chukwa/datacollection/collector/TestDelayedAcks.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.datacollection.collector;

import java.io.File;
import java.util.*;
import java.util.regex.*;
import org.apache.hadoop.chukwa.*;
import org.apache.hadoop.chukwa.datacollection.adaptor.AdaptorShutdownPolicy;
import org.apache.hadoop.chukwa.datacollection.adaptor.TestDirTailingAdaptor;
import org.apache.hadoop.chukwa.datacollection.adaptor.filetailer.FileTailingAdaptor;
import org.apache.hadoop.chukwa.datacollection.agent.AdaptorResetThread;
import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
import org.apache.hadoop.chukwa.datacollection.collector.servlet.CommitCheckServlet;
import org.apache.hadoop.chukwa.datacollection.collector.servlet.ServletCollector;
import org.apache.hadoop.chukwa.datacollection.connector.ChunkCatcherConnector;
import org.apache.hadoop.chukwa.datacollection.connector.http.HttpConnector;
import org.apache.hadoop.chukwa.datacollection.sender.*;
import org.apache.hadoop.chukwa.datacollection.sender.ChukwaHttpSender.CommitListEntry;
import org.apache.hadoop.chukwa.datacollection.writer.ChukwaWriter;
import org.apache.hadoop.chukwa.datacollection.writer.SeqFileWriter;
import org.apache.hadoop.chukwa.extraction.archive.SinkArchiver;
import org.apache.hadoop.chukwa.util.ConstRateAdaptor;
import org.apache.hadoop.chukwa.util.ConstRateValidator.ByteRange;
import org.apache.hadoop.chukwa.util.ConstRateValidator.ValidatorSM;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.mortbay.jetty.Server;
import org.mortbay.jetty.servlet.Context;
import org.mortbay.jetty.servlet.ServletHolder;
import junit.framework.TestCase;
import static org.apache.hadoop.chukwa.datacollection.sender.AsyncAckSender.DelayedCommit;
import static org.apache.hadoop.chukwa.util.TempFileUtil.*;

public class TestDelayedAcks extends TestCase {
  
  static final int PORTNO = 9993;
  static int END2END_TEST_SECS = 30;
  static int SEND_RATE = 180* 1000; //bytes/sec
  static int CLIENT_SCANPERIOD = 1000;
  static int SERVER_SCANPERIOD = 1000;
  static int ROTATEPERIOD = 2000;
  
  int ACK_TIMEOUT = 200;
  

//start an adaptor -- chunks should appear in the connector
    //wait for timeout.  More chunks should appear.
  public void testAdaptorTimeout() throws Exception {
    Configuration conf = new Configuration();
    conf.set("chukwaAgent.control.port", "0");
    conf.setBoolean("chukwaAgent.checkpoint.enabled", false);
    conf.setInt("chukwaAgent.adaptor.context.switch.time", 500);
    conf.setInt(AdaptorResetThread.TIMEOUT_OPT, ACK_TIMEOUT);

    ChukwaAgent agent = ChukwaAgent.getAgent(conf);
    agent.start();
    ChunkCatcherConnector chunks = new ChunkCatcherConnector();
    chunks.start();
    assertEquals(0, agent.adaptorCount());
    File testFile = makeTestFile("testDA", 50, new File(System.getProperty("test.build.data", "/tmp")));
    long len = testFile.length();
    System.out.println("wrote data to " + testFile);
    AdaptorResetThread restart = new AdaptorResetThread(conf, agent);
    //start timeout thread
    agent.processAddCommand("add fta = "+ FileTailingAdaptor.class.getCanonicalName()
        + " testdata " + testFile.getCanonicalPath() + " 0" );
    
    
    assertEquals(1, agent.adaptorCount());
    Chunk c1 = chunks.waitForAChunk();
    assertNotNull(c1);
    List<CommitListEntry> pendingAcks = new ArrayList<CommitListEntry>();
    pendingAcks.add(new DelayedCommit(c1.getInitiator(), c1.getSeqID(),
        c1.getData().length, "foo", c1.getSeqID(), agent.getAdaptorName(c1.getInitiator())));
    restart.reportPending(pendingAcks);

    assertEquals(len, c1.getData().length);
    Thread.sleep(ACK_TIMEOUT*2);
    int resetCount = restart.resetTimedOutAdaptors(ACK_TIMEOUT);
    Chunk c2 = chunks.waitForAChunk(1000);
    assertNotNull(c2);
    assertEquals(len, c2.getData().length);
    assertTrue(resetCount > 0);
    agent.shutdown();
    
    testFile.delete();
  }
  
  /*
   * Checks the CommitCheckServlet works correctly with a one-chunk file.
   */
  public void testDelayedAck() throws Exception {
    Configuration conf = new Configuration();

    SeqFileWriter writer = new SeqFileWriter();

    conf.set("writer.hdfs.filesystem", "file:///");
    
    File tempDir = new File(System.getProperty("test.build.data", "/tmp"));
    if (!tempDir.exists()) {
      tempDir.mkdirs();
    }
    
    String outputDirectory = tempDir.getPath() + "/test_DA" + System.currentTimeMillis();

    String seqWriterOutputDir = outputDirectory +"/seqWriter/seqOutputDir";
    conf.set(SeqFileWriter.OUTPUT_DIR_OPT, seqWriterOutputDir );

    writer.init(conf);
    ArrayList<Chunk> oneChunk = new ArrayList<Chunk>();
    oneChunk.add(new ChunkImpl("dt", "name", 1, new byte[] {'b'}, null));

    ChukwaWriter.CommitStatus cs = writer.add(oneChunk);
    writer.close();
    
    File seqWriterFile = null;
    File directory = new File(seqWriterOutputDir);
    String[] files = directory.list();
    for(String file: files) {
      if ( file.endsWith(".done") ){
        seqWriterFile = new File(directory, file);
        break;
      }
    }
    long lenWritten = seqWriterFile.length();
    System.out.println("wrote " + lenWritten+ " bytes");
    assertTrue(cs instanceof ChukwaWriter.COMMIT_PENDING);
    ChukwaWriter.COMMIT_PENDING pending = (ChukwaWriter.COMMIT_PENDING) cs;
    assertTrue(pending.pendingEntries.size() == 1);
    String res = pending.pendingEntries.get(0);
    System.out.println("result was " + res);
    
    Pattern expectedPat= Pattern.compile(".* ([0-9]+)\n");
    Matcher match = expectedPat.matcher(res);
    assertTrue(match.matches());
    long bytesPart = Long.parseLong(match.group(1));
    assertEquals(bytesPart, lenWritten);
  }
  

  public static Server startCollectorOnPort(Configuration conf, int port, 
      ServletCollector collector) throws Exception {
    Server server = new Server(port);
    
    Context root = new Context(server, "/", Context.SESSIONS);
    root.addServlet(new ServletHolder(collector), "/*");
    root.addServlet(new ServletHolder(new CommitCheckServlet(conf)), "/"+CommitCheckServlet.DEFAULT_PATH);

    server.start();
    server.setStopAtShutdown(false);
    return server;
  }
  

  public static String buildConf(Configuration conf) {
    File tempDir = new File(System.getProperty("test.build.data", "/tmp"));
    if (!tempDir.exists()) {
      tempDir.mkdirs();
    }
    
    String outputDirectory = tempDir.getPath() + "/test_DA" + System.currentTimeMillis() ;

    conf.setInt("chukwaCollector.rotateInterval", ROTATEPERIOD);
    conf.set("writer.hdfs.filesystem", "file:///");
    String seqWriterOutputDir = outputDirectory +"/chukwa_sink";
    conf.set(SeqFileWriter.OUTPUT_DIR_OPT, seqWriterOutputDir );
    conf.setInt(AsyncAckSender.POLLPERIOD_OPT, CLIENT_SCANPERIOD);
    conf.setInt(CommitCheckServlet.SCANPERIOD_OPT, SERVER_SCANPERIOD);
    conf.setBoolean(HttpConnector.ASYNC_ACKS_OPT, true);
    conf.setInt(HttpConnector.MIN_POST_INTERVAL_OPT, 100);
    conf.setInt(HttpConnector.MAX_SIZE_PER_POST_OPT, 10 * 1000*1000);
    conf.setInt(SeqFileWriter.STAT_PERIOD_OPT, 60*60*24); 
    conf.setBoolean("chukwaAgent.checkpoint.enabled", false);
    //turn off stats reporting thread, so we can use Writer.dataSize
    conf.set(AsyncAckSender.POLLHOSTS_OPT, "afilethatdoesntexist");
      //so that it won't try to read conf/collectors
    conf.setInt("chukwaAgent.control.port", 0);
    return outputDirectory;
  }
  
  public void testEndToEnd() {
    try {
      Configuration conf = new Configuration();

      String outputDirectory = buildConf(conf);
      ServletCollector collector = new ServletCollector(conf);
      Server collectorServ = startCollectorOnPort(conf, PORTNO, collector);
      Thread.sleep(1000);
      
      ChukwaAgent agent = ChukwaAgent.getAgent(conf);
      agent.start();
      HttpConnector conn = new HttpConnector(agent, "http://localhost:"+PORTNO+"/");
      conn.start();
      String resp = agent.processAddCommand("add constSend = " + ConstRateAdaptor.class.getCanonicalName() + 
          " testData "+ SEND_RATE + " 0");
      assertTrue("adaptor_constSend".equals(resp));
      Thread.sleep(END2END_TEST_SECS * 1000);

      //do the shutdown directly, here, so that acks are still processed.
      assertNotNull(agent.getAdaptor("adaptor_constSend"));
      long bytesOutput = agent.getAdaptor("adaptor_constSend").shutdown(AdaptorShutdownPolicy.GRACEFULLY);
      Thread.sleep(CLIENT_SCANPERIOD + SERVER_SCANPERIOD + ROTATEPERIOD + 3000);
      
      String[] stat = agent.getAdaptorList().get("adaptor_constSend").split(" ");
      long bytesCommitted = Long.valueOf(stat[stat.length -1]);
      
      long bytesPerSec = bytesOutput / (1000 * END2END_TEST_SECS);
      System.out.println("data rate was " + bytesPerSec + " kb /second");
   
      //all data should be committed
      System.out.println(bytesCommitted + " bytes committed");
      System.out.println(bytesOutput + " bytes output");
      System.out.println("difference is " + (bytesOutput - bytesCommitted));
      ChukwaWriter w = collector.getWriter();
      long bytesWritten = ((SeqFileWriter)w).getBytesWritten();
      System.out.println("collector wrote " + bytesWritten);

      assertEquals(bytesCommitted, bytesOutput);
      assertEquals(bytesWritten, bytesCommitted);
      //We need a little imprecision here, since the send rate is a bit bursty,
      //and since some acks got lost after the adaptor was stopped.
      assertTrue(bytesPerSec > 9 * SEND_RATE/ 1000 / 10);
      AsyncAckSender sender = (AsyncAckSender)conn.getSender();
      assertEquals(0, sender.adaptorReset.getResetCount());
      
      agent.shutdown();
      collectorServ.stop();
      conn.shutdown();
      Thread.sleep(5000); //for collector to shut down
      TestDirTailingAdaptor.nukeDirContents(new File(outputDirectory));
      
    } catch (Exception e) {
      e.printStackTrace();
      fail(e.toString());
    }

  }


}