From f01ad9ada813ea26c3604b8fbbcdd4170624caa6 Mon Sep 17 00:00:00 2001 From: gojomo Date: Thu, 19 Nov 2009 23:18:15 +0000 Subject: [PATCH] [HER-1546] Springify(5): Update checkpointing to work smoothly with spring-configured crawls 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 --- .../org/archive/io/GenerationFileHandler.java | 1 + .../archive/util/ObjectIdentityMemCache.java | 9 ++ .../crawler/reporting/CrawlStatSnapshot.java | 16 +++ .../reporting/CrawlerLoggerModule.java | 21 ++-- .../crawler/reporting/StatisticsTracker.java | 114 ++++++++++++++++-- .../org/archive/crawler/util/TopNSet.java | 9 +- 6 files changed, 152 insertions(+), 18 deletions(-) diff --git a/commons/src/main/java/org/archive/io/GenerationFileHandler.java b/commons/src/main/java/org/archive/io/GenerationFileHandler.java index a8cb5ac7..bbb6fb26 100644 --- a/commons/src/main/java/org/archive/io/GenerationFileHandler.java +++ b/commons/src/main/java/org/archive/io/GenerationFileHandler.java @@ -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); diff --git a/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java b/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java index c6fc329e..5d5445a8 100644 --- a/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java +++ b/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java @@ -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 { public void sync() { // do nothing } + + /** + * Offer raw map access for convenience of checkpoint/recovery. + * @return Map + */ + public Map getMap() { + return map; + } } diff --git a/engine/src/main/java/org/archive/crawler/reporting/CrawlStatSnapshot.java b/engine/src/main/java/org/archive/crawler/reporting/CrawlStatSnapshot.java index 736a3e72..d65373df 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/CrawlStatSnapshot.java +++ b/engine/src/main/java/org/archive/crawler/reporting/CrawlStatSnapshot.java @@ -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); + } } diff --git a/engine/src/main/java/org/archive/crawler/reporting/CrawlerLoggerModule.java b/engine/src/main/java/org/archive/crawler/reporting/CrawlerLoggerModule.java index 7b72686f..8e75f356 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/CrawlerLoggerModule.java +++ b/engine/src/main/java/org/archive/crawler/reporting/CrawlerLoggerModule.java @@ -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. * *

Default access only to be called by Checkpointer. * @throws Exception */ - public void checkpoint(File checkpointDir, List 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; } diff --git a/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java b/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java index bec46bb9..2665a95a 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java +++ b/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java @@ -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; + } + } diff --git a/engine/src/main/java/org/archive/crawler/util/TopNSet.java b/engine/src/main/java/org/archive/crawler/util/TopNSet.java index 01040298..f9e1c4bd 100644 --- a/engine/src/main/java/org/archive/crawler/util/TopNSet.java +++ b/engine/src/main/java/org/archive/crawler/util/TopNSet.java @@ -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 + */ + public HashMap getTopSet() { + return set; } }