ZOOKEEPER-3137: add a utility to truncate logs to a zxid

Author: Brian Nixon <nixon@fb.com>

Reviewers: Michael Han <hanm@apache.org>, Andor Molnár <andor@apache.org>

Closes #615 from enixon/add-chop

docs/zookeeperAdmin.html

@@ -2527,7 +2527,8 @@ server.3=zoo3:2888:3888</pre>
 <p>Using older log and snapshot files, you can look at the previous
         state of ZooKeeper servers and even restore that state. The
         LogFormatter class allows an administrator to look at the transactions
-        in a log.</p>
+        in a log. The LogChopper class allows an administrator to create
+         truncated versions of transaction logs.</p>
 <p>The ZooKeeper server creates snapshot and log files, but
         never deletes them. The retention policy of the data and log
         files is implemented outside of the ZooKeeper server. The

src/java/main/org/apache/zookeeper/server/persistence/TxnLogToolkit.java

@@ -30,6 +30,7 @@ import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.Record;
 import org.apache.zookeeper.server.ExitCode;
 import org.apache.zookeeper.server.TraceFormatter;
+import org.apache.zookeeper.server.util.LogChopper;
 import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.txn.CreateContainerTxn;
 import org.apache.zookeeper.txn.CreateTTLTxn;
@@ -37,6 +38,8 @@ import org.apache.zookeeper.txn.CreateTxn;
 import org.apache.zookeeper.txn.SetDataTxn;
 import org.apache.zookeeper.txn.TxnHeader;
 
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
 import java.io.Closeable;
 import java.io.EOFException;
 import java.io.File;
@@ -44,6 +47,8 @@ import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.text.DateFormat;
 import java.util.Date;
 import java.util.Scanner;
@@ -96,13 +101,20 @@ public class TxnLogToolkit implements Closeable {
     private FilePadding filePadding = new FilePadding();
     private boolean force = false;
 
+    // chop mode
+    private long zxid = -1L;
+
     /**
      * @param args Command line arguments
      */
     public static void main(String[] args) throws Exception {
         try (final TxnLogToolkit lt = parseCommandLine(args)) {
-            lt.dump(new Scanner(System.in));
-            lt.printStat();
+            if (lt.isDumpMode()) {
+                lt.dump(new Scanner(System.in));
+                lt.printStat();
+            } else {
+                lt.chop();
+            }
         } catch (TxnLogToolkitParseException e) {
             System.err.println(e.getMessage() + "\n");
             printHelpAndExit(e.getExitCode(), e.getOptions());
@@ -117,10 +129,7 @@ public class TxnLogToolkit implements Closeable {
         this.recoveryMode = recoveryMode;
         this.verbose = verbose;
         this.force = force;
-        txnLogFile = new File(txnLogFileName);
-        if (!txnLogFile.exists() || !txnLogFile.canRead()) {
-            throw new TxnLogToolkitException(ExitCode.UNEXPECTED_ERROR.getValue(), "File doesn't exist or not readable: %s", txnLogFile);
-        }
+        txnLogFile = loadTxnFile(txnLogFileName);
         if (recoveryMode) {
             recoveryLogFile = new File(txnLogFile.toString() + ".fixed");
             if (recoveryLogFile.exists()) {
@@ -134,6 +143,19 @@ public class TxnLogToolkit implements Closeable {
         }
     }
 
+    public TxnLogToolkit(String txnLogFileName, String zxidName) throws TxnLogToolkitException {
+        txnLogFile = loadTxnFile(txnLogFileName);
+        zxid = Long.decode(zxidName);
+    }
+
+    private File loadTxnFile(String txnLogFileName) throws TxnLogToolkitException {
+        File logFile = new File(txnLogFileName);
+        if (!logFile.exists() || !logFile.canRead()) {
+            throw new TxnLogToolkitException(ExitCode.UNEXPECTED_ERROR.getValue(), "File doesn't exist or not readable: %s", logFile);
+        }
+        return logFile;
+    }
+
     public void dump(Scanner scanner) throws Exception {
         crcFixed = 0;
 
@@ -204,6 +226,24 @@ public class TxnLogToolkit implements Closeable {
         }
     }
 
+    public void chop() {
+        File targetFile = new File(txnLogFile.getParentFile(), txnLogFile.getName() + ".chopped" + zxid);
+        try (
+                InputStream is = new BufferedInputStream(new FileInputStream(txnLogFile));
+                OutputStream os = new BufferedOutputStream(new FileOutputStream(targetFile))
+        ) {
+            if (!LogChopper.chop(is, os, zxid)) {
+                throw new TxnLogToolkitException(ExitCode.INVALID_INVOCATION.getValue(), "Failed to chop %s", txnLogFile.getName());
+            }
+        } catch (Exception e) {
+            System.out.println("Got exception: " + e.getMessage());
+        }
+    }
+
+    public boolean isDumpMode() {
+        return zxid < 0;
+    }
+
     private boolean askForFix(Scanner scanner) throws TxnLogToolkitException {
         while (true) {
             System.out.print("Would you like to fix it (Yes/No/Abort) ? ");
@@ -320,6 +360,12 @@ public class TxnLogToolkit implements Closeable {
         Option forceOpt = new Option("y", "yes", false, "Non-interactive mode: repair all CRC errors without asking");
         options.addOption(forceOpt);
 
+        // Chop mode options
+        Option chopOpt = new Option("c", "chop", false, "Chop mode. Chop txn file to a zxid.");
+        Option zxidOpt = new Option("z", "zxid", true, "Used with chop. Zxid to which to chop.");
+        options.addOption(chopOpt);
+        options.addOption(zxidOpt);
+
         try {
             CommandLine cli = parser.parse(options, args);
             if (cli.hasOption("help")) {
@@ -328,6 +374,9 @@ public class TxnLogToolkit implements Closeable {
             if (cli.getArgs().length < 1) {
                 printHelpAndExit(1, options);
             }
+            if (cli.hasOption("chop") && cli.hasOption("zxid")) {
+                return new TxnLogToolkit(cli.getArgs()[0], cli.getOptionValue("zxid"));
+            }
             return new TxnLogToolkit(cli.hasOption("recover"), cli.hasOption("verbose"), cli.getArgs()[0], cli.hasOption("yes"));
         } catch (ParseException e) {
             throw new TxnLogToolkitParseException(options, ExitCode.UNEXPECTED_ERROR.getValue(), e.getMessage());
@@ -336,7 +385,7 @@ public class TxnLogToolkit implements Closeable {
 
     private static void printHelpAndExit(int exitCode, Options options) {
         HelpFormatter help = new HelpFormatter();
-        help.printHelp(120,"TxnLogToolkit [-dhrv] <txn_log_file_name>", "", options, "");
+        help.printHelp(120,"TxnLogToolkit [-dhrvc] <txn_log_file_name> (-z <zxid>)", "", options, "");
         System.exit(exitCode);
     }
 

src/java/main/org/apache/zookeeper/server/util/LogChopper.java

@@ -0,0 +1,160 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.util;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.Record;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.server.ExitCode;
+import org.apache.zookeeper.server.persistence.FileHeader;
+import org.apache.zookeeper.server.persistence.FileTxnLog;
+import org.apache.zookeeper.txn.TxnHeader;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.EOFException;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.zip.Adler32;
+import java.util.zip.Checksum;
+
+/**
+ * this class will chop the log at the specified zxid
+ */
+@InterfaceAudience.Public
+public class LogChopper {
+    public static void main(String args[]) {
+        ExitCode rc = ExitCode.INVALID_INVOCATION;
+        if (args.length != 3) {
+            System.out.println("Usage: LogChopper zxid_to_chop_to txn_log_to_chop chopped_filename");
+            System.out.println("    this program will read the txn_log_to_chop file and copy all the transactions");
+            System.out.println("    from it up to (and including) the given zxid into chopped_filename.");
+            System.exit(rc.getValue());
+        }
+        String txnLog = args[1];
+        String choppedLog = args[2];
+
+        try (
+            InputStream is = new BufferedInputStream(new FileInputStream(txnLog));
+            OutputStream os = new BufferedOutputStream(new FileOutputStream(choppedLog))
+        ) {
+            long zxid = Long.decode(args[0]);
+
+            if (chop(is, os, zxid)) {
+                rc = ExitCode.EXECUTION_FINISHED;
+            }
+        } catch (Exception e) {
+            System.out.println("Got exception: " + e.getMessage());
+        }
+        System.exit(rc.getValue());
+    }
+
+    public static boolean chop(InputStream is, OutputStream os, long zxid) throws IOException {
+        BinaryInputArchive logStream = BinaryInputArchive.getArchive(is);
+        BinaryOutputArchive choppedStream = BinaryOutputArchive.getArchive(os);
+        FileHeader fhdr = new FileHeader();
+        fhdr.deserialize(logStream, "fileheader");
+
+        if (fhdr.getMagic() != FileTxnLog.TXNLOG_MAGIC) {
+            System.err.println("Invalid magic number in txn log file");
+            return false;
+        }
+        System.out.println("ZooKeeper Transactional Log File with dbid "
+                + fhdr.getDbid() + " txnlog format version "
+                + fhdr.getVersion());
+
+        fhdr.serialize(choppedStream, "fileheader");
+        int count = 0;
+        boolean hasZxid = false;
+        long previousZxid = -1;
+        while (true) {
+            long crcValue;
+            byte[] bytes;
+            try {
+                crcValue = logStream.readLong("crcvalue");
+
+                bytes = logStream.readBuffer("txnEntry");
+            } catch (EOFException e) {
+                System.out.println("EOF reached after " + count + " txns.");
+                // returning false because nothing was chopped
+                return false;
+            }
+            if (bytes.length == 0) {
+                // Since we preallocate, we define EOF to be an
+                // empty transaction
+                System.out.println("EOF reached after " + count + " txns.");
+                // returning false because nothing was chopped
+                return false;
+            }
+
+            Checksum crc = new Adler32();
+            crc.update(bytes, 0, bytes.length);
+            if (crcValue != crc.getValue()) {
+                throw new IOException("CRC doesn't match " + crcValue +
+                        " vs " + crc.getValue());
+            }
+            TxnHeader hdr = new TxnHeader();
+            Record txn = SerializeUtils.deserializeTxn(bytes, hdr);
+            if (logStream.readByte("EOR") != 'B') {
+                System.out.println("Last transaction was partial.");
+                throw new EOFException("Last transaction was partial.");
+            }
+
+            final long txnZxid = hdr.getZxid();
+            if (txnZxid == zxid) {
+                hasZxid = true;
+            }
+
+            // logging the gap to make the inconsistency investigation easier
+            if (previousZxid != -1 && txnZxid != previousZxid + 1) {
+                long txnEpoch = ZxidUtils.getEpochFromZxid(txnZxid);
+                long txnCounter = ZxidUtils.getCounterFromZxid(txnZxid);
+                long previousEpoch = ZxidUtils.getEpochFromZxid(previousZxid);
+                if (txnEpoch == previousEpoch) {
+                    System.out.println(
+                            String.format("There is intra-epoch gap between %x and %x",
+                                    previousZxid, txnZxid));
+                } else if (txnCounter != 1) {
+                    System.out.println(
+                        String.format("There is inter-epoch gap between %x and %x",
+                                previousZxid, txnZxid));
+                }
+            }
+            previousZxid = txnZxid;
+
+            if (txnZxid > zxid) {
+                if (count == 0 || !hasZxid) {
+                    System.out.println(String.format("This log does not contain zxid %x", zxid));
+                    return false;
+                }
+                System.out.println(String.format("Chopping at %x new log has %d records", zxid, count));
+                return true;
+            }
+
+            choppedStream.writeLong(crcValue, "crcvalue");
+            choppedStream.writeBuffer(bytes, "txnEntry");
+            choppedStream.writeByte((byte)'B', "EOR");
+
+            count++;
+        }
+    }
+}

src/java/test/org/apache/zookeeper/test/LogChopperTest.java

@@ -0,0 +1,129 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.test;
+
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.server.persistence.FileTxnLog;
+import org.apache.zookeeper.server.persistence.TxnLog;
+import org.apache.zookeeper.server.util.LogChopper;
+import org.apache.zookeeper.txn.DeleteTxn;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+
+class Pair<V1, V2> {
+    private V1 v1;
+    private V2 v2;
+    Pair(V1 v1, V2 v2) {
+        this.v1 = v1;
+        this.v2 = v2;
+    }
+    public V1 getFirst() {
+        return v1;
+    }
+    public V2 getSecond() {
+        return v2;
+    }
+}
+
+public class LogChopperTest extends ClientBase {
+
+    void rmr(File dir) throws IOException {
+        Files.walkFileTree(dir.toPath(), new SimpleFileVisitor<Path>() {
+            @Override
+            public FileVisitResult visitFile(Path file, BasicFileAttributes a) throws IOException {
+                Files.delete(file);
+                return FileVisitResult.CONTINUE;
+            }
+
+            @Override
+            public FileVisitResult postVisitDirectory(Path dir, IOException e) throws IOException {
+                Files.delete(dir);
+                return FileVisitResult.CONTINUE;
+            }
+        });
+    }
+
+    Pair<Long, Long> getFirstLastZxid(File logFile) throws IOException {
+        File tmp = createTmpDir();
+        Files.copy(logFile.toPath(), new File(tmp, "log.0").toPath());
+        FileTxnLog txnLog = new FileTxnLog(tmp);
+        TxnLog.TxnIterator it = txnLog.read(0);
+        long firstZxid = it.getHeader().getZxid();
+        long lastZxid = firstZxid;
+        while (it.next()) {
+            lastZxid = it.getHeader().getZxid();
+        }
+        txnLog.close();
+        rmr(tmp);
+        return new Pair<Long, Long>(firstZxid, lastZxid);
+    }
+
+    @Test
+    public void testChopper() throws IOException {
+        long clientId = 17;
+        int cxid = 77;
+        long zxid = 1000;
+        long time = 1;
+        int type = ZooDefs.OpCode.delete;
+        DeleteTxn txn = new DeleteTxn("/foo");
+        File tmpDir = createTmpDir();
+        FileTxnLog txnLog = new FileTxnLog(tmpDir);
+
+        for (int i = 0; i < 100; i++) {
+            TxnHeader hdr = new TxnHeader(clientId, cxid, ++zxid, ++time, type);
+            txnLog.append(hdr, txn);
+        }
+
+        // append a txn with gap
+        TxnHeader hdr = new TxnHeader(clientId, cxid, zxid + 10, ++time, type);
+        txnLog.append(hdr, txn);
+
+        txnLog.commit();
+
+        // now find the log we just created.
+        final File logFile = new File(tmpDir, "log." + Integer.toHexString(1001));
+        Pair<Long, Long> firstLast = getFirstLastZxid(logFile);
+        Assert.assertEquals(1001, (long)firstLast.getFirst());
+        Assert.assertEquals(1110, (long)firstLast.getSecond());
+
+        File choppedFile = new File(tmpDir, "chopped_failed");
+        Assert.assertFalse(LogChopper.chop(
+                new FileInputStream(logFile),
+                new FileOutputStream(choppedFile), 1107));
+
+        choppedFile = new File(tmpDir, "chopped");
+        Assert.assertTrue(LogChopper.chop(
+                new FileInputStream(logFile),
+                new FileOutputStream(choppedFile), 1017));
+
+        firstLast = getFirstLastZxid(choppedFile);
+        Assert.assertEquals(1001, (long)firstLast.getFirst());
+        Assert.assertEquals(1017, (long)firstLast.getSecond());
+    }
+}