From 345e927d3208301b030ecaae0ff4c4e7c2e64e39 Mon Sep 17 00:00:00 2001 From: Alex Osborne Date: Fri, 25 Oct 2024 14:27:38 +0900 Subject: [PATCH] Remove HBase from contrib The transitive dependencies pulled in via hbase are causing many security alerts for unpatched vulnerabilities in Dependabot, Maven Central and other tools. The newest version of hbase-client these modules seem to compile against still has many vulnerable transitive dependencies and the latest version of HBase appears to have many breaking API changes. These can be re-added if someone volunteers to take over maintaining this code and updates it so that it doesn't pull in vulnerable dependencies. --- contrib/pom.xml | 35 -- .../modules/recrawl/FetchHistoryHelper.java | 32 -- .../archive/modules/recrawl/hbase/HBase.java | 126 ------ .../hbase/HBaseContentDigestHistory.java | 303 -------------- .../hbase/HBasePersistLoadProcessor.java | 88 ---- .../recrawl/hbase/HBasePersistProcessor.java | 32 -- .../hbase/HBasePersistStoreProcessor.java | 133 ------ .../modules/recrawl/hbase/HBaseTable.java | 159 -------- .../modules/recrawl/hbase/HBaseTableBean.java | 78 ---- .../hbase/MultiColumnRecrawlDataSchema.java | 140 ------- .../recrawl/hbase/RecrawlDataSchema.java | 35 -- .../recrawl/hbase/RecrawlDataSchemaBase.java | 123 ------ .../SingleColumnJsonRecrawlDataSchema.java | 174 -------- .../recrawl/hbase/SingleHBaseTable.java | 380 ------------------ 14 files changed, 1838 deletions(-) delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/HBase.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseContentDigestHistory.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistLoadProcessor.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistProcessor.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistStoreProcessor.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTable.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTableBean.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/MultiColumnRecrawlDataSchema.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchema.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchemaBase.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleColumnJsonRecrawlDataSchema.java delete mode 100644 contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleHBaseTable.java diff --git a/contrib/pom.xml b/contrib/pom.xml index 03584ed3..52a20335 100644 --- a/contrib/pom.xml +++ b/contrib/pom.xml @@ -24,41 +24,6 @@ ${project.version} compile - - org.apache.hbase - hbase-client - 0.98.6-cdh5.3.5 - - - jets3t - net.java.dev.jets3t - - - junit - junit - - - - jdk.tools - jdk.tools - - - - com.google.guava - guava - - - commons-io - commons-io - - - commons-collections - commons-collections - - - com.rabbitmq amqp-client diff --git a/contrib/src/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java b/contrib/src/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java index a4807a52..7719c288 100644 --- a/contrib/src/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java +++ b/contrib/src/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java @@ -18,10 +18,6 @@ */ package org.archive.modules.recrawl; -import java.text.DateFormat; -import java.text.ParseException; -import java.text.SimpleDateFormat; -import java.util.Date; import java.util.HashMap; import java.util.Map; @@ -82,35 +78,7 @@ public class FetchHistoryHelper { return null; } - protected static final DateFormat HTTP_DATE_FORMAT = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss zzz"); - protected FetchHistoryHelper() { } - /** - * converts time in HTTP Date format {@code dateStr} to seconds - * since epoch. - * @param dateStr time in HTTP Date format. - * @return seconds since epoch - */ - public static long parseHttpDate(String dateStr) { - synchronized (HTTP_DATE_FORMAT) { - try { - Date d = HTTP_DATE_FORMAT.parse(dateStr); - return d.getTime() / 1000; - } catch (ParseException ex) { - if (logger.isDebugEnabled()) - logger.debug("bad HTTP DATE: " + dateStr); - return 0; - } - } - } - - public static String formatHttpDate(long time) { - synchronized (HTTP_DATE_FORMAT) { - // format(Date) is not thread safe either - return HTTP_DATE_FORMAT.format(new Date(time * 1000)); - } - } - } \ No newline at end of file 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 b72f7d7a..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") || scheme.equals("sftp"))) { - 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"). - *
    - *
  • {@code s}: fetch status (as integer text)
    • - *
  • {@code d}: content digest (with {@code sha1:} prefix, Base32 text)
    • - *
  • {@code e}: ETag (enclosing quotes stripped)
    • - *
  • {@code m}: last-modified date-time (as integer timestamp, binary format)
    • - *
  • {@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.
    • - *
- * - * @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}. - *
    - *
  • configuring single column family name
    • - *
  • formatting/parsing HTTP date text
    • - *
  • constructing row key
    • - *
  • preparing fetch-history array
    • - *
- * @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. - *
    - *
  • {@code r}: re-crawl data in JSON format
    • - *
  • {@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.
    • - *
- * @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(); - } -}