diff --git a/commons/src/main/java/org/archive/util/Base32.java b/commons/src/main/java/org/archive/util/Base32.java index addfd11e..ad06309f 100644 --- a/commons/src/main/java/org/archive/util/Base32.java +++ b/commons/src/main/java/org/archive/util/Base32.java @@ -18,142 +18,32 @@ */ package org.archive.util; +import com.google.common.io.BaseEncoding; + /** - * Base32 - encodes and decodes RFC3548 Base32 - * (see http://www.faqs.org/rfcs/rfc3548.html ) - * - * Imported public-domain code of Bitzi. - * - * @author Robert Kaye - * @author Gordon Mohr + * @deprecated Use {@link com.google.common.io.BaseEncoding#base32()} */ +@Deprecated public class Base32 { - private static final String base32Chars = - "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"; - private static final int[] base32Lookup = - { 0xFF,0xFF,0x1A,0x1B,0x1C,0x1D,0x1E,0x1F, // '0', '1', '2', '3', '4', '5', '6', '7' - 0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF, // '8', '9', ':', ';', '<', '=', '>', '?' - 0xFF,0x00,0x01,0x02,0x03,0x04,0x05,0x06, // '@', 'A', 'B', 'C', 'D', 'E', 'F', 'G' - 0x07,0x08,0x09,0x0A,0x0B,0x0C,0x0D,0x0E, // 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O' - 0x0F,0x10,0x11,0x12,0x13,0x14,0x15,0x16, // 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W' - 0x17,0x18,0x19,0xFF,0xFF,0xFF,0xFF,0xFF, // 'X', 'Y', 'Z', '[', '\', ']', '^', '_' - 0xFF,0x00,0x01,0x02,0x03,0x04,0x05,0x06, // '`', 'a', 'b', 'c', 'd', 'e', 'f', 'g' - 0x07,0x08,0x09,0x0A,0x0B,0x0C,0x0D,0x0E, // 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o' - 0x0F,0x10,0x11,0x12,0x13,0x14,0x15,0x16, // 'p', 'q', 'r', 's', 't', 'u', 'v', 'w' - 0x17,0x18,0x19,0xFF,0xFF,0xFF,0xFF,0xFF // 'x', 'y', 'z', '{', '|', '}', '~', 'DEL' - }; - /** - * Encodes byte array to Base32 String. - * - * @param bytes Bytes to encode. - * @return Encoded byte array bytes as a String. - * + * @deprecated Use {@link com.google.common.io.BaseEncoding#base32()} */ + @Deprecated static public String encode(final byte[] bytes) { - int i = 0, index = 0, digit = 0; - int currByte, nextByte; - StringBuffer base32 = new StringBuffer((bytes.length + 7) * 8 / 5); - - while (i < bytes.length) { - currByte = (bytes[i] >= 0) ? bytes[i] : (bytes[i] + 256); // unsign - - /* Is the current digit going to span a byte boundary? */ - if (index > 3) { - if ((i + 1) < bytes.length) { - nextByte = - (bytes[i + 1] >= 0) ? bytes[i + 1] : (bytes[i + 1] + 256); - } else { - nextByte = 0; - } - - digit = currByte & (0xFF >> index); - index = (index + 5) % 8; - digit <<= index; - digit |= nextByte >> (8 - index); - i++; - } else { - digit = (currByte >> (8 - (index + 5))) & 0x1F; - index = (index + 5) % 8; - if (index == 0) - i++; - } - base32.append(base32Chars.charAt(digit)); - } - - return base32.toString(); + return BaseEncoding.base32() + .omitPadding() + .lowerCase() + .encode(bytes) + .toUpperCase(); } - /** - * Decodes the given Base32 String to a raw byte array. - * - * @param base32 - * @return Decoded base32 String as a raw byte array. + * @deprecated Use {@link com.google.common.io.BaseEncoding#base32()} */ - static public byte[] decode(final String base32) { - int i, index, lookup, offset, digit; - byte[] bytes = new byte[base32.length() * 5 / 8]; - - for (i = 0, index = 0, offset = 0; i < base32.length(); i++) { - lookup = base32.charAt(i) - '0'; - - /* Skip chars outside the lookup table */ - if (lookup < 0 || lookup >= base32Lookup.length) { - continue; - } - - digit = base32Lookup[lookup]; - - /* If this digit is not in the table, ignore it */ - if (digit == 0xFF) { - continue; - } - - if (index <= 3) { - index = (index + 5) % 8; - if (index == 0) { - bytes[offset] |= digit; - offset++; - if (offset >= bytes.length) - break; - } else { - bytes[offset] |= digit << (8 - index); - } - } else { - index = (index + 5) % 8; - bytes[offset] |= (digit >>> index); - offset++; - - if (offset >= bytes.length) { - break; - } - bytes[offset] |= digit << (8 - index); - } - } - return bytes; - } - - /** For testing, take a command-line argument in Base32, decode, print in hex, - * encode, print - * - * @param args - */ - static public void main(String[] args) { - if (args.length == 0) { - System.out.println("Supply a Base32-encoded argument."); - return; - } - System.out.println(" Original: " + args[0]); - byte[] decoded = Base32.decode(args[0]); - System.out.print(" Hex: "); - for (int i = 0; i < decoded.length; i++) { - int b = decoded[i]; - if (b < 0) { - b += 256; - } - System.out.print((Integer.toHexString(b + 256)).substring(1)); - } - System.out.println(); - System.out.println("Reencoded: " + Base32.encode(decoded)); + @Deprecated + static public byte[] decode(final String base32) { + return BaseEncoding.base32() + .omitPadding() + .lowerCase() + .decode(base32.toLowerCase()); } } diff --git a/commons/src/main/java/org/archive/util/BloomFilter64bit.java b/commons/src/main/java/org/archive/util/BloomFilter64bit.java index eaf57874..9c048a59 100644 --- a/commons/src/main/java/org/archive/util/BloomFilter64bit.java +++ b/commons/src/main/java/org/archive/util/BloomFilter64bit.java @@ -27,91 +27,30 @@ package org.archive.util; import java.io.Serializable; +import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.security.SecureRandom; import java.util.Random; -/** A Bloom filter. - * - * ADAPTED/IMPROVED VERSION OF MG4J it.unimi.dsi.mg4j.util.BloomFilter - * - *

KEY CHANGES: - * - *

- * - *
- * - *

Instances of this class represent a set of character sequences (with - * false positives) using a Bloom filter. Because of the way Bloom filters work, - * you cannot remove elements. - * - *

Bloom filters have an expected error rate, depending on the number - * of hash functions used, on the filter size and on the number of elements in - * the filter. This implementation uses a variable optimal number of hash - * functions, depending on the expected number of elements. More precisely, a - * Bloom filter for n character sequences with d hash - * functions will use ln 2 dn ≈ - * 1.44 dn bits; false positives will happen with - * probability 2-d. - * - *

Hash functions are generated at creation time using universal hashing. - * Each hash function uses {@link #NUMBER_OF_WEIGHTS} random integers, which - * are cyclically multiplied by the character codes in a character sequence. - * The resulting integers are XOR-ed together. - * - *

This class exports access methods that are very similar to those of - * {@link java.util.Set}, but it does not implement that interface, as too - * many non-optional methods would be unimplementable (e.g., iterators). - * - * @author Sebastiano Vigna - * @author Gordon Mohr - */ +import com.google.common.annotations.VisibleForTesting; +import com.google.common.hash.Funnels; +import com.google.common.primitives.Ints; + public class BloomFilter64bit implements Serializable, BloomFilter { - private static final long serialVersionUID = 2L; + private static final long serialVersionUID = 3L; - /** The number of weights used to create hash functions. */ - protected final static int NUMBER_OF_WEIGHTS = 2083; // CHANGED FROM 16 - /** The number of bits in this filter. */ - final protected long m; - /** if bitfield is an exact power of 2 in length, it is this power */ - protected int power = -1; /** The expected number of inserts; determines calculated size */ - final protected long expectedInserts; - /** The number of hash functions used by this filter. */ - final protected int d; - /** The underlying bit vector */ - final protected long[][] bits; - /** The random integers used to generate the hash functions. */ - final protected long[][] weight; + private final long expectedInserts; /** The number of elements currently in the filter. It may be * smaller than the actual number of additions of distinct character * sequences because of false positives. */ - protected int size; + private int size; - /** The natural logarithm of 2, used in the computation of the number of bits. */ - protected final static double NATURAL_LOG_OF_2 = Math.log( 2 ); - - /** power-of-two to use as maximum size of bitfield subarrays */ - protected final static int SUBARRAY_POWER_OF_TWO = 26; // 512MiB of longs - /** number of longs in one subarray */ - protected final static int SUBARRAY_LENGTH_IN_LONGS = 1 << SUBARRAY_POWER_OF_TWO; - /** mask for lowest SUBARRAY_POWER_OF_TWO bits */ - protected final static int SUBARRAY_MASK = SUBARRAY_LENGTH_IN_LONGS - 1; //0x0FFFFFFF - - protected final static boolean DEBUG = false; + private final com.google.common.hash.BloomFilter delegate; + private final long bitSize; + private final int numHashFunctions; /** Creates a new Bloom filter with given number of hash functions and * expected number of elements. @@ -141,45 +80,18 @@ public class BloomFilter64bit implements Serializable, BloomFilter { * @param roundUp if true, round bit size up to next-nearest-power-of-2 */ public BloomFilter64bit(final long n, final int d, Random weightsGenerator, boolean roundUp ) { + delegate = com.google.common.hash.BloomFilter.create(Funnels.unencodedCharsFunnel(), Ints.saturatedCast(n), 0.0000003); this.expectedInserts = n; - this.d = d; - long lenInLongs = (long)Math.ceil( ( (long)n * (long)d / NATURAL_LOG_OF_2 ) / 64L ); - if ( lenInLongs > (1L<<48) ) { - throw new IllegalArgumentException( - "This filter would require " + lenInLongs + " longs, " + - "greater than this classes maximum of 2^48 longs (2PiB)." ); - } - long lenInBits = lenInLongs * 64L; - - if(roundUp) { - int pow = 0; - while((1L<s. - * @param k a hash function index (smaller than {@link #d}). - * @return the position in the filter corresponding to s for the hash function k. - */ - protected long hash( final CharSequence s, final int l, final int k ) { - final long[] w = weight[ k ]; - long h = 0; - int i = l; - while( i-- != 0 ) h ^= s.charAt( i ) * w[ i % NUMBER_OF_WEIGHTS ]; - long retVal; - if(power>0) { - retVal = h >>> (64-power); - } else { - // ####----####---- - retVal = ( h & 0x7FFFFFFFFFFFFFFFL ) % m; - } - return retVal; - } - - public long[] bitIndexesFor(CharSequence s) { - long[] ret = new long[d]; - for(int i = 0; i < d; i++) { - ret[i] = hash(s,s.length(),i); - } - return ret; - } - /** Checks whether the given character sequence is in this filter. * *

Note that this method may return true on a character sequence that is has @@ -237,9 +119,7 @@ public class BloomFilter64bit implements Serializable, BloomFilter { */ public boolean contains( final CharSequence s ) { - int i = d, l = s.length(); - while( i-- != 0 ) if ( ! getBit( hash( s, l, i ) ) ) return false; - return true; + return delegate.mightContain(s); } /** Adds a character sequence to the filter. @@ -249,79 +129,19 @@ public class BloomFilter64bit implements Serializable, BloomFilter { */ public boolean add( final CharSequence s ) { - boolean result = false; - int i = d, l = s.length(); - long h; - while( i-- != 0 ) { - h = hash( s, l, i ); - if ( ! setGetBit( h ) ) { - result = true; - } - } - if ( result ) size++; - return result; - } - - protected final static long ADDRESS_BITS_PER_UNIT = 6; // 64=2^6 - protected final static long BIT_INDEX_MASK = (1<<6)-1; // = 63 = 2^BITS_PER_UNIT - 1; - - /** - * Returns from the local bitvector the value of the bit with - * the specified index. The value is true if the bit - * with the index bitIndex is currently set; otherwise, - * returns false. - * - * (adapted from cern.colt.bitvector.QuickBitVector) - * - * @param bitIndex the bit index. - * @return the value of the bit with the specified index. - */ - public boolean getBit(long bitIndex) { - long longIndex = bitIndex >>> ADDRESS_BITS_PER_UNIT; - int arrayIndex = (int) (longIndex >>> SUBARRAY_POWER_OF_TWO); - int subarrayIndex = (int) (longIndex & SUBARRAY_MASK); - return ((bits[arrayIndex][subarrayIndex] & (1L << (bitIndex & BIT_INDEX_MASK))) != 0); + boolean added = delegate.put(s); + if (added) { + size++; + } + return added; } - /** - * Changes the bit with index bitIndex in local bitvector. - * - * (adapted from cern.colt.bitvector.QuickBitVector) - * - * @param bitIndex the index of the bit to be set. + /* (non-Javadoc) + * @see org.archive.util.BloomFilter#getSizeBytes() */ - protected void setBit( long bitIndex) { - long longIndex = bitIndex >>> ADDRESS_BITS_PER_UNIT; - int arrayIndex = (int) (longIndex >>> SUBARRAY_POWER_OF_TWO); - int subarrayIndex = (int) (longIndex & SUBARRAY_MASK); - bits[arrayIndex][subarrayIndex] |= (1L << (bitIndex & BIT_INDEX_MASK)); + public long getSizeBytes() { + return bitSize / 8; } - - /** - * Sets the bit with index bitIndex in local bitvector -- - * returning the old value. - * - * (adapted from cern.colt.bitvector.QuickBitVector) - * - * @param bitIndex the index of the bit to be set. - */ - protected boolean setGetBit( long bitIndex) { - long longIndex = bitIndex >>> ADDRESS_BITS_PER_UNIT; - int arrayIndex = (int) (longIndex >>> SUBARRAY_POWER_OF_TWO); - int subarrayIndex = (int) (longIndex & SUBARRAY_MASK); - long mask = 1L << (bitIndex & BIT_INDEX_MASK); - boolean ret = (bits[arrayIndex][subarrayIndex] & mask)!=0; - bits[arrayIndex][subarrayIndex] |= mask; - return ret; - } - - /* (non-Javadoc) - * @see org.archive.util.BloomFilter#getSizeBytes() - */ - public long getSizeBytes() { - // account for ragged-sized last array - return 8*(((bits.length-1)*bits[0].length)+bits[bits.length-1].length); - } @Override public long getExpectedInserts() { @@ -330,6 +150,20 @@ public class BloomFilter64bit implements Serializable, BloomFilter { @Override public long getHashCount() { - return d; + return numHashFunctions; + } + + @VisibleForTesting + public boolean getBit(long bitIndex) { + try { + Field bitsField = delegate.getClass().getDeclaredField("bits"); + bitsField.setAccessible(true); + Object bitarray = bitsField.get(delegate); + Method getBitMethod = bitarray.getClass().getDeclaredMethod("get", long.class); + getBitMethod.setAccessible(true); + return (boolean) getBitMethod.invoke(bitarray, bitIndex); + } catch (Exception e) { + throw new RuntimeException(e); + } } } diff --git a/contrib/pom.xml b/contrib/pom.xml index 18586aaa..a0e094c0 100644 --- a/contrib/pom.xml +++ b/contrib/pom.xml @@ -13,28 +13,6 @@ UTF-8 - - org.apache.hbase - hbase-client - 0.98.6-cdh5.3.5 - - - jets3t - net.java.dev.jets3t - - - junit - junit - - - - jdk.tools - jdk.tools - - - org.archive.heritrix heritrix-engine diff --git a/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java b/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java index 661e0669..a85211df 100644 --- a/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java +++ b/contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java @@ -33,6 +33,7 @@ import java.io.UnsupportedEncodingException; import java.net.URI; import java.nio.channels.Channels; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; @@ -44,6 +45,7 @@ import java.util.logging.Level; import java.util.logging.Logger; import org.apache.commons.httpclient.URIException; +import org.archive.crawler.frontier.AMQPUrlReceiver; import org.archive.crawler.reporting.CrawlerLoggerModule; import org.archive.format.warc.WARCConstants.WARCRecordType; import org.archive.io.warc.WARCRecordInfo; @@ -112,22 +114,60 @@ public class ExtractorYoutubeDL extends Extractor protected static final int MAX_VIDEOS_PER_PAGE = 1000; + // for shouldExtract + protected HashMap seedsYDLd = new HashMap(); + protected transient Logger ydlLogger = null; // unnamed toethread-local temporary file protected transient ThreadLocal tempfile = new ThreadLocal() { protected RandomAccessFile initialValue() { - File t; - try { - t = File.createTempFile("ydl", ".json"); - RandomAccessFile f = new RandomAccessFile(t, "rw"); - t.delete(); - return f; - } catch (IOException e) { - throw new RuntimeException(e); - } + return null; } }; + protected void closeLocalTempFile() { + RandomAccessFile localTemp = tempfile.get(); + if(localTemp == null || !isOpen(localTemp)) + return; // avoid making a new temp file just to close it immediately + try { + getLocalTempFile().close(); + tempfile.set(null); + } + catch (Exception e) { + logger.log(Level.WARNING, "problem closing ydl temp file " + e); + } + } + protected RandomAccessFile getLocalTempFile() { + RandomAccessFile localTemp = tempfile.get(); + if(localTemp == null || !isOpen(localTemp)) { + localTemp = openNewTempFile(); + tempfile.set(localTemp); + } + logger.info("Getting youtube-dl temp file "); + return localTemp; + } + protected boolean isOpen(RandomAccessFile f) { + try { + f.length(); + return true; + } + catch (IOException e) { + logger.info("youtube-dl temp file is not open"); + return false ; + } + } + protected RandomAccessFile openNewTempFile() { + logger.info("Opening New youtube-dl temp file "); + File t; + try { + t = File.createTempFile("ydl", ".json"); + RandomAccessFile f = new RandomAccessFile(t, "rw"); + t.delete(); + return f; + } catch (IOException e) { + throw new RuntimeException(e); + } + } protected CrawlerLoggerModule crawlerLoggerModule; public CrawlerLoggerModule getCrawlerLoggerModule() { @@ -419,7 +459,8 @@ public class ExtractorYoutubeDL extends Extractor * https://github.com/ytdl-org/youtube-dl/blob/master/README.md#format-selection */ ProcessBuilder pb = new ProcessBuilder("youtube-dl", "--ignore-config", - "--simulate", "--dump-single-json", "--format=best", + "--simulate", "--dump-single-json", "--format=best[height <=? 576]", + "--no-cache-dir", "--no-playlist", "--playlist-end=" + MAX_VIDEOS_PER_PAGE, uri.toString()); logger.info("running: " + String.join(" ", pb.command())); @@ -446,7 +487,7 @@ public class ExtractorYoutubeDL extends Extractor } }); - YoutubeDLResults results = new YoutubeDLResults(tempfile.get()); + YoutubeDLResults results = new YoutubeDLResults(getLocalTempFile()); try { try { @@ -507,6 +548,11 @@ public class ExtractorYoutubeDL extends Extractor return false; } + // skip crawl uris received from umbra + if (uri.getAnnotations().contains(AMQPUrlReceiver.A_RECEIVED_FROM_AMQP)) { + return false; + } + String mime = uri.getContentType().toLowerCase(); if (mime.startsWith("text/html") || mime.startsWith("application/xhtml") @@ -524,7 +570,14 @@ public class ExtractorYoutubeDL extends Extractor // should build record for containing page, which has an // annotation like "youtube-dl:3" (no slash) String annotation = findYdlAnnotation(uri); - return annotation != null && !annotation.contains("/"); + boolean shouldBuild = (annotation != null && !annotation.contains("/")); + + // If we processed this uri, then we have an open temp file that won't get closed + // for us by the warc writer + if(!shouldBuild) + closeLocalTempFile(); + + return shouldBuild; } @Override @@ -545,10 +598,10 @@ public class ExtractorYoutubeDL extends Extractor recordInfo.setMimetype("application/vnd.youtube-dl_formats+json;charset=utf-8"); recordInfo.setEnforceLength(true); - tempfile.get().seek(0); - InputStream inputStream = Channels.newInputStream(tempfile.get().getChannel()); + getLocalTempFile().seek(0); + InputStream inputStream = Channels.newInputStream(getLocalTempFile().getChannel()); recordInfo.setContentStream(inputStream); - recordInfo.setContentLength(tempfile.get().length()); + recordInfo.setContentLength(getLocalTempFile().length()); logger.info("built record timestamp=" + timestamp + " url=" + recordInfo.getUrl()); @@ -574,7 +627,7 @@ public class ExtractorYoutubeDL extends Extractor ExtractorYoutubeDL e = new ExtractorYoutubeDL(); FileInputStream in = new FileInputStream("/tmp/ydl-single-video.json"); - YoutubeDLResults results = new YoutubeDLResults(e.tempfile.get()); + YoutubeDLResults results = new YoutubeDLResults(e.getLocalTempFile()); e.streamYdlOutput(in, results); System.out.println("video urls: " + results.videoUrls); System.out.println("page urls: " + results.pageUrls); @@ -590,7 +643,7 @@ public class ExtractorYoutubeDL extends Extractor } in = new FileInputStream("/tmp/ydl-uncgreensboro-limited.json"); - results = new YoutubeDLResults(e.tempfile.get()); + results = new YoutubeDLResults(e.getLocalTempFile()); e.streamYdlOutput(in, results); System.out.println("video urls: " + results.videoUrls); System.out.println("page urls: " + results.pageUrls); diff --git a/contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java b/contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java deleted file mode 100644 index ea381c85..00000000 --- a/contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java +++ /dev/null @@ -1,255 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.postprocessor; - -import java.io.UnsupportedEncodingException; -import java.util.Map; -import java.util.Properties; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.ThreadFactory; -import java.util.logging.Logger; - -import org.apache.commons.collections.Closure; -import org.apache.kafka.clients.producer.Callback; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; -import org.apache.kafka.common.serialization.ByteArraySerializer; -import org.apache.kafka.common.serialization.StringSerializer; -import org.archive.crawler.framework.Frontier; -import org.archive.crawler.frontier.AbstractFrontier; -import org.archive.crawler.frontier.BdbFrontier; -import org.archive.crawler.io.UriProcessingFormatter; -import org.archive.modules.CrawlURI; -import org.archive.modules.Processor; -import org.archive.modules.net.ServerCache; -import org.json.JSONObject; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.Lifecycle; - -/** - * For Kafka 0.8.x. Sends messages in asynchronous mode (producer.type=async) - * and does not wait for acknowledgment from kafka (request.required.acks=0). - * Sends messages with no key. These things could be configurable if needed. - * - * @see UriProcessingFormatter - * @author nlevitt - */ -public class KafkaCrawlLogFeed extends Processor implements Lifecycle { - - protected static final Logger logger = Logger.getLogger(KafkaCrawlLogFeed.class.getName()); - - protected Frontier frontier; - public Frontier getFrontier() { - return this.frontier; - } - /** Autowired frontier, needed to determine when a url is finished. */ - @Autowired - public void setFrontier(Frontier frontier) { - this.frontier = frontier; - } - - protected ServerCache serverCache; - public ServerCache getServerCache() { - return this.serverCache; - } - @Autowired - public void setServerCache(ServerCache serverCache) { - this.serverCache = serverCache; - } - - protected Map extraFields; - public Map getExtraFields() { - return extraFields; - } - public void setExtraFields(Map extraFields) { - this.extraFields = extraFields; - } - - protected boolean dumpPendingAtClose = false; - public boolean getDumpPendingAtClose() { - return dumpPendingAtClose; - } - /** - * If true, publish all pending urls (i.e. queued urls still in the - * frontier) when crawl job is stopping. They are recognizable by the status - * field which has the value 0. - * - * @see BdbFrontier#setDumpPendingAtClose(boolean) - */ - public void setDumpPendingAtClose(boolean dumpPendingAtClose) { - this.dumpPendingAtClose = dumpPendingAtClose; - } - - protected String brokerList = "localhost:9092"; - /** Kafka broker list (kafka property "metadata.broker.list"). */ - public void setBrokerList(String brokerList) { - this.brokerList = brokerList; - } - public String getBrokerList() { - return brokerList; - } - - protected String topic = "heritrix-crawl-log"; - public void setTopic(String topic) { - this.topic = topic; - } - public String getTopic() { - return topic; - } - - protected byte[] buildMessage(CrawlURI curi) { - JSONObject jo = CrawlLogJsonBuilder.buildJson(curi, getExtraFields(), getServerCache()); - try { - return jo.toString().getBytes("UTF-8"); - } catch (UnsupportedEncodingException e) { - throw new RuntimeException(e); - } - } - - @Override - protected boolean shouldProcess(CrawlURI curi) { - if (frontier instanceof AbstractFrontier) { - return !((AbstractFrontier) frontier).needsReenqueuing(curi); - } else { - return false; - } - } - - private transient long pendingDumpedCount = 0l; - @Override - public synchronized void stop() { - if (!isRunning) { - return; - } - - if (dumpPendingAtClose) { - if (frontier instanceof BdbFrontier) { - - Closure closure = new Closure() { - public void execute(Object curi) { - try { - innerProcess((CrawlURI) curi); - pendingDumpedCount++; - } catch (InterruptedException e) { - } - } - }; - - logger.info("dumping " + frontier.queuedUriCount() + " queued urls to kafka feed"); - ((BdbFrontier) frontier).forAllPendingDo(closure); - logger.info("dumped " + pendingDumpedCount + " queued urls to kafka feed"); - } else { - logger.warning("frontier is not a BdbFrontier, cannot dumpPendingAtClose"); - } - } - - String rateStr = String.format("%1.1f", 0.01 * stats.errors / stats.total); - logger.info("final error count: " + stats.errors + "/" + stats.total + " (" + rateStr + "%)"); - - if (kafkaProducer != null) { - kafkaProducer.close(); - kafkaProducer = null; - } - if (kafkaProducerThreads != null) { - kafkaProducerThreads.destroy(); - kafkaProducerThreads = null; - } - - super.stop(); - } - - private transient ThreadGroup kafkaProducerThreads; - - transient protected KafkaProducer kafkaProducer; - protected KafkaProducer kafkaProducer() { - if (kafkaProducer == null) { - synchronized (this) { - if (kafkaProducer == null) { - final Properties props = new Properties(); - props.put("bootstrap.servers", getBrokerList()); - props.put("acks", "1"); - props.put("producer.type", "async"); - props.put("key.serializer", StringSerializer.class.getName()); - props.put("value.serializer", ByteArraySerializer.class.getName()); - - /* - * XXX This mess here exists so that the kafka producer - * thread is in a thread group that is not the ToePool, - * so that it doesn't get interrupted at the end of the - * crawl in ToePool.cleanup(). - */ - kafkaProducerThreads = new ThreadGroup(Thread.currentThread().getThreadGroup().getParent(), "KafkaProducerThreads"); - ThreadFactory threadFactory = new ThreadFactory() { - public Thread newThread(Runnable r) { - return new Thread(kafkaProducerThreads, r); - } - }; - Callable> task = new Callable>() { - public KafkaProducer call() throws InterruptedException { - return new KafkaProducer(props); - } - }; - ExecutorService executorService = Executors.newFixedThreadPool(1, threadFactory); - Future> future = executorService.submit(task); - try { - kafkaProducer = future.get(); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } catch (ExecutionException e) { - throw new RuntimeException(e); - } finally { - executorService.shutdown(); - } - } - } - } - return kafkaProducer; - } - - protected final class StatsCallback implements Callback { - public long errors = 0l; - public long total = 0l; - - @Override - public void onCompletion(RecordMetadata metadata, Exception exception) { - total++; - if (exception != null) { - errors++; - } - - if (total % 10000 == 0) { - String rateStr = String.format("%1.1f", 0.01 * errors / total); - logger.info("error count so far: " + errors + "/" + total + " (" + rateStr + "%)"); - } - } - } - protected StatsCallback stats = new StatsCallback(); - - @Override - protected void innerProcess(CrawlURI curi) throws InterruptedException { - byte[] message = buildMessage(curi); - ProducerRecord producerRecord = new ProducerRecord(getTopic(), message); - kafkaProducer().send(producerRecord, stats); - } -} diff --git a/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java b/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java index defe81f3..e43672a3 100644 --- a/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java +++ b/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java @@ -93,7 +93,7 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle { protected static final Logger logger = Logger.getLogger(TroughCrawlLogFeed.class.getName()); - protected static final int BATCH_MAX_TIME_MS = 60 * 1000; + protected static final int BATCH_MAX_TIME_MS = 20 * 1000; protected static final int BATCH_MAX_SIZE = 400; protected AtomicInteger crawledBatchSize = new AtomicInteger(0); protected AtomicInteger uncrawledBatchSize = new AtomicInteger(0); @@ -333,28 +333,29 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle { } } if(batch != null && batch.size() > 0) { - StringBuffer sqlTmpl = new StringBuffer(); - sqlTmpl.append( - "insert into uncrawled_url (timestamp, url, hop_path, status_code, via, seed, host)" - + " values (%s, %s, %s, %s, %s, %s, %s)"); + StringBuffer sqlTmpl = new StringBuffer(); + sqlTmpl.append( + "insert into uncrawled_url (timestamp, url, hop_path, status_code, via, seed, host)" + + " values (%s, %s, %s, %s, %s, %s, %s)"); - for (int i = 1; i < batch.size(); i++) { - sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s)"); - } - - Object[] flattenedValues = new Object[7 * batch.size()]; - for (int i = 0; i < batch.size(); i++) { - System.arraycopy(batch.get(i),0,flattenedValues,7 * i, 7); - } - - try { - troughClient().write(getSegmentId(), sqlTmpl.toString(), flattenedValues); - } catch (Exception e) { - logger.log(Level.WARNING, "problem posting batch of " + batch.size() + " uncrawled urls to trough segment " + getSegmentId(), e); - } - - uncrawledBatchLastTime = System.currentTimeMillis(); - uncrawledBatch.clear(); + for (int i = 1; i < batch.size(); i++) { + sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s)"); } + + Object[] flattenedValues = new Object[7 * batch.size()]; + for (int i = 0; i < batch.size(); i++) { + System.arraycopy(batch.get(i),0,flattenedValues,7 * i, 7); + } + + try { + troughClient().write(getSegmentId(), sqlTmpl.toString(), flattenedValues); + } catch (Exception e) { + logger.log(Level.WARNING, "problem posting batch of " + batch.size() + " uncrawled urls to trough segment " + getSegmentId(), e); + } + + uncrawledBatchLastTime = System.currentTimeMillis(); + uncrawledBatch.clear(); + } + } } diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBase.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBase.java deleted file mode 100644 index 751d6d53..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBase.java +++ /dev/null @@ -1,126 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - -import java.io.IOException; -import java.util.Map; -import java.util.Map.Entry; -import java.util.logging.Logger; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.springframework.context.Lifecycle; - -/** - * Represents a deployment of HBase. (An instance, a database, an HBase...) - * - * @author nlevitt - */ -public class HBase implements Lifecycle { - - private static final Logger logger = - Logger.getLogger(HBase.class.getName()); - - protected Configuration conf = null; - - private Map properties; - - public Map getProperties() { - return properties; - } - - public void setProperties(Map properties) { - this.properties = properties; - - if (conf == null) { - conf = HBaseConfiguration.create(); - } - for (Entry entry: getProperties().entrySet()) { - conf.set(entry.getKey(), entry.getValue()); - } - } - - public synchronized Configuration configuration() { - if (conf == null) { - conf = HBaseConfiguration.create(); - } - - return conf; - } - - // convenience setters - public void setZookeeperQuorum(String value) { - configuration().set("hbase.zookeeper.quorum", value); - } - public void setZookeeperClientPort(int port) { - configuration().setInt("hbase.zookeeper.property.clientPort", port); - } - - protected transient HBaseAdmin admin; - - public synchronized HBaseAdmin admin() throws IOException { - if (admin == null) { - admin = new HBaseAdmin(configuration()); - } - - return admin; - } - - @Override - public synchronized void stop() { - isRunning = false; - if (admin != null) { - try { - admin.close(); - } catch (IOException e) { - logger.warning("problem closing HBaseAdmin " + admin + " - " + e); - } - - admin = null; - } - if (conf != null) { - // HConnectionManager.deleteConnection(conf); // XXX? - conf = null; - } - } - - protected transient boolean isRunning = false; - @Override - public boolean isRunning() { - return isRunning; - } - - @Override - public void start() { - isRunning = true; - } - - public synchronized void reset() { - if (admin != null) { - try { - admin.close(); - } catch (IOException e) { - logger.warning("problem closing HBaseAdmin " + admin + " - " + e); - } - - admin = null; - } - } -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseContentDigestHistory.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseContentDigestHistory.java deleted file mode 100644 index 27a97e85..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseContentDigestHistory.java +++ /dev/null @@ -1,303 +0,0 @@ -package org.archive.modules.recrawl.hbase; - -import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_CONTENT_DIGEST_COUNT; -import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_DATE; -import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_URL; -import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILENAME; -import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILE_OFFSET; -import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_RECORD_ID; - -import java.io.IOException; -import java.util.HashMap; -import java.util.Iterator; -import java.util.Map; -import java.util.Map.Entry; -import java.util.logging.Level; -import java.util.logging.Logger; - -import org.apache.hadoop.hbase.HColumnDescriptor; -import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Result; -import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException; -import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException; -import org.apache.hadoop.hbase.util.Bytes; -import org.archive.modules.CrawlURI; -import org.archive.modules.recrawl.AbstractContentDigestHistory; -import org.json.JSONException; -import org.json.JSONObject; -import org.springframework.context.Lifecycle; - -import com.google.common.collect.BiMap; -import com.google.common.collect.HashBiMap; - -/** - * HBase content digest history store. Must be a toplevel bean in - * crawler-beans.cxml in order to receive {@link Lifecycle} events. - * - * @see AbstractContentDigestHistory - * @author nlevitt - */ -public class HBaseContentDigestHistory extends AbstractContentDigestHistory implements Lifecycle { - - private static final Logger logger = - Logger.getLogger(HBaseContentDigestHistory.class.getName()); - - protected static final byte[] COLUMN_FAMILY = Bytes.toBytes("f"); - protected static final byte[] COLUMN = Bytes.toBytes("c"); - - protected static final BiMap JSON_KEYS_MAP = HashBiMap.create(); - static { - JSON_KEYS_MAP.put(A_CONTENT_DIGEST_COUNT, "c"); - JSON_KEYS_MAP.put(A_ORIGINAL_URL, "u"); - JSON_KEYS_MAP.put(A_WARC_RECORD_ID, "i"); - JSON_KEYS_MAP.put(A_WARC_FILENAME, "f"); - JSON_KEYS_MAP.put(A_WARC_FILE_OFFSET, "o"); - JSON_KEYS_MAP.put(A_ORIGINAL_DATE, "d"); - } - - protected HBaseTable table; - public void setTable(HBaseTable table) { - this.table = table; - } - - protected boolean addColumnFamily = false; - public boolean getAddColumnFamily() { - return addColumnFamily; - } - /** - * Add the expected column family - * {@link #COLUMN_FAMILY} to the HBase table if the - * table doesn't already have it. - */ - public void setAddColumnFamily(boolean addColumnFamily) { - this.addColumnFamily = addColumnFamily; - } - - protected int retryIntervalMs = 10*1000; - public int getRetryIntervalMs() { - return retryIntervalMs; - } - public void setRetryIntervalMs(int retryIntervalMs) { - this.retryIntervalMs = retryIntervalMs; - } - - protected int maxTries = 1; - public int getMaxTries() { - return maxTries; - } - public void setMaxTries(int maxTries) { - this.maxTries = maxTries; - } - - protected String keySuffix = null; - public String getKeySuffix() { - return keySuffix; - } - - /** - * If not null, keySuffix is appended to the lookup key when loading and - * storing digest history. Thus the key looks like {digest}{keySuffix}, e.g. - * "sha1:22SFHXERHNFOEY6WK7YOUN4PFIPZSB4D-1193". The purpose is to support - * multiple namespaces in a single hbase table, to avoid proliferation of - * small tables. The reason we use a suffix instead of a prefix is to leave - * open the possibility of deduplication across these different namespaces - * at some point in the future. - * - * @param keySuffix - */ - public void setKeySuffix(String keySuffix) { - this.keySuffix = keySuffix; - } - - @Override - protected String persistKeyFor(CrawlURI curi) { - if (keySuffix != null) { - return super.persistKeyFor(curi) + keySuffix; - } else { - return super.persistKeyFor(curi); - } - } - - protected synchronized void addColumnFamily() { - try { - HTableDescriptor oldDesc = table.getHtableDescriptor(); - if (oldDesc.getFamily(COLUMN_FAMILY) == null) { - HTableDescriptor newDesc = new HTableDescriptor(oldDesc); - newDesc.addFamily(new HColumnDescriptor(COLUMN_FAMILY)); - logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc); - HBaseAdmin hbaseAdmin = table.getHbase().admin(); - hbaseAdmin.disableTable(table.getName()); - hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc); - hbaseAdmin.enableTable(table.getName()); - } - } catch (IOException e) { - logger.warning("problem adding column family: " + e); - } - } - - private boolean isRunning; - @Override - public void start() { - // add column family here to avoid disabling table while another - // ToeThread is trying to use it - if (getAddColumnFamily()) { - addColumnFamily(); - } - this.isRunning = true; - } - @Override - public void stop() { - this.isRunning = false; - } - @Override - public boolean isRunning() { - return isRunning; - } - - @Override - public void load(CrawlURI curi) { - // make this call in all cases so that the value is initialized and - // WARCWriterProcessor knows it should put the info in there - HashMap contentDigestHistory = curi.getContentDigestHistory(); - - byte[] key = Bytes.toBytes(persistKeyFor(curi)); - Result hbaseResult = tryHbaseGet(curi, new Get(key)); - - if (hbaseResult != null) { - Map loadedHistory = parseHbaseResult(curi, hbaseResult); - - if (loadedHistory != null) { - if (logger.isLoggable(Level.FINER)) { - logger.finer("loaded history by digest " + persistKeyFor(curi) - + " for uri " + curi + " - " + loadedHistory); - } - contentDigestHistory.putAll(loadedHistory); - } - } - } - - protected Result tryHbaseGet(CrawlURI curi, Get hbaseGet) { - try { - return table.get(hbaseGet); - } catch (IOException e) { - logger.warning("problem retrieving persist data from hbase, proceeding without, for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e); - return null; - } - } - - protected Map parseHbaseResult(CrawlURI curi, Result hbaseResult) { - HashMap loadedHistory = null; - // no data for uri is indicated by empty Result - if (!hbaseResult.isEmpty()) { - byte[] jsonBytes = hbaseResult.getValue(COLUMN_FAMILY, COLUMN); - if (jsonBytes != null) { - JSONObject json = null; - try { - json = new JSONObject(Bytes.toString(jsonBytes)); - loadedHistory = new HashMap(); - @SuppressWarnings("unchecked") - Iterator keyIter = json.keys(); - while (keyIter.hasNext()) { - String jsonKey = keyIter.next(); - Object jsonValue = json.get(jsonKey); - String historyMapKey = JSON_KEYS_MAP.inverse().get(jsonKey); - if (historyMapKey == null) { - logger.warning("unknown key \"" + jsonKey + "\" found in hbase json for digest " + persistKeyFor(curi)); - historyMapKey = jsonKey; - } - loadedHistory.put(historyMapKey, jsonValue); - } - } catch (JSONException e) { - logger.warning("problem parsing json for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e); - } - } else { - // shouldn't happen? result.isEmpty() is normal case - logger.fine("[jsonBytes==null] no persist data for digest " + persistKeyFor(curi) + " uri " + curi); - } - } else { - logger.finest("[result.isEmpty()] no persist data for digest " + persistKeyFor(curi) + " uri " + curi); - } - - return loadedHistory; - } - - @Override - public void store(CrawlURI curi) { - if (!curi.hasContentDigestHistory() - || curi.getContentDigestHistory().isEmpty()) { - return; - } - if (logger.isLoggable(Level.FINER)) { - logger.finer("storing history by digest " + persistKeyFor(curi) - + " for uri " + curi + " - " - + curi.getContentDigestHistory()); - } - - Put hbasePut = createHbasePut(curi); - tryHbasePut(curi, hbasePut); - } - - protected void tryHbasePut(CrawlURI curi, Put p) { - int tryCount = 0; - do { - tryCount++; - try { - table.put(p); - return; - } catch (RetriesExhaustedWithDetailsException e) { - if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) { - addColumnFamily(); - tryCount--; - } else { - logger.warning("put failed " + "(try " + tryCount + " of " - + getMaxTries() + ")" + " for " + curi + " - " + e); - } - } catch (IOException e) { - logger.warning("put failed " + "(try " + tryCount + " of " - + getMaxTries() + ")" + " for " + curi + " - " + e); - } catch (NullPointerException e) { - // HTable.put() throws NullPointerException while connection is lost. - logger.warning("put failed " + "(try " + tryCount + " of " - + getMaxTries() + ")" + " for " + curi + " - " + e); - } - - if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) { - try { - Thread.sleep(getRetryIntervalMs()); - } catch (InterruptedException ex) { - logger.warning("thread interrupted. aborting retry for " + curi); - return; - } - } - } while (tryCount < getMaxTries() && isRunning()); - - if (isRunning()) { - logger.warning("giving up after " + tryCount + " tries on put for " + curi); - } - } - - protected Put createHbasePut(CrawlURI curi) { - byte[] key = Bytes.toBytes(persistKeyFor(curi)); - Put hbasePut = new Put(key); - try { - JSONObject json = new JSONObject(); - for (Entry entry: curi.getContentDigestHistory().entrySet()) { - String jsonKey = JSON_KEYS_MAP.get(entry.getKey()); - if (jsonKey == null) { - logger.warning("unknown key \"" + entry.getKey() + "\" found in content digest history map for " + curi); - jsonKey = entry.getKey(); - } - json.put(jsonKey, entry.getValue()); - } - hbasePut.add(COLUMN_FAMILY, COLUMN, Bytes.toBytes(json.toString())); - } catch (JSONException e) { - // should not happen - all values are either primitive or String. - logger.log(Level.SEVERE, "problem creating json object for digest " + persistKeyFor(curi) + " uri " + curi, e); - } - return hbasePut; - } - -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistLoadProcessor.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistLoadProcessor.java deleted file mode 100644 index 6a967236..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistLoadProcessor.java +++ /dev/null @@ -1,88 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - -import java.io.IOException; -import java.util.logging.Level; -import java.util.logging.Logger; - -import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.Result; -import org.archive.modules.CrawlURI; -import org.archive.modules.ProcessResult; -import org.archive.modules.Processor; -import org.archive.modules.recrawl.FetchHistoryProcessor; - -/** - * A {@link Processor} for retrieving recrawl info from HBase table. - * See {@link HBasePersistProcessor} for table schema. - * As with other fetch history processors, this needs to be combined with {@link FetchHistoryProcessor} - * (set up after FetchHTTP, before WarcWriter) to work. - * @see HBasePersistStoreProcessor - * @author kenji - */ -public class HBasePersistLoadProcessor extends HBasePersistProcessor { - private static final Logger logger = - Logger.getLogger(HBasePersistLoadProcessor.class.getName()); - - @Override - protected ProcessResult innerProcessResult(CrawlURI uri) throws InterruptedException { - byte[] key = rowKeyForURI(uri); - Get g = new Get(key); - try { - Result r = table.get(g); - // no data for uri is indicated by empty Result - if (r.isEmpty()) { - if (logger.isLoggable(Level.FINE)) { - logger.fine(uri + ": "); - } - return ProcessResult.PROCEED; - } - schema.load(r, uri); - if (uri.getFetchStatus() < 0) { - return ProcessResult.FINISH; - } - } catch (IOException e) { - logger.warning("problem retrieving persist data from hbase, proceeding without, for " + uri + " - " + e); - } catch (Exception ex) { - // get() throws RuntimeException upon ZooKeeper connection failures. - // no crawl history load failure should make fetch of URL fail. - logger.log(Level.WARNING, "Get failed for " + uri + ": ", ex); - } - return ProcessResult.PROCEED; - } - - /** - * unused. - */ - @Override - protected void innerProcess(CrawlURI uri) throws InterruptedException { - } - - @Override - protected boolean shouldProcess(CrawlURI uri) { - // TODO: we want deduplicate robots.txt, too. - //if (uri.isPrerequisite()) return false; - String scheme = uri.getUURI().getScheme(); - if (!(scheme.equals("http") || scheme.equals("https") || scheme.equals("ftp"))) { - return false; - } - return true; - } -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistProcessor.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistProcessor.java deleted file mode 100644 index 6d84dd81..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistProcessor.java +++ /dev/null @@ -1,32 +0,0 @@ -package org.archive.modules.recrawl.hbase; - -import org.archive.modules.CrawlURI; -import org.archive.modules.recrawl.AbstractPersistProcessor; -import org.springframework.beans.factory.annotation.Required; - -/** - * A base class for processors for keeping de-duplication data in HBase. - * Table schema is defined by {@link RecrawlDataSchema} implementation. - * @author kenji - */ -public abstract class HBasePersistProcessor extends AbstractPersistProcessor { - - protected HBaseTableBean table; - @Required - public void setTable(HBaseTableBean table) { - this.table = table; - } - - protected RecrawlDataSchema schema; - public RecrawlDataSchema getSchema() { - return schema; - } - @Required - public void setSchema(RecrawlDataSchema schema) { - this.schema = schema; - } - - protected byte[] rowKeyForURI(CrawlURI curi) { - return schema.rowKeyForURI(curi); - } -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistStoreProcessor.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistStoreProcessor.java deleted file mode 100644 index bd611380..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistStoreProcessor.java +++ /dev/null @@ -1,133 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - -import java.io.IOException; -import java.util.logging.Logger; - -import org.apache.hadoop.hbase.HColumnDescriptor; -import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException; -import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException; -import org.apache.hadoop.hbase.util.Bytes; -import org.archive.modules.CrawlURI; -import org.archive.modules.fetcher.FetchStatusCodes; -import org.archive.modules.recrawl.RecrawlAttributeConstants; - -/** - * @author kenji - */ -public class HBasePersistStoreProcessor extends HBasePersistProcessor implements FetchStatusCodes, RecrawlAttributeConstants { - private static final Logger logger = Logger.getLogger(HBasePersistStoreProcessor.class.getName()); - - protected boolean addColumnFamily = false; - public boolean getAddColumnFamily() { - return addColumnFamily; - } - /** - * Add the expected column family - * {@link HBaseContentDigestHistory#COLUMN_FAMILY} to the HBase table if the - * table doesn't already have it. - */ - public void setAddColumnFamily(boolean addColumnFamily) { - this.addColumnFamily = addColumnFamily; - } - - protected int retryIntervalMs = 10*1000; - public int getRetryIntervalMs() { - return retryIntervalMs; - } - public void setRetryIntervalMs(int retryIntervalMs) { - this.retryIntervalMs = retryIntervalMs; - } - - protected int maxTries = 1; - public int getMaxTries() { - return maxTries; - } - public void setMaxTries(int maxTries) { - this.maxTries = maxTries; - } - - protected synchronized void addColumnFamily() { - try { - HTableDescriptor oldDesc = table.getHtableDescriptor(); - byte[] columnFamily = Bytes.toBytes(schema.getColumnFamily()); - if (oldDesc.getFamily(columnFamily) == null) { - HTableDescriptor newDesc = new HTableDescriptor(oldDesc); - newDesc.addFamily(new HColumnDescriptor(columnFamily)); - logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc); - HBaseAdmin hbaseAdmin = table.getHbase().admin(); - hbaseAdmin.disableTable(table.getName()); - hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc); - hbaseAdmin.enableTable(table.getName()); - } - } catch (IOException e) { - logger.warning("problem adding column family: " + e); - } - } - - @Override - protected void innerProcess(CrawlURI uri) { - Put p = schema.createPut(uri); - int tryCount = 0; - do { - tryCount++; - try { - table.put(p); - return; - } catch (RetriesExhaustedWithDetailsException e) { - if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) { - addColumnFamily(); - tryCount--; - } else { - logger.warning("put failed " + "(try " + tryCount + " of " - + getMaxTries() + ")" + " for " + uri + " - " + e); - } - } catch (IOException e) { - logger.warning("put failed " + "(try " + tryCount + " of " - + getMaxTries() + ")" + " for " + uri + " - " + e); - } catch (NullPointerException e) { - // HTable.put() throws NullPointerException while connection is lost. - logger.warning("put failed " + "(try " + tryCount + " of " - + getMaxTries() + ")" + " for " + uri + " - " + e); - } - - if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) { - try { - Thread.sleep(getRetryIntervalMs()); - } catch (InterruptedException ex) { - logger.warning("thread interrupted. aborting retry for " + uri); - return; - } - } - } while (tryCount < getMaxTries() && isRunning()); - - if (isRunning()) { - logger.warning("giving up after " + tryCount + " tries on put for " + uri); - } - } - - @Override - protected boolean shouldProcess(CrawlURI curi) { - return super.shouldStore(curi); - } -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTable.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTable.java deleted file mode 100644 index 3032c2df..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTable.java +++ /dev/null @@ -1,159 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - -import java.io.IOException; -import java.util.logging.Level; -import java.util.logging.Logger; - -import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.TableName; -import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HBaseAdmin; -import org.apache.hadoop.hbase.client.HConnection; -import org.apache.hadoop.hbase.client.HConnectionManager; -import org.apache.hadoop.hbase.client.HTableInterface; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Result; - -/** - * @author kenji - * @author nlevitt - */ -public class HBaseTable extends HBaseTableBean { - - static final Logger logger = - Logger.getLogger(HBaseTable.class.getName()); - - protected boolean create = false; - protected HConnection hconn = null; - protected ThreadLocal htable = new ThreadLocal(); - - public boolean getCreate() { - return create; - } - /** Create the named table if it doesn't exist. */ - public void setCreate(boolean create) { - this.create = create; - } - - public HBaseTable() { - } - - protected synchronized HConnection hconnection() throws IOException { - if (hconn == null) { - hconn = HConnectionManager.createConnection(hbase.configuration()); - } - return hconn; - } - - protected HTableInterface htable() throws IOException { - if (htable.get() == null) { - htable.set(hconnection().getTable(htableName)); - } - return htable.get(); - } - - @Override - public void put(Put p) throws IOException { - try { - htable().put(p); - } catch (IOException e) { - reset(); - throw e; - } - } - - @Override - public Result get(Get g) throws IOException { - try { - return htable().get(g); - } catch (IOException e) { - reset(); - throw e; - } - } - - public HTableDescriptor getHtableDescriptor() throws IOException { - try { - return htable().getTableDescriptor(); - } catch (IOException e) { - reset(); - throw e; - } - } - - @Override - public void start() { - if (getCreate()) { - int attempt = 1; - while (true) { - try { - HBaseAdmin admin = hbase.admin(); - if (!admin.tableExists(htableName)) { - HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(htableName)); - logger.info("hbase table '" + htableName + "' does not exist, creating it... " + desc); - admin.createTable(desc); - } - break; - } catch (IOException e) { - logger.log(Level.WARNING, "(attempt " + attempt + ") problem creating hbase table " + htableName, e); - attempt++; - reset(); - // back off up to 60 seconds between retries - try { - Thread.sleep(Math.min(attempt * 1000, 60000)); - } catch (InterruptedException e1) { - } - } - } - } - - super.start(); - } - - protected void reset() { - if (htable.get() != null) { - try { - htable.get().close(); - } catch (IOException e) { - logger.log(Level.WARNING, "htablename='" + htableName + "' htable.close() threw " + e, e); - } - htable.remove(); - } - - if (hconn != null) { - try { - hconn.close(); - } catch (IOException e) { - logger.log(Level.WARNING, "hconn.close() threw " + e, e); - } - // HConnectionManager.deleteStaleConnection(hconn); - hconn = null; - } - - hbase.reset(); - } - - @Override - public synchronized void stop() { - super.stop(); - reset(); - } -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTableBean.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTableBean.java deleted file mode 100644 index 1ad3164e..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTableBean.java +++ /dev/null @@ -1,78 +0,0 @@ -package org.archive.modules.recrawl.hbase; - -import java.io.IOException; - -import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Result; -import org.archive.modules.recrawl.PersistOnlineProcessor; -import org.springframework.context.Lifecycle; - -/** - * base class for different types of HBaseTable Spring bean implementations. - * @author kenji - * @author nlevitt - * - */ -public abstract class HBaseTableBean implements Lifecycle { - - protected String htableName = PersistOnlineProcessor.URI_HISTORY_DBNAME; - protected HBase hbase = new HBase(); - protected transient boolean isRunning = false; - - // - public void setName(String name) { - this.htableName = name; - } - - public String getName() { - return htableName; - } - // - - /** - * set name of single HTable this instance accesses. - * @param htableName - */ - public void setHtableName(String htableName) { - this.htableName = htableName; - } - public String getHtableName() { - return htableName; - } - - public HBaseTableBean() { - super(); - } - - public void setHbase(HBase hbase) { - this.hbase = hbase; - } - - public HBase getHbase() { - return hbase; - } - - public abstract void put(Put p) throws IOException; - - public abstract Result get(Get g) throws IOException; - - public abstract HTableDescriptor getHtableDescriptor() throws IOException; - - @Override - public boolean isRunning() { - return isRunning; - } - - @Override - public void start() { - isRunning = true; - } - - @Override - public synchronized void stop() { - isRunning = false; - } - -} \ No newline at end of file diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/MultiColumnRecrawlDataSchema.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/MultiColumnRecrawlDataSchema.java deleted file mode 100644 index 18ea44a9..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/MultiColumnRecrawlDataSchema.java +++ /dev/null @@ -1,140 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - -import java.util.Map; -import java.util.logging.Logger; - -import org.apache.hadoop.hbase.KeyValue; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Result; -import org.apache.hadoop.hbase.util.Bytes; -import org.archive.modules.CrawlURI; -import org.archive.modules.fetcher.FetchStatusCodes; -import org.archive.modules.recrawl.FetchHistoryHelper; -import org.archive.modules.recrawl.RecrawlAttributeConstants; - -/** - * RecrawlDataSchema that stores each recrawl data properties in a separate column in single column - * family, whose name may be configured with {@link #setColumnFamily(String)} (default "f"). - *

- * - * @author kenji - */ -public class MultiColumnRecrawlDataSchema extends RecrawlDataSchemaBase implements RecrawlDataSchema, RecrawlAttributeConstants { - static final Logger logger = Logger.getLogger(MultiColumnRecrawlDataSchema.class.getName()); - - public static final byte[] COLUMN_STATUS = Bytes.toBytes("s"); - public static final byte[] COLUMN_CONTENT_DIGEST = Bytes.toBytes("d"); - public static final byte[] COLUMN_ETAG = Bytes.toBytes("e"); - public static final byte[] COLUMN_LAST_MODIFIED = Bytes.toBytes("m"); - - /* (non-Javadoc) - * @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut() - */ - public Put createPut(CrawlURI uri) { - byte[] uriBytes = rowKeyForURI(uri); - byte[] key = uriBytes; - Put p = new Put(key); - String digest = uri.getContentDigestSchemeString(); - if (digest != null) { - p.add(columnFamily, COLUMN_CONTENT_DIGEST, Bytes.toBytes(digest)); - } - p.add(columnFamily, COLUMN_STATUS, Bytes.toBytes(Integer.toString(uri.getFetchStatus()))); - - if (uri.isHttpTransaction()) { - String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER); - if (etag != null) { - // Etqg is usually quoted - if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"') - etag = etag.substring(1, etag.length() - 1); - p.add(columnFamily, COLUMN_ETAG, Bytes.toBytes(etag)); - } - String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER); - if (lastmod != null) { - long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod); - if (lastmod_sec == 0) { - try { - lastmod_sec = uri.getFetchCompletedTime(); - } catch (NullPointerException ex) { - logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine()); - } - } - if (lastmod_sec != 0) - p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(lastmod_sec)); - } else { - try { - long completed = uri.getFetchCompletedTime(); - if (completed != 0) - p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(completed)); - } catch (NullPointerException ex) { - logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine()); - } - } - } - return p; - } - - /* (non-Javadoc) - * @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(java.util.Map, org.apache.hadoop.hbase.client.Result) - */ - public void load(Result result, CrawlURI curi) { - // check for "do-not-crawl" flag - any non-empty data tells not to crawl this - // URL. - byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL); - if (nocrawl != null && nocrawl.length > 0) { - // fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag - // is primarily intended for preventing crawler from stepping on traps. - curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF); - curi.getAnnotations().add("nocrawl"); - return; - } - // all column should have identical timestamp. - KeyValue rkv = result.getColumnLatest(columnFamily, COLUMN_STATUS); - long timestamp = rkv.getTimestamp(); - Map history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength); - // FetchHTTP ignores history with status <= 0 - byte[] status = result.getValue(columnFamily, COLUMN_STATUS); - if (status != null) { - // Note that status is stored as integer text. It's typically three-chars - // that is less than 4-byte integer bits. - history.put(RecrawlAttributeConstants.A_STATUS, Integer.parseInt(Bytes.toString(status))); - byte[] etag = result.getValue(columnFamily, COLUMN_ETAG); - if (etag != null) { - history.put(RecrawlAttributeConstants.A_ETAG_HEADER, Bytes.toString(etag)); - } - byte[] lastmod = result.getValue(columnFamily, COLUMN_LAST_MODIFIED); - if (lastmod != null) { - long lastmod_sec = Bytes.toLong(lastmod); - history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod_sec)); - } - byte[] digest = result.getValue(columnFamily, COLUMN_CONTENT_DIGEST); - if (digest != null) { - history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, Bytes.toString(digest)); - } - } - } - -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchema.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchema.java deleted file mode 100644 index 720d93d3..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchema.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - - -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Result; -import org.archive.modules.CrawlURI; - -/** - * @author kenji - */ -public interface RecrawlDataSchema { - public String getColumnFamily(); - public Put createPut(CrawlURI uri); - public void load(Result result, CrawlURI curi); - // TODO: drop this method by revising createPut(CrawlURI) method. - public byte[] rowKeyForURI(CrawlURI curi); -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchemaBase.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchemaBase.java deleted file mode 100644 index 62e6f3de..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchemaBase.java +++ /dev/null @@ -1,123 +0,0 @@ -package org.archive.modules.recrawl.hbase; - -import java.util.Map; -import java.util.logging.Logger; - -import org.apache.hadoop.hbase.util.Bytes; -import org.archive.modules.CrawlURI; -import org.archive.modules.canonicalize.CanonicalizationRule; -import org.archive.modules.recrawl.FetchHistoryHelper; -import org.archive.modules.recrawl.FetchHistoryProcessor; -import org.archive.modules.recrawl.PersistProcessor; - -/** - * implements common utility methods for implementing {@link RecrawlDataSchema}. - * - * @author kenji - */ -abstract public class RecrawlDataSchemaBase implements RecrawlDataSchema { - private static final Logger logger = Logger.getLogger(RecrawlDataSchemaBase.class.getName()); - - /** - * default value for {@link #columnFamily}. - */ - public static final byte[] DEFAULT_COLUMN_FAMILY = Bytes.toBytes("f"); - protected byte[] columnFamily = DEFAULT_COLUMN_FAMILY; - - public static final byte[] COLUMN_NOCRAWL = Bytes.toBytes("z"); - - /** - * default value for {@link #useCanonicalString}. - */ - public static boolean DEFAULT_USE_CANONICAL_STRING = true; - - private boolean useCanonicalString = DEFAULT_USE_CANONICAL_STRING; - private CanonicalizationRule keyRule = null; - - protected int historyLength = 2; - - public RecrawlDataSchemaBase() { - super(); - } - - public void setColumnFamily(String colf) { - columnFamily = Bytes.toBytes(colf); - } - - public boolean isUseCanonicalString() { - return useCanonicalString; - } - /** - * if set to true, canonicalized string will be used as row key, rather than URI - * @param useCanonicalString - */ - public void setUseCanonicalString(boolean useCanonicalString) { - this.useCanonicalString = useCanonicalString; - } - - public String getColumnFamily() { - return Bytes.toString(columnFamily); - } - - - public CanonicalizationRule getKeyRule() { - return keyRule; - } - /** - * alternative canonicalization rule for generating row key from URI. - * TODO: currently unused. - * @param keyRule - */ - public void setKeyRule(CanonicalizationRule keyRule) { - this.keyRule = keyRule; - } - - public int getHistoryLength() { - return historyLength; - } - - /** - * maximum number of crawl history entries to retain in {@link CrawlURI}. - * when more than this number of crawl history entry is being added by - * {@link #getFetchHistory(CrawlURI, long)}, oldest entry will be discarded. - * {@code historyLength} should be the same number as - * {@link FetchHistoryProcessor#setHistoryLength(int)}, or FetchHistoryProcessor will - * reallocate the crawl history array. - * @param historyLength - * @see FetchHistoryProcessor#setHistoryLength(int) - */ - public void setHistoryLength(int historyLength) { - this.historyLength = historyLength; - } - - /** - * calls {@link FetchHistoryHelper#getFetchHistory(CrawlURI, long, int)} with {@link #historyLength}. - * @param uri CrawlURI from which fetch history is obtained. - * @return Map object for storing re-crawl data (never null). - * @see FetchHistoryHelper#getFetchHistory(CrawlURI, long, int) - * @see FetchHistoryProcessor - */ - protected Map getFetchHistory(CrawlURI uri, long timestamp) { - return FetchHistoryHelper.getFetchHistory(uri, timestamp, historyLength); - } - - /** - * return row key for {@code curi}. - * TODO: move this to HBasePersistProcessor by redesigning {@link RecrawlDataSchema}. - * @param curi {@link CrawlURI} for which a row is being fetched. - * @return row key - */ - public byte[] rowKeyForURI(CrawlURI curi) { - if (useCanonicalString) { - // TODO: use keyRule if specified. - return Bytes.toBytes(PersistProcessor.persistKeyFor(curi)); - } else { - return Bytes.toBytes(curi.toString()); - } - } -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleColumnJsonRecrawlDataSchema.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleColumnJsonRecrawlDataSchema.java deleted file mode 100644 index 7cbff0c2..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleColumnJsonRecrawlDataSchema.java +++ /dev/null @@ -1,174 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - -import java.util.Map; -import java.util.logging.Level; -import java.util.logging.Logger; - -import org.apache.commons.httpclient.HttpMethod; -import org.apache.hadoop.hbase.KeyValue; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Result; -import org.apache.hadoop.hbase.util.Bytes; -import org.archive.modules.CrawlURI; -import org.archive.modules.fetcher.FetchStatusCodes; -import org.archive.modules.recrawl.FetchHistoryHelper; -import org.archive.modules.recrawl.RecrawlAttributeConstants; -import org.json.JSONException; -import org.json.JSONObject; - -/** - * {@linkplain SingleColumnJsonRecrawlDataSchema} stores all re-crawl data properties in a single column, - * in JSON format. As HBase stores each column paired with the row key, it takes a lot of space to store - * each re-crawl data property in its own column. - * - * @author Kenji Nagahashi - */ -public class SingleColumnJsonRecrawlDataSchema extends RecrawlDataSchemaBase -implements RecrawlDataSchema { - static final Logger logger = Logger.getLogger(SingleColumnJsonRecrawlDataSchema.class.getName()); - - public static byte[] DEFAULT_COLUMN = Bytes.toBytes("r"); - - // JSON property names for re-crawl data properties - public static final String PROPERTY_STATUS = "s"; - public static final String PROPERTY_CONTENT_DIGEST = "d"; - public static final String PROPERTY_ETAG = "e"; - public static final String PROPERTY_LAST_MODIFIED = "m"; - - // SHA1 scheme is assumed. - public static final String CONTENT_DIGEST_SCHEME = "sha1:"; - - // single column for storing JSON of re-crawl data - protected byte[] column = DEFAULT_COLUMN; - public void setColumn(String column) { - this.column = Bytes.toBytes(column); - } - public String getColumn() { - return Bytes.toString(column); - } - - /* (non-Javadoc) - * @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut(org.archive.modules.CrawlURI) - */ - public Put createPut(CrawlURI uri) { - byte[] key = rowKeyForURI(uri); - Put p = new Put(key); - JSONObject jo = new JSONObject(); - try { - // TODO should we post warning message when scheme != "sha1"? - String digest = uri.getContentDigestString(); - if (digest != null) { - jo.put(PROPERTY_CONTENT_DIGEST, digest); - } - jo.put(PROPERTY_STATUS, uri.getFetchStatus()); - if (uri.isHttpTransaction()) { - String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER); - if (etag != null) { - // Etag is usually quoted - if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"') - etag = etag.substring(1, etag.length() - 1); - jo.put(PROPERTY_ETAG, etag); - } - String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER); - if (lastmod != null) { - long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod); - if (lastmod_sec == 0) { - try { - lastmod_sec = uri.getFetchCompletedTime(); - } catch (NullPointerException ex) { - logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine()); - } - } - } else { - try { - long completed = uri.getFetchCompletedTime(); - if (completed != 0) - jo.put(PROPERTY_LAST_MODIFIED, completed); - } catch (NullPointerException ex) { - logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine()); - } - } - } - } catch (JSONException ex) { - // should not happen - all values are either primitive or String. - logger.log(Level.SEVERE, "JSON translation failed", ex); - } - p.add(columnFamily, column, Bytes.toBytes(jo.toString())); - return p; - } - - /* (non-Javadoc) - * @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(org.apache.hadoop.hbase.client.Result) - */ - public void load(Result result, CrawlURI curi) { - // check for "do-not-crawl" flag - any non-empty data tells not to crawl this - // URL. - byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL); - if (nocrawl != null && nocrawl.length > 0) { - // fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag - // is primarily intended for preventing crawler from stepping on traps. - curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF); - curi.getAnnotations().add("nocrawl"); - return; - } - - KeyValue rkv = result.getColumnLatest(columnFamily, column); - long timestamp = rkv.getTimestamp(); - Map history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength); - if (history == null) { - // crawl history array is fully occupied by crawl history entries - // newer than timestamp. - return; - } - byte[] jsonBytes = rkv.getValue(); - if (jsonBytes != null) { - JSONObject jo = null; - try { - jo = new JSONObject(Bytes.toString(jsonBytes)); - } catch (JSONException ex) { - logger.warning(String.format("JSON parsing failed for key %1s: %2s", - result.getRow(), ex.getMessage())); - } - if (jo != null) { - int status = jo.optInt(PROPERTY_STATUS, -1); - if (status >= 0) { - history.put(RecrawlAttributeConstants.A_STATUS, status); - } - String digest = jo.optString(PROPERTY_CONTENT_DIGEST); - if (digest != null) { - history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, CONTENT_DIGEST_SCHEME + digest); - } - String etag = jo.optString(PROPERTY_ETAG); - if (etag != null) { - history.put(RecrawlAttributeConstants.A_ETAG_HEADER, etag); - } - long lastmod = jo.optLong(PROPERTY_LAST_MODIFIED); - if (lastmod > 0) { - history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod)); - } - } - } - } - -} diff --git a/contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleHBaseTable.java b/contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleHBaseTable.java deleted file mode 100644 index 7a6d3f9e..00000000 --- a/contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleHBaseTable.java +++ /dev/null @@ -1,380 +0,0 @@ -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You 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. - */ -package org.archive.modules.recrawl.hbase; - -import java.io.IOException; -import java.util.LinkedHashMap; -import java.util.Map; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantReadWriteLock; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.NotServingRegionException; -import org.apache.hadoop.hbase.TableNotFoundException; -import org.apache.hadoop.hbase.client.Get; -import org.apache.hadoop.hbase.client.HTable; -import org.apache.hadoop.hbase.client.HTableInterface; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Result; -import org.apache.hadoop.hbase.util.Bytes; - -/** - * simple HTable wrapper that shares single instance of HTable among threads. - * If you only perform get on HTable, this implementation - * should be good enough. If multiple threads performs Put, {@link HBaseTable} - * would be more efficient. - *

when HBase I/O fails due to issue with network/region server/zookeeper, this - * class waits for preset time (see {@link #setReconnectInterval(int)}) - * before trying to reestablish HBase connection. During this hold-ff period, all - * {@link #get(Get)} and {@link #put(Put)} calls will fail. - * - * @author kenji - */ -public class SingleHBaseTable extends HBaseTableBean { - private static final Log LOG = LogFactory.getLog(SingleHBaseTable.class); - - private HTableInterface table; - private volatile long tableError; - private ReentrantReadWriteLock tableUseLock = new ReentrantReadWriteLock(); - - boolean autoReconnect = true; - - public boolean isAutoReconnect() { - return autoReconnect; - } - /** - * if set to {@code true}, HBaseClient tries to reconnect to the HBase master - * immediately when Put request failed due to connection loss (note {@link #put(Put)} - * still throws IOException even if autoReconnect is enabled.) - * @param autoReconnect true to enable auto-reconnect - */ - public void setAutoReconnect(boolean autoReconnect) { - this.autoReconnect = autoReconnect; - } - - protected boolean autoFlush = true; - /** - * passed on to HTable's autoFlush property upon creation. - * @return true for enabling auto-flush. - */ - public boolean isAutoFlush() { - return autoFlush; - } - public void setAutoFlush(boolean autoFlush) { - this.autoFlush = autoFlush; - } - - // default 3 minutes - private int reconnectInterval = 1000 * 3 * 60; - - public int getReconnectInterval() { - return reconnectInterval; - } - /** - * set hold-off interval upon communication errors. - * @param reconnectInterval hold-off interval in milliseconds. - */ - public void setReconnectInterval(int reconnectInterval) { - this.reconnectInterval = reconnectInterval; - } - - // counters - - protected AtomicLong getCount = new AtomicLong(); - // count of GET/PUT failures (i.e. not counting connection failures). - protected AtomicLong getErrorCount = new AtomicLong(); - protected AtomicLong getSkipCount = new AtomicLong(); - - protected AtomicLong putCount = new AtomicLong(); - protected AtomicLong putErrorCount = new AtomicLong(); - protected AtomicLong putSkipCount = new AtomicLong(); - - protected AtomicLong connectCount = new AtomicLong(); - - public long getGetCount() { return getCount.get(); } - public long getGetErrorCount() { return getErrorCount.get(); } - public long getGetSkipCount() { return getSkipCount.get(); } - public long getPutCount() { return putCount.get(); } - public long getConnectCount() { return connectCount.get(); } - - // for diagnosing deadlock situation - public Map getTableLockState() { - Map m = new LinkedHashMap(); - m.put("readLockCount", tableUseLock.getReadLockCount()); - m.put("queueLength", tableUseLock.getQueueLength()); - m.put("writeLocked", tableUseLock.isWriteLocked()); - return m; - } - - public SingleHBaseTable() { - } - - /** - * attempts to reconnect to HBase if table is null. - * must not be called with read-lock. - * @return existing or newly opened HTableInterface. - */ - protected HTableInterface getTable() { - if (table == null && autoReconnect) - openTable(); - return table; - } - /** - * close HTable {@code table}, set current time to tableError if closing because - * of a communication error. should be called with write lock. - * @param htable HTable to close. - * @param byError true if closing because of an error. - */ - protected void closeTable(HTableInterface htable, boolean byError) { - if (htable == null) return; - if (table != htable) { - // other thread did closeTable on htable. don't close table. - return; - } - try { - table = null; - htable.close(); - } catch (IOException ex) { - LOG.warn("error closing " + htable + " - some commits may have been lost"); - } - if (byError) { - tableError = System.currentTimeMillis(); - } - } - - public void put(Put p) throws IOException { - putCount.incrementAndGet(); - // trigger reconnection if necessary. as table can be modified before - // read lock is acquired, we don't read table variable here. - getTable(); - boolean htableFailed = false; - HTableInterface htable = null; - Lock readLock = tableUseLock.readLock(); - try { - if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) { - putSkipCount.incrementAndGet(); - throw new IOException("could not acquire read lock for HTable."); - } - } catch (InterruptedException ex) { - throw new IOException("interrupted while acquiring read lock", ex); - } - try { - htable = table; - if (htable == null) { - putSkipCount.incrementAndGet(); - throw new IOException("HBase connection is unvailable."); - } - // HTable.put() buffers Puts and access to the buffer is not - // synchronized. - synchronized (htable) { - try { - htable.put(p); - } catch (NullPointerException ex) { - // HTable.put() throws NullPointerException when connection is lost. - // It is somewhat weird, so translate it to IOException. - putErrorCount.incrementAndGet(); - htableFailed = true; - throw new IOException("hbase connection is lost", ex); - } catch (NotServingRegionException ex) { - putErrorCount.incrementAndGet(); - // no need to close HTable. - throw ex; - } catch (IOException ex) { - putErrorCount.incrementAndGet(); - htableFailed = true; - throw ex; - } - } - } finally { - readLock.unlock(); - if (htableFailed) { - closeTable(htable, true); - } - } - } - - public Result get(Get g) throws IOException { - getCount.incrementAndGet(); - // trigger reconnection if necessary. as table can be modified before - // read lock is acquired, we don't read table variable here. - getTable(); - boolean htableFailed = false; - HTableInterface htable = null; - Lock readLock = tableUseLock.readLock(); - try { - if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) { - getSkipCount.incrementAndGet(); - throw new IOException("could not acquire read lock for HTable."); - } - } catch (InterruptedException ex) { - throw new IOException("interrupted while acquiring read lock", ex); - } - try { - htable = table; - if (htable == null) { - getSkipCount.incrementAndGet(); - throw new IOException("HBase connection is unvailable."); - } - try { - return htable.get(g); - } catch (NotServingRegionException ex) { - // caused by disruption to HBase cluster. no need to - // refresh HBase connection, since connection itself - // is working okay. - // TODO: should we need to back-off for a while? other - // regions may still be accessible. - getErrorCount.incrementAndGet(); - throw ex; - } catch (IOException ex) { - getErrorCount.incrementAndGet(); - htableFailed = true; - throw ex; - } - } finally { - readLock.unlock(); - if (htableFailed) { - closeTable(htable, true); - } - } - } - - @Override - public HTableDescriptor getHtableDescriptor() throws IOException { - HTableInterface table = getTable(); - if (table == null) { - throw new IOException("HBase connection is unavailable."); - } - return table.getTableDescriptor(); - } - - public boolean inBackoffPeriod() { - return (tableError > 0 && - (System.currentTimeMillis() - tableError) < reconnectInterval); - } - - /** - * timestamp of the last Put/Get error. - * @return timestamp in ms. - */ - public long getTableErrorTime() { - return tableError; - } - /** - * connect to HBase. - * it does nothing if table is non-null, or it is in the back-off period since - * the last error. - * should be called with write lock. - */ - protected boolean openTable() { - if (table != null) return true; - // fail immediately if we're in back-off period. - if (inBackoffPeriod()) return false; - try { - HTable t = new HTable(hbase.configuration(), Bytes.toBytes(htableName)); - connectCount.incrementAndGet(); - t.setAutoFlush(autoFlush); - table = t; - tableError = 0; - return true; - } catch (TableNotFoundException ex) { - // ex.getMessage() only has table name. be a little bit more friendly. - LOG.warn("failed to connect to HTable \"" + htableName + "\": Table Not Found"); - tableError = System.currentTimeMillis(); - return false; - } catch (IOException ex) { - LOG.warn("failed to connect to HTable \"" + htableName + "\" (" + ex.getMessage() + ")"); - tableError = System.currentTimeMillis(); - return false; - } - } - /** - * number of seconds to wait for acquiring read lock. - * if read lock is not acquired within this many seconds (probably - * due to deadlock situation on write-lock side), {@link #get(Get)} will - * silently fail. - */ - public final static long TRY_READ_LOCK_TIMEOUT = 5; - /** - * number of seconds to wait for acquiring write lock. - */ - public final static long TRY_WRITE_LOCK_TIMEOUT = 10; - - /** - * close current connection and establish new connection. - * fails silently if back-off period is in effect. - */ - protected void reconnect(boolean onerror) throws IOException, InterruptedException { - // avoid deadlock situation caused by attempting - // to acquire write lock while holding read lock. - // there'd be no real dead-lock now that timeout on write lock is implemented, - // but it's nice to know there's a bug in locking. - if (tableUseLock.getReadHoldCount() > 0) { - LOG.warn("avoiding deadlock: reconnect() called by thread with read lock."); - return; - } - Lock writeLock = tableUseLock.writeLock(); - if (!writeLock.tryLock(TRY_WRITE_LOCK_TIMEOUT, TimeUnit.SECONDS)) { - LOG.warn("reconnect() could not acquire write lock on tableUseLock for " + - TRY_WRITE_LOCK_TIMEOUT + "s, giving up."); - return; - } - try { - closeTable(table, onerror); - openTable(); - } finally { - writeLock.unlock(); - } - } - - /** - * close current connection and establish new connection. - * for refreshing stale connection through scripting. - * resets tableErrorTime to zero (it will be set to non-zero if - * reconnection attempt fails). - * @throws IOException - * @throws InterruptedException - */ - public void reconnect() throws IOException, InterruptedException { - tableError = 0; - reconnect(false); - } - -// public boolean isRunning() { -// return table != null; -// } - public void start() { - super.start(); - openTable(); - } - public void stop() { - if (table != null) { - try { - table.close(); - } catch (IOException ex) { - LOG.warn("table.close() failed", ex); - } - } - table = null; - super.stop(); - } -} diff --git a/engine/src/main/java/org/archive/crawler/restlet/JobResource.java b/engine/src/main/java/org/archive/crawler/restlet/JobResource.java index 5b4ab6c7..27a92aa5 100644 --- a/engine/src/main/java/org/archive/crawler/restlet/JobResource.java +++ b/engine/src/main/java/org/archive/crawler/restlet/JobResource.java @@ -224,7 +224,12 @@ public class JobResource extends BaseResource { } else if ("pause".equals(action)) { cj.getCrawlController().requestCrawlPause(); } else if ("unpause".equals(action)) { - cj.getCrawlController().requestCrawlResume(); + try { + cj.getCrawlController().requestCrawlResume(); + } catch (Exception e){ + System.err.println(getName() + ": exception " + e + " during unpause."); + e.printStackTrace(); + } } else if ("checkpoint".equals(action)) { String cp = cj.getCheckpointService().requestCrawlCheckpoint(); if (StringUtils.isNotEmpty(cp)) { diff --git a/engine/src/main/java/org/archive/crawler/restlet/PagedRepresentation.java b/engine/src/main/java/org/archive/crawler/restlet/PagedRepresentation.java index aabd0d46..b032f9c1 100644 --- a/engine/src/main/java/org/archive/crawler/restlet/PagedRepresentation.java +++ b/engine/src/main/java/org/archive/crawler/restlet/PagedRepresentation.java @@ -144,7 +144,6 @@ public class PagedRepresentation extends CharacterRepresentation { pw.println(""); emitControls(pw); - pw.close(); } /** diff --git a/engine/src/main/java/org/archive/crawler/restlet/RateLimitGuard.java b/engine/src/main/java/org/archive/crawler/restlet/RateLimitGuard.java index 7d9cd703..bde86bfd 100644 --- a/engine/src/main/java/org/archive/crawler/restlet/RateLimitGuard.java +++ b/engine/src/main/java/org/archive/crawler/restlet/RateLimitGuard.java @@ -53,6 +53,7 @@ public class RateLimitGuard extends DigestAuthenticator { long now = System.currentTimeMillis(); long sleepMs = (lastFailureTime+MIN_MS_BETWEEN_ATTEMPTS)-now; if(sleepMs>0) { + System.out.println(new java.util.Date() + " " + getName() + ": trying to sleep"); try { Thread.sleep(sleepMs); } catch (InterruptedException e) { diff --git a/engine/src/test/java/org/archive/crawler/util/BloomUriUniqFilterTest.java b/engine/src/test/java/org/archive/crawler/util/BloomUriUniqFilterTest.java index 28390704..81e3c1a5 100644 --- a/engine/src/test/java/org/archive/crawler/util/BloomUriUniqFilterTest.java +++ b/engine/src/test/java/org/archive/crawler/util/BloomUriUniqFilterTest.java @@ -67,7 +67,7 @@ implements UriUniqFilter.CrawlUriReceiver { this.filter.addForce(this.getUri(), new CrawlURI(UURIFactory.getInstance(this.getUri()))); // Should only have add 'this' once. - assertTrue("Count is off", this.filter.count() == 1); + assertEquals("Count is off", 1, this.filter.count()); } /** @@ -104,8 +104,7 @@ implements UriUniqFilter.CrawlUriReceiver { logger.fine("Readded subset " + list.size() + " in " + (System.currentTimeMillis() - start)); - assertTrue("Count is off: " + filter.count(), - filter.count() == MAX_COUNT); + assertEquals("Count is off", MAX_COUNT, filter.count()); } public void testNote() { diff --git a/modules/pom.xml b/modules/pom.xml index ff3816ae..1c7759b1 100644 --- a/modules/pom.xml +++ b/modules/pom.xml @@ -1,5 +1,6 @@ - + org.archive heritrix diff --git a/modules/src/main/java/org/archive/modules/CrawlURI.java b/modules/src/main/java/org/archive/modules/CrawlURI.java index 82863698..8391be3c 100644 --- a/modules/src/main/java/org/archive/modules/CrawlURI.java +++ b/modules/src/main/java/org/archive/modules/CrawlURI.java @@ -859,6 +859,7 @@ implements Reporter, Serializable, OverlayContext, Comparable { this.httpRecorder = null; this.fetchStatus = S_UNATTEMPTED; this.setPrerequisite(false); + this.clearPrerequisiteUri(); this.contentSize = UNCALCULATED; this.contentLength = UNCALCULATED; // Clear 'links extracted' flag. diff --git a/modules/src/main/java/org/archive/modules/extractor/ExtractorHTML.java b/modules/src/main/java/org/archive/modules/extractor/ExtractorHTML.java index ebcaf47a..b26eed3f 100644 --- a/modules/src/main/java/org/archive/modules/extractor/ExtractorHTML.java +++ b/modules/src/main/java/org/archive/modules/extractor/ExtractorHTML.java @@ -981,7 +981,8 @@ public class ExtractorHTML extends ContentExtractor implements InitializingBean } else if ("refresh".equalsIgnoreCase(httpEquiv) && content != null) { int urlIndex = content.indexOf("=") + 1; if(urlIndex>0) { - String refreshUri = content.substring(urlIndex); + // strip any quotes ("') characters from the URL value. + String refreshUri = TextUtils.replaceAll("[\"']", content.substring(urlIndex), ""); try { int max = getExtractorParameters().getMaxOutlinks(); addRelativeToBase(curi, max, refreshUri, diff --git a/modules/src/main/java/org/archive/modules/extractor/ExtractorMultipleRegex.java b/modules/src/main/java/org/archive/modules/extractor/ExtractorMultipleRegex.java index e3afa0fb..536ff7b2 100644 --- a/modules/src/main/java/org/archive/modules/extractor/ExtractorMultipleRegex.java +++ b/modules/src/main/java/org/archive/modules/extractor/ExtractorMultipleRegex.java @@ -192,6 +192,7 @@ public class ExtractorMultipleRegex extends Extractor { while (matcher.find()) { add(new GroupList(matcher)); } + TextUtils.recycleMatcher(matcher); } public MatchList(GroupList... groupList) { for (GroupList x: groupList) { @@ -219,6 +220,7 @@ public class ExtractorMultipleRegex extends Extractor { matchLists = new LinkedHashMap(); matchLists.put("uriRegex", new MatchList(new GroupList(matcher))); } else { + TextUtils.recycleMatcher(matcher); return; // if uri regex doesn't match, we're done } @@ -229,6 +231,7 @@ public class ExtractorMultipleRegex extends Extractor { curi.getNonFatalFailures().add(e); LOGGER.log(Level.WARNING, "Failed get of replay char sequence in " + Thread.currentThread().getName(), e); + TextUtils.recycleMatcher(matcher); return; } @@ -237,6 +240,7 @@ public class ExtractorMultipleRegex extends Extractor { String regex = getContentRegexes().get(regexName); MatchList matchList = new MatchList(regex, cs); if (matchList.isEmpty()) { + TextUtils.recycleMatcher(matcher); return; // no match found for regex, so we can stop now } matchLists.put(regexName, matchList); @@ -257,6 +261,7 @@ public class ExtractorMultipleRegex extends Extractor { Map bindings = makeBindings(matchLists, regexNames, i); buildAndAddOutlink(curi, bindings); } + TextUtils.recycleMatcher(matcher); } // bindings are the variables available to populate the template diff --git a/modules/src/main/java/org/archive/modules/writer/ARCWriterProcessor.java b/modules/src/main/java/org/archive/modules/writer/ARCWriterProcessor.java index 7a568579..e6944937 100644 --- a/modules/src/main/java/org/archive/modules/writer/ARCWriterProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/ARCWriterProcessor.java @@ -129,8 +129,7 @@ public class ARCWriterProcessor extends WriterPoolProcessor { // We just closed the file because it was larger than maxBytes. // Add to the totalBytesWritten the size of the first record // in the file, if any. - setTotalBytesWritten(getTotalBytesWritten() + - (writer.getPosition() - position)); + addTotalBytesWritten(writer.getPosition() - position); position = writer.getPosition(); } @@ -155,8 +154,7 @@ public class ARCWriterProcessor extends WriterPoolProcessor { throw e; } finally { if (writer != null) { - setTotalBytesWritten(getTotalBytesWritten() + - (writer.getPosition() - position)); + addTotalBytesWritten(writer.getPosition() - position); getPool().returnFile(writer); String filename = writer.getFile().getName(); diff --git a/modules/src/main/java/org/archive/modules/writer/BaseWARCWriterProcessor.java b/modules/src/main/java/org/archive/modules/writer/BaseWARCWriterProcessor.java index e9b2c262..b5178233 100644 --- a/modules/src/main/java/org/archive/modules/writer/BaseWARCWriterProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/BaseWARCWriterProcessor.java @@ -211,7 +211,7 @@ abstract public class BaseWARCWriterProcessor extends WriterPoolProcessor + WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK) + " bytes to " + writer.getFile().getName() + " for " + curi); } - setTotalBytesWritten(getTotalBytesWritten() + (writer.getPosition() - startPosition)); + addTotalBytesWritten(writer.getPosition() - startPosition); curi.addExtraInfo("warcFilename", writer.getFilenameWithoutOccupiedSuffix()); curi.addExtraInfo("warcFileOffset", startPosition); diff --git a/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java b/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java index 4fc45558..9b86057d 100644 --- a/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java @@ -1,12 +1,16 @@ package org.archive.modules.writer; +import java.io.InputStream; import java.io.IOException; import java.net.URI; import java.util.Arrays; import java.util.List; +import java.util.HashMap; +import java.util.Map; import java.util.logging.Level; import java.util.logging.Logger; +import org.apache.commons.io.IOUtils; import org.archive.io.warc.WARCRecordInfo; import org.archive.io.warc.WARCWriter; import org.archive.modules.CrawlURI; @@ -22,6 +26,8 @@ import org.archive.modules.warc.RevisitRecordBuilder; import org.archive.modules.warc.WARCRecordBuilder; import org.archive.modules.warc.WhoisResponseRecordBuilder; import org.archive.spring.HasKeyedProperties; +import org.json.JSONException; +import org.json.JSONObject; /** * WARC writer processor. The types of records that to be written can be @@ -123,8 +129,7 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements // We rolled over to a new warc and wrote a warcinfo record. // Tally stats and reset temp stats, to avoid including warcinfo // record in stats for current url. - setTotalBytesWritten(getTotalBytesWritten() + - (writer.getPosition() - position)); + addTotalBytesWritten(writer.getPosition() - position); addStats(writer.getTmpStats()); writer.resetTmpStats(); writer.resetTmpRecordLog(); @@ -159,6 +164,17 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements WARCRecordInfo record = recordBuilder.buildRecord(curi, concurrentTo); if (record != null) { writer.writeRecord(record); + InputStream is = null; + try { + is = record.getContentStream(); + is.close(); + } + catch (Exception e){ + logger.log(Level.WARNING, "problem closing Warc Record Content Stream " + e); + } + finally { + IOUtils.closeQuietly(record.getContentStream()); //Closing one way or the other seems to leave some file handles open. Calling close() and using closeQuietly() handles both FileStreams and FileChannels + } if (concurrentTo == null) { concurrentTo = record.getRecordId(); } @@ -166,4 +182,42 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements } } } + @Override + protected JSONObject toCheckpointJson() throws JSONException { + JSONObject json = super.toCheckpointJson(); + json.put("urlsWritten", urlsWritten); + json.put("stats", stats); + return json; + } + + @Override + protected void fromCheckpointJson(JSONObject json) throws JSONException { + super.fromCheckpointJson(json); + + // conditionals below are for backward compatibility with old checkpoints + + if (json.has("urlsWritten")) { + urlsWritten.set(json.getLong("urlsWritten")); + } + + if (json.has("stats")) { + HashMap> cpStats = new HashMap>(); + JSONObject jsonStats = json.getJSONObject("stats"); + if (JSONObject.getNames(jsonStats) != null) { + for (String key1: JSONObject.getNames(jsonStats)) { + JSONObject jsonSubstats = jsonStats.getJSONObject(key1); + if (!cpStats.containsKey(key1)) { + cpStats.put(key1, new HashMap()); + } + Map substats = cpStats.get(key1); + + for (String key2: JSONObject.getNames(jsonSubstats)) { + long value = jsonSubstats.getLong(key2); + substats.put(key2, value); + } + } + addStats(cpStats); + } + } + } } diff --git a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java index 4726a007..5585fc56 100644 --- a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java @@ -166,8 +166,7 @@ public class WARCWriterProcessor extends BaseWARCWriterProcessor implements WARC // We rolled over to a new warc and wrote a warcinfo record. // Tally stats and reset temp stats, to avoid including warcinfo // record in stats for current url. - setTotalBytesWritten(getTotalBytesWritten() + - (writer.getPosition() - position)); + addTotalBytesWritten(writer.getPosition() - position); addStats(writer.getTmpStats()); writer.resetTmpStats(); writer.resetTmpRecordLog(); @@ -647,6 +646,7 @@ public class WARCWriterProcessor extends BaseWARCWriterProcessor implements WARC protected JSONObject toCheckpointJson() throws JSONException { JSONObject json = super.toCheckpointJson(); json.put("urlsWritten", urlsWritten); + json.put("totalBytesWritten", getTotalBytesWritten()); json.put("stats", stats); return json; } @@ -660,6 +660,9 @@ public class WARCWriterProcessor extends BaseWARCWriterProcessor implements WARC if (json.has("urlsWritten")) { urlsWritten.set(json.getLong("urlsWritten")); } + if (json.has("totalBytesWritten")) { + setTotalBytesWritten(json.getLong("totalBytesWritten")); + } if (json.has("stats")) { HashMap> cpStats = new HashMap>(); diff --git a/modules/src/main/java/org/archive/modules/writer/WriterPoolProcessor.java b/modules/src/main/java/org/archive/modules/writer/WriterPoolProcessor.java index a5a030e8..b066bbdd 100644 --- a/modules/src/main/java/org/archive/modules/writer/WriterPoolProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WriterPoolProcessor.java @@ -31,6 +31,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Logger; import org.archive.checkpointing.Checkpoint; @@ -42,6 +43,7 @@ import org.archive.modules.CrawlMetadata; import org.archive.modules.CrawlURI; import org.archive.modules.ProcessResult; import org.archive.modules.Processor; +import org.archive.modules.deciderules.DecideResult; import org.archive.modules.deciderules.recrawl.IdenticalDigestDecideRule; import org.archive.modules.net.CrawlHost; import org.archive.modules.net.ServerCache; @@ -273,7 +275,7 @@ implements Lifecycle, Checkpointable, WriterPoolSettings { /** * Total number of bytes written to disc. */ - private long totalBytesWritten = 0; + private AtomicLong totalBytesWritten = new AtomicLong(); private AtomicInteger serial = new AtomicInteger(); @@ -315,7 +317,7 @@ implements Lifecycle, Checkpointable, WriterPoolSettings { if (max <= 0) { return ProcessResult.PROCEED; } - if (max <= this.totalBytesWritten) { + if (max <= getTotalBytesWritten()) { return ProcessResult.FINISH; // FIXME: Specify reason // controller.requestCrawlStop(CrawlStatus.FINISHED_WRITE_LIMIT); } @@ -362,6 +364,12 @@ implements Lifecycle, Checkpointable, WriterPoolSettings { return false; } + if (getShouldProcessRule().decisionFor(curi) == DecideResult.REJECT) { + curi.getAnnotations().add(ANNOTATION_UNWRITTEN + ":rejected(" + + getShouldProcessRule().getClass() + ")"); + return false; + } + return true; } @@ -435,11 +443,14 @@ implements Lifecycle, Checkpointable, WriterPoolSettings { } protected long getTotalBytesWritten() { - return totalBytesWritten; + return totalBytesWritten.get(); } protected void setTotalBytesWritten(long totalBytesWritten) { - this.totalBytesWritten = totalBytesWritten; + this.totalBytesWritten.set(totalBytesWritten); + } + protected void addTotalBytesWritten(long bytesWritten) { + this.totalBytesWritten.addAndGet(bytesWritten); } public abstract List getMetadata();