Added -signaturelength option

Bug corrections on saved state loading
Bug corrections on file reading

src/main/java/info/fetter/logstashforwarder/FileReader.java

@@ -52,7 +52,7 @@ public class FileReader {
 		eventList = new ArrayList<Event>(spoolSize);
 	}
 
-	public int readFiles(Collection<FileState> fileList) throws IOException, AdapterException {
+	public int readFiles(Collection<FileState> fileList) throws AdapterException {
 		int eventCount = 0;
 		if(logger.isTraceEnabled()) {
 			logger.trace("Reading " + fileList.size() + " file(s)");
@@ -71,12 +71,12 @@ public class FileReader {
 		return eventCount; // Return number of events sent to adapter
 	}
 
-	private int readFile(FileState state, int spaceLeftInSpool) throws IOException {
+	private int readFile(FileState state, int spaceLeftInSpool) {
 		int eventListSizeBefore = eventList.size();
 		File file = state.getFile();
 		long pointer = state.getPointer();
 		if(logger.isTraceEnabled()) {
-			logger.trace("File : " + file.getCanonicalPath() + " pointer : " + pointer);
+			logger.trace("File : " + file + " pointer : " + pointer);
 			logger.trace("Space left in spool : " + spaceLeftInSpool);
 		}
 		pointer = readLines(state, spaceLeftInSpool);
@@ -84,22 +84,27 @@ public class FileReader {
 		return eventList.size() - eventListSizeBefore; // Return number of events read
 	}
 
-	private long readLines(FileState state, int spaceLeftInSpool) throws IOException {
+	private long readLines(FileState state, int spaceLeftInSpool) {
 		RandomAccessFile reader = state.getRandomAccessFile();
 		long pos = state.getPointer();
-		reader.seek(pos);
-		String line = readLine(reader);
-		while (line != null && spaceLeftInSpool > 0) {
-			if(logger.isTraceEnabled()) {
-				logger.trace("-- Read line : " + line);
-				logger.trace("-- Space left in spool : " + spaceLeftInSpool);
+		try {
+			reader.seek(pos);
+			String line = readLine(reader);
+			while (line != null && spaceLeftInSpool > 0) {
+				if(logger.isTraceEnabled()) {
+					logger.trace("-- Read line : " + line);
+					logger.trace("-- Space left in spool : " + spaceLeftInSpool);
+				}
+				pos = reader.getFilePointer();
+				addEvent(state, pos, line);
+				line = readLine(reader);
+				spaceLeftInSpool--;
 			}
-			pos = reader.getFilePointer();
-			addEvent(state, pos, line);
-			line = readLine(reader);
-			spaceLeftInSpool--;
+			reader.seek(pos); // Ensure we can re-read if necessary
+		} catch(IOException e) {
+			logger.warn("Exception raised while reading file : " + state.getFile());
+			e.printStackTrace();
 		}
-		reader.seek(pos); // Ensure we can re-read if necessary
 		return pos;
 	}
 

src/main/java/info/fetter/logstashforwarder/FileState.java

@@ -18,6 +18,7 @@ package info.fetter.logstashforwarder;
  */
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 
@@ -47,10 +48,10 @@ public class FileState {
 	private FileState oldFileState;
 	@JsonIgnore
 	private Event fields;
-	
+
 	public FileState() {
 	}
-	
+
 	public FileState(File file) throws IOException {
 		this.file = file;
 		directory = file.getCanonicalFile().getParent();
@@ -59,9 +60,16 @@ public class FileState {
 		lastModified = file.lastModified();
 		size = file.length();
 	}
-	
-	private void setFileFromDirectoryAndName() {
-		this.file = new File(directory + File.separator + fileName);
+
+	private void setFileFromDirectoryAndName() throws FileNotFoundException {
+		file = new File(directory + File.separator + fileName);
+		if(file.exists()) {
+			randomAccessFile = new RandomAccessFile(file, "r");
+			lastModified = file.lastModified();
+			size = file.length();
+		} else {
+			deleted = true;
+		}
 	}
 
 	public File getFile() {
@@ -75,41 +83,41 @@ public class FileState {
 	public long getSize() {
 		return size;
 	}
-	
+
 	public String getDirectory() {
 		return directory;
 	}
-	
-	public void setDirectory(String directory) {
+
+	public void setDirectory(String directory) throws FileNotFoundException {
 		this.directory = directory;
 		if(fileName != null && directory != null) {
 			setFileFromDirectoryAndName();
 		}
 	}
-	
+
 	public String getFileName() {
 		return fileName;
 	}
-	
-	public void setFileName(String fileName) {
+
+	public void setFileName(String fileName) throws FileNotFoundException {
 		this.fileName = fileName;
 		if(fileName != null && directory != null) {
 			setFileFromDirectoryAndName();
 		}
 	}
-	
+
 	public boolean isDeleted() {
 		return deleted;
 	}
-	
+
 	public void setDeleted() {
 		deleted = true;
 	}
-	
+
 	public boolean hasChanged() {
 		return changed;
 	}
-	
+
 	public void setChanged(boolean changed) {
 		this.changed = changed;
 	}
@@ -129,7 +137,7 @@ public class FileState {
 	public long getPointer() {
 		return pointer;
 	}
-	
+
 	public void setPointer(long pointer) {
 		this.pointer = pointer;
 	}
@@ -157,16 +165,16 @@ public class FileState {
 	public void setFields(Event fields) {
 		this.fields = fields;
 	}
-	
+
 	@Override
 	public String toString() {
-	     return new ToStringBuilder(this).
-	    	       append("fileName", fileName).
-	    	       append("directory", directory).
-	    	       append("pointer", pointer).
-	    	       append("signature", signature).
-	    	       append("signatureLength", signatureLength).
-	    	       toString();
+		return new ToStringBuilder(this).
+				append("fileName", fileName).
+				append("directory", directory).
+				append("pointer", pointer).
+				append("signature", signature).
+				append("signatureLength", signatureLength).
+				toString();
 	}
-	
+
 }

src/main/java/info/fetter/logstashforwarder/FileWatcher.java

@@ -39,10 +39,10 @@ public class FileWatcher {
 	private static final Logger logger = Logger.getLogger(FileWatcher.class);
 	private List<FileAlterationObserver> observerList = new ArrayList<FileAlterationObserver>();
 	public static final int ONE_DAY = 24 * 3600 * 1000;
-	private Map<File,FileState> watchMap = new HashMap<File,FileState>();
-	private Map<File,FileState> changedWatchMap = new HashMap<File,FileState>();
+	private Map<File,FileState> oldWatchMap = new HashMap<File,FileState>();
+	private Map<File,FileState> newWatchMap = new HashMap<File,FileState>();
 	private FileState[] savedStates;
-	private static int MAX_SIGNATURE_LENGTH = 1024;
+	private int maxSignatureLength;
 
 	public FileWatcher() {
 		try {
@@ -52,6 +52,17 @@ public class FileWatcher {
 		}
 	}
 
+	public void initialize() throws IOException {
+		logger.debug("Initializing FileWatcher");
+		if(savedStates != null) {
+			for(FileState state : savedStates) {
+				oldWatchMap.put(state.getFile(), state);
+			}
+		}
+		processModifications();
+		printWatchMap();
+	}
+
 	public void addFilesToWatch(String fileToWatch, Event fields, int deadTime) {
 		try {
 			if(fileToWatch.equals("-")) {
@@ -79,15 +90,15 @@ public class FileWatcher {
 	public int readFiles(FileReader reader) throws IOException, AdapterException {
 		logger.debug("Reading files");
 		logger.trace("==============");
-		int numberOfLinesRead = reader.readFiles(watchMap.values());
-		Registrar.writeStateToJson(watchMap.values());
+		int numberOfLinesRead = reader.readFiles(oldWatchMap.values());
+		Registrar.writeStateToJson(oldWatchMap.values());
 		return numberOfLinesRead;
 	}
 
 	private void processModifications() throws IOException {
 
-		for(File file : changedWatchMap.keySet()) {
-			FileState state = changedWatchMap.get(file);
+		for(File file : newWatchMap.keySet()) {
+			FileState state = newWatchMap.get(file);
 			if(logger.isTraceEnabled()) {
 				logger.trace("Checking file : " + file.getCanonicalPath());
 				logger.trace("-- Last modified : " + state.getLastModified());
@@ -97,7 +108,7 @@ public class FileWatcher {
 			}
 
 			logger.trace("Determine if file has just been written to");
-			FileState oldState = watchMap.get(file);
+			FileState oldState = oldWatchMap.get(file);
 			if(oldState != null) {
 				if(oldState.getSize() > state.getSize()) {
 					logger.trace("File shorter : file can't be the same");
@@ -123,8 +134,8 @@ public class FileWatcher {
 
 			if(state.getOldFileState() == null) {
 				logger.trace("Determine if file has been renamed and/or written to");
-				for(File otherFile : watchMap.keySet()) {
-					FileState otherState = watchMap.get(otherFile);
+				for(File otherFile : oldWatchMap.keySet()) {
+					FileState otherState = oldWatchMap.get(otherFile);
 					if(otherState != null && state.getSize() >= otherState.getSize() && state.getDirectory().equals(otherState.getDirectory())) {
 						if(logger.isTraceEnabled()) {
 							logger.trace("Comparing to : " + otherFile.getCanonicalPath());
@@ -151,29 +162,31 @@ public class FileWatcher {
 		}
 
 		logger.trace("Refreshing file state");
-		for(File file : changedWatchMap.keySet()) {
+		for(File file : newWatchMap.keySet()) {
 			if(logger.isTraceEnabled()) {
 				logger.trace("Refreshing file : " + file.getCanonicalPath());
 			}
-			FileState state = changedWatchMap.get(file);
+			FileState state = newWatchMap.get(file);
 			FileState oldState = state.getOldFileState();
 			if(oldState == null) {
 				logger.trace("File has been truncated or created, not retrieving pointer");
 			} else {
 				logger.trace("File has not been truncated or created, retrieving pointer");
 				state.setPointer(oldState.getPointer());
-				oldState.getRandomAccessFile().close();
+				try {
+					oldState.getRandomAccessFile().close();
+				} catch(Exception e) {}
 			}
 		}
 
 		logger.trace("Replacing old state");
-		for(File file : changedWatchMap.keySet()) {
-			FileState state = changedWatchMap.get(file);
-			watchMap.put(file, state);
+		for(File file : newWatchMap.keySet()) {
+			FileState state = newWatchMap.get(file);
+			oldWatchMap.put(file, state);
 		}
 
 		// Truncating changedWatchMap
-		changedWatchMap.clear();
+		newWatchMap.clear();
 
 		removeMarkedFilesFromWatchMap();
 	}
@@ -216,46 +229,7 @@ public class FileWatcher {
 		observerList.add(observer);
 		observer.initialize();
 		for(File file : FileUtils.listFiles(directory, fileFilter, null)) {
-			FileState savedState = null;
-			if(savedStates != null) {
-				for(FileState state : savedStates) {
-					logger.trace("Comparing file : " + file + " with saved file : " + state.getFile());
-					if(file.equals(state.getFile())) {
-						savedState = state;
-						logger.debug("Match found with saved file " + state.getFile());
-					}
-				}
-			}
-			if(savedState == null) {
-				addFileToWatchMap(watchMap, file, fields);
-			} else {
-				addSavedFileToWatchMap(savedState, fields);
-			}
-		}
-	}
-
-	private void addSavedFileToWatchMap(FileState savedFileState, Event fields) {
-		try {
-			File file = savedFileState.getFile();
-			FileState state = new FileState(file);
-			state.setFields(fields);
-			int savedSignatureLength = savedFileState.getSignatureLength();
-			state.setSignatureLength(savedSignatureLength);
-			long savedSignature = savedFileState.getSignature();
-			int signatureLength = (int) (state.getSize() > MAX_SIGNATURE_LENGTH ? MAX_SIGNATURE_LENGTH : state.getSize());
-			long signature = FileSigner.computeSignature(state.getRandomAccessFile(), signatureLength);
-			if(signature == savedSignature) { 
-				state.setPointer(savedFileState.getPointer());
-				logger.debug("Restoring signature of size : " + savedSignatureLength + " on file : " + file + " : " + savedSignature);
-			} else {
-				logger.debug("File " + file + " signature has changed");
-				logger.trace("Setting signature of size : " + signatureLength + " on file : " + file + " : " + signature);
-			}
-			state.setSignatureLength(signatureLength);
-			state.setSignature(signature);
-			watchMap.put(file, state);
-		} catch (IOException e) {
-			logger.error("Caught IOException : " + e.getMessage());
+			addFileToWatchMap(newWatchMap, file, fields);
 		}
 	}
 
@@ -263,7 +237,7 @@ public class FileWatcher {
 		try {
 			FileState state = new FileState(file);
 			state.setFields(fields);
-			int signatureLength = (int) (state.getSize() > MAX_SIGNATURE_LENGTH ? MAX_SIGNATURE_LENGTH : state.getSize());
+			int signatureLength = (int) (state.getSize() > maxSignatureLength ? maxSignatureLength : state.getSize());
 			state.setSignatureLength(signatureLength);
 			long signature = FileSigner.computeSignature(state.getRandomAccessFile(), signatureLength);
 			state.setSignature(signature);
@@ -277,7 +251,7 @@ public class FileWatcher {
 	public void onFileChange(File file, Event fields) {
 		try {
 			logger.debug("Change detected on file : " + file.getCanonicalPath());
-			addFileToWatchMap(changedWatchMap, file, fields);
+			addFileToWatchMap(newWatchMap, file, fields);
 		} catch (IOException e) {
 			logger.error("Caught IOException : " + e.getMessage());
 		}	
@@ -286,7 +260,7 @@ public class FileWatcher {
 	public void onFileCreate(File file, Event fields) {
 		try {
 			logger.debug("Create detected on file : " + file.getCanonicalPath());
-			addFileToWatchMap(changedWatchMap, file, fields);
+			addFileToWatchMap(newWatchMap, file, fields);
 		} catch (IOException e) {
 			logger.error("Caught IOException : " + e.getMessage());
 		}
@@ -295,7 +269,7 @@ public class FileWatcher {
 	public void onFileDelete(File file) {
 		try {
 			logger.debug("Delete detected on file : " + file.getCanonicalPath());
-			watchMap.get(file).setDeleted();
+			oldWatchMap.get(file).setDeleted();
 		} catch (IOException e) {
 			logger.error("Caught IOException : " + e.getMessage());
 		}
@@ -304,8 +278,8 @@ public class FileWatcher {
 	private void printWatchMap() throws IOException {
 		if(logger.isTraceEnabled()) {
 			logger.trace("WatchMap contents : ");
-			for(File file : watchMap.keySet()) {
-				FileState state = watchMap.get(file);
+			for(File file : oldWatchMap.keySet()) {
+				FileState state = oldWatchMap.get(file);
 				logger.trace("\tFile : " + file.getCanonicalPath() + " marked for deletion : " + state.isDeleted());
 			}
 		}
@@ -314,8 +288,8 @@ public class FileWatcher {
 	private void removeMarkedFilesFromWatchMap() throws IOException {
 		logger.trace("Removing deleted files from watchMap");
 		List<File> markedList = null;
-		for(File file : watchMap.keySet()) {
-			FileState state = watchMap.get(file);
+		for(File file : oldWatchMap.keySet()) {
+			FileState state = oldWatchMap.get(file);
 			if(state.isDeleted()) {
 				if(markedList == null) {
 					markedList = new ArrayList<File>();
@@ -325,19 +299,29 @@ public class FileWatcher {
 		}
 		if(markedList != null) {
 			for(File file : markedList) {
-				FileState state = watchMap.remove(file);
-				state.getRandomAccessFile().close();
-				logger.trace("\tFile : " + file.getCanonicalFile() + " removed");
+				FileState state = oldWatchMap.remove(file);
+				try {
+					state.getRandomAccessFile().close();
+				} catch(Exception e) {}
+				logger.trace("\tFile : " + file + " removed");
 			}
 		}
 	}
 
 	public void close() throws IOException {
 		logger.debug("Closing all files");
-		for(File file : watchMap.keySet()) {
-			FileState state = watchMap.get(file);
+		for(File file : oldWatchMap.keySet()) {
+			FileState state = oldWatchMap.get(file);
 			state.getRandomAccessFile().close();
 		}
 	}
 
+	public int getMaxSignatureLength() {
+		return maxSignatureLength;
+	}
+
+	public void setMaxSignatureLength(int maxSignatureLength) {
+		this.maxSignatureLength = maxSignatureLength;
+	}
+
 }

src/main/java/info/fetter/logstashforwarder/Forwarder.java

@@ -52,6 +52,7 @@ public class Forwarder {
 	private static Level logLevel = INFO;
 	private static ProtocolAdapter adapter;
 	private static Random random = new Random();
+	private static int signatureLength = 4096;
 
 	public static void main(String[] args) {
 		try {
@@ -62,6 +63,7 @@ public class Forwarder {
 			//			Logger.getLogger(FileReader.class).setLevel(TRACE);
 			//			Logger.getLogger(FileReader.class).setAdditivity(false);
 			watcher = new FileWatcher();
+			watcher.setMaxSignatureLength(signatureLength);
 			configManager = new ConfigurationManager(config);
 			configManager.readConfiguration();
 			for(FilesSection files : configManager.getConfig().getFiles()) {
@@ -69,6 +71,7 @@ public class Forwarder {
 					watcher.addFilesToWatch(path, new Event(files.getFields()), FileWatcher.ONE_DAY);
 				}
 			}
+			watcher.initialize();
 			reader = new FileReader(spoolSize);
 			connectToServer();
 			infiniteLoop();
@@ -128,6 +131,10 @@ public class Forwarder {
 				.isRequired()
 				.withDescription("path to logstash-forwarder configuration file")
 				.create("config");
+		Option signatureLengthOption = OptionBuilder.withArgName("signature length")
+				.hasArg()
+				.withDescription("Maximum length of file signature")
+				.create("signaturelength");
 
 		options.addOption(helpOption)
 		.addOption(idleTimeoutOption)
@@ -135,6 +142,7 @@ public class Forwarder {
 		.addOption(quiet)
 		.addOption(debug)
 		.addOption(trace)
+		.addOption(signatureLengthOption)
 		.addOption(configOption);	
 		CommandLineParser parser = new GnuParser();
 		try {
@@ -148,6 +156,9 @@ public class Forwarder {
 			if(line.hasOption("config")) {
 				config = line.getOptionValue("config");
 			}
+			if(line.hasOption("signaturelength")) {
+				signatureLength = Integer.parseInt(line.getOptionValue("signaturelength"));
+			}
 			if(line.hasOption("quiet")) {
 				logLevel = ERROR;
 			}

src/test/java/info/fetter/logstashforwarder/FileWatcherTest.java

@@ -54,17 +54,18 @@ public class FileWatcherTest {
 		}
 	}
 	
-	//@Test
+	@Test
 	public void testWildcardWatch() throws InterruptedException, IOException {
 		FileWatcher watcher = new FileWatcher();
-		watcher.addFilesToWatch("./test*.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY);
+		watcher.addFilesToWatch("./testFileWatcher*.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY);
+		watcher.initialize();
 
-		File file1 = new File("test1.txt");
-		File file2 = new File("test2.txt");
+		File file1 = new File("testFileWatcher1.txt");
+		File file2 = new File("testFileWatcher2.txt");
 		//File file3 = new File("test3.txt");
 		//File file4 = new File("test4.txt");
 		
-		File testDir = new File("test");
+		//File testDir = new File("testFileWatcher");
 		//FileUtils.forceMkdir(new File("test"));
 		
 		watcher.checkFiles();
@@ -76,29 +77,22 @@ public class FileWatcherTest {
 		FileUtils.write(file2, "file 2 line 1\n", true);
 		Thread.sleep(1000);
 		watcher.checkFiles();
-		FileUtils.moveFileToDirectory(file1, testDir, true);
-		FileUtils.write(file2, "file 2 line 2\n", true);
-		FileUtils.moveFile(file2, file1);
-		FileUtils.write(file2, "file 3 line 1\n", true);
-//		FileUtils.touch(file3);
+//		FileUtils.moveFileToDirectory(file1, testDir, true);
+//		FileUtils.write(file2, "file 2 line 2\n", true);
+//		FileUtils.moveFile(file2, file1);
+//		FileUtils.write(file2, "file 3 line 1\n", true);
+//
+//		Thread.sleep(1000);
+//		watcher.checkFiles();
+//		
+//		
+//		watcher.close();
 //		FileUtils.forceDelete(file1);
 //		FileUtils.forceDelete(file2);
-		Thread.sleep(1000);
-		watcher.checkFiles();
+//		FileUtils.forceDelete(testDir);
 		
 		
-		watcher.close();
-		FileUtils.forceDelete(file1);
-		FileUtils.forceDelete(file2);
-		FileUtils.forceDelete(testDir);
-		
-		
-//		FileUtils.moveFile(file3, file4);
-//		FileUtils.touch(file3);
-//		Thread.sleep(500);
-//		watcher.checkFiles();
-//		FileUtils.forceDelete(file3);
-//		FileUtils.forceDelete(file4);
+
 	}
 	
 	@Test