Add directory watchService with regex file name

src/main/java/com/airfrance/diqmqs/logparser/Application.java

@@ -0,0 +1,100 @@
+package com.airfrance.diqmqs.logparser;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Scanner;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class Application extends Thread {
+
+	private ParserEngine engine;
+	private boolean onlyOneTime;
+	private Pattern pattern;
+	private Matcher matcher;
+	private Log log;
+	private File paramFile;
+	private Watchdog watcher;
+
+	public Application(String patternPath, boolean onlyOneTime) throws IOException {
+		engine = new ParserEngine(patternPath);
+		this.onlyOneTime = onlyOneTime;
+		log = Log.getLogger(Application.class.getName());
+		watcher = new Watchdog(this);
+	}
+
+	void startParserFromCLI(String[] filesName, String[] regexName, String applicationName, boolean readFromStart)
+			throws IOException {
+		for (int i = 0; i < filesName.length; i++) {
+			File f = new File(filesName[i]);
+			engine.addNewParser(f, regexName[i], applicationName, readFromStart, onlyOneTime);
+
+		}
+	}
+
+	void stopParser(File file) {
+		engine.stopOneParser(file.getAbsolutePath());
+	}
+
+	void startParser(File file, String regexName, String applicationName) {
+		engine.addNewParser(file, regexName, applicationName, true, false);
+	}
+
+	void startFromParamFile(File paramFile) throws IOException {
+
+		this.paramFile = paramFile;
+		startParserFromFile(paramFile);
+		log.info("All parsers started");
+		watcher.register(paramFile.getParentFile().toPath());
+		watcher.start();
+
+	}
+
+	public void startParserFromFile(File file) throws IOException {
+
+		try {
+			Scanner scanner = new Scanner(file);
+			while (scanner.hasNextLine()) {
+				String line = scanner.nextLine();
+				boolean readFromStart = false;
+				if (!line.startsWith("#")) {
+					String[] arguments = line.split("\\t");
+					if (arguments.length >= 4) {
+						String applicationName = arguments[0].toLowerCase();
+						Paths pathDirectories = new Paths("/", arguments[1]);
+						String regexFile = arguments[2];
+						String regexGrokName = arguments[3];
+						if (arguments.length == 5) {
+							if (arguments[4].equalsIgnoreCase("true")) {
+								readFromStart = true;
+							}
+						}
+						for (File dir : pathDirectories.getFiles()) {
+							pattern = Pattern.compile(regexFile);
+							for (File f : dir.listFiles()) {
+								matcher = pattern.matcher(f.getName());
+								if (matcher.find()) {
+									engine.addNewParser(f, regexGrokName, applicationName, readFromStart, onlyOneTime);
+								}
+							}
+							watcher.register(dir.toPath(), regexFile, regexGrokName, applicationName);
+						}
+					}
+					else
+					{
+						log.error("Missing field on line : " + line );
+					}
+				}
+			}
+			scanner.close();
+		} catch (FileNotFoundException fnfe) {
+			log.error(fnfe.getMessage());
+		}
+	}
+
+	public File getParamFile() {
+		return paramFile;
+	}
+
+}

src/main/java/com/airfrance/diqmqs/logparser/Cli.java

@@ -1,8 +1,7 @@
 package com.airfrance.diqmqs.logparser;
 
 import java.io.File;
-import java.io.FileNotFoundException;
+import java.io.IOException;
-import java.util.Scanner;
 
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
@@ -14,7 +13,7 @@ import org.apache.commons.cli.ParseException;
 
 public class Cli {
 
-	public final static String VER		 	= "1.1";
+	public final static String VER = "2.0";
 	public static Options options = new Options();
 	public static Option help = new Option("help", "Print this message");
 	public static Option version = new Option("version", "Print the version information and exit");
@@ -27,14 +26,13 @@ public class Cli {
 	public static Option debugOption = new Option("debug", "Active debug output message");
 	public static Option infoOption = new Option("info", "Active info output message");
 	public static Option oneTime = new Option("oneTime", "Read the file once time");
-	private static  long lastModifiedTime	= 0L;
+
 	public Cli() {
 		// TODO Auto-generated constructor stub
 	}
 
 	public static void main(String[] args) {
 
-		
 		application.setArgName("application name");
 		patternFilePath.setArgName("path to the pattern file");
 		logFile.setArgName("path to the logs file");
@@ -55,16 +53,15 @@ public class Cli {
 		boolean readFromStart = false;
 		boolean onlyOneTime = false;
 		CommandLineParser parser = new DefaultParser();
+
 		CommandLine cmd = null;
 		try {
 			cmd = parser.parse(options, args);
-		}
+		} catch (ParseException e) {
-		catch (ParseException e) {
 			System.err.println("Command parse error : " + e.getMessage());
 			System.exit(0);
 		}
 
-
 		if (cmd.hasOption("help")) {
 			formatter.printHelp("logParser", options);
 			System.exit(0);
@@ -86,9 +83,7 @@ public class Cli {
 		String patternPath = "";
 		if (cmd.hasOption("pattern")) {
 			patternPath = cmd.getOptionValue("pattern");
-		}
+		} else {
-		else
-		{
 			System.err.println("patternFilePath application missing");
 			System.exit(0);
 		}
@@ -97,47 +92,18 @@ public class Cli {
 			onlyOneTime = true;
 		}
 
+		Application app;
+		try {
+			app = new Application(patternPath, onlyOneTime);
+
 			if (cmd.hasOption("paramfile")) {
-			String fileParamName = cmd.getOptionValue("paramfile");
+				File f = new File(cmd.getOptionValue("paramfile"));
-			File param = new File(fileParamName);
+				app.startFromParamFile(f);
-			ParserEngine engine = new ParserEngine(patternPath);
+			} else {
-			Log log = Log.getLogger(Cli.class.getName());
-			try {
-				startParserFromFile(param, engine , onlyOneTime);
-				lastModifiedTime = param.lastModified();
-				log.info("All parsers started");
-				// Check for reload the param file
-				while (true) {
-		            try {
-						Thread.sleep(5000);
-						if (param.exists())
-						{
-							if ( param.lastModified() != lastModifiedTime )
-							{
-								engine.stopAllParser();
-								startParserFromFile(param, engine , onlyOneTime);
-								lastModifiedTime = param.lastModified();
-							}
-						}
-						
-					} catch (InterruptedException e) {
-						e.printStackTrace();
-					}
-		        }
-				
-			} catch (FileNotFoundException e) {
-				System.err.println("Execption : " +  e.getMessage());
-				System.exit(0);
-			}			
-		}
-		else
-		{
 				String applicationName = "";
 				if (cmd.hasOption("application")) {
 					applicationName = cmd.getOptionValue("application");
-			}
+				} else {
-			else
-			{
 					System.err.println("Option application missing");
 					System.exit(0);
 				}
@@ -145,9 +111,7 @@ public class Cli {
 				String[] regexName = null;
 				if (cmd.hasOption("regex")) {
 					regexName = cmd.getOptionValues("regex");
-			}
+				} else {
-			else
-			{
 					System.err.println("regex application missing");
 					System.exit(0);
 				}
@@ -157,45 +121,15 @@ public class Cli {
 				}
 
 				if (cmd.hasOption("logfile")) {
-				ParserEngine engine = new ParserEngine(patternPath);
 					String[] filesName = cmd.getOptionValues("logfile");
-				for (int i = 0; i < filesName.length ; i++) {
+					app.startParserFromCLI(filesName, regexName, applicationName, readFromStart);
-					File f = new File (filesName[i]);
-					engine.addNewParser(f, regexName[i], applicationName, readFromStart, onlyOneTime);
 				}
 			}
+		} catch (IOException e) {
+			System.err.println(e.getMessage());
+			System.exit(0);
 		}
 
 	}
 
-	
-	static void startParserFromFile(File file, ParserEngine engine, boolean onceTime) throws FileNotFoundException {
-		
-		Scanner scanner = new Scanner(file);
-		
-		while (scanner.hasNextLine()) {
-			 String line = scanner.nextLine();
-			 boolean readFromStart = false;
-			 if (!line.startsWith("#"))
-			 {
-				String[] arguments =  line.split("\\t");
-				String applicationName = arguments[0].toLowerCase();
-				Paths paths = new Paths("/", arguments[1]);
-				String regexName = arguments[2];
-				if (arguments.length == 4) {
-					if ( arguments[3].equalsIgnoreCase("true") ){
-						readFromStart = true;
-					} 
-				}
-					
-				for (File f : paths.getFiles()) {
-					engine.addNewParser(f, regexName, applicationName, readFromStart, onceTime);
-			 	}
-		
-			 }
-		}
-		scanner.close();
-		
-	}
-
 }

src/main/java/com/airfrance/diqmqs/logparser/ParserEngine.java

@@ -8,6 +8,7 @@ import java.net.InetAddress;
 import java.net.URL;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
@@ -32,8 +33,6 @@ public class ParserEngine implements Runnable{
 	private ScheduledExecutorService scheduler;
 	private String hostname;
 
-
-
 	@SuppressWarnings("unused")
 	private ScheduledFuture<?> timerHandle;
 
@@ -44,7 +43,7 @@ public class ParserEngine implements Runnable{
 		try {
 			hostname = InetAddress.getLocalHost().getHostName().split("\\.")[0];
 		} catch (UnknownHostException e) {
-			e.printStackTrace();
+			log.error(e.getMessage());
 		}
 		this.timerHandle = scheduler.scheduleWithFixedDelay(this, 1, 1, TimeUnit.SECONDS);
 		log.info("Create ParserEngine with pattern file : " + patternPath);
@@ -56,7 +55,7 @@ public class ParserEngine implements Runnable{
 		Tailer tailer = new Tailer(f, listener, delay, !readFromStart, true, 8192);
 		Thread thread = new Thread(tailer);
 		thread.setDaemon(true);
-	    thread.setName("Parser - " + threadsParser.size() );
+		thread.setName(f.getAbsolutePath());
 		thread.start();
 		threadsParser.add(thread);
 		this.onceTime = onceTime;
@@ -71,9 +70,19 @@ public class ParserEngine implements Runnable{
 		threadsParser.clear();
 	}
 
+	void stopOneParser(String name) {
+		for (Iterator<Thread> ite = threadsParser.iterator(); ite.hasNext();) {
+			Thread thread = ite.next();
+			if (thread.getName().equals(name)) {
+				thread.interrupt();
+				log.info("Stop Thread " + thread.getName());
+				ite.remove();
+			}
+		}
+	}
+
 	// Read data from the Queue and send it to influxdb
 	public void run() {
-		
 		StringBuilder postData = new StringBuilder();
 		log.info("Actual queue size  " + queue.size());
 		int nbElement = queue.drainTo(res, 2000);
@@ -81,10 +90,8 @@ public class ParserEngine implements Runnable{
 			for (String s : res) {
 				postData.append(s + "\n");
 			}
-		}
+		} else {
-		else {
+			if (onceTime) {
-			if (onceTime)
-			{
 				stopAllParser();
 				System.exit(0);
 			}
@@ -92,11 +99,9 @@ public class ParserEngine implements Runnable{
 		}
 		try {
 			if (postData.length() > 0) {
-				if (sendMetricToInfluxdb(postData.toString()))
+				if (sendMetricToInfluxdb(postData.toString())) {
-				{
 					res.clear();
-				}
+				} else {
-				else{
 
 				}
 			}
@@ -146,7 +151,6 @@ public class ParserEngine implements Runnable{
 		return patternPath;
 	}
 
-
 	public void setPatternPath(String patternPath) {
 		this.patternPath = patternPath;
 	}

src/main/java/com/airfrance/diqmqs/logparser/Watchdog.java

@@ -0,0 +1,166 @@
+package com.airfrance.diqmqs.logparser;
+
+import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
+import static java.nio.file.StandardWatchEventKinds.ENTRY_DELETE;
+import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
+
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+import java.nio.file.StandardWatchEventKinds;
+import java.nio.file.WatchEvent;
+import java.nio.file.WatchKey;
+import java.nio.file.WatchService;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class Watchdog extends Thread {
+
+	private final WatchService watcher;
+	private final Map<WatchKey, BeanFile> keys;
+	private Pattern pattern;
+	private Matcher matcher;
+	private Log log;
+	private Application app;
+	private boolean interrupted = false;
+
+	public Watchdog(Application app) throws IOException {
+		this.watcher = FileSystems.getDefault().newWatchService();
+		this.keys = new HashMap<WatchKey, BeanFile>();
+		log = Log.getLogger(Watchdog.class.getName());
+		this.app = app;
+	}
+
+	@SuppressWarnings("rawtypes")
+	public void start() {
+
+		while (!interrupted) {
+			WatchKey key;
+			try {
+				key = watcher.take();
+				BeanFile bean = keys.get(key);
+				for (WatchEvent<?> event : key.pollEvents()) {
+					WatchEvent.Kind kind = event.kind();
+					WatchEvent pathEvent = (WatchEvent) event;
+					String filename = ((Path)pathEvent.context()).toString();
+					Path file = bean.getDir().resolve((Path)pathEvent.context());
+					
+					if (kind.equals(StandardWatchEventKinds.ENTRY_CREATE)) {
+						// New file
+						for ( int i = 0 ; i < bean.getRegexFiles().size() ; i++ ) {
+							pattern = Pattern.compile(bean.getRegexFiles().get(i));
+							matcher = pattern.matcher(filename);
+							if (matcher.find()) {
+								app.startParser(file.toFile(), bean.getRegexGroks().get(i), bean.getApplicationNames().get(i));
+							}
+						}
+					} else if (kind.equals(StandardWatchEventKinds.ENTRY_DELETE)) {
+						// File deleted, stop the parser
+						app.stopParser(file.toFile());
+					} else if (kind.equals(StandardWatchEventKinds.ENTRY_MODIFY)) {						
+						if ( file.toFile().equals(app.getParamFile()) ) {
+							// Reload all
+							log.info("Reload all parsers from ParamFile");
+							try {
+								app.startParserFromFile(file.toFile());
+							} catch (IOException e) {
+								log.error("Error when loading param file to start Parser : " + e.getMessage());
+							}
+						}
+					}
+				}
+				key.reset();
+
+			} catch (InterruptedException x) {
+				interrupted = true;
+			}
+		}
+		try {
+			watcher.close();
+			log.info("Watch service closed.");
+		} catch (IOException e) {
+			log.info("Watch service close error : " + e.getMessage());
+		}
+
+	}
+
+	public void register(Path dir) throws IOException {
+		log.info("Register : " + dir.toString() + " for file " + app.getParamFile());
+		WatchKey key = dir.register(watcher, ENTRY_MODIFY);
+		BeanFile bf = new BeanFile(dir, null, null, null);
+		keys.put(key, bf);
+	}
+
+	/**
+	 * Register the given directory with the WatchService
+	 */
+	public void register(Path dir, String regexFile, String regexGrok, String applicationName) throws IOException {
+		log.info("Register : " + dir.toString() + " with regex on file " + regexFile);
+		WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE);
+		if (keys.containsKey(key)) {
+			log.info("Folder already register, add new Grok regex : " + regexGrok + " on regex file " + regexFile + " for application " + applicationName);
+			BeanFile bf = keys.get(key);
+			bf.addApplicationNames(applicationName);
+			bf.addRegexFiles(regexFile);
+			bf.addRegexGroks(regexGrok);
+			keys.put(key, bf);
+		}
+		else {
+			BeanFile bf = new BeanFile(dir, regexFile, regexGrok, applicationName);
+			keys.put(key, bf);
+		}
+	
+	}
+
+	class BeanFile {
+		public Path getDir() {
+			return dir;
+		}
+
+		public void setDir(Path dir) {
+			this.dir = dir;
+		}
+
+		public List<String> getRegexGroks() {
+			return regexGroks;
+		}
+
+		public void addRegexGroks(String regexGrok) {
+			regexGroks.add(regexGrok);
+		}
+
+		public List<String> getApplicationNames() {
+			return applicationNames;
+		}
+
+		public void addApplicationNames(String applicationName) {
+			applicationNames.add(applicationName);
+		}
+
+		public List<String> getRegexFiles() {
+			return regexFiles;
+		}
+
+		public void addRegexFiles(String regexFile) {
+			regexFiles.add(regexFile);
+		}
+	
+
+		List<String> regexGroks = new ArrayList<String>();
+		List<String> applicationNames = new ArrayList<String>();
+		List<String> regexFiles = new ArrayList<String>();
+		Path dir;
+
+		BeanFile(Path dir, String regexFile, String regexGrok, String applicationName) {
+			this.dir = dir;
+			applicationNames.add(applicationName);
+			regexGroks.add(regexGrok);
+			regexFiles.add(regexFile);
+		}
+	}
+
+}