Implements #19018 'Run CollecTor without crontab'. Added scheduler logic,... (4685cf10) · Commits · Legacy / gitolite / user / iwakeh / Collector

build.xml

+5 −3

Original line number	Diff line number	Diff line
		@@ -210,7 +210,7 @@
		<classpath refid="cobertura.test.classpath" />
		<formatter type="xml" />
		<batchtest toDir="${testresult}" >
		<fileset dir="${testclasses}" />
		<fileset dir="${testclasses}" includes="*/Test.class" />
		</batchtest>
		</junit>
		<cobertura-report format="html" destdir="${coverageresult}" >
		@@ -218,8 +218,10 @@
		<include name="*/.java" />
		</fileset>
		</cobertura-report>
		<cobertura-check branchrate="0" totallinerate="15" totalbranchrate="5" >
		<cobertura-check branchrate="0" totallinerate="4" totalbranchrate="1" >
		<regex pattern="org.torproject.collector.conf.*" branchrate="100" linerate="100"/>
		<regex pattern="org.torproject.collector.cron.*" branchrate="66" linerate="73"/>
		<regex pattern="org.torproject.collector.Main" branchrate="66" linerate="94"/>
		</cobertura-check>
		</target>
		<target name="test" depends="compile,compile-tests">
		@@ -231,7 +233,7 @@
		<classpath refid="test.classpath"/>
		<formatter type="plain" usefile="false"/>
		<batchtest>
		<fileset dir="${testclasses}" />
		<fileset dir="${testclasses}" includes="*/Test.class" />
		</batchtest>
		</junit>
		</target>

src/main/java/org/torproject/collector/Main.java

+18 −36

