From 01072b4b462fd56c2cce4ea04658a62f0fe12231 Mon Sep 17 00:00:00 2001 From: Kenji Nagahashi Date: Wed, 6 Mar 2013 17:28:38 -0800 Subject: [PATCH] support mixing fetch histories from multiple sources (hbase, WBM). --- .../modules/recrawl/FetchHistoryHelper.java | 99 +++++++++++++++++++ .../recrawl/wbm/WbmPersistLoadProcessor.java | 79 ++++++++++----- .../wbm/WbmPersistLoadProcessorTest.java | 41 +++++++- 3 files changed, 187 insertions(+), 32 deletions(-) create mode 100644 main/java/org/archive/modules/recrawl/FetchHistoryHelper.java diff --git a/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java b/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java new file mode 100644 index 00000000..bc659a79 --- /dev/null +++ b/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java @@ -0,0 +1,99 @@ +package org.archive.modules.recrawl; + +import java.text.DateFormat; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.archive.modules.CrawlURI; + +public class FetchHistoryHelper { + private static final Log logger = LogFactory.getLog(FetchHistoryHelper.class); + /** + * key for storing timestamp in crawl history map. + */ + public static final String A_TIMESTAMP = ".ts"; + + public static String getHeaderValue(org.apache.commons.httpclient.HttpMethod method, String name) { + org.apache.commons.httpclient.Header header = method.getResponseHeader(name); + return header != null ? header.getValue() : null; + } + + /** + * returns a Map to store recrawl data, positioned property in CrawlURI's + * fetch history array property, according to {@code timestamp}. this makes it possible + * to import crawl history data from multiple sources. + * @param uri target {@link CrawlURI} + * @param timestamp timestamp (in ms) of crawl history to be added. + * @return Map object to store recrawl data, or null if {@code timestamp} is older + * than existing crawl history entry and there's no room for it. + * @see #setHistoryLength(int) + */ + @SuppressWarnings("unchecked") + public static Map getFetchHistory(CrawlURI uri, long timestamp, int historyLength) { + Map data = uri.getData(); + Map[] history = (Map[])data.get(RecrawlAttributeConstants.A_FETCH_HISTORY); + if (history == null) { + // there's no history records at all. + // FetchHistoryProcessor assumes history is HashMap[], not Map[]. + history = new HashMap[historyLength]; + data.put(RecrawlAttributeConstants.A_FETCH_HISTORY, history); + } + for (int i = 0; i < history.length; i++) { + if (history[i] == null) { + history[i] = new HashMap(); + history[i].put(A_TIMESTAMP, timestamp); + return history[i]; + } + Object ts = history[i].get(A_TIMESTAMP); + // no timestamp value is regarded as older than anything. + if (!(ts instanceof Long) || timestamp > (Long)ts) { + if (i < history.length - 2) { + System.arraycopy(history, i, history, i + 1, history.length - i - 1); + } else if (i == history.length - 2) { + history[i + 1] = history[i]; + } + history[i] = new HashMap(); + history[i].put(A_TIMESTAMP, timestamp); + return history[i]; + } + } + return null; + } + + protected static final DateFormat HTTP_DATE_FORMAT = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss zzz"); + + protected FetchHistoryHelper() { + } + + /** + * converts time in HTTP Date format {@code dateStr} to seconds + * since epoch. + * @param dateStr time in HTTP Date format. + * @return seconds since epoch + */ + public static long parseHttpDate(String dateStr) { + synchronized (HTTP_DATE_FORMAT) { + try { + Date d = HTTP_DATE_FORMAT.parse(dateStr); + return d.getTime() / 1000; + } catch (ParseException ex) { + if (logger.isDebugEnabled()) + logger.debug("bad HTTP DATE: " + dateStr); + return 0; + } + } + } + + public static String formatHttpDate(long time) { + synchronized (HTTP_DATE_FORMAT) { + // format is not thread safe either + return HTTP_DATE_FORMAT.format(new Date(time * 1000)); + } + } + +} \ No newline at end of file diff --git a/main/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessor.java b/main/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessor.java index 7fed5288..f58409ba 100644 --- a/main/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessor.java +++ b/main/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessor.java @@ -23,7 +23,9 @@ import java.io.InputStream; import java.io.UnsupportedEncodingException; import java.net.URLEncoder; import java.nio.ByteBuffer; +import java.text.ParseException; import java.util.Date; +import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicLong; @@ -49,16 +51,17 @@ import org.archive.modules.CrawlURI; import org.archive.modules.ProcessResult; import org.archive.modules.Processor; import org.archive.modules.hq.GzipInflatingHttpEntityWrapper; -import org.archive.modules.hq.recrawl.RecrawlDataSchemaBase; +import org.archive.modules.recrawl.FetchHistoryHelper; import org.archive.modules.recrawl.RecrawlAttributeConstants; import org.archive.util.ArchiveUtils; +import org.archive.util.DateUtils; /** * A {@link Processor} for retrieving recrawl info from remote Wayback Machine index. * This is currently in the early stage of experiment. Both low-level protocol and WBM API * semantics will certainly undergo several revisions. *

Current interface:

- *

http://web-beta.archive.org/cdx/search?url=archive.org&startDate=1999 will return raw + *

http://web-beta.archive.org/cdx/search/cdx?url=archive.org&startDate=1999 will return raw * CDX lines for archive.org, since 1999-01-01 00:00:00. *

*

As index is updated in a separate batch processing job, there's no "Store" counterpart.

@@ -69,6 +72,15 @@ public class WbmPersistLoadProcessor extends Processor { private HttpClient client; + private int historyLength = 2; + + public void setHistoryLength(int historyLength) { + this.historyLength = historyLength; + } + public int getHistoryLength() { + return historyLength; + } + // ~Jan 2, 2013 //private String queryURL = "http://web-beta.archive.org/cdx/search"; private String queryURL = "http://web.archive.org/cdx/search/cdx"; @@ -249,17 +261,19 @@ public class WbmPersistLoadProcessor extends Processor { return ProcessResult.PROCEED; } InputStream is = null; - String hash = null; + Map info = null; try { - hash = getLastHash(is = entity.getContent()); + info = getLastCrawl(is = entity.getContent()); } catch (IOException ex) { } finally { if (is != null) ArchiveUtils.closeQuietly(is); } - if (hash != null) { - Map history = RecrawlDataSchemaBase.getFetchHistory(curi); - history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, contentDigestScheme + hash); + if (info != null) { + Map history = FetchHistoryHelper.getFetchHistory(curi, + (Long)info.get(FetchHistoryHelper.A_TIMESTAMP), historyLength); + if (history != null) + history.putAll(info); loadedCount.incrementAndGet(); } else { failedCount.incrementAndGet(); @@ -267,34 +281,37 @@ public class WbmPersistLoadProcessor extends Processor { return ProcessResult.PROCEED; } - protected String getLastHash(InputStream is) throws IOException { + protected HashMap getLastCrawl(InputStream is) throws IOException { // read CDX lines, save most recent (at the end) hash. ByteBuffer buffer = ByteBuffer.allocate(32); + ByteBuffer tsbuffer = ByteBuffer.allocate(14); int field = 0; int c; do { - if (field == 5) { - buffer.clear(); - while (buffer.remaining() > 0) { + c = is.read(); + if (field == 1) { + // 14-digits timestamp + tsbuffer.clear(); + while (Character.isDigit(c) && tsbuffer.remaining() > 0) { + tsbuffer.put((byte)c); c = is.read(); - if (c >= 'A' && c <= 'Z' || c >= '0' && c <= '9') { - buffer.put((byte)c); - } else { - break; - } } - if (buffer.remaining() == 0) { + if (c != ' ' || tsbuffer.position() != 14) { + tsbuffer.clear(); + } + // fall through to skip the rest + } else if (field == 5) { + buffer.clear(); + while ((c >= 'A' && c <= 'Z' || c >= '0' && c <= '9') && buffer.remaining() > 0) { + buffer.put((byte)c); c = is.read(); - if (c == ' ') { - field++; - continue; - } - } else { + } + if (c != ' ' || buffer.position() != 32) { buffer.clear(); } + // fall through to skip the rest } - while (true) { - c = is.read(); + while (true) { if (c == -1) { break; } else if (c == '\n') { @@ -304,13 +321,21 @@ public class WbmPersistLoadProcessor extends Processor { field++; break; } + c = is.read(); } } while (c != -1); + + HashMap info = new HashMap(); if (buffer.remaining() == 0) { - return new String(buffer.array()); - } else { - return null; + info.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, contentDigestScheme + new String(buffer.array())); } + if (tsbuffer.remaining() == 0) { + try { + info.put(FetchHistoryHelper.A_TIMESTAMP, DateUtils.parse14DigitDate(new String(tsbuffer.array())).getTime()); + } catch (ParseException ex) { + } + } + return info.isEmpty() ? null : info; } /** diff --git a/test/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessorTest.java b/test/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessorTest.java index aa4750b0..55a5aa8e 100644 --- a/test/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessorTest.java +++ b/test/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessorTest.java @@ -8,6 +8,7 @@ import java.io.BufferedReader; import java.io.InputStream; import java.io.InputStreamReader; import java.net.URLEncoder; +import java.util.HashMap; import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -23,9 +24,10 @@ import org.apache.http.message.BasicHttpResponse; import org.apache.http.message.BasicStatusLine; import org.archive.modules.CrawlURI; import org.archive.modules.ProcessResult; -import org.archive.modules.hq.recrawl.RecrawlDataSchemaBase; +import org.archive.modules.recrawl.FetchHistoryHelper; import org.archive.modules.recrawl.RecrawlAttributeConstants; import org.archive.net.UURIFactory; +import org.archive.util.DateUtils; import org.easymock.EasyMock; import org.junit.Test; @@ -37,6 +39,7 @@ import com.google.common.util.concurrent.ExecutionList; * * TODO: *
    + *
  • test for pathological cases: illegal chars, incorrect length, etc. *
  • test if connection is properly released back to the pool for all possible cases. *
* @author kenji @@ -57,16 +60,17 @@ public class WbmPersistLoadProcessorTest { } /** - * stub HttpResponse for normal case + * stub HttpResponse for normal case. * @author kenji * */ public static class TestNormalHttpResponse extends BasicHttpResponse { + public static final String EXPECTED_TS = "20121101155310"; public static final String EXPECTED_HASH = "GHN5VKF3TBKNSEZTASOM23BJRTKFFNJK"; public TestNormalHttpResponse() { super(new BasicStatusLine(new ProtocolVersion("HTTP", 1, 0), 200, "OK")); setEntity(new ByteArrayEntity( - ("org,archive)/ 20121101155310 http://archive.org/ text/html 200 "+ + ("org,archive)/ "+EXPECTED_TS+" http://archive.org/ text/html 200 "+ EXPECTED_HASH+" - - 6908 982548871 "+ "google.es-20121101-155506/IA-FOC-google.es-20121101073708-00001.warc.gz\n" ).getBytes() @@ -74,6 +78,15 @@ public class WbmPersistLoadProcessorTest { } } + protected Map getFetchHistory(CrawlURI curi, int idx) { + Map data = curi.getData(); + @SuppressWarnings("unchecked") + Map[] historyArray = (Map[])data.get(RecrawlAttributeConstants.A_FETCH_HISTORY); + assertNotNull(historyArray); + Map history = historyArray[idx]; + return history; + } + @Test public void testInnerProcessResultSingleShotWithMock() throws Exception { // because AbstractHttpClient marks most execute(...) methods final, it is very tiresome to implement @@ -88,13 +101,31 @@ public class WbmPersistLoadProcessorTest { t.setHttpClient(client); t.setContentDigestScheme(CONTENT_DIGEST_SCHEME); CrawlURI curi = new CrawlURI(UURIFactory.getInstance("http://archive.org/")); + + // put history entry newer than being loaded + long expected_ts = DateUtils.parse14DigitDate(TestNormalHttpResponse.EXPECTED_TS).getTime(); + Map[] fetchHistory = (Map[])curi.getData().get(RecrawlAttributeConstants.A_FETCH_HISTORY); + if (fetchHistory == null) { + fetchHistory = new HashMap[2]; + curi.getData().put(RecrawlAttributeConstants.A_FETCH_HISTORY, fetchHistory); + } + fetchHistory[0] = new HashMap(); + fetchHistory[0].put(FetchHistoryHelper.A_TIMESTAMP, expected_ts + 2000); + fetchHistory[1] = new HashMap(); + fetchHistory[1].put(FetchHistoryHelper.A_TIMESTAMP, expected_ts - 2000); + ProcessResult result = t.innerProcessResult(curi); assertEquals("result is PROCEED", ProcessResult.PROCEED, result); - Map history = RecrawlDataSchemaBase.getFetchHistory(curi); + // newly loaded history entry should fall in between two existing entries (index=1) + Map history = getFetchHistory(curi, 1); assertNotNull("history", history); String hash = (String)history.get(RecrawlAttributeConstants.A_CONTENT_DIGEST); assertEquals("CONTENT_DIGEST", CONTENT_DIGEST_SCHEME+TestNormalHttpResponse.EXPECTED_HASH, hash); + + Long ts = (Long)history.get(FetchHistoryHelper.A_TIMESTAMP); + assertNotNull("ts is non-null", ts); + assertEquals("'ts' has expected timestamp", expected_ts, ts.longValue()); } @Test @@ -103,7 +134,7 @@ public class WbmPersistLoadProcessorTest { //CrawlURI curi = new CrawlURI(UURIFactory.getInstance("http://archive.org/")); CrawlURI curi = new CrawlURI(UURIFactory.getInstance("http://www.mext.go.jp/null.gif")); ProcessResult result = t.innerProcessResult(curi); - Map history = RecrawlDataSchemaBase.getFetchHistory(curi); + Map history = getFetchHistory(curi, 0); assertNotNull("getFetchHistory returns non-null", history); String hash = (String)history.get(RecrawlAttributeConstants.A_CONTENT_DIGEST); assertNotNull("CONTENT_DIGEST is non-null", hash);