add warc stats to CrawlURI.getData() dynamic attributes list after writing records; new method WARCWriterProcessor.getStats() to expose cumulative stats

modules/src/main/java/org/archive/modules/CoreAttributeConstants.java

@@ -124,5 +124,6 @@ public interface CoreAttributeConstants {
     
     // arbitrary additions to WARC response record headers
     public static final String A_WARC_RESPONSE_HEADERS = "warc-response-headers";
-
+    
+    public static final String A_WARC_STATS = "warc-stats";
 }

modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java

@@ -37,6 +37,7 @@ import static org.archive.modules.CoreAttributeConstants.A_FTP_CONTROL_CONVERSAT
 import static org.archive.modules.CoreAttributeConstants.A_FTP_FETCH_STATUS;
 import static org.archive.modules.CoreAttributeConstants.A_SOURCE_TAG;
 import static org.archive.modules.CoreAttributeConstants.A_WARC_RESPONSE_HEADERS;
+import static org.archive.modules.CoreAttributeConstants.A_WARC_STATS;
 import static org.archive.modules.CoreAttributeConstants.HEADER_TRUNC;
 import static org.archive.modules.CoreAttributeConstants.LENGTH_TRUNC;
 import static org.archive.modules.CoreAttributeConstants.TIMER_TRUNC;
@@ -104,6 +105,9 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit
         Logger.getLogger(WARCWriterProcessor.class.getName());
 
     private ConcurrentMap<String, ConcurrentMap<String, AtomicLong>> stats = new ConcurrentHashMap<String, ConcurrentMap<String, AtomicLong>>();
+    public ConcurrentMap<String, ConcurrentMap<String, AtomicLong>> getStats() {
+        return stats;
+    }
 
     private AtomicLong urlsWritten = new AtomicLong();
     
@@ -204,28 +208,28 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit
             final CrawlURI curi)
     throws IOException {
         WARCWriter writer = (WARCWriter) getPool().borrowFile();
-      
+
+        // Reset writer temp stats so they reflect only this set of records.
+        writer.resetTmpStats();
+        writer.resetTmpRecordLog();
+
         long position = writer.getPosition();
         try {
-            // See if we need to open a new file because we've exceeded maxBytes.
-            // Call to checkFileSize will open new file if we're at maximum for
-            // current file.
+            // Roll over to new warc file if we've exceeded maxBytes.
             writer.checkSize();
             if (writer.getPosition() != position) {
-                // 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.
+                // 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));
+                addStats(writer.getTmpStats());
+                writer.resetTmpStats();
+                writer.resetTmpRecordLog();
+
                 position = writer.getPosition();
             }
-                       
-            // Reset writer temp stats so they reflect only this set of records.
-            // They'll be added to totals below, in finally block, after records
-            // have been written.
-            writer.resetTmpStats();
-            writer.resetTmpRecordLog();
-            
+
             // Write a request, response, and metadata all in the one
             // 'transaction'.
             final URI baseid = getRecordID();
@@ -256,9 +260,19 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit
                 getPool().returnFile(writer);
             }
         }
+        // XXX this looks wrong, check should happen *before* writing the
+        // record, the way checkBytesWritten() currently works
         return checkBytesWritten();
     }
     
+    protected Map<String, Map<String, Long>> copyStats(Map<String, Map<String, Long>> orig) {
+        Map<String, Map<String, Long>> copy = new HashMap<String, Map<String, Long>>(orig.size());
+        for (String k: orig.keySet()) {
+            copy.put(k, new HashMap<String, Long>(orig.get(k)));
+        }
+        return copy;
+    }
+
     protected void updateMetadataAfterWrite(final CrawlURI curi,
             WARCWriter writer, long startPosition) {
         if (WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.NUM_RECORDS) > 0l) {
@@ -275,6 +289,8 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit
         curi.addExtraInfo("warcFilename", writer.getFilenameWithoutOccupiedSuffix());
         curi.addExtraInfo("warcFileOffset", startPosition);
 
+        curi.getData().put(A_WARC_STATS, copyStats(writer.getTmpStats()));
+
         // history for uri-based dedupe
         Map<String,Object>[] history = curi.getFetchHistory();
         if (history != null && history[0] != null) {