[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]
[tor-commits] [metrics-web/master] Split up ArchiveReader in two parts.
commit af7f95dcc86328b558f4711a9b8e4b72688df27f
Author: Karsten Loesing <karsten.loesing@xxxxxxx>
Date: Thu Dec 13 13:50:33 2012 +0100
Split up ArchiveReader in two parts.
---
src/org/torproject/ernie/cron/ArchiveReader.java | 173 --------------------
.../ernie/cron/BridgeStatsFileHandler.java | 68 +++++++-
src/org/torproject/ernie/cron/Main.java | 15 ++-
.../cron/RelayDescriptorDatabaseImporter.java | 135 +++++++++++++++-
4 files changed, 207 insertions(+), 184 deletions(-)
diff --git a/src/org/torproject/ernie/cron/ArchiveReader.java b/src/org/torproject/ernie/cron/ArchiveReader.java
deleted file mode 100644
index 9126787..0000000
--- a/src/org/torproject/ernie/cron/ArchiveReader.java
+++ /dev/null
@@ -1,173 +0,0 @@
-/* Copyright 2011, 2012 The Tor Project
- * See LICENSE for licensing information */
-package org.torproject.ernie.cron;
-
-import java.io.File;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.logging.Logger;
-
-import org.apache.commons.codec.DecoderException;
-import org.apache.commons.codec.binary.Hex;
-import org.apache.commons.codec.digest.DigestUtils;
-import org.torproject.descriptor.Descriptor;
-import org.torproject.descriptor.DescriptorFile;
-import org.torproject.descriptor.DescriptorReader;
-import org.torproject.descriptor.DescriptorSourceFactory;
-import org.torproject.descriptor.ExtraInfoDescriptor;
-import org.torproject.descriptor.NetworkStatusEntry;
-import org.torproject.descriptor.RelayNetworkStatusConsensus;
-import org.torproject.descriptor.ServerDescriptor;
-
-/**
- * Read in all files in a given directory and pass buffered readers of
- * them to the relay descriptor parser.
- */
-public class ArchiveReader {
-
- /**
- * Stats file handler that accepts parse results for bridge statistics.
- */
- private BridgeStatsFileHandler bsfh;
-
- /**
- * Relay descriptor database importer that stores relay descriptor
- * contents for later evaluation.
- */
- private RelayDescriptorDatabaseImporter rddi;
-
- /**
- * Logger for this class.
- */
- private Logger logger;
-
- public ArchiveReader(RelayDescriptorDatabaseImporter rddi,
- BridgeStatsFileHandler bsfh, File archivesDirectory,
- File statsDirectory, boolean keepImportHistory) {
-
- if (archivesDirectory == null ||
- statsDirectory == null) {
- throw new IllegalArgumentException();
- }
-
- this.rddi = rddi;
- this.bsfh = bsfh;
-
- this.logger = Logger.getLogger(ArchiveReader.class.getName());
- if (archivesDirectory.exists()) {
- logger.fine("Importing files in directory " + archivesDirectory
- + "/...");
- DescriptorReader reader =
- DescriptorSourceFactory.createDescriptorReader();
- reader.addDirectory(archivesDirectory);
- if (keepImportHistory) {
- reader.setExcludeFiles(new File(statsDirectory,
- "relay-descriptor-history"));
- }
- Iterator<DescriptorFile> descriptorFiles = reader.readDescriptors();
- while (descriptorFiles.hasNext()) {
- DescriptorFile descriptorFile = descriptorFiles.next();
- if (descriptorFile.getDescriptors() != null) {
- for (Descriptor descriptor : descriptorFile.getDescriptors()) {
- if (descriptor instanceof RelayNetworkStatusConsensus) {
- this.addRelayNetworkStatusConsensus(
- (RelayNetworkStatusConsensus) descriptor);
- } else if (descriptor instanceof ServerDescriptor) {
- this.addServerDescriptor((ServerDescriptor) descriptor);
- } else if (descriptor instanceof ExtraInfoDescriptor) {
- this.addExtraInfoDescriptor(
- (ExtraInfoDescriptor) descriptor);
- }
- }
- }
- }
- }
-
- logger.info("Finished importing relay descriptors.");
- }
-
- private void addRelayNetworkStatusConsensus(
- RelayNetworkStatusConsensus consensus) {
- for (NetworkStatusEntry statusEntry :
- consensus.getStatusEntries().values()) {
- this.rddi.addStatusEntry(consensus.getValidAfterMillis(),
- statusEntry.getNickname(),
- statusEntry.getFingerprint().toLowerCase(),
- statusEntry.getDescriptor().toLowerCase(),
- statusEntry.getPublishedMillis(), statusEntry.getAddress(),
- statusEntry.getOrPort(), statusEntry.getDirPort(),
- statusEntry.getFlags(), statusEntry.getVersion(),
- statusEntry.getBandwidth(), statusEntry.getPortList(),
- statusEntry.getStatusEntryBytes());
- try {
- this.bsfh.addHashedRelay(DigestUtils.shaHex(Hex.decodeHex(
- statusEntry.getFingerprint().toCharArray())).toUpperCase());
- } catch (DecoderException e) {
- }
- }
- this.rddi.addConsensus(consensus.getValidAfterMillis(),
- consensus.getRawDescriptorBytes());
- }
-
- private void addServerDescriptor(ServerDescriptor descriptor) {
- this.rddi.addServerDescriptor(descriptor.getServerDescriptorDigest(),
- descriptor.getNickname(), descriptor.getAddress(),
- descriptor.getOrPort(), descriptor.getDirPort(),
- descriptor.getFingerprint(), descriptor.getBandwidthRate(),
- descriptor.getBandwidthBurst(), descriptor.getBandwidthObserved(),
- descriptor.getPlatform(), descriptor.getPublishedMillis(),
- descriptor.getUptime(), descriptor.getExtraInfoDigest(),
- descriptor.getRawDescriptorBytes());
- }
-
- private void addExtraInfoDescriptor(ExtraInfoDescriptor descriptor) {
- if (descriptor.getDirreqV3Reqs() != null) {
- int allUsers = 0;
- Map<String, String> obs = new HashMap<String, String>();
- for (Map.Entry<String, Integer> e :
- descriptor.getDirreqV3Reqs().entrySet()) {
- String country = e.getKey();
- int users = e.getValue() - 4;
- allUsers += users;
- obs.put(country, "" + users);
- }
- obs.put("zy", "" + allUsers);
- this.rddi.addDirReqStats(descriptor.getFingerprint(),
- descriptor.getDirreqStatsEndMillis(),
- descriptor.getDirreqStatsIntervalLength(), obs);
- }
- if (descriptor.getConnBiDirectStatsEndMillis() >= 0L) {
- this.rddi.addConnBiDirect(descriptor.getFingerprint(),
- descriptor.getConnBiDirectStatsEndMillis(),
- descriptor.getConnBiDirectStatsIntervalLength(),
- descriptor.getConnBiDirectBelow(),
- descriptor.getConnBiDirectRead(),
- descriptor.getConnBiDirectWrite(),
- descriptor.getConnBiDirectBoth());
- }
- List<String> bandwidthHistoryLines = new ArrayList<String>();
- if (descriptor.getWriteHistory() != null) {
- bandwidthHistoryLines.add(descriptor.getWriteHistory().getLine());
- }
- if (descriptor.getReadHistory() != null) {
- bandwidthHistoryLines.add(descriptor.getReadHistory().getLine());
- }
- if (descriptor.getDirreqWriteHistory() != null) {
- bandwidthHistoryLines.add(
- descriptor.getDirreqWriteHistory().getLine());
- }
- if (descriptor.getDirreqReadHistory() != null) {
- bandwidthHistoryLines.add(
- descriptor.getDirreqReadHistory().getLine());
- }
- this.rddi.addExtraInfoDescriptor(descriptor.getExtraInfoDigest(),
- descriptor.getNickname(),
- descriptor.getFingerprint().toLowerCase(),
- descriptor.getPublishedMillis(),
- descriptor.getRawDescriptorBytes(), bandwidthHistoryLines);
- }
-}
-
diff --git a/src/org/torproject/ernie/cron/BridgeStatsFileHandler.java b/src/org/torproject/ernie/cron/BridgeStatsFileHandler.java
index 4534d2f..085394c 100644
--- a/src/org/torproject/ernie/cron/BridgeStatsFileHandler.java
+++ b/src/org/torproject/ernie/cron/BridgeStatsFileHandler.java
@@ -29,11 +29,16 @@ import java.util.TreeSet;
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
+import org.apache.commons.codec.digest.DigestUtils;
import org.torproject.descriptor.Descriptor;
import org.torproject.descriptor.DescriptorFile;
import org.torproject.descriptor.DescriptorReader;
import org.torproject.descriptor.DescriptorSourceFactory;
import org.torproject.descriptor.ExtraInfoDescriptor;
+import org.torproject.descriptor.NetworkStatusEntry;
+import org.torproject.descriptor.RelayNetworkStatusConsensus;
import org.torproject.descriptor.ServerDescriptor;
/**
@@ -117,7 +122,11 @@ public class BridgeStatsFileHandler {
private File statsDirectory;
- private boolean keepImportHistory;
+ private boolean keepBridgeDescriptorImportHistory;
+
+ private File archivesDirectory;
+
+ private boolean keepRelayDescriptorImportHistory;
/**
* Initializes this class, including reading in intermediate results
@@ -125,14 +134,21 @@ public class BridgeStatsFileHandler {
* <code>stats/hashed-relay-identities</code>.
*/
public BridgeStatsFileHandler(String connectionURL,
- File bridgesDir, File statsDirectory, boolean keepImportHistory) {
+ File bridgesDir, File statsDirectory,
+ boolean keepBridgeDescriptorImportHistory, File archivesDirectory,
+ boolean keepRelayDescriptorImportHistory) {
- if (bridgesDir == null || statsDirectory == null) {
- throw new IllegalArgumentException();
+ if (bridgesDir == null || statsDirectory == null ||
+ archivesDirectory == null || statsDirectory == null) {
+ throw new IllegalArgumentException();
}
this.bridgesDir = bridgesDir;
this.statsDirectory = statsDirectory;
- this.keepImportHistory = keepImportHistory;
+ this.keepBridgeDescriptorImportHistory =
+ keepBridgeDescriptorImportHistory;
+ this.archivesDirectory = archivesDirectory;
+ this.keepRelayDescriptorImportHistory =
+ keepRelayDescriptorImportHistory;
/* Initialize set of known countries. */
this.countries = new TreeSet<String>();
@@ -356,7 +372,7 @@ public class BridgeStatsFileHandler {
DescriptorReader reader =
DescriptorSourceFactory.createDescriptorReader();
reader.addDirectory(bridgesDir);
- if (keepImportHistory) {
+ if (keepBridgeDescriptorImportHistory) {
reader.setExcludeFiles(new File(statsDirectory,
"bridge-descriptor-history"));
}
@@ -424,6 +440,46 @@ public class BridgeStatsFileHandler {
}
}
+ public void importRelayDescriptors() {
+ if (archivesDirectory.exists()) {
+ logger.fine("Importing files in directory " + archivesDirectory
+ + "/...");
+ DescriptorReader reader =
+ DescriptorSourceFactory.createDescriptorReader();
+ reader.addDirectory(archivesDirectory);
+ if (keepRelayDescriptorImportHistory) {
+ reader.setExcludeFiles(new File(statsDirectory,
+ "relay-descriptor-history"));
+ }
+ Iterator<DescriptorFile> descriptorFiles = reader.readDescriptors();
+ while (descriptorFiles.hasNext()) {
+ DescriptorFile descriptorFile = descriptorFiles.next();
+ if (descriptorFile.getDescriptors() != null) {
+ for (Descriptor descriptor : descriptorFile.getDescriptors()) {
+ if (descriptor instanceof RelayNetworkStatusConsensus) {
+ this.addRelayNetworkStatusConsensus(
+ (RelayNetworkStatusConsensus) descriptor);
+ }
+ }
+ }
+ }
+ }
+
+ logger.info("Finished importing relay descriptors.");
+ }
+
+ private void addRelayNetworkStatusConsensus(
+ RelayNetworkStatusConsensus consensus) {
+ for (NetworkStatusEntry statusEntry :
+ consensus.getStatusEntries().values()) {
+ try {
+ this.addHashedRelay(DigestUtils.shaHex(Hex.decodeHex(
+ statusEntry.getFingerprint().toCharArray())).toUpperCase());
+ } catch (DecoderException e) {
+ }
+ }
+ }
+
/**
* Writes the list of hashed relay identities and bridge user numbers as
* observed by single bridges to disk, aggregates per-day statistics for
diff --git a/src/org/torproject/ernie/cron/Main.java b/src/org/torproject/ernie/cron/Main.java
index 457433f..a3f38e6 100644
--- a/src/org/torproject/ernie/cron/Main.java
+++ b/src/org/torproject/ernie/cron/Main.java
@@ -37,7 +37,9 @@ public class Main {
new BridgeStatsFileHandler(
config.getRelayDescriptorDatabaseJDBC(),
new File(config.getSanitizedBridgesDirectory()),
- statsDirectory, config.getKeepSanitizedBridgesImportHistory()) :
+ statsDirectory, config.getKeepSanitizedBridgesImportHistory(),
+ new File(config.getDirectoryArchivesDirectory()),
+ config.getKeepDirectoryArchiveImportHistory()) :
null;
// Import relay descriptors
@@ -49,11 +51,16 @@ public class Main {
config.getWriteRelayDescriptorDatabase() ?
config.getRelayDescriptorDatabaseJDBC() : null,
config.getWriteRelayDescriptorsRawFiles() ?
- config.getRelayDescriptorRawFilesDirectory() : null) : null;
- new ArchiveReader(rddi, bsfh,
+ config.getRelayDescriptorRawFilesDirectory() : null,
new File(config.getDirectoryArchivesDirectory()),
statsDirectory,
- config.getKeepDirectoryArchiveImportHistory());
+ config.getKeepDirectoryArchiveImportHistory()) : null;
+ if (rddi != null) {
+ rddi.importRelayDescriptors();
+ }
+ if (bsfh != null) {
+ bsfh.importRelayDescriptors();
+ }
rddi.closeConnection();
}
diff --git a/src/org/torproject/ernie/cron/RelayDescriptorDatabaseImporter.java b/src/org/torproject/ernie/cron/RelayDescriptorDatabaseImporter.java
index 3e7e694..c9d9bc4 100644
--- a/src/org/torproject/ernie/cron/RelayDescriptorDatabaseImporter.java
+++ b/src/org/torproject/ernie/cron/RelayDescriptorDatabaseImporter.java
@@ -16,8 +16,11 @@ import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.ParseException;
import java.text.SimpleDateFormat;
+import java.util.ArrayList;
import java.util.Calendar;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -28,6 +31,14 @@ import java.util.logging.Level;
import java.util.logging.Logger;
import org.postgresql.util.PGbytea;
+import org.torproject.descriptor.Descriptor;
+import org.torproject.descriptor.DescriptorFile;
+import org.torproject.descriptor.DescriptorReader;
+import org.torproject.descriptor.DescriptorSourceFactory;
+import org.torproject.descriptor.ExtraInfoDescriptor;
+import org.torproject.descriptor.NetworkStatusEntry;
+import org.torproject.descriptor.RelayNetworkStatusConsensus;
+import org.torproject.descriptor.ServerDescriptor;
/**
* Parse directory data.
@@ -207,12 +218,25 @@ public final class RelayDescriptorDatabaseImporter {
private boolean importIntoDatabase;
private boolean writeRawImportFiles;
+ private File archivesDirectory;
+ private File statsDirectory;
+ private boolean keepImportHistory;
+
/**
* Initialize database importer by connecting to the database and
* preparing statements.
*/
public RelayDescriptorDatabaseImporter(String connectionURL,
- String rawFilesDirectory) {
+ String rawFilesDirectory, File archivesDirectory,
+ File statsDirectory, boolean keepImportHistory) {
+
+ if (archivesDirectory == null ||
+ statsDirectory == null) {
+ throw new IllegalArgumentException();
+ }
+ this.archivesDirectory = archivesDirectory;
+ this.statsDirectory = statsDirectory;
+ this.keepImportHistory = keepImportHistory;
/* Initialize logger. */
this.logger = Logger.getLogger(
@@ -979,6 +1003,115 @@ public final class RelayDescriptorDatabaseImporter {
}
}
+ public void importRelayDescriptors() {
+ if (archivesDirectory.exists()) {
+ logger.fine("Importing files in directory " + archivesDirectory
+ + "/...");
+ DescriptorReader reader =
+ DescriptorSourceFactory.createDescriptorReader();
+ reader.addDirectory(archivesDirectory);
+ if (keepImportHistory) {
+ reader.setExcludeFiles(new File(statsDirectory,
+ "relay-descriptor-history"));
+ }
+ Iterator<DescriptorFile> descriptorFiles = reader.readDescriptors();
+ while (descriptorFiles.hasNext()) {
+ DescriptorFile descriptorFile = descriptorFiles.next();
+ if (descriptorFile.getDescriptors() != null) {
+ for (Descriptor descriptor : descriptorFile.getDescriptors()) {
+ if (descriptor instanceof RelayNetworkStatusConsensus) {
+ this.addRelayNetworkStatusConsensus(
+ (RelayNetworkStatusConsensus) descriptor);
+ } else if (descriptor instanceof ServerDescriptor) {
+ this.addServerDescriptor((ServerDescriptor) descriptor);
+ } else if (descriptor instanceof ExtraInfoDescriptor) {
+ this.addExtraInfoDescriptor(
+ (ExtraInfoDescriptor) descriptor);
+ }
+ }
+ }
+ }
+ }
+
+ logger.info("Finished importing relay descriptors.");
+ }
+
+ private void addRelayNetworkStatusConsensus(
+ RelayNetworkStatusConsensus consensus) {
+ for (NetworkStatusEntry statusEntry :
+ consensus.getStatusEntries().values()) {
+ this.addStatusEntry(consensus.getValidAfterMillis(),
+ statusEntry.getNickname(),
+ statusEntry.getFingerprint().toLowerCase(),
+ statusEntry.getDescriptor().toLowerCase(),
+ statusEntry.getPublishedMillis(), statusEntry.getAddress(),
+ statusEntry.getOrPort(), statusEntry.getDirPort(),
+ statusEntry.getFlags(), statusEntry.getVersion(),
+ statusEntry.getBandwidth(), statusEntry.getPortList(),
+ statusEntry.getStatusEntryBytes());
+ }
+ this.addConsensus(consensus.getValidAfterMillis(),
+ consensus.getRawDescriptorBytes());
+ }
+
+ private void addServerDescriptor(ServerDescriptor descriptor) {
+ this.addServerDescriptor(descriptor.getServerDescriptorDigest(),
+ descriptor.getNickname(), descriptor.getAddress(),
+ descriptor.getOrPort(), descriptor.getDirPort(),
+ descriptor.getFingerprint(), descriptor.getBandwidthRate(),
+ descriptor.getBandwidthBurst(), descriptor.getBandwidthObserved(),
+ descriptor.getPlatform(), descriptor.getPublishedMillis(),
+ descriptor.getUptime(), descriptor.getExtraInfoDigest(),
+ descriptor.getRawDescriptorBytes());
+ }
+
+ private void addExtraInfoDescriptor(ExtraInfoDescriptor descriptor) {
+ if (descriptor.getDirreqV3Reqs() != null) {
+ int allUsers = 0;
+ Map<String, String> obs = new HashMap<String, String>();
+ for (Map.Entry<String, Integer> e :
+ descriptor.getDirreqV3Reqs().entrySet()) {
+ String country = e.getKey();
+ int users = e.getValue() - 4;
+ allUsers += users;
+ obs.put(country, "" + users);
+ }
+ obs.put("zy", "" + allUsers);
+ this.addDirReqStats(descriptor.getFingerprint(),
+ descriptor.getDirreqStatsEndMillis(),
+ descriptor.getDirreqStatsIntervalLength(), obs);
+ }
+ if (descriptor.getConnBiDirectStatsEndMillis() >= 0L) {
+ this.addConnBiDirect(descriptor.getFingerprint(),
+ descriptor.getConnBiDirectStatsEndMillis(),
+ descriptor.getConnBiDirectStatsIntervalLength(),
+ descriptor.getConnBiDirectBelow(),
+ descriptor.getConnBiDirectRead(),
+ descriptor.getConnBiDirectWrite(),
+ descriptor.getConnBiDirectBoth());
+ }
+ List<String> bandwidthHistoryLines = new ArrayList<String>();
+ if (descriptor.getWriteHistory() != null) {
+ bandwidthHistoryLines.add(descriptor.getWriteHistory().getLine());
+ }
+ if (descriptor.getReadHistory() != null) {
+ bandwidthHistoryLines.add(descriptor.getReadHistory().getLine());
+ }
+ if (descriptor.getDirreqWriteHistory() != null) {
+ bandwidthHistoryLines.add(
+ descriptor.getDirreqWriteHistory().getLine());
+ }
+ if (descriptor.getDirreqReadHistory() != null) {
+ bandwidthHistoryLines.add(
+ descriptor.getDirreqReadHistory().getLine());
+ }
+ this.addExtraInfoDescriptor(descriptor.getExtraInfoDigest(),
+ descriptor.getNickname(),
+ descriptor.getFingerprint().toLowerCase(),
+ descriptor.getPublishedMillis(),
+ descriptor.getRawDescriptorBytes(), bandwidthHistoryLines);
+ }
+
/**
* Close the relay descriptor database connection.
*/
_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits