core/src/test/java/org/apache/hadoop/chukwa/datacollection/adaptor/TestDirTailingAdaptor.java - chukwa - Git at Google

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

 import java.io.File;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
 import org.apache.hadoop.conf.*;
 import org.apache.log4j.Level;
 import junit.framework.TestCase;

 public class TestDirTailingAdaptor extends TestCase {

   ChukwaAgent agent;
   File baseDir;
   static final int SCAN_INTERVAL = 1000;

   /**
    * This test is exactly the same as testDirTailed except that it applies filtering and<br/>
    * creates a file that should not be read inorder to test the filter.
    * @throws IOException
    * @throws ChukwaAgent.AlreadyRunningException
    * @throws InterruptedException
    */
   public void testDirTailerFiltering() throws IOException,
   ChukwaAgent.AlreadyRunningException, InterruptedException {

     DirTailingAdaptor.log.setLevel(Level.DEBUG);

     Configuration conf = new Configuration();
     baseDir = new File(System.getProperty("test.build.data", "/tmp")).getCanonicalFile();
     File checkpointDir = new File(baseDir, "dirtailerTestCheckpoints");
     createEmptyDir(checkpointDir);

     conf.setInt("adaptor.dirscan.intervalMs", SCAN_INTERVAL);
     conf.set("chukwaAgent.checkpoint.dir", checkpointDir.getCanonicalPath());
     conf.set("chukwaAgent.checkpoint.name", "checkpoint_");
     conf.setInt("chukwaAgent.control.port", 0);
     conf.setBoolean("chukwaAgent.checkpoint.enabled", false);

     agent = ChukwaAgent.getAgent(conf);
     agent.start();
     File emptyDir = new File(baseDir, "emptyDir2");
     createEmptyDir(emptyDir);

     assertEquals(0, agent.adaptorCount());
     //check filtering with empty directory
     agent.processAddCommand("add emptydir2= DirTailingAdaptor raw " + emptyDir + " *file filetailer.CharFileTailingAdaptorUTF8 0");
     assertEquals(1, agent.adaptorCount());

     File dirWithFile = new File(baseDir, "dir3");
     dirWithFile.delete();
     assertFalse("temp directory not empty",dirWithFile.exists());

     //this file should be found by the filter
     dirWithFile.mkdir();
     File inDir = File.createTempFile("atemp", "file", dirWithFile);
     inDir.deleteOnExit();
     //This file should not be found by the filter
     File noreadFile = File.createTempFile("atemp", "noread", dirWithFile);
     noreadFile.deleteOnExit();

     //apply filter *file
     agent.processAddCommand("add dir3= DirTailingAdaptor raw " + dirWithFile + " *file filetailer.CharFileTailingAdaptorUTF8 0");
     Thread.sleep(3000);
     assertEquals(3, agent.adaptorCount());
     agent.shutdown();

     conf.setBoolean("chukwaAgent.checkpoint.enabled", true);
     Thread.sleep(1500); //wait a little bit to make sure new file ts is > last checkpoint time.
     File anOldFile = File.createTempFile("oldXYZ","file", dirWithFile);
     File aNewFile = File.createTempFile("new", "file", dirWithFile);
     anOldFile.deleteOnExit();
     aNewFile.deleteOnExit();
     anOldFile.setLastModified(10);//just after epoch
     agent = ChukwaAgent.getAgent(conf); //restart agent.
     agent.start();

    Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
    assertTrue(aNewFile.exists());

     //make sure we started tailing the new, not the old, file.
     for(Map.Entry<String, String> adaptors : agent.getAdaptorList().entrySet()) {
       System.out.println(adaptors.getKey() +": " + adaptors.getValue());
       assertFalse(adaptors.getValue().contains("oldXYZ"));
     }
     Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
     //should be four adaptors: the DirTailer on emptyDir, the DirTailer on the full dir,
     //and FileTailers for File inDir and file newfile and not the noread file.
     assertEquals(4, agent.adaptorCount());
     agent.shutdown();
     Thread.sleep(1500); //wait a little bit to make sure new file ts is > last checkpoint time.

     nukeDirContents(checkpointDir);//nuke dir
     checkpointDir.delete();
     emptyDir.delete();
     nukeDirContents(dirWithFile);
     dirWithFile.delete();
   }


   public void testDirTailer() throws IOException,
   ChukwaAgent.AlreadyRunningException, InterruptedException {

     DirTailingAdaptor.log.setLevel(Level.DEBUG);

     Configuration conf = new Configuration();
     baseDir = new File(System.getProperty("test.build.data", "/tmp")).getCanonicalFile();
     File checkpointDir = new File(baseDir, "dirtailerTestCheckpoints");
     createEmptyDir(checkpointDir);

     conf.setInt("adaptor.dirscan.intervalMs", SCAN_INTERVAL);
     conf.set("chukwaAgent.checkpoint.dir", checkpointDir.getCanonicalPath());
     conf.set("chukwaAgent.checkpoint.name", "checkpoint_");
     conf.setInt("chukwaAgent.control.port", 0);
     conf.setBoolean("chukwaAgent.checkpoint.enabled", false);

     boolean retry = true;
     while(retry) {
       try {
         retry = false;
         agent = ChukwaAgent.getAgent(conf);
         agent.start();
       } catch(Exception e) {
         retry = true;
       }
     }
     File emptyDir = new File(baseDir, "emptyDir");
     createEmptyDir(emptyDir);

     assertEquals(0, agent.adaptorCount());
     agent.processAddCommand("add emptydir= DirTailingAdaptor raw " + emptyDir + " filetailer.CharFileTailingAdaptorUTF8 0");
     assertEquals(1, agent.adaptorCount());

     File dirWithFile = new File(baseDir, "dir2");
     dirWithFile.delete();
     assertFalse("temp directory not empty",dirWithFile.exists());

     dirWithFile.mkdir();
     File inDir = File.createTempFile("atemp", "file", dirWithFile);
     inDir.deleteOnExit();
     agent.processAddCommand("add dir2= DirTailingAdaptor raw " + dirWithFile + " *file filetailer.CharFileTailingAdaptorUTF8 0");
     Thread.sleep(3000);
     assertEquals(3, agent.adaptorCount());
     System.out.println("DirTailingAdaptor looks OK before restart");
     agent.shutdown();

     conf.setBoolean("chukwaAgent.checkpoint.enabled", true);
     Thread.sleep(1500); //wait a little bit to make sure new file ts is > last checkpoint time.
     File anOldFile = File.createTempFile("oldXYZ","file", dirWithFile);
     File aNewFile = File.createTempFile("new", "file", dirWithFile);
     anOldFile.deleteOnExit();
     aNewFile.deleteOnExit();
     anOldFile.setLastModified(10);//just after epoch
     agent = ChukwaAgent.getAgent(conf); //restart agent.
     agent.start();

    Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
    assertTrue(aNewFile.exists());

     //make sure we started tailing the new, not the old, file.
     for(Map.Entry<String, String> adaptors : agent.getAdaptorList().entrySet()) {
       System.out.println(adaptors.getKey() +": " + adaptors.getValue());
       assertFalse(adaptors.getValue().contains("oldXYZ"));
     }
     //should be four adaptors: the DirTailer on emptyDir, the DirTailer on the full dir,
     //and FileTailers for File inDir and file newfile
     Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
     assertEquals(4, agent.adaptorCount());
     agent.shutdown();

     nukeDirContents(checkpointDir);//nuke dir
     checkpointDir.delete();
     emptyDir.delete();
     nukeDirContents(dirWithFile);
     dirWithFile.delete();
   }

   //returns true if dir exists
   public static boolean nukeDirContents(File dir) {
     if(dir.exists()) {
       if(dir.isDirectory()) {
         for(File f: dir.listFiles()) {
           nukeDirContents(f);
           f.delete();
         }
       } else
         dir.delete();

       return true;
     }
     return false;
   }

   public static void createEmptyDir(File dir) {
     if(!nukeDirContents(dir))
       dir.mkdir();
     assertTrue(dir.isDirectory() && dir.listFiles().length == 0);
   }

 }
	/*
	* 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.adaptor;

	import java.io.File;
	import java.io.IOException;
	import java.util.Map;
	import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
	import org.apache.hadoop.conf.*;
	import org.apache.log4j.Level;
	import junit.framework.TestCase;

	public class TestDirTailingAdaptor extends TestCase {

	ChukwaAgent agent;
	File baseDir;
	static final int SCAN_INTERVAL = 1000;

	/**
	* This test is exactly the same as testDirTailed except that it applies filtering and<br/>
	* creates a file that should not be read inorder to test the filter.
	* @throws IOException
	* @throws ChukwaAgent.AlreadyRunningException
	* @throws InterruptedException
	*/
	public void testDirTailerFiltering() throws IOException,
	ChukwaAgent.AlreadyRunningException, InterruptedException {

	DirTailingAdaptor.log.setLevel(Level.DEBUG);

	Configuration conf = new Configuration();
	baseDir = new File(System.getProperty("test.build.data", "/tmp")).getCanonicalFile();
	File checkpointDir = new File(baseDir, "dirtailerTestCheckpoints");
	createEmptyDir(checkpointDir);

	conf.setInt("adaptor.dirscan.intervalMs", SCAN_INTERVAL);
	conf.set("chukwaAgent.checkpoint.dir", checkpointDir.getCanonicalPath());
	conf.set("chukwaAgent.checkpoint.name", "checkpoint_");
	conf.setInt("chukwaAgent.control.port", 0);
	conf.setBoolean("chukwaAgent.checkpoint.enabled", false);

	agent = ChukwaAgent.getAgent(conf);
	agent.start();
	File emptyDir = new File(baseDir, "emptyDir2");
	createEmptyDir(emptyDir);

	assertEquals(0, agent.adaptorCount());
	//check filtering with empty directory
	agent.processAddCommand("add emptydir2= DirTailingAdaptor raw " + emptyDir + " *file filetailer.CharFileTailingAdaptorUTF8 0");
	assertEquals(1, agent.adaptorCount());

	File dirWithFile = new File(baseDir, "dir3");
	dirWithFile.delete();
	assertFalse("temp directory not empty",dirWithFile.exists());

	//this file should be found by the filter
	dirWithFile.mkdir();
	File inDir = File.createTempFile("atemp", "file", dirWithFile);
	inDir.deleteOnExit();
	//This file should not be found by the filter
	File noreadFile = File.createTempFile("atemp", "noread", dirWithFile);
	noreadFile.deleteOnExit();

	//apply filter *file
	agent.processAddCommand("add dir3= DirTailingAdaptor raw " + dirWithFile + " *file filetailer.CharFileTailingAdaptorUTF8 0");
	Thread.sleep(3000);
	assertEquals(3, agent.adaptorCount());
	agent.shutdown();

	conf.setBoolean("chukwaAgent.checkpoint.enabled", true);
	Thread.sleep(1500); //wait a little bit to make sure new file ts is > last checkpoint time.
	File anOldFile = File.createTempFile("oldXYZ","file", dirWithFile);
	File aNewFile = File.createTempFile("new", "file", dirWithFile);
	anOldFile.deleteOnExit();
	aNewFile.deleteOnExit();
	anOldFile.setLastModified(10);//just after epoch
	agent = ChukwaAgent.getAgent(conf); //restart agent.
	agent.start();

	Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
	assertTrue(aNewFile.exists());

	//make sure we started tailing the new, not the old, file.
	for(Map.Entry<String, String> adaptors : agent.getAdaptorList().entrySet()) {
	System.out.println(adaptors.getKey() +": " + adaptors.getValue());
	assertFalse(adaptors.getValue().contains("oldXYZ"));
	}
	Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
	//should be four adaptors: the DirTailer on emptyDir, the DirTailer on the full dir,
	//and FileTailers for File inDir and file newfile and not the noread file.
	assertEquals(4, agent.adaptorCount());
	agent.shutdown();
	Thread.sleep(1500); //wait a little bit to make sure new file ts is > last checkpoint time.

	nukeDirContents(checkpointDir);//nuke dir
	checkpointDir.delete();
	emptyDir.delete();
	nukeDirContents(dirWithFile);
	dirWithFile.delete();
	}


	public void testDirTailer() throws IOException,
	ChukwaAgent.AlreadyRunningException, InterruptedException {

	DirTailingAdaptor.log.setLevel(Level.DEBUG);

	Configuration conf = new Configuration();
	baseDir = new File(System.getProperty("test.build.data", "/tmp")).getCanonicalFile();
	File checkpointDir = new File(baseDir, "dirtailerTestCheckpoints");
	createEmptyDir(checkpointDir);

	conf.setInt("adaptor.dirscan.intervalMs", SCAN_INTERVAL);
	conf.set("chukwaAgent.checkpoint.dir", checkpointDir.getCanonicalPath());
	conf.set("chukwaAgent.checkpoint.name", "checkpoint_");
	conf.setInt("chukwaAgent.control.port", 0);
	conf.setBoolean("chukwaAgent.checkpoint.enabled", false);

	boolean retry = true;
	while(retry) {
	try {
	retry = false;
	agent = ChukwaAgent.getAgent(conf);
	agent.start();
	} catch(Exception e) {
	retry = true;
	}
	}
	File emptyDir = new File(baseDir, "emptyDir");
	createEmptyDir(emptyDir);

	assertEquals(0, agent.adaptorCount());
	agent.processAddCommand("add emptydir= DirTailingAdaptor raw " + emptyDir + " filetailer.CharFileTailingAdaptorUTF8 0");
	assertEquals(1, agent.adaptorCount());

	File dirWithFile = new File(baseDir, "dir2");
	dirWithFile.delete();
	assertFalse("temp directory not empty",dirWithFile.exists());

	dirWithFile.mkdir();
	File inDir = File.createTempFile("atemp", "file", dirWithFile);
	inDir.deleteOnExit();
	agent.processAddCommand("add dir2= DirTailingAdaptor raw " + dirWithFile + " *file filetailer.CharFileTailingAdaptorUTF8 0");
	Thread.sleep(3000);
	assertEquals(3, agent.adaptorCount());
	System.out.println("DirTailingAdaptor looks OK before restart");
	agent.shutdown();

	conf.setBoolean("chukwaAgent.checkpoint.enabled", true);
	Thread.sleep(1500); //wait a little bit to make sure new file ts is > last checkpoint time.
	File anOldFile = File.createTempFile("oldXYZ","file", dirWithFile);
	File aNewFile = File.createTempFile("new", "file", dirWithFile);
	anOldFile.deleteOnExit();
	aNewFile.deleteOnExit();
	anOldFile.setLastModified(10);//just after epoch
	agent = ChukwaAgent.getAgent(conf); //restart agent.
	agent.start();

	Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
	assertTrue(aNewFile.exists());

	//make sure we started tailing the new, not the old, file.
	for(Map.Entry<String, String> adaptors : agent.getAdaptorList().entrySet()) {
	System.out.println(adaptors.getKey() +": " + adaptors.getValue());
	assertFalse(adaptors.getValue().contains("oldXYZ"));
	}
	//should be four adaptors: the DirTailer on emptyDir, the DirTailer on the full dir,
	//and FileTailers for File inDir and file newfile
	Thread.sleep(3 * SCAN_INTERVAL); //wait a bit for the new file to be detected.
	assertEquals(4, agent.adaptorCount());
	agent.shutdown();

	nukeDirContents(checkpointDir);//nuke dir
	checkpointDir.delete();
	emptyDir.delete();
	nukeDirContents(dirWithFile);
	dirWithFile.delete();
	}

	//returns true if dir exists
	public static boolean nukeDirContents(File dir) {
	if(dir.exists()) {
	if(dir.isDirectory()) {
	for(File f: dir.listFiles()) {
	nukeDirContents(f);
	f.delete();
	}
	} else
	dir.delete();

	return true;
	}
	return false;
	}

	public static void createEmptyDir(File dir) {
	if(!nukeDirContents(dir))
	dir.mkdir();
	assertTrue(dir.isDirectory() && dir.listFiles().length == 0);
	}

	}