Merge pull request #1 from Sentido-Labs/master

Merge multiline supports

.classpath

@@ -27,5 +27,10 @@
 			<attribute name="maven.pomderived" value="true"/>
 		</attributes>
 	</classpathentry>
+	<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
 	<classpathentry kind="output" path="target/classes"/>
 </classpath>

.settings/org.eclipse.core.resources.prefs

@@ -0,0 +1,6 @@
+eclipse.preferences.version=1
+encoding//src/main/java=UTF-8
+encoding//src/main/resources=UTF-8
+encoding//src/test/java=UTF-8
+encoding//src/test/resources=UTF-8
+encoding/<project>=UTF-8

README.md

@@ -43,6 +43,7 @@ For help run :
 The configuration file is the same (json format), but there are a few differences :
   - the ssl ca parameter points to a java [keystore](https://github.com/didfet/logstash-forwarder-java/blob/master/HOWTO-KEYSTORE.md) containing the root certificate of the server, not a PEM file
   - comments are C-style comments
+  - multiline support with attributes "pattern", "negate" (true/false) and what (Previous/Next)
 
 ### Command-line options
 
@@ -61,4 +62,3 @@ There are a few more options :
   - logfile : send logs to this file instead of stdout
   - logfilesize : maximum size of each log file (default 10M)
   - logfilenumber : number of rotated log files (default 5)
-

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

@@ -25,10 +25,12 @@ import org.apache.commons.io.monitor.FileAlterationObserver;
 public class FileModificationListener implements FileAlterationListener {
 	private Event fields;
 	private FileWatcher watcher;
+	private Multiline multiline;
 
-	public FileModificationListener(FileWatcher watcher, Event fields) {
+	public FileModificationListener(FileWatcher watcher, Event fields, Multiline multiline) {
 		this.watcher = watcher;
 		this.fields = fields;
+		this.multiline = multiline;
 	}
 
 	public void onDirectoryChange(File file) {
@@ -44,11 +46,11 @@ public class FileModificationListener implements FileAlterationListener {
 	}
 
 	public void onFileChange(File file) {
-		watcher.onFileChange(file, fields);
+		watcher.onFileChange(file, fields, multiline);
 	}
 
 	public void onFileCreate(File file) {
-		watcher.onFileCreate(file, fields);
+		watcher.onFileCreate(file, fields, multiline);
 	}
 
 	public void onFileDelete(File file) {

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

@@ -23,10 +23,12 @@ import info.fetter.logstashforwarder.util.RandomAccessFile;
 import java.io.File;
 import java.io.IOException;
 //import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.commons.lang.ArrayUtils;
 
 import org.apache.log4j.Logger;
 
@@ -122,22 +124,75 @@ public class FileReader extends Reader {
 		return false;
 	}
 
+	private static byte[] extractBytes(ByteBuffer byteBuffer)
+	{
+		byte[] bytes = new byte[byteBuffer.position()];
+		byteBuffer.rewind();
+		byteBuffer.get(bytes);
+		byteBuffer.clear();
+		return bytes;
+	}
+
 	private long readLines(FileState state, int spaceLeftInSpool) {
 		RandomAccessFile reader = state.getRandomAccessFile();
 		long pos = state.getPointer();
+		Multiline multiline = state.getMultiline();
 		try {
 			reader.seek(pos);
 			byte[] line = readLine(reader);
+			ByteBuffer bufferedLines = ByteBuffer.allocate(BYTEBUFFER_CAPACITY);
+			bufferedLines.clear();
 			while (line != null && spaceLeftInSpool > 0) {
 				if(logger.isTraceEnabled()) {
 					logger.trace("-- Read line : " + new String(line));
 					logger.trace("-- Space left in spool : " + spaceLeftInSpool);
 				}
 				pos = reader.getFilePointer();
+				if (multiline == null) {
 					addEvent(state, pos, line);
-				line = readLine(reader);
 					spaceLeftInSpool--;
 				}
+				else {
+					if (logger.isTraceEnabled()) {
+						logger.trace("-- Multiline : " + multiline);
+						logger.trace("-- Multiline : matches " + multiline.isPatternFound(line));
+					}
+					if (multiline.isPatternFound(line)) {
+						// buffer the line
+						if (bufferedLines.position() > 0) {
+							bufferedLines.put(Multiline.JOINT);
+						}
+						bufferedLines.put(line);
+					}
+					else {
+						if (multiline.isPrevious()) {
+							// did not match, so new event started
+							if (bufferedLines.position() > 0) {
+								addEvent(state, pos, extractBytes(bufferedLines));
+								spaceLeftInSpool--;
+							}
+							bufferedLines.put(line);
+						}
+						else {
+							// did not match, add the current line
+							if (bufferedLines.position() > 0) {
+								bufferedLines.put(Multiline.JOINT);
+								bufferedLines.put(line);
+								addEvent(state, pos, extractBytes(bufferedLines));
+								spaceLeftInSpool--;
+							}
+							else {
+								addEvent(state, pos, line);
+								spaceLeftInSpool--;
+							}
+						}
+					}
+				}
+				line = readLine(reader);
+			}
+			if (bufferedLines.position() > 0) {
+				addEvent(state, pos, extractBytes(bufferedLines)); // send any buffered lines left
+			}
 			reader.seek(pos); // Ensure we can re-read if necessary
 		} catch(IOException e) {
 			logger.warn("Exception raised while reading file : " + state.getFile(), e);
@@ -152,10 +207,7 @@ public class FileReader extends Reader {
 		while((ch=reader.read()) != -1) {
 			switch(ch) {
 			case '\n':
-				byte[] line = new byte[byteBuffer.position()];
+				return extractBytes(byteBuffer);
-				byteBuffer.rewind();
-				byteBuffer.get(line);
-				return line;
 			case '\r':
 				seenCR = true;
 				break;

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

@@ -29,6 +29,7 @@ import java.io.IOException;
 import org.apache.commons.lang.builder.ToStringBuilder;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import java.util.Map;
 
 public class FileState {
 	@JsonIgnore
@@ -53,6 +54,8 @@ public class FileState {
 	@JsonIgnore
 	private Event fields;
 	@JsonIgnore
+	private Multiline multiline;
+	@JsonIgnore
 	private boolean matchedToNewFile = false;
 
 	public FileState() {
@@ -180,6 +183,14 @@ public class FileState {
 		this.fields = fields;
 	}
 
+	public Multiline getMultiline() {
+		return multiline;
+	}
+
+	public void setMultiline(Multiline multiline) {
+		this.multiline = multiline;
+	}
+
 	public boolean isMatchedToNewFile() {
 		return matchedToNewFile;
 	}

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

@@ -70,14 +70,14 @@ public class FileWatcher {
 		printWatchMap();
 	}
 
-	public void addFilesToWatch(String fileToWatch, Event fields, int deadTime) {
+	public void addFilesToWatch(String fileToWatch, Event fields, long deadTime, Multiline multiline) {
 		try {
 			if(fileToWatch.equals("-")) {
 				addStdIn(fields);
 			} else if(fileToWatch.contains("*")) {
-				addWildCardFiles(fileToWatch, fields, deadTime);
+				addWildCardFiles(fileToWatch, fields, deadTime, multiline);
 			} else {
-				addSingleFile(fileToWatch, fields, deadTime);
+				addSingleFile(fileToWatch, fields, deadTime, multiline);
 			}
 		} catch(Exception e) {
 			throw new RuntimeException(e);
@@ -223,7 +223,7 @@ public class FileWatcher {
 		removeMarkedFilesFromWatchMap();
 	}
 
-	private void addSingleFile(String fileToWatch, Event fields, int deadTime) throws Exception {
+	private void addSingleFile(String fileToWatch, Event fields, long deadTime, Multiline multiline) throws Exception {
 		logger.info("Watching file : " + new File(fileToWatch).getCanonicalPath());
 		String directory = FilenameUtils.getFullPath(fileToWatch);
 		String fileName = FilenameUtils.getName(fileToWatch);
@@ -231,10 +231,10 @@ public class FileWatcher {
 				FileFilterUtils.fileFileFilter(),
 				FileFilterUtils.nameFileFilter(fileName),
 				new LastModifiedFileFilter(deadTime));
-		initializeWatchMap(new File(directory), fileFilter, fields);
+		initializeWatchMap(new File(directory), fileFilter, fields, multiline);
 	}
 
-	private void addWildCardFiles(String filesToWatch, Event fields, int deadTime) throws Exception {
+	private void addWildCardFiles(String filesToWatch, Event fields, long deadTime, Multiline multiline) throws Exception {
 		logger.info("Watching wildcard files : " + filesToWatch);
 		String directory = FilenameUtils.getFullPath(filesToWatch);
 		String wildcard = FilenameUtils.getName(filesToWatch);
@@ -243,7 +243,7 @@ public class FileWatcher {
 				FileFilterUtils.fileFileFilter(),
 				new WildcardFileFilter(wildcard),
 				new LastModifiedFileFilter(deadTime));
-		initializeWatchMap(new File(directory), fileFilter, fields);
+		initializeWatchMap(new File(directory), fileFilter, fields, multiline);
 	}
 
 	private void addStdIn(Event fields) {
@@ -252,22 +252,22 @@ public class FileWatcher {
 		stdinConfigured = true;
 	}
 
-	private void initializeWatchMap(File directory, IOFileFilter fileFilter, Event fields) throws Exception {
+	private void initializeWatchMap(File directory, IOFileFilter fileFilter, Event fields, Multiline multiline) throws Exception {
 		if(!directory.isDirectory()) {
 			logger.warn("Directory " + directory + " does not exist");
 			return;
 		}
 		FileAlterationObserver observer = new FileAlterationObserver(directory, fileFilter);
-		FileModificationListener listener = new FileModificationListener(this, fields);
+		FileModificationListener listener = new FileModificationListener(this, fields, multiline);
 		observer.addListener(listener);
 		observerList.add(observer);
 		observer.initialize();
 		for(File file : FileUtils.listFiles(directory, fileFilter, null)) {
-			addFileToWatchMap(newWatchMap, file, fields);
+			addFileToWatchMap(newWatchMap, file, fields, multiline);
 		}
 	}
 
-	private void addFileToWatchMap(Map<File,FileState> map, File file, Event fields) {
+	private void addFileToWatchMap(Map<File,FileState> map, File file, Event fields, Multiline multiline) {
 		try {
 			FileState state = new FileState(file);
 			state.setFields(fields);
@@ -276,25 +276,26 @@ public class FileWatcher {
 			long signature = FileSigner.computeSignature(state.getRandomAccessFile(), signatureLength);
 			state.setSignature(signature);
 			logger.trace("Setting signature of size : " + signatureLength + " on file : " + file + " : " + signature);
+			state.setMultiline(multiline);
 			map.put(file, state);
 		} catch(IOException e) {
 			logger.error("Caught IOException : " + e.getMessage());
 		}
 	}
 
-	public void onFileChange(File file, Event fields) {
+	public void onFileChange(File file, Event fields, Multiline multiline) {
 		try {
 			logger.debug("Change detected on file : " + file.getCanonicalPath());
-			addFileToWatchMap(newWatchMap, file, fields);
+			addFileToWatchMap(newWatchMap, file, fields, multiline);
 		} catch (IOException e) {
 			logger.error("Caught IOException : " + e.getMessage());
 		}
 	}
 
-	public void onFileCreate(File file, Event fields) {
+	public void onFileCreate(File file, Event fields, Multiline multiline) {
 		try {
 			logger.debug("Create detected on file : " + file.getCanonicalPath());
-			addFileToWatchMap(newWatchMap, file, fields);
+			addFileToWatchMap(newWatchMap, file, fields, multiline);
 		} catch (IOException e) {
 			logger.error("Caught IOException : " + e.getMessage());
 		}

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

@@ -80,7 +80,7 @@ public class Forwarder {
 			configManager.readConfiguration();
 			for(FilesSection files : configManager.getConfig().getFiles()) {
 				for(String path : files.getPaths()) {
-					watcher.addFilesToWatch(path, new Event(files.getFields()), files.getDeadTimeInSeconds() * 1000);
+					watcher.addFilesToWatch(path, new Event(files.getFields()), files.getDeadTimeInSeconds() * 1000, files.getMultiline());
 				}
 			}
 			watcher.initialize();

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

@@ -0,0 +1,90 @@
+package info.fetter.logstashforwarder;
+
+/*
+ * Copyright 2015 Didier Fetter
+ *
+ * Licensed 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.
+ *
+ */
+
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+import java.util.regex.Pattern;
+import org.apache.commons.lang.builder.ToStringBuilder;
+
+public class Multiline {
+	public enum WhatType { Previous, Next };
+	public static byte JOINT = (byte) ' ';
+
+	private Pattern pattern = null;
+	private boolean negate = false;
+	private WhatType what = WhatType.Previous;
+
+	public Multiline() {
+	}
+
+	public Multiline(Multiline event) {
+		if(event != null) {
+			this.negate = event.negate;
+			this.pattern = event.pattern;
+			this.what = event.what;
+		}
+	}
+
+	public Multiline(Map<String,String> fields) throws UnsupportedEncodingException {
+		String strPattern = "";
+		for(String key : fields.keySet()) {
+			if ("pattern".equals(key))
+				strPattern = fields.get(key);
+			else if ("negate".equals(key))
+				negate = Boolean.parseBoolean(fields.get(key));
+			else if ("what".equals(key))
+				what = WhatType.valueOf(fields.get(key));
+			else
+				throw new UnsupportedEncodingException(key + " not supported");
+		}
+		pattern = Pattern.compile(strPattern);
+
+	}
+
+	public Pattern getPattern() {
+		return pattern;
+	}
+
+	public boolean isNegate() {
+		return negate;
+	}
+
+	public WhatType getWhat() {
+		return what;
+	}
+
+	public boolean isPrevious() {
+		return what == WhatType.Previous;
+	}
+
+	public boolean isPatternFound (byte[] line) {
+		boolean result = pattern.matcher(new String(line)).find();
+		if (negate) return !result;
+		return result;
+	}
+
+	@Override
+	public String toString() {
+		return new ToStringBuilder(this).
+			append("pattern", pattern).
+			append("negate", negate).
+			append("what", what).
+			toString();
+	}
+}

src/main/java/info/fetter/logstashforwarder/config/FilesSection.java

@@ -19,16 +19,20 @@ package info.fetter.logstashforwarder.config;
 
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 import org.apache.commons.lang.builder.ToStringBuilder;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import info.fetter.logstashforwarder.Multiline;
+import java.io.UnsupportedEncodingException;
 
 public class FilesSection {
 	private List<String> paths;
 	private Map<String,String> fields;
 	@JsonProperty("dead time")
 	private String deadTime = "24h";
+	private Multiline multiline;
 
 	public List<String> getPaths() {
 		return paths;
@@ -50,8 +54,8 @@ public class FilesSection {
 		return deadTime;
 	}
 
-	public int getDeadTimeInSeconds() {
+	public long getDeadTimeInSeconds() {
-		int deadTimeInSeconds = 0;
+		long deadTimeInSeconds = 0;
 		String remaining = deadTime;
 
 		if(deadTime.contains("h")) {
@@ -79,12 +83,21 @@ public class FilesSection {
 		this.deadTime = deadTime;
 	}
 
+	public Multiline getMultiline() {
+		return multiline;
+	}
+
+	public void setMultiline(Map<String, String> multilineMap) throws UnsupportedEncodingException {
+		this.multiline = new Multiline(multilineMap);
+	}
+
 	@Override
 	public String toString() {
 		return new ToStringBuilder(this).
 				append("paths", paths).
 				append("fields", fields).
 				append("dead time", deadTime).
+				append("multiline", multiline).
 				toString();
 	}
 }

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

@@ -23,7 +23,9 @@ import info.fetter.logstashforwarder.util.AdapterException;
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.log4j.BasicConfigurator;
@@ -54,12 +56,41 @@ public class FileReaderTest {
 		List<FileState> fileList = new ArrayList<FileState>(1);
 		File file1 = new File("testFileReader1.txt");
 		FileUtils.write(file1, "testFileReader1 line1\n");
+		FileUtils.write(file1, " nl line12\n", true);
 		FileUtils.write(file1, "testFileReader1 line2\n", true);
 		FileUtils.write(file1, "testFileReader1 line3\n", true);
 		Thread.sleep(500);
 		FileState state = new FileState(file1);
 		fileList.add(state);
 		state.setFields(new Event().addField("testFileReader1", "testFileReader1"));
+		Map<String, String> m = new HashMap<String, String>();
+		m.put("pattern", " nl");
+		m.put("negate", "false");
+		state.setMultiline(new Multiline(m));
+		reader.readFiles(fileList);
+		reader.readFiles(fileList);
+		reader.readFiles(fileList);
+		//FileUtils.forceDelete(file1);
+	}
+
+        @Test
+	public void testFileReader2() throws IOException, InterruptedException, AdapterException {
+		FileReader reader = new FileReader(2);
+		reader.setAdapter(new MockProtocolAdapter());
+		List<FileState> fileList = new ArrayList<FileState>(1);
+		File file1 = new File("testFileReader1.txt");
+		FileUtils.write(file1, "testFileReader1 line1\n");
+		FileUtils.write(file1, " nl line12\n", true);
+		FileUtils.write(file1, "testFileReader1 line2\n", true);
+		FileUtils.write(file1, "testFileReader1 line3\n", true);
+		Thread.sleep(500);
+		FileState state = new FileState(file1);
+		fileList.add(state);
+		state.setFields(new Event().addField("testFileReader1", "testFileReader1"));
+		Map<String, String> m = new HashMap<String, String>();
+		m.put("pattern", "testFileReader1");
+		m.put("negate", "true");
+		state.setMultiline(new Multiline(m));
 		reader.readFiles(fileList);
 		reader.readFiles(fileList);
 		reader.readFiles(fileList);

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

@@ -21,6 +21,8 @@ import static org.apache.log4j.Level.*;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.log4j.BasicConfigurator;
@@ -47,21 +49,32 @@ public class FileWatcherTest {
 	//@Test
 	public void testFileWatch() throws InterruptedException, IOException {
 		FileWatcher watcher = new FileWatcher();
-		watcher.addFilesToWatch("./test.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY);
+		watcher.addFilesToWatch("./test.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY, null);
 		for(int i = 0; i < 100; i++) {
 			Thread.sleep(1000);
 			watcher.checkFiles();
 		}
 	}
 
-	@Test
+	//@Test
+	public void testFileWatchWithMultilines() throws InterruptedException, IOException {
+		FileWatcher watcher = new FileWatcher();
+		Multiline multiline = new Multiline();
+		watcher.addFilesToWatch("./test.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY, multiline);
+		for(int i = 0; i < 100; i++) {
+			Thread.sleep(1000);
+			watcher.checkFiles();
+		}
+	}
+
+	//@Test
 	public void testWildcardWatch() throws InterruptedException, IOException {
 		if(System.getProperty("os.name").toLowerCase().contains("win")) {
 			logger.warn("Not executing this test on windows");
 			return;
 		}
 		FileWatcher watcher = new FileWatcher();
-		watcher.addFilesToWatch("./testFileWatcher*.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY);
+		watcher.addFilesToWatch("./testFileWatcher*.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY, null);
 		watcher.initialize();
 
 		File file1 = new File("testFileWatcher1.txt");
@@ -97,6 +110,58 @@ public class FileWatcherTest {
 
 
 
+	}
+
+        @Test
+	public void testWildcardWatchMultiline() throws InterruptedException, IOException {
+		if(System.getProperty("os.name").toLowerCase().contains("win")) {
+			logger.warn("Not executing this test on windows");
+			return;
+		}
+		FileWatcher watcher = new FileWatcher();
+                Map<String, String> m = new HashMap<String, String>();
+                m.put("pattern", " nl");
+                m.put("negate", "false");
+                Multiline multiline = new Multiline(m);
+		watcher.addFilesToWatch("./testFileWatcher*.txt", new Event().addField("test", "test"), FileWatcher.ONE_DAY, multiline);
+		watcher.initialize();
+
+		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("testFileWatcher");
+		//FileUtils.forceMkdir(new File("test"));
+
+		watcher.checkFiles();
+		Thread.sleep(100);
+		FileUtils.write(file1, "file 1 line 1\n nl line 1-2", true);
+		Thread.sleep(100);
+		watcher.checkFiles();
+		FileUtils.write(file1, "file 1 line 2\n", true);
+                Thread.sleep(100);
+		watcher.checkFiles();
+		FileUtils.write(file1, " nl line 3\n", true);
+		//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(file1, file2);
+//		FileUtils.write(file2, "file 3 line 1\n", true);
+//
+		Thread.sleep(1000);
+		watcher.checkFiles();
+//
+//
+		watcher.close();
+		FileUtils.deleteQuietly(file1);
+		FileUtils.deleteQuietly(file2);
+//		FileUtils.forceDelete(testDir);
+
+
+
 	}
 
 	@Test

src/test/java/info/fetter/logstashforwarder/config/ConfigurationManagerTest.java

@@ -57,6 +57,7 @@ public class ConfigurationManagerTest {
 			for(String path : files.getPaths()) {
 				logger.debug(" - Path : " + path);
 			}
+			logger.debug(" - Multiline : " + files.getMultiline());
 			logger.debug(" - Dead time : " + files.getDeadTimeInSeconds());
 			if(files.getDeadTime().equals("24h")) {
 				assertEquals(86400, files.getDeadTimeInSeconds());

src/test/resources/config1.json

@@ -53,6 +53,7 @@
         "/var/log/apache/error-*.log"
       ],
       "fields": { "type": "error" },
+      "multiline": { "pattern": "^[0-9]{4}", "negate": "true" },
       "dead time": "8h32m50s" 
     }
   ]