diff --git a/commons/src/main/java/org/archive/bdb/BdbModule.java b/commons/src/main/java/org/archive/bdb/BdbModule.java index 41ec7d4c..8d6ca0c4 100644 --- a/commons/src/main/java/org/archive/bdb/BdbModule.java +++ b/commons/src/main/java/org/archive/bdb/BdbModule.java @@ -102,7 +102,7 @@ Serializable, Closeable { boolean allowCreate; boolean sortedDuplicates; boolean transactional; - + boolean deferredWrite = true; public BdbConfig() { } @@ -129,7 +129,7 @@ Serializable, Closeable { public DatabaseConfig toDatabaseConfig() { DatabaseConfig result = new DatabaseConfig(); - result.setDeferredWrite(true); + result.setDeferredWrite(deferredWrite); result.setTransactional(transactional); result.setAllowCreate(allowCreate); result.setSortedDuplicates(sortedDuplicates); @@ -145,6 +145,11 @@ Serializable, Closeable { public void setTransactional(boolean transactional) { this.transactional = transactional; } + + + public void setDeferredWrite(boolean b) { + this.deferredWrite = true; + } } diff --git a/commons/src/main/java/org/archive/util/IoUtils.java b/commons/src/main/java/org/archive/util/IoUtils.java index f6288708..e0048d5d 100644 --- a/commons/src/main/java/org/archive/util/IoUtils.java +++ b/commons/src/main/java/org/archive/util/IoUtils.java @@ -37,6 +37,8 @@ import java.io.ObjectOutputStream; import java.io.OutputStream; import java.io.Reader; import java.io.Serializable; +import java.net.URL; +import java.net.URLConnection; import java.nio.charset.Charset; import java.util.Iterator; import java.util.List; @@ -352,4 +354,22 @@ public class IoUtils { } return new BufferedReader(new InputStreamReader(is)); } + + /** + * Get a BufferedReader on the crawler journal given. + * + * @param source URL journal + * @return journal buffered reader. + * @throws IOException + */ + public static BufferedReader getBufferedReader(URL source) throws IOException { + URLConnection conn = source.openConnection(); + boolean isGzipped = conn.getContentType() != null && conn.getContentType().equalsIgnoreCase("application/x-gzip") + || conn.getContentEncoding() != null && conn.getContentEncoding().equalsIgnoreCase("gzip"); + InputStream uis = conn.getInputStream(); + return new BufferedReader(isGzipped? + new InputStreamReader(new GZIPInputStream(uis)): + new InputStreamReader(uis)); + + } } diff --git a/modules/src/main/java/org/archive/modules/recrawl/PersistLoadProcessor.java b/modules/src/main/java/org/archive/modules/recrawl/PersistLoadProcessor.java index e037d461..3e4038b8 100644 --- a/modules/src/main/java/org/archive/modules/recrawl/PersistLoadProcessor.java +++ b/modules/src/main/java/org/archive/modules/recrawl/PersistLoadProcessor.java @@ -22,9 +22,19 @@ */ package org.archive.modules.recrawl; +import java.io.IOException; import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.apache.commons.lang.StringUtils; +import org.archive.modules.recrawl.PersistLoadProcessor; +import org.archive.modules.recrawl.PersistProcessor; import org.archive.modules.ProcessorURI; +import org.archive.spring.ConfigFile; +import org.archive.spring.ConfigPath; + +import com.sleepycat.je.DatabaseException; /** * Loads CrawlURI attributes from previous fetch from persistent storage for @@ -35,10 +45,28 @@ import org.archive.modules.ProcessorURI; */ public class PersistLoadProcessor extends PersistOnlineProcessor { private static final long serialVersionUID = -1917169316015093131L; + private static final Logger logger = + Logger.getLogger(PersistLoadProcessor.class.getName()); // class description: "PersistLoadProcessor. Loads CrawlURI attributes " + // "from a previous crawl for current consultation." + + /** + * A source (either log file or BDB directory) from which to copy + * history information into the current store at startup. (Whenever + * possible, it would be better to ensure the original history DB + * is in its own independent BDB environment, and then copy and + * reuse that environment in the followup crawl(s).) + */ + ConfigPath preloadSource = + new ConfigFile("preload source",""); + public ConfigPath getPreloadSource() { + return preloadSource; + } + public void setPreloadSource(ConfigPath preloadSource) { + this.preloadSource = preloadSource; + } public PersistLoadProcessor() { } @@ -57,4 +85,21 @@ public class PersistLoadProcessor extends PersistOnlineProcessor { protected boolean shouldProcess(ProcessorURI uri) { return shouldLoad(uri); } -} \ No newline at end of file + + @Override + public void start() { + super.start(); + String psource = getPreloadSource().getPath(); + if (StringUtils.isNotBlank(psource)) { + try { + int count = PersistProcessor.copyPersistSourceToHistoryMap( + null, preloadSource.getFile().getAbsolutePath(),store); + logger.info("Loaded deduplication information for " + count + " previously fetched urls from " + preloadSource); + } catch (IOException ioe) { + logger.log(Level.WARNING, "Problem loading " + preloadSource + ", proceeding without deduplication! " + ioe); + } catch(DatabaseException de) { + logger.log(Level.WARNING, "Problem loading " + preloadSource + ", proceeding without deduplication! " + de); + } + } + } +} //EOC \ No newline at end of file diff --git a/modules/src/main/java/org/archive/modules/recrawl/PersistProcessor.java b/modules/src/main/java/org/archive/modules/recrawl/PersistProcessor.java index f79b89c9..5fd9a0ea 100644 --- a/modules/src/main/java/org/archive/modules/recrawl/PersistProcessor.java +++ b/modules/src/main/java/org/archive/modules/recrawl/PersistProcessor.java @@ -21,12 +21,15 @@ package org.archive.modules.recrawl; import java.io.BufferedReader; import java.io.File; -import java.io.FileNotFoundException; import java.io.IOException; import java.io.UnsupportedEncodingException; +import java.net.MalformedURLException; +import java.net.URL; import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; +import java.util.logging.ConsoleHandler; +import java.util.logging.Handler; import java.util.logging.Level; import java.util.logging.Logger; @@ -36,8 +39,9 @@ import org.apache.commons.lang.SerializationUtils; import org.archive.bdb.BdbModule; import org.archive.modules.Processor; import org.archive.modules.ProcessorURI; -import org.archive.util.ArchiveUtils; +import org.archive.util.FileUtils; import org.archive.util.IoUtils; +import org.archive.util.OneLineSimpleLogger; import org.archive.util.SURT; import org.archive.util.bdbje.EnhancedEnvironment; import org.archive.util.iterator.LineReadingIterator; @@ -60,6 +64,9 @@ import com.sleepycat.je.EnvironmentConfig; * @author gojomo */ public abstract class PersistProcessor extends Processor { + + private static final long serialVersionUID = 1L; + private static final Logger logger = Logger.getLogger(PersistProcessor.class.getName()); @@ -73,6 +80,7 @@ public abstract class PersistProcessor extends Processor { BdbModule.BdbConfig dbConfig = new BdbModule.BdbConfig(); dbConfig.setTransactional(false); dbConfig.setAllowCreate(true); + dbConfig.setDeferredWrite(true); return dbConfig; } @@ -87,6 +95,7 @@ public abstract class PersistProcessor extends Processor { * @return String key */ public static String persistKeyFor(ProcessorURI curi) { + // use a case-sensitive SURT for uniqueness and sorting benefits return persistKeyFor(curi.getUURI().toString()); } @@ -94,7 +103,7 @@ public abstract class PersistProcessor extends Processor { // use a case-sensitive SURT for uniqueness and sorting benefits return SURT.fromURI(uri,true); } - + /** * Whether the current CrawlURI's state should be persisted (to log or * direct to database) @@ -118,20 +127,206 @@ public abstract class PersistProcessor extends Processor { return true; } + /** + * Copies entries from an existing environment db to a new one. If + * historyMap is not provided, only logs the entries that would have been + * copied. + * + * @param sourceDir existing environment database directory + * @param historyMap new environment db (or null for a dry run) + * @return number of records + * @throws DatabaseException + */ + private static int copyPersistEnv(File sourceDir, StoredSortedMap historyMap) + throws DatabaseException { + int count = 0; + + // open the source env history DB, copying entries to target env + EnhancedEnvironment sourceEnv = setupEnvironment(sourceDir); + StoredClassCatalog sourceClassCatalog = sourceEnv.getClassCatalog(); + Database sourceHistoryDB = sourceEnv.openDatabase( + null, URI_HISTORY_DBNAME, historyDatabaseConfig().toDatabaseConfig()); + StoredSortedMap sourceHistoryMap = new StoredSortedMap(sourceHistoryDB, + new StringBinding(), new SerialBinding(sourceClassCatalog, + Map.class), true); + + Iterator> iter = sourceHistoryMap.entrySet().iterator(); + while (iter.hasNext()) { + Entry item = iter.next(); + if (logger.isLoggable(Level.FINE)) { + logger.fine(item.getKey() + " " + item.getValue()); + } + + if (historyMap != null) { + historyMap.put(item.getKey(), item.getValue()); + } + count++; + } + StoredIterator.close(iter); + sourceHistoryDB.close(); + sourceEnv.close(); + + return count; + } + + /** + * Populates an environment db from a persist log. If historyMap is + * not provided, only logs the entries that would have been populated. + * + * @param persistLogReader + * persist log + * @param historyMap + * new environment db (or null for a dry run) + * @return number of records + * @throws UnsupportedEncodingException + * @throws DatabaseException + */ + private static int populatePersistEnvFromLog(BufferedReader persistLogReader, StoredSortedMap historyMap) + throws UnsupportedEncodingException, DatabaseException { + int count = 0; + + Iterator iter = new LineReadingIterator(persistLogReader); + while (iter.hasNext()) { + String line = iter.next(); + if (line.length() == 0) { + continue; + } + String[] splits = line.split(" "); + if (splits.length != 2) { + logger.severe("bad line: " + line); + continue; + } + + Map alist = (Map) SerializationUtils.deserialize(Base64.decodeBase64(splits[1].getBytes("UTF-8"))); + + if (logger.isLoggable(Level.FINE)) { + logger.fine(splits[0] + " " + alist); + } + + if (historyMap != null) { + // XXX allow RuntimeExceptions to propagate + historyMap.put(splits[0], alist); + } + count++; + } + IOUtils.closeQuietly(persistLogReader); + + return count; + } + + /** + * Populates a new environment db from an old environment db or a persist + * log. If path to new environment is not provided, only logs the entries + * that would have been populated. + * + * @param sourcePath + * source of old entries: can be a path to an existing + * environment db, or a URL or path to a persist log + * @param envFile + * path to new environment db (or null for a dry run) + * @return number of records + * @throws DatabaseException + * @throws IOException + */ + public static int populatePersistEnv(String sourcePath, File envFile) + throws DatabaseException, IOException { + int count = 0; + StoredSortedMap historyMap = null; + EnhancedEnvironment targetEnv = null; + StoredClassCatalog classCatalog = null; + Database historyDB = null; + + if (envFile != null) { + // set up target environment + if (!envFile.exists()) { + envFile.mkdirs(); + } + targetEnv = setupEnvironment(envFile); + classCatalog = targetEnv.getClassCatalog(); + historyDB = targetEnv.openDatabase(null, URI_HISTORY_DBNAME, + historyDatabaseConfig().toDatabaseConfig()); + historyMap = new StoredSortedMap(historyDB, + new StringBinding(), new SerialBinding(classCatalog, + Map.class), true); + } + + try { + count = copyPersistSourceToHistoryMap(null, sourcePath, historyMap); + } finally { + // in finally block so that we unlock the target env even if we + // failed to populate it + if (envFile != null) { + logger.info(count + " records imported from " + sourcePath + " to BDB env " + envFile); + historyDB.sync(); + historyDB.close(); + targetEnv.close(); + } else { + logger.info(count + " records found in " + sourcePath); + } + } + + return count; + } + + /** + * Populates a given StoredSortedMap (history map) from an old + * environment db or a persist log. If a map is not provided, only + * logs the entries that would have been populated. + * + * @param sourcePath + * source of old entries: can be a path to an existing + * environment db, or a URL or path to a persist log + * @param historyMap + * map to populate (or null for a dry run) + * @return number of records + * @throws DatabaseException + * @throws IOException + */ + public static int copyPersistSourceToHistoryMap(File context, + String sourcePath, + StoredSortedMap historyMap) + throws DatabaseException, IOException, MalformedURLException, + UnsupportedEncodingException { + int count; + // delegate depending on the source + File sourceFile = FileUtils.maybeRelative(context, sourcePath); + if (sourceFile.isDirectory()) { + count = copyPersistEnv(sourceFile, historyMap); + } else { + BufferedReader persistLogReader = null; + if (sourceFile.isFile()) { + persistLogReader = IoUtils.getBufferedReader(sourceFile); + } else { + URL sourceUrl = new URL(sourcePath); + persistLogReader = IoUtils.getBufferedReader(sourceUrl); + } + count = populatePersistEnvFromLog(persistLogReader, historyMap); + } + return count; + } + /** * Utility main for importing a log into a BDB-JE environment or moving a * database between environments (2 arguments), or simply dumping a log - * to stdout in a more readable format (1 argument). + * to stderr in a more readable format (1 argument). * * @param args command-line arguments * @throws DatabaseException * @throws IOException */ public static void main(String[] args) throws DatabaseException, IOException { - if(args.length==2) { - main2args(args); - } else if (args.length==1) { - main1arg(args); + Handler handler = new ConsoleHandler(); + handler.setLevel(Level.ALL); + handler.setFormatter(new OneLineSimpleLogger()); + logger.addHandler(handler); + logger.setUseParentHandlers(false); + + if (args.length == 2) { + logger.setLevel(Level.INFO); + populatePersistEnv(args[0], new File(args[1])); + } else if (args.length == 1) { + logger.setLevel(Level.FINE); + populatePersistEnv(args[0], null); } else { System.out.println("Arguments: "); System.out.println(" source [target]"); @@ -141,160 +336,8 @@ public abstract class PersistProcessor extends Processor { "and target, if present, is a BDB env dir. "); return; } - } - /** - * Move the history information in the first argument (either the path - * to a log or to an environment containing a uri_history database) to - * the environment in the second environment (path; environment will - * be created if it dow not already exist). - * - * @param args command-line arguments - * @throws DatabaseException - * @throws FileNotFoundException - * @throws UnsupportedEncodingException - * @throws IOException - */ - @SuppressWarnings("unchecked") - private static void main2args(String[] args) throws DatabaseException, FileNotFoundException, UnsupportedEncodingException, IOException { - File source = new File(args[0]); - File env = new File(args[1]); - if(!env.exists()) { - env.mkdirs(); - } - - // setup target environment - EnhancedEnvironment targetEnv = setupEnvironment(env); - StoredClassCatalog classCatalog = targetEnv.getClassCatalog(); - Database historyDB = targetEnv.openDatabase( - null,URI_HISTORY_DBNAME,historyDatabaseConfig().toDatabaseConfig()); - StoredSortedMap historyMap = new StoredSortedMap(historyDB, - new StringBinding(), new SerialBinding(classCatalog, - Map.class), true); - - int count = 0; - - if(source.isFile()) { - // scan log, writing to database - BufferedReader br = IoUtils.getBufferedReader(source); - Iterator iter = new LineReadingIterator(br); - while(iter.hasNext()) { - String line = (String) iter.next(); - line = line.trim(); - if(line.length()==0) { - continue; - } - String[] splits = line.split(" "); - if(splits.length!=2) { - logger.severe("bad line: "+line); - continue; - } - try { - historyMap.put( - splits[0], - SerializationUtils.deserialize( - Base64.decodeBase64(splits[1].getBytes("UTF8")))); - } catch (RuntimeException e) { - logger.log(Level.SEVERE,"problem with line: "+line, e); - } - count++; - } - IOUtils.closeQuietly(br); - } else { - // open the source env history DB, copying entries to target env - EnhancedEnvironment sourceEnv = setupEnvironment(source); - StoredClassCatalog sourceClassCatalog = sourceEnv.getClassCatalog(); - Database sourceHistoryDB = sourceEnv.openDatabase( - null,URI_HISTORY_DBNAME,historyDatabaseConfig().toDatabaseConfig()); - StoredSortedMap sourceHistoryMap = new StoredSortedMap(sourceHistoryDB, - new StringBinding(), new SerialBinding(sourceClassCatalog, - Map.class), true); - Iterator iter = sourceHistoryMap.entrySet().iterator(); - while(iter.hasNext()) { - Entry item = (Entry) iter.next(); - historyMap.put(item.getKey(), item.getValue()); - count++; - } - StoredIterator.close(iter); - sourceHistoryDB.close(); - sourceEnv.close(); - } - - // cleanup - historyDB.sync(); - historyDB.close(); - targetEnv.close(); - System.out.println(count+" records imported from "+source+" to BDB env "+env); - } - - /** - * Dump the contents of the argument (path to a persist log) to stdout - * in a slightly more readable format. - * - * @param args command-line arguments - * @throws DatabaseException - * @throws FileNotFoundException - * @throws UnsupportedEncodingException - * @throws IOException - */ - @SuppressWarnings("unchecked") - private static void main1arg(String[] args) throws DatabaseException, FileNotFoundException, UnsupportedEncodingException, IOException { - File source = new File(args[0]); - - int count = 0; - - if(source.isFile()) { - // scan log, writing to database - BufferedReader br = IoUtils.getBufferedReader(source); - Iterator iter = new LineReadingIterator(br); - while(iter.hasNext()) { - String line = (String) iter.next(); - line = line.trim(); - if(line.length()==0) { - continue; - } - String[] splits = line.split(" "); - if(splits.length!=2) { - logger.severe("bad line: "+line); - continue; - } - try { - Map alist = (Map)SerializationUtils.deserialize( - Base64.decodeBase64(splits[1].getBytes("UTF8"))); - System.out.println( - splits[0] + " " + ArchiveUtils.prettyString(alist)); - } catch (RuntimeException e) { - logger.log(Level.SEVERE,"problem with line: "+line, e); - } - count++; - } - IOUtils.closeQuietly(br); - } else { - // open the source env history DB, copying entries to target env - EnhancedEnvironment sourceEnv = setupEnvironment(source); - StoredClassCatalog sourceClassCatalog = sourceEnv.getClassCatalog(); - Database sourceHistoryDB = sourceEnv.openDatabase( - null,URI_HISTORY_DBNAME,historyDatabaseConfig().toDatabaseConfig()); - StoredSortedMap sourceHistoryMap = new StoredSortedMap(sourceHistoryDB, - new StringBinding(), new SerialBinding(sourceClassCatalog, - Map.class), true); - Iterator iter = sourceHistoryMap.entrySet().iterator(); - while(iter.hasNext()) { - Entry item = (Entry) iter.next(); -// Map alist = (Map)item.getValue(); - Map alist = (Map) sourceHistoryMap.get(item.getKey()); - System.out.println(item.getKey() + " " + ArchiveUtils.prettyString(alist)); - count++; - } - StoredIterator.close(iter); - sourceHistoryDB.close(); - sourceEnv.close(); - } - - System.out.println(count+" records dumped from "+source); - } - public static EnhancedEnvironment setupEnvironment(File env) throws DatabaseException { EnvironmentConfig envConfig = new EnvironmentConfig(); envConfig.setAllowCreate(true);