Original line number	Diff line number	Diff line
		@@ -5,6 +5,9 @@ package org.torproject.collector;

		import org.torproject.collector.bridgedescs.SanitizedBridgesWriter;
		import org.torproject.collector.conf.Configuration;
		import org.torproject.collector.conf.Key;
		import org.torproject.collector.cron.CollecTorMain;
		import org.torproject.collector.cron.Scheduler;
		import org.torproject.collector.exitlists.ExitListDownloader;
		import org.torproject.collector.index.CreateIndexJson;
		import org.torproject.collector.relaydescs.ArchiveWriter;
		@@ -37,34 +40,31 @@ public class Main {
		/** All possible main classes.
		* If a new CollecTorMain class is available, just add it to this map.
		*/
		static final Map<String, Class> collecTorMains = new HashMap<>();
		static final Map<Key, Class<? extends CollecTorMain>> collecTorMains = new HashMap<>();

		static { // add a new main class here
		collecTorMains.put("bridgedescs", SanitizedBridgesWriter.class);
		collecTorMains.put("exitlists", ExitListDownloader.class);
		collecTorMains.put("updateindex", CreateIndexJson.class);
		collecTorMains.put("relaydescs", ArchiveWriter.class);
		collecTorMains.put("torperf", TorperfDownloader.class);
		collecTorMains.put(Key.BridgedescsActivated, SanitizedBridgesWriter.class);
		collecTorMains.put(Key.ExitlistsActivated, ExitListDownloader.class);
		collecTorMains.put(Key.UpdateindexActivated, CreateIndexJson.class);
		collecTorMains.put(Key.RelaydescsActivated, ArchiveWriter.class);
		collecTorMains.put(Key.TorperfActivated, TorperfDownloader.class);
		}

		private static final String modules = collecTorMains.keySet().toString()
		.replace("[", "").replace("]", "").replaceAll(", ", "\|");

		private static Configuration conf = new Configuration();

		/**
		* One argument is necessary.
		* At most one argument.
		* See class description {@link Main}.
		*/
		public static void main(String[] args) throws Exception {
		File confFile = null;
		if (null == args \|\| args.length < 1 \|\| args.length > 2) {
		printUsage("CollecTor needs one or two arguments.");
		return;
		} else if (args.length == 1) {
		if (args == null \|\| args.length == 0) {
		confFile = new File(CONF_FILE);
		} else if (args.length == 2) {
		confFile = new File(args[1]);
		} else if (args.length == 1) {
		confFile = new File(args[0]);
		} else {
		printUsage("CollecTor takes at most one argument.");
		return;
		}
		if (!confFile.exists() \|\| confFile.length() < 1L) {
		writeDefaultConfig(confFile);
		@@ -72,12 +72,12 @@ public class Main {
		} else {
		readConfigurationFrom(confFile);
		}
		invokeGivenMain(args[0]);
		Scheduler.getInstance().scheduleModuleRuns(collecTorMains, conf);
		}

		private static void printUsage(String msg) {
		final String usage = "Usage:\njava -jar collector.jar "
		+ "<" + modules + "> [path/to/configFile]";
		+ "[path/to/configFile]";
		System.out.println(msg + "\n" + usage);
		}

		@@ -105,23 +105,5 @@ public class Main {
		}
		}

		private static void invokeGivenMain(String mainId) {
		Class clazz = collecTorMains.get(mainId);
		if (null == clazz) {
		printUsage("Unknown argument: " + mainId);
		}
		invokeMainOnClass(clazz);
		}

		private static void invokeMainOnClass(Class clazz) {
		try {
		clazz.getMethod("main", new Class[] { Configuration.class })
		.invoke(null, (Object) conf);
		} catch (NoSuchMethodException \| IllegalAccessException
		\| InvocationTargetException e) {
		log.error("Cannot invoke 'main' method on "
		+ clazz.getName() + ". " + e, e);
		}
		}
		}

src/main/java/org/torproject/collector/bridgedescs/SanitizedBridgesWriter.java

+5 −27

Original line number	Diff line number	Diff line
		@@ -6,7 +6,7 @@ package org.torproject.collector.bridgedescs;
		import org.torproject.collector.conf.Configuration;
		import org.torproject.collector.conf.ConfigurationException;
		import org.torproject.collector.conf.Key;
		import org.torproject.collector.main.LockFile;
		import org.torproject.collector.cron.CollecTorMain;

		import org.apache.commons.codec.DecoderException;
		import org.apache.commons.codec.binary.Base64;
		@@ -49,36 +49,12 @@ import java.util.TreeMap;
		* by the bridge to advertise their capabilities), and extra-info
		* descriptors (published by the bridge, mainly for statistical analysis).</p>
		*/
		public class SanitizedBridgesWriter extends Thread {
		public class SanitizedBridgesWriter extends CollecTorMain {

		private static Logger logger = LoggerFactory.getLogger(SanitizedBridgesWriter.class);

		/** Executes the bridge-descriptors module using the given
		* configuration. */
		public static void main(Configuration config) throws ConfigurationException {

		logger.info("Starting bridge-descriptors module of CollecTor.");

		// Use lock file to avoid overlapping runs
		LockFile lf = new LockFile(config.getPath(Key.LockFilePath).toString(), "bridge-descriptors");
		lf.acquireLock();

		// Sanitize bridge descriptors
		new SanitizedBridgesWriter(config).run();

		// Remove lock file
		lf.releaseLock();

		logger.info("Terminating bridge-descriptors module of CollecTor.");
		}

		private Configuration config;

		/**
		* Initializes this class.
		*/
		public SanitizedBridgesWriter(Configuration config) {
		this.config = config;
		super(config);
		}

		private String rsyncCatString;
		@@ -106,12 +82,14 @@ public class SanitizedBridgesWriter extends Thread {

		@Override
		public void run() {
		logger.info("Starting bridge-descriptors module of CollecTor.");
		try {
		startProcessing();
		} catch (ConfigurationException ce) {
		logger.error("Configuration failed: " + ce, ce);
		throw new RuntimeException(ce);
		}
		logger.info("Terminating bridge-descriptors module of CollecTor.");
		}

		private void startProcessing() throws ConfigurationException {

src/main/java/org/torproject/collector/cron/CollecTorMain.java

0 → 100644

+28 −0

Original line number	Diff line number	Diff line
		/* Copyright 2016 The Tor Project
		* See LICENSE for licensing information */

		package org.torproject.collector.cron;

		import org.torproject.collector.conf.Configuration;
		import org.torproject.collector.conf.Key;

		import org.slf4j.Logger;
		import org.slf4j.LoggerFactory;

		import java.lang.reflect.InvocationTargetException;
		import java.util.Calendar;
		import java.util.Map;
		import java.util.concurrent.Executors;
		import java.util.concurrent.ScheduledExecutorService;
		import java.util.concurrent.TimeUnit;

		public abstract class CollecTorMain implements Runnable {

		protected Configuration config;

		public CollecTorMain( Configuration conf) {
		this.config = conf;
		}

		}

src/main/java/org/torproject/collector/cron/Scheduler.java

0 → 100644

+102 −0

Original line number	Diff line number	Diff line
		/* Copyright 2016 The Tor Project
		* See LICENSE for licensing information */

		package org.torproject.collector.cron;

		import org.torproject.collector.conf.Configuration;
		import org.torproject.collector.conf.ConfigurationException;
		import org.torproject.collector.conf.Key;
		import org.torproject.collector.cron.CollecTorMain;

		import org.slf4j.Logger;
		import org.slf4j.LoggerFactory;

		import java.lang.reflect.InvocationTargetException;
		import java.util.Calendar;
		import java.util.List;
		import java.util.Map;
		import java.util.concurrent.Executors;
		import java.util.concurrent.ScheduledExecutorService;
		import java.util.concurrent.TimeUnit;

		/**
		* Scheduler that starts the modules configured in collector.properties.
		*/
		public class Scheduler {

		public static final String ACTIVATED = "Activated";
		public static final String PERIODMIN = "PeriodMinutes";
		public static final String OFFSETMIN = "OffsetMinutes";

		private final Logger log = LoggerFactory.getLogger(Scheduler.class);

		private final ScheduledExecutorService scheduler =
		Executors.newScheduledThreadPool(1);

		private static Scheduler instance = new Scheduler();

		private Scheduler(){}

		public static Scheduler getInstance() {
		return instance;
		}

		/**
		* Schedule all classes given according to the parameters in the
		* the configuration.
		*/
		public void scheduleModuleRuns(Map<Key,
		Class<? extends CollecTorMain>> collecTorMains, Configuration conf) {
		for ( Map.Entry<Key, Class<? extends CollecTorMain>> ctmEntry
		: collecTorMains.entrySet() ) {
		try {
		if ( conf.getBool(ctmEntry.getKey()) ) {
		String prefix = ctmEntry.getKey().name().replace(ACTIVATED, "");
		CollecTorMain ctm = ctmEntry.getValue()
		.getConstructor(Configuration.class).newInstance(conf);
		scheduleExecutions(ctm,
		conf.getInt(Key.valueOf(prefix + OFFSETMIN)),
		conf.getInt(Key.valueOf(prefix + PERIODMIN)));
		}
		} catch (ConfigurationException \| IllegalAccessException
		\| InstantiationException \| InvocationTargetException
		\| NoSuchMethodException \| RuntimeException ex) {
		log.error("Cannot schedule " + ctmEntry.getValue().getName()
		+ ". Reason: " + ex.getMessage(), ex);
		shutdownScheduler();
		throw new RuntimeException("Halted scheduling.", ex);
		}
		}
		}

		private void scheduleExecutions(CollecTorMain ctm, int offset, int period) {
		this.log.info("Periodic updater started for " + ctm.getClass().getName()
		+ "; offset=" + offset + ", period=" + period + ".");
		int currentMinute = Calendar.getInstance().get(Calendar.MINUTE);
		int initialDelay = (60 - currentMinute + offset) % 60;

		/* Run after initialDelay delay and then every period min. */
		this.log.info("Periodic updater will start every " + period + "th min "
		+ "at minute " + ((currentMinute + initialDelay) % 60) + ".");
		this.scheduler.scheduleAtFixedRate(ctm, initialDelay, 60,
		TimeUnit.MINUTES);
		}

		/**
		* Try to shutdown smoothly, i.e., wait for running tasks to terminate.
		*/
		public void shutdownScheduler() {
		try {
		scheduler.shutdown();
		scheduler.awaitTermination(20L, java.util.concurrent.TimeUnit.MINUTES);
		log.info("Shutdown of all scheduled tasks completed successfully.");
		} catch ( InterruptedException ie ) {
		List<Runnable> notTerminated = scheduler.shutdownNow();
		log.error("Regular shutdown failed for: " + notTerminated);
		if ( !notTerminated.isEmpty() ) {
		log.error("Forced shutdown failed for: " + notTerminated);
		}
		}
		}
		}