commit 8b45d2fb00af489e4ff9ebb74a892e5f91485844 Author: Karsten Loesing <karsten.loesing@gmx.net> Date: Thu May 24 09:34:26 2012 +0200 Merge Torperf files into new .tpf file format. Implements part of #3036. --- config.template | 12 +- src/org/torproject/ernie/db/Configuration.java | 39 +-- src/org/torproject/ernie/db/Main.java | 3 +- src/org/torproject/ernie/db/TorperfDownloader.java | 484 ++++++++++++++++++-- 4 files changed, 467 insertions(+), 71 deletions(-) diff --git a/config.template b/config.template index 5034e12..aecc67d 100644 --- a/config.template +++ b/config.template @@ -136,11 +136,9 @@ ## times) #TorperfSource torperf http://torperf.torproject.org/ # -## Torperf .data files available on a given source (option can be -## contained multiple times) -#TorperfDataFiles torperf 50kb.data 1mb.data 5mb.data -# -## Torperf .extradata files available on a given source (option can be -## contained multiple times) -#TorperfExtradataFiles torperf 50kb.extradata 1mb.extradata 5mb.extradata +## Torperf measurement file size in bytes, .data file, and .extradata file +## available on a given source (option can be contained multiple times) +#TorperfFiles torperf 51200 50kb.data 50kb.extradata +#TorperfFiles torperf 1048576 1mb.data 1mb.extradata +#TorperfFiles torperf 5242880 5mb.data 5mb.extradata diff --git a/src/org/torproject/ernie/db/Configuration.java b/src/org/torproject/ernie/db/Configuration.java index 6ab9f44..63349c8 100644 --- a/src/org/torproject/ernie/db/Configuration.java +++ b/src/org/torproject/ernie/db/Configuration.java @@ -59,8 +59,7 @@ public class Configuration { private boolean processTorperfFiles = false; private String torperfOutputDirectory = "torperf/"; private SortedMap<String, String> torperfSources = null; - private SortedMap<String, List<String>> torperfDataFiles = null; - private SortedMap<String, List<String>> torperfExtradataFiles = null; + private List<String> torperfFiles = null; private boolean provideFilesViaRsync = false; private String rsyncDirectory = "rsync"; public Configuration() { @@ -194,29 +193,18 @@ public class Configuration { String sourceName = parts[1]; String baseUrl = parts[2]; this.torperfSources.put(sourceName, baseUrl); - } else if (line.startsWith("TorperfDataFiles")) { - if (this.torperfDataFiles == null) { - this.torperfDataFiles = new TreeMap<String, List<String>>(); + } else if (line.startsWith("TorperfFiles")) { + if (this.torperfFiles == null) { + this.torperfFiles = new ArrayList<String>(); } String[] parts = line.split(" "); - String sourceName = parts[1]; - List<String> dataFiles = new ArrayList<String>(); - for (int i = 2; i < parts.length; i++) { - dataFiles.add(parts[i]); - } - this.torperfDataFiles.put(sourceName, dataFiles); - } else if (line.startsWith("TorperfExtradataFiles")) { - if (this.torperfExtradataFiles == null) { - this.torperfExtradataFiles = - new TreeMap<String, List<String>>(); - } - String[] parts = line.split(" "); - String sourceName = parts[1]; - List<String> extradataFiles = new ArrayList<String>(); - for (int i = 2; i < parts.length; i++) { - extradataFiles.add(parts[i]); + if (parts.length != 5) { + logger.severe("Configuration file contains TorperfFiles " + + "option with wrong number of values in line '" + line + + "'! Exiting!"); + System.exit(1); } - this.torperfExtradataFiles.put(sourceName, extradataFiles); + this.torperfFiles.add(line); } else if (line.startsWith("ProvideFilesViaRsync")) { this.provideFilesViaRsync = Integer.parseInt( line.split(" ")[1]) != 0; @@ -381,11 +369,8 @@ public class Configuration { public SortedMap<String, String> getTorperfSources() { return this.torperfSources; } - public SortedMap<String, List<String>> getTorperfDataFiles() { - return this.torperfDataFiles; - } - public SortedMap<String, List<String>> getTorperfExtradataFiles() { - return this.torperfExtradataFiles; + public List<String> getTorperfFiles() { + return this.torperfFiles; } public boolean getProvideFilesViaRsync() { return this.provideFilesViaRsync; diff --git a/src/org/torproject/ernie/db/Main.java b/src/org/torproject/ernie/db/Main.java index e44eac0..4fa9e37 100644 --- a/src/org/torproject/ernie/db/Main.java +++ b/src/org/torproject/ernie/db/Main.java @@ -139,8 +139,7 @@ public class Main { // Process Torperf files if (config.getProcessTorperfFiles()) { new TorperfDownloader(new File(config.getTorperfOutputDirectory()), - config.getTorperfSources(), config.getTorperfDataFiles(), - config.getTorperfExtradataFiles()); + config.getTorperfSources(), config.getTorperfFiles()); } // Copy recently published files to a local directory that can then diff --git a/src/org/torproject/ernie/db/TorperfDownloader.java b/src/org/torproject/ernie/db/TorperfDownloader.java index 8ee0fe9..e7c55a1 100644 --- a/src/org/torproject/ernie/db/TorperfDownloader.java +++ b/src/org/torproject/ernie/db/TorperfDownloader.java @@ -11,65 +11,172 @@ import java.io.IOException; import java.io.InputStreamReader; import java.net.HttpURLConnection; import java.net.URL; +import java.text.SimpleDateFormat; import java.util.List; import java.util.Map; import java.util.SortedMap; +import java.util.TimeZone; +import java.util.TreeMap; import java.util.logging.Level; import java.util.logging.Logger; /* Download possibly truncated Torperf .data and .extradata files from - * configured sources and append them to the files we already have. */ + * configured sources, append them to the files we already have, and merge + * the two files into the .tpf format. */ public class TorperfDownloader { private File torperfOutputDirectory = null; private SortedMap<String, String> torperfSources = null; - private SortedMap<String, List<String>> torperfDataFiles = null; - private SortedMap<String, List<String>> torperfExtradataFiles = null; + private List<String> torperfFilesLines = null; private Logger logger = null; + private SimpleDateFormat dateFormat; public TorperfDownloader(File torperfOutputDirectory, SortedMap<String, String> torperfSources, - SortedMap<String, List<String>> torperfDataFiles, - SortedMap<String, List<String>> torperfExtradataFiles) { + List<String> torperfFilesLines) { if (torperfOutputDirectory == null) { throw new IllegalArgumentException(); } this.torperfOutputDirectory = torperfOutputDirectory; this.torperfSources = torperfSources; - this.torperfDataFiles = torperfDataFiles; - this.torperfExtradataFiles = torperfExtradataFiles; + this.torperfFilesLines = torperfFilesLines; if (!this.torperfOutputDirectory.exists()) { this.torperfOutputDirectory.mkdirs(); } this.logger = Logger.getLogger(TorperfDownloader.class.getName()); - this.downloadAndMergeFiles(this.torperfDataFiles, true); - this.downloadAndMergeFiles(this.torperfExtradataFiles, false); + this.dateFormat = new SimpleDateFormat("yyyy-MM-dd"); + this.dateFormat.setTimeZone(TimeZone.getTimeZone("UTC")); + this.readLastMergedTimestamps(); + for (String torperfFilesLine : this.torperfFilesLines) { + this.downloadAndMergeFiles(torperfFilesLine); + } + this.writeLastMergedTimestamps(); } - private void downloadAndMergeFiles( - SortedMap<String, List<String>> dataOrExtradataFiles, - boolean isDataFile) { - for (Map.Entry<String, List<String>> e : - dataOrExtradataFiles.entrySet()) { - String sourceName = e.getKey(); - String sourceBaseUrl = torperfSources.get(sourceName); - List<String> files = e.getValue(); - for (String file : files) { - String url = sourceBaseUrl + file; - File outputFile = new File(torperfOutputDirectory, - sourceName + "-" + file); - this.downloadAndMergeFile(url, outputFile, isDataFile); + private File torperfLastMergedFile = + new File("stats/torperf-last-merged"); + SortedMap<String, String> lastMergedTimestamps = + new TreeMap<String, String>(); + private void readLastMergedTimestamps() { + if (!this.torperfLastMergedFile.exists()) { + return; + } + try { + BufferedReader br = new BufferedReader(new FileReader( + this.torperfLastMergedFile)); + String line; + while ((line = br.readLine()) != null) { + String[] parts = line.split(" "); + String fileName = null, timestamp = null; + if (parts.length == 2) { + try { + Double.parseDouble(parts[1]); + fileName = parts[0]; + timestamp = parts[1]; + } catch (NumberFormatException e) { + /* Handle below. */ + } + } + if (fileName == null || timestamp == null) { + this.logger.log(Level.WARNING, "Invalid line '" + line + "' in " + + this.torperfLastMergedFile.getAbsolutePath() + ". " + + "Ignoring past history of merging .data and .extradata " + + "files."); + this.lastMergedTimestamps.clear(); + break; + } + this.lastMergedTimestamps.put(fileName, timestamp); } + br.close(); + } catch (IOException e) { + this.logger.log(Level.WARNING, "Error while reading '" + + this.torperfLastMergedFile.getAbsolutePath() + ". Ignoring " + + "past history of merging .data and .extradata files."); + this.lastMergedTimestamps.clear(); + } + } + + private void writeLastMergedTimestamps() { + try { + BufferedWriter bw = new BufferedWriter(new FileWriter( + this.torperfLastMergedFile)); + for (Map.Entry<String, String> e : + this.lastMergedTimestamps.entrySet()) { + String fileName = e.getKey(); + String timestamp = e.getValue(); + bw.write(fileName + " " + timestamp + "\n"); + } + bw.close(); + } catch (IOException e) { + this.logger.log(Level.WARNING, "Error while writing '" + + this.torperfLastMergedFile.getAbsolutePath() + ". This may " + + "result in ignoring history of merging .data and .extradata " + + "files in the next execution.", e); + } + } + + private void downloadAndMergeFiles(String torperfFilesLine) { + String[] parts = torperfFilesLine.split(" "); + String sourceName = parts[1]; + int fileSize = -1; + try { + fileSize = Integer.parseInt(parts[2]); + } catch (NumberFormatException e) { + this.logger.log(Level.WARNING, "Could not parse file size in " + + "TorperfFiles configuration line '" + torperfFilesLine + + "'."); + return; + } + + /* Download and append the .data file. */ + String dataFileName = parts[3]; + String sourceBaseUrl = torperfSources.get(sourceName); + String dataUrl = sourceBaseUrl + dataFileName; + String dataOutputFileName = sourceName + "-" + dataFileName; + File dataOutputFile = new File(torperfOutputDirectory, + dataOutputFileName); + boolean downloadedDataFile = this.downloadAndAppendFile(dataUrl, + dataOutputFile, true); + + /* Download and append the .extradata file. */ + String extradataFileName = parts[4]; + String extradataUrl = sourceBaseUrl + extradataFileName; + String extradataOutputFileName = sourceName + "-" + extradataFileName; + File extradataOutputFile = new File(torperfOutputDirectory, + extradataOutputFileName); + boolean downloadedExtradataFile = this.downloadAndAppendFile( + extradataUrl, extradataOutputFile, false); + + /* Merge both files into .tpf format. */ + if (!downloadedDataFile && !downloadedExtradataFile) { + return; + } + String skipUntil = null; + if (this.lastMergedTimestamps.containsKey(dataOutputFileName)) { + skipUntil = this.lastMergedTimestamps.get(dataOutputFileName); + } + try { + skipUntil = this.mergeFiles(dataOutputFile, extradataOutputFile, + sourceName, fileSize, skipUntil); + } catch (IOException e) { + this.logger.log(Level.WARNING, "Failed merging " + dataOutputFile + + " and " + extradataOutputFile + ".", e); + } + if (skipUntil != null) { + this.lastMergedTimestamps.put(dataOutputFileName, skipUntil); } } - private void downloadAndMergeFile(String url, File outputFile, + private boolean downloadAndAppendFile(String url, File outputFile, boolean isDataFile) { + + /* Read an existing output file to determine which line will be the + * first to append to it. */ String lastTimestampLine = null; int linesAfterLastTimestampLine = 0; if (outputFile.exists() && outputFile.lastModified() > System.currentTimeMillis() - 330L * 60L * 1000L) { - return; + return false; } else if (outputFile.exists()) { try { BufferedReader br = new BufferedReader(new FileReader( @@ -85,10 +192,10 @@ public class TorperfDownloader { } br.close(); } catch (IOException e) { - logger.log(Level.WARNING, "Failed reading '" - + outputFile.getAbsolutePath() + "' to find the last line to " - + "append to.", e); - return; + this.logger.log(Level.WARNING, "Failed reading '" + + outputFile.getAbsolutePath() + "' to determine the first " + + "line to append to it.", e); + return false; } } try { @@ -120,14 +227,14 @@ public class TorperfDownloader { bw.close(); br.close(); } catch (IOException e) { - logger.log(Level.WARNING, "Failed downloading and merging '" + url - + "'.", e); - return; + this.logger.log(Level.WARNING, "Failed downloading and/or merging '" + + url + "'.", e); + return false; } if (lastTimestampLine == null) { - logger.warning("'" + outputFile.getAbsolutePath() + "' doesn't " - + "contain any timestamp lines. Unable to check whether that " - + "file is stale or not."); + this.logger.warning("'" + outputFile.getAbsolutePath() + + "' doesn't contain any timestamp lines. Unable to check " + + "whether that file is stale or not."); } else { long lastTimestampMillis = -1L; if (isDataFile) { @@ -141,11 +248,318 @@ public class TorperfDownloader { } if (lastTimestampMillis < System.currentTimeMillis() - 330L * 60L * 1000L) { - logger.warning("The last timestamp in '" + this.logger.warning("The last timestamp in '" + outputFile.getAbsolutePath() + "' is more than 5:30 hours " + "old: " + lastTimestampMillis); } } + return true; + } + + private String mergeFiles(File dataFile, File extradataFile, + String source, int fileSize, String skipUntil) throws IOException { + SortedMap<String, String> config = new TreeMap<String, String>(); + config.put("SOURCE", source); + config.put("FILESIZE", String.valueOf(fileSize)); + if (!dataFile.exists() || !extradataFile.exists()) { + this.logger.warning("File " + dataFile.getAbsolutePath() + " or " + + extradataFile.getAbsolutePath() + " is missing."); + return null; + } + this.logger.fine("Merging " + dataFile.getAbsolutePath() + " and " + + extradataFile.getAbsolutePath() + " into .tpf format."); + BufferedReader brD = new BufferedReader(new FileReader(dataFile)), + brE = new BufferedReader(new FileReader(extradataFile)); + String lineD = brD.readLine(), lineE = brE.readLine(); + int d = 1, e = 1; + String maxDataComplete = null, maxUsedAt = null; + while (lineD != null) { + + /* Parse .data line. Every valid .data line will go into the .tpf + * format, either with additional information from the .extradata + * file or without it. */ + if (lineD.isEmpty()) { + this.logger.finer("Skipping empty line " + dataFile.getName() + ":" + + d++ + "."); + lineD = brD.readLine(); + continue; + } + SortedMap<String, String> data = this.parseDataLine(lineD); + if (data == null) { + this.logger.warning("Skipping illegal line .data:" + d++ + " '" + + lineD + "'."); + lineD = brD.readLine(); + continue; + } + String dataComplete = data.get("DATACOMPLETE"); + double dataCompleteSeconds = Double.parseDouble(dataComplete); + if (skipUntil != null && dataComplete.compareTo(skipUntil) < 0) { + this.logger.finer("Skipping " + dataFile.getName() + ":" + + d++ + " which we already processed before."); + lineD = brD.readLine(); + continue; + } + maxDataComplete = dataComplete; + + /* Parse .extradata line if available and try to find the one that + * matches the .data line. */ + SortedMap<String, String> extradata = null; + while (lineE != null) { + if (lineE.isEmpty()) { + this.logger.finer("Skipping " + extradataFile.getName() + ":" + + e++ + " which is empty."); + lineE = brE.readLine(); + continue; + } + if (lineE.startsWith("BUILDTIMEOUT_SET ")) { + this.logger.finer("Skipping " + extradataFile.getName() + ":" + + e++ + " which is a BUILDTIMEOUT_SET line."); + lineE = brE.readLine(); + continue; + } else if (lineE.startsWith("ok ") || + lineE.startsWith("error ")) { + this.logger.finer("Skipping " + extradataFile.getName() + ":" + + e++ + " which is in the old format."); + lineE = brE.readLine(); + continue; + } + extradata = this.parseExtradataLine(lineE); + if (extradata == null) { + this.logger.warning("Skipping Illegal line .extradata:" + e++ + + " '" + lineE + "'."); + lineE = brE.readLine(); + continue; + } + if (!extradata.containsKey("USED_AT")) { + this.logger.finer("Skipping " + extradataFile.getName() + ":" + + e++ + " which doesn't contain a USED_AT element."); + lineE = brE.readLine(); + continue; + } + String usedAt = extradata.get("USED_AT"); + double usedAtSeconds = Double.parseDouble(usedAt); + if (skipUntil != null && usedAt.compareTo(skipUntil) < 0) { + this.logger.finer("Skipping " + extradataFile.getName() + ":" + + e++ + " which we already processed before."); + lineE = brE.readLine(); + continue; + } + maxUsedAt = usedAt; + if (Math.abs(usedAtSeconds - dataCompleteSeconds) <= 1.0) { + this.logger.fine("Merging " + extradataFile.getName() + ":" + + e++ + " into the current .data line."); + lineE = brE.readLine(); + break; + } else if (usedAtSeconds > dataCompleteSeconds) { + this.logger.finer("Comparing " + extradataFile.getName() + + " to the next .data line."); + extradata = null; + break; + } else { + this.logger.finer("Skipping " + extradataFile.getName() + ":" + + e++ + " which is too old to be merged with .data:" + d + + "."); + lineE = brE.readLine(); + continue; + } + } + + /* Write output line to .tpf file. */ + SortedMap<String, String> keysAndValues = + new TreeMap<String, String>(); + if (extradata != null) { + keysAndValues.putAll(extradata); + } + keysAndValues.putAll(data); + keysAndValues.putAll(config); + this.logger.fine("Writing " + dataFile.getName() + ":" + d++ + "."); + lineD = brD.readLine(); + try { + this.writeTpfLine(source, fileSize, keysAndValues); + } catch (IOException ex) { + this.logger.log(Level.WARNING, "Error writing output line. " + + "Aborting to merge " + dataFile.getName() + " and " + + extradataFile.getName() + ".", e); + break; + } + } + brD.close(); + brE.close(); + this.writeCachedTpfLines(); + if (maxDataComplete == null) { + return maxUsedAt; + } else if (maxUsedAt == null) { + return maxDataComplete; + } else if (maxDataComplete.compareTo(maxUsedAt) > 0) { + return maxUsedAt; + } else { + return maxDataComplete; + } + } + + private SortedMap<Integer, String> dataTimestamps; + private SortedMap<String, String> parseDataLine(String line) { + String[] parts = line.trim().split(" "); + if (line.length() == 0 || parts.length < 20) { + return null; + } + if (this.dataTimestamps == null) { + this.dataTimestamps = new TreeMap<Integer, String>(); + this.dataTimestamps.put(0, "START"); + this.dataTimestamps.put(2, "SOCKET"); + this.dataTimestamps.put(4, "CONNECT"); + this.dataTimestamps.put(6, "NEGOTIATE"); + this.dataTimestamps.put(8, "REQUEST"); + this.dataTimestamps.put(10, "RESPONSE"); + this.dataTimestamps.put(12, "DATAREQUEST"); + this.dataTimestamps.put(14, "DATARESPONSE"); + this.dataTimestamps.put(16, "DATACOMPLETE"); + this.dataTimestamps.put(21, "DATAPERC10"); + this.dataTimestamps.put(23, "DATAPERC20"); + this.dataTimestamps.put(25, "DATAPERC30"); + this.dataTimestamps.put(27, "DATAPERC40"); + this.dataTimestamps.put(29, "DATAPERC50"); + this.dataTimestamps.put(31, "DATAPERC60"); + this.dataTimestamps.put(33, "DATAPERC70"); + this.dataTimestamps.put(35, "DATAPERC80"); + this.dataTimestamps.put(37, "DATAPERC90"); + } + SortedMap<String, String> data = new TreeMap<String, String>(); + try { + for (Map.Entry<Integer, String> e : this.dataTimestamps.entrySet()) { + int i = e.getKey(); + if (parts.length > i + 1) { + String key = e.getValue(); + String value = String.format("%s.%02d", parts[i], + Integer.parseInt(parts[i + 1]) / 10000); + data.put(key, value); + } + } + } catch (NumberFormatException e) { + return null; + } + data.put("WRITEBYTES", parts[18]); + data.put("READBYTES", parts[19]); + if (parts.length >= 21) { + data.put("DIDTIMEOUT", parts[20]); + } + return data; + } + + private SortedMap<String, String> parseExtradataLine(String line) { + String[] parts = line.split(" "); + SortedMap<String, String> extradata = new TreeMap<String, String>(); + String previousKey = null; + for (String part : parts) { + String[] keyAndValue = part.split("=", -1); + if (keyAndValue.length == 2) { + String key = keyAndValue[0]; + previousKey = key; + String value = keyAndValue[1]; + if (value.contains(".") && value.lastIndexOf(".") == + value.length() - 2) { + /* Make sure that all floats have two trailing digits. */ + value += "0"; + } + extradata.put(key, value); + } else if (keyAndValue.length == 1 && previousKey != null) { + String value = keyAndValue[0]; + if (previousKey.equals("STREAM_FAIL_REASONS") && + (value.equals("MISC") || value.equals("EXITPOLICY") || + value.equals("RESOURCELIMIT") || + value.equals("RESOLVEFAILED"))) { + extradata.put(previousKey, extradata.get(previousKey) + ":" + + value); + } else { + return null; + } + } else { + return null; + } + } + return extradata; + } + + private String cachedSource; + private int cachedFileSize; + private String cachedStartDate; + private SortedMap<String, String> cachedTpfLines; + private void writeTpfLine(String source, int fileSize, + SortedMap<String, String> keysAndValues) throws IOException { + StringBuilder sb = new StringBuilder(); + int written = 0; + for (Map.Entry<String, String> keyAndValue : + keysAndValues.entrySet()) { + String key = keyAndValue.getKey(); + String value = keyAndValue.getValue(); + sb.append((written++ > 0 ? " " : "") + key + "=" + value); + } + String line = sb.toString(); + String startString = keysAndValues.get("START"); + long startMillis = Long.parseLong(startString.substring(0, + startString.indexOf("."))) * 1000L; + String startDate = dateFormat.format(startMillis); + if (this.cachedTpfLines == null || !source.equals(this.cachedSource) || + fileSize != this.cachedFileSize || + !startDate.equals(this.cachedStartDate)) { + this.writeCachedTpfLines(); + this.readTpfLinesToCache(source, fileSize, startDate); + } + if (!this.cachedTpfLines.containsKey(startString) || + line.length() > this.cachedTpfLines.get(startString).length()) { + this.cachedTpfLines.put(startString, line); + } + } + + private void readTpfLinesToCache(String source, int fileSize, + String startDate) throws IOException { + this.cachedTpfLines = new TreeMap<String, String>(); + this.cachedSource = source; + this.cachedFileSize = fileSize; + this.cachedStartDate = startDate; + File tpfFile = new File(torperfOutputDirectory, + startDate.replaceAll("-", "/") + "/" + + source + "-" + String.valueOf(fileSize) + "-" + startDate + + ".tpf"); + if (!tpfFile.exists()) { + return; + } + BufferedReader br = new BufferedReader(new FileReader(tpfFile)); + String line; + while ((line = br.readLine()) != null) { + if (line.startsWith("@type ")) { + continue; + } + if (line.contains("START=")) { + String startString = line.substring(line.indexOf("START=") + + "START=".length()).split(" ")[0]; + this.cachedTpfLines.put(startString, line); + } + } + br.close(); + } + + private void writeCachedTpfLines() throws IOException { + if (this.cachedSource == null || this.cachedFileSize == 0 || + this.cachedStartDate == null || this.cachedTpfLines == null) { + return; + } + File tpfFile = new File(torperfOutputDirectory, + this.cachedStartDate.replaceAll("-", "/") + + "/" + this.cachedSource + "-" + + String.valueOf(this.cachedFileSize) + "-" + + this.cachedStartDate + ".tpf"); + tpfFile.getParentFile().mkdirs(); + BufferedWriter bw = new BufferedWriter(new FileWriter(tpfFile)); + bw.write("@type torperf 1.0\n"); + for (String line : this.cachedTpfLines.values()) { + bw.write(line + "\n"); + } + bw.close(); + this.cachedSource = null; + this.cachedFileSize = 0; + this.cachedStartDate = null; + this.cachedTpfLines = null; } }