lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-23 06:05:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

[HER-1546] Springify(5): Update checkpointing to work smoothly with spring-configured crawls

Browse Source 
stats & logging checkpoint support
* StatisticsTracker.java
    implement Checkpointable; save small state in JSON, recycle BDB data when recovering
* CrawlerLoggerModule.java
    implement Checkpointable; rotate logs on checkpoint
* GenerationFileHandler.java
    avoid clobbering leftover logs from earlier futures via moveAside
* CrawlStatSnapshot.java
    (sameProgressAs) test for lack-of-progress
* ObjectIdentityMemCache.java, TopNSet.java
    offer deeper access to assist checkpoint/resume
This commit is contained in:
gojomo
2009-11-19 23:18:15 +00:00
parent b63c2bdc77
commit f01ad9ada8
6 changed files with 152 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/main/java/org/archive/io/GenerationFileHandler.java
+1
View File
@@ -100,6 +100,7 @@ public class GenerationFileHandler extends FileHandler {
storeSuffix;
File activeFile = new File(filename);
File storeFile = new File(storeFilename);
FileUtils.moveAsideIfExists(storeFile);
if (!activeFile.renameTo(storeFile)) {
throw new IOException("Unable to move " + filename + " to " +
storeFilename);
commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java
+9
View File
@@ -19,6 +19,7 @@
package org.archive.util;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -72,4 +73,12 @@ implements ObjectIdentityCache<String,V> {
public void sync() {
// do nothing
}
/**
* Offer raw map access for convenience of checkpoint/recovery.
* @return Map<String, V>
*/
public Map<String, V> getMap() {
return map;
}
}
engine/src/main/java/org/archive/crawler/reporting/CrawlStatSnapshot.java
+16
View File
@@ -152,4 +152,20 @@ public class CrawlStatSnapshot {
}
return (int) (100 * finishedUriCount / total);
}
/**
* Return true if this snapshot shows no tangible progress in
* its URI counts over the supplied snapshot. May be used to
* suppress unnecessary redundant reporting/checkpointing.
* @param lastSnapshot
* @return true if this snapshot stats are essentially same as previous given
*/
public boolean sameProgressAs(CrawlStatSnapshot lastSnapshot) {
if(lastSnapshot==null) {
return false;
}
return (finishedUriCount == lastSnapshot.finishedUriCount)
&& (queuedUriCount == lastSnapshot.queuedUriCount)
&& (downloadDisregards == lastSnapshot.downloadDisregards);
}
}
engine/src/main/java/org/archive/crawler/reporting/CrawlerLoggerModule.java
+14 -7
View File
@@ -23,7 +23,6 @@ import java.io.File;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.FileHandler;
import java.util.logging.Formatter;
@@ -33,7 +32,7 @@ import java.util.logging.SimpleFormatter;
import org.apache.commons.httpclient.URIException;
import org.archive.checkpointing.Checkpointable;
import org.archive.checkpointing.RecoverAction;
import org.archive.crawler.framework.Checkpoint;
import org.archive.crawler.framework.Engine;
import org.archive.crawler.io.NonFatalErrorFormatter;
import org.archive.crawler.io.RuntimeErrorFormatter;
@@ -48,6 +47,7 @@ import org.archive.net.UURIFactory;
import org.archive.spring.ConfigPath;
import org.archive.util.ArchiveUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.Lifecycle;
/**
@@ -342,20 +342,27 @@ public class CrawlerLoggerModule
manifest.append(type + (bundle? "+": "-") + " " + file + "\n");
}
public void startCheckpoint(Checkpoint checkpointInProgress) {}
/**
* Run checkpointing.
*
* <p>Default access only to be called by Checkpointer.
* @throws Exception
*/
public void checkpoint(File checkpointDir, List<RecoverAction> actions)
throws IOException {
public void doCheckpoint(Checkpoint checkpointInProgress) throws IOException {
// Rotate off crawler logs.
rotateLogFiles("." + checkpointDir.getName());
// this.checkpointer.getNextCheckpointName());
rotateLogFiles("." + checkpointInProgress.getShortName());
}
public void finishCheckpoint(Checkpoint checkpointInProgress) {}
Checkpoint recoveryCheckpoint;
@Autowired(required=false)
public void setRecoveryCheckpoint(Checkpoint checkpoint) {
this.recoveryCheckpoint = checkpoint;
}
public Logger getNonfatalErrors() {
return nonfatalErrors;
}
engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java
+106 -8
View File
@@ -25,7 +25,6 @@ import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.Serializable;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
@@ -41,9 +40,11 @@ import java.util.logging.Logger;
import org.archive.bdb.BdbModule;
import org.archive.bdb.TempStoredSortedMap;
import org.archive.checkpointing.Checkpointable;
import org.archive.crawler.event.CrawlStateEvent;
import org.archive.crawler.event.CrawlURIDispositionEvent;
import org.archive.crawler.event.StatSnapshotEvent;
import org.archive.crawler.framework.Checkpoint;
import org.archive.crawler.framework.CrawlController;
import org.archive.crawler.framework.Engine;
import org.archive.crawler.util.CrawledBytesHistotable;
@@ -54,12 +55,16 @@ import org.archive.modules.seeds.SeedListener;
import org.archive.modules.seeds.SeedModule;
import org.archive.spring.ConfigPath;
import org.archive.util.ArchiveUtils;
import org.archive.util.JSONUtils;
import org.archive.util.MimetypeUtils;
import org.archive.util.ObjectIdentityCache;
import org.archive.util.ObjectIdentityMemCache;
import org.archive.util.PaddingStringBuffer;
import org.archive.util.Supplier;
import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
@@ -127,7 +132,8 @@ public class StatisticsTracker
SeedListener,
Lifecycle,
Runnable,
Serializable {
Checkpointable,
BeanNameAware {
private static final long serialVersionUID = 5L;
protected SeedModule seeds;
@@ -351,25 +357,63 @@ public class StatisticsTracker
dumpReports();
}
@SuppressWarnings("unchecked")
public void start() {
isRunning = true;
boolean isRecover = (recoveryCheckpoint != null);
try {
this.sourceHostDistribution = bdb.getObjectCache("sourceHostDistribution",
false, ConcurrentMap.class);
isRecover, ConcurrentMap.class);
this.hostsDistribution = bdb.getObjectCache("hostsDistribution",
false, AtomicLong.class);
this.hostsBytes = bdb.getObjectCache("hostsBytes", false,
AtomicLong.class);
isRecover, AtomicLong.class);
this.hostsBytes = bdb.getObjectCache("hostsBytes",
isRecover, AtomicLong.class);
this.hostsLastFinished = bdb.getObjectCache("hostsLastFinished",
false, AtomicLong.class);
isRecover, AtomicLong.class);
this.processedSeedsRecords = bdb.getObjectCache("processedSeedsRecords",
false, SeedRecord.class);
isRecover, SeedRecord.class);
this.hostsDistributionTop = new TopNSet(getLiveHostReportSize());
this.hostsBytesTop = new TopNSet(getLiveHostReportSize());
this.hostsLastFinishedTop = new TopNSet(getLiveHostReportSize());
if(isRecover) {
JSONObject json = recoveryCheckpoint.loadJson(beanName);
crawlStartTime = json.getLong("crawlStartTime");
crawlEndTime = json.getLong("crawlEndTime");
crawlTotalPausedTime = json.getLong("crawlTotalPausedTime");
crawlPauseStarted = json.getLong("crawlPauseStarted");
tallyCurrentPause();
JSONUtils.putAllLongs(
hostsDistributionTop.getTopSet(),
json.getJSONObject("hostsDistributionTop"));
JSONUtils.putAllLongs(
hostsBytesTop.getTopSet(),
json.getJSONObject("hostsBytesTop"));
JSONUtils.putAllLongs(
hostsLastFinishedTop.getTopSet(),
json.getJSONObject("hostsLastFinishedTop"));
JSONUtils.putAllAtomicLongs(
((ObjectIdentityMemCache)mimeTypeDistribution).getMap(),
json.getJSONObject("mimeTypeDistribution"));
JSONUtils.putAllAtomicLongs(
((ObjectIdentityMemCache)mimeTypeBytes).getMap(),
json.getJSONObject("mimeTypeBytes"));
JSONUtils.putAllAtomicLongs(
((ObjectIdentityMemCache)statusCodeDistribution).getMap(),
json.getJSONObject("statusCodeDistribution"));
JSONUtils.putAllLongs(
crawledBytes,
json.getJSONObject("crawledBytes"));
}
} catch (DatabaseException e) {
throw new IllegalStateException(e);
} catch (JSONException e) {
throw new IllegalStateException(e);
}
// Log the legend
this.controller.logProgressStatistics(progressStatisticsLegend());
@@ -395,6 +439,13 @@ public class StatisticsTracker
" dl-failures busy-thread mem-use-KB heap-size-KB " +
" congestion max-depth avg-depth";
}
public String getProgressStamp() {
return
progressStatisticsLegend()
+ "\n"
+ getSnapshot().getProgressStatisticsLine();
}
/**
* Notify tracker that crawl has begun. Must be called
@@ -1051,4 +1102,51 @@ public class StatisticsTracker
public void concludedSeedBatch() {
// do nothing;
}
// BeanNameAware
String beanName;
public void setBeanName(String name) {
this.beanName = name;
}
// Checkpointable
// CrawlController's only interest is in knowing that a Checkpoint is
// being recovered
public void startCheckpoint(Checkpoint checkpointInProgress) {}
public void doCheckpoint(Checkpoint checkpointInProgress) throws IOException {
JSONObject json = new JSONObject();
try {
json.put("crawlStartTime",crawlStartTime);
json.put("crawlEndTime",crawlEndTime);
long virtualCrawlPauseStarted = crawlPauseStarted;
if(virtualCrawlPauseStarted<1) {
// TODO: use instant checkpoint started?
virtualCrawlPauseStarted = System.currentTimeMillis();
}
json.put("crawlPauseStarted",virtualCrawlPauseStarted);
json.put("crawlTotalPausedTime",crawlTotalPausedTime);
json.put("hostsDistributionTop", hostsDistributionTop.getTopSet());
json.put("hostsBytesTop", hostsBytesTop.getTopSet());
json.put("hostsLastFinishedTop", hostsLastFinishedTop.getTopSet());
json.put("mimeTypeDistribution", ((ObjectIdentityMemCache)mimeTypeDistribution).getMap());
json.put("mimeTypeBytes", ((ObjectIdentityMemCache)mimeTypeBytes).getMap());
json.put("statusCodeDistribution", ((ObjectIdentityMemCache)statusCodeDistribution).getMap());
json.put("crawledBytes", crawledBytes);
// TODO: save crawledBytesHistotable
checkpointInProgress.saveJson(beanName, json);
} catch (JSONException e) {
// impossible
throw new RuntimeException(e);
}
}
public void finishCheckpoint(Checkpoint checkpointInProgress) {}
Checkpoint recoveryCheckpoint;
public void setRecoveryCheckpoint(Checkpoint recoveryCheckpoint) {
this.recoveryCheckpoint = recoveryCheckpoint;
}
}
engine/src/main/java/org/archive/crawler/util/TopNSet.java
+6 -3
View File
@@ -88,8 +88,11 @@ public class TopNSet implements Serializable {
}
}
public String[] keySet(){
return set.keySet().toArray(new String[0]);
/**
* Make internal map available (for checkpoint/restore purposes).
* @return HashMap<String,Long>
*/
public HashMap<String,Long> getTopSet() {
return set;
}
}