test/unit/org/apache/cassandra/utils/binlog/ExternalArchiverTest.java - cassandra - Git at Google

 /*
  * 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.cassandra.utils.binlog;

 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.attribute.PosixFilePermission;
 import java.nio.file.attribute.PosixFilePermissions;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;

 import com.google.common.collect.Sets;
 import org.junit.Test;

 import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
 import org.apache.cassandra.utils.Pair;

 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;

 public class ExternalArchiverTest
 {
     @Test
     public void testArchiver() throws IOException, InterruptedException
     {
         Pair<String, String> s = createScript();
         String script = s.left;
         String dir = s.right;
         Path logdirectory = Files.createTempDirectory("logdirectory");
         File logfileToArchive = Files.createTempFile(logdirectory, "logfile", "xyz").toFile();
         Files.write(logfileToArchive.toPath(), "content".getBytes());

         ExternalArchiver ea = new ExternalArchiver(script+" %path", null, 10);
         ea.onReleased(1, logfileToArchive);
         while (logfileToArchive.exists())
         {
             Thread.sleep(100);
         }

         File movedFile = new File(dir, logfileToArchive.getName());
         assertTrue(movedFile.exists());
         movedFile.deleteOnExit();
         ea.stop();
         assertEquals(0, logdirectory.toFile().listFiles().length);
     }

     @Test
     public void testArchiveExisting() throws IOException, InterruptedException
     {
         Pair<String, String> s = createScript();
         String script = s.left;
         String moveDir = s.right;
         List<File> existingFiles = new ArrayList<>();
         Path dir = Files.createTempDirectory("archive");
         for (int i = 0; i < 10; i++)
         {
             File logfileToArchive = Files.createTempFile(dir, "logfile", SingleChronicleQueue.SUFFIX).toFile();
             logfileToArchive.deleteOnExit();
             Files.write(logfileToArchive.toPath(), ("content"+i).getBytes());
             existingFiles.add(logfileToArchive);
         }

         ExternalArchiver ea = new ExternalArchiver(script + " %path", dir, 10);
         boolean allGone = false;
         while (!allGone)
         {
             allGone = true;
             for (File f : existingFiles)
             {
                 if (f.exists())
                 {
                     allGone = false;
                     Thread.sleep(100);
                     break;
                 }
                 File movedFile = new File(moveDir, f.getName());
                 assertTrue(movedFile.exists());
                 movedFile.deleteOnExit();
             }
         }
         ea.stop();
         assertEquals(0, dir.toFile().listFiles().length);
     }

     @Test
     public void testArchiveOnShutdown() throws IOException, InterruptedException
     {
         Pair<String, String> s = createScript();
         String script = s.left;
         String moveDir = s.right;
         Path dir = Files.createTempDirectory("archive");
         ExternalArchiver ea = new ExternalArchiver(script + " %path", dir, 10);
         List<File> existingFiles = new ArrayList<>();
         for (int i = 0; i < 10; i++)
         {
             File logfileToArchive = Files.createTempFile(dir, "logfile", SingleChronicleQueue.SUFFIX).toFile();
             logfileToArchive.deleteOnExit();
             Files.write(logfileToArchive.toPath(), ("content"+i).getBytes());
             existingFiles.add(logfileToArchive);
         }
         // ea.stop will archive all .cq4 files in the directory
         ea.stop();
         for (File f : existingFiles)
         {
             assertFalse(f.exists());
             File movedFile = new File(moveDir, f.getName());
             assertTrue(movedFile.exists());
             movedFile.deleteOnExit();
         }
     }

     /**
      * Make sure retries work
      * 1. create a script that will fail two times before executing the command
      * 2. create an ExternalArchiver that retries two times (this means we execute the script 3 times, meaning the last one will be successful)
      * 3. make sure the file is on disk until the script has been executed 3 times
      * 4. make sure the file is gone and that the command was executed successfully
      */
     @Test
     public void testRetries() throws IOException, InterruptedException
     {
         Pair<String, String> s = createFailingScript(2);
         String script = s.left;
         String moveDir = s.right;
         Path logdirectory = Files.createTempDirectory("logdirectory");
         File logfileToArchive = Files.createTempFile(logdirectory, "logfile", "xyz").toFile();
         Files.write(logfileToArchive.toPath(), "content".getBytes());
         AtomicInteger tryCounter = new AtomicInteger();
         AtomicBoolean success = new AtomicBoolean();
         ExternalArchiver ea = new ExternalArchiver(script + " %path", null, 1000, 2, (cmd) ->
         {
             tryCounter.incrementAndGet();
             ExternalArchiver.exec(cmd);
             success.set(true);
         });
         ea.onReleased(0, logfileToArchive);
         while (tryCounter.get() < 2) // while we have only executed this 0 or 1 times, the file should still be on disk
         {
             Thread.sleep(100);
             assertTrue(logfileToArchive.exists());
         }

         while (!success.get())
             Thread.sleep(100);

         // there will be 3 attempts in total, 2 failing ones, then the successful one:
         assertEquals(3, tryCounter.get());
         assertFalse(logfileToArchive.exists());
         File movedFile = new File(moveDir, logfileToArchive.getName());
         assertTrue(movedFile.exists());
         ea.stop();
     }


     /**
      * Makes sure that max retries is honored
      *
      * 1. create a script that will fail 3 times before actually executing the command
      * 2. create an external archiver that retries 2 times (this means that the script will get executed 3 times)
      * 3. make sure the file is still on disk and that we have not successfully executed the script
      *
      */
     @Test
     public void testMaxRetries() throws IOException, InterruptedException
     {
         Pair<String, String> s = createFailingScript(3);
         String script = s.left;
         String moveDir = s.right;
         Path logdirectory = Files.createTempDirectory("logdirectory");
         File logfileToArchive = Files.createTempFile(logdirectory, "logfile", "xyz").toFile();
         Files.write(logfileToArchive.toPath(), "content".getBytes());

         AtomicInteger tryCounter = new AtomicInteger();
         AtomicBoolean success = new AtomicBoolean();
         ExternalArchiver ea = new ExternalArchiver(script + " %path", null, 1000, 2, (cmd) ->
         {
             try
             {
                 ExternalArchiver.exec(cmd);
                 success.set(true);
             }
             catch (Throwable t)
             {
                 tryCounter.incrementAndGet();
                 throw t;
             }
         });
         ea.onReleased(0, logfileToArchive);
         while (tryCounter.get() < 3)
             Thread.sleep(500);
         assertTrue(logfileToArchive.exists());
         // and the file should not get moved:
         Thread.sleep(5000);
         assertTrue(logfileToArchive.exists());
         assertFalse(success.get());
         File [] fs = new File(moveDir).listFiles(f ->
                                                  {
                                                      if (f.getName().startsWith("file."))
                                                      {
                                                          f.deleteOnExit();
                                                          return true;
                                                      }
                                                      throw new AssertionError("There should be no other files in the directory");
                                                  });
         assertEquals(3, fs.length); // maxRetries + the first try
         ea.stop();
     }


     private Pair<String, String> createScript() throws IOException
     {
         File f = Files.createTempFile("script", "", PosixFilePermissions.asFileAttribute(Sets.newHashSet(PosixFilePermission.OWNER_WRITE,
                                                                                                          PosixFilePermission.OWNER_READ,
                                                                                                          PosixFilePermission.OWNER_EXECUTE))).toFile();
         f.deleteOnExit();
         File dir = Files.createTempDirectory("archive").toFile();
         dir.deleteOnExit();
         String script = "#!/bin/sh\nmv $1 "+dir.getAbsolutePath();
         Files.write(f.toPath(), script.getBytes());
         return Pair.create(f.getAbsolutePath(), dir.getAbsolutePath());
     }

     private Pair<String, String> createFailingScript(int failures) throws IOException
     {
         File f = Files.createTempFile("script", "", PosixFilePermissions.asFileAttribute(Sets.newHashSet(PosixFilePermission.OWNER_WRITE,
                                                                                                          PosixFilePermission.OWNER_READ,
                                                                                                          PosixFilePermission.OWNER_EXECUTE))).toFile();
         f.deleteOnExit();
         File dir = Files.createTempDirectory("archive").toFile();
         dir.deleteOnExit();
         // this script counts files in dir.getAbsolutePath, then if there are more than failures files in there, it moves the actual file
         String script = "#!/bin/bash%n" +
                         "DIR=%s%n" +
                         "shopt -s nullglob%n" +
                         "numfiles=($DIR/*)%n" +
                         "numfiles=${#numfiles[@]}%n" +
                         "if (( $numfiles < %d )); then%n" +
                         "    mktemp $DIR/file.XXXXX%n" +
                         "    exit 1%n" +
                         "else%n" +
                         "    mv $1 $DIR%n"+
                         "fi%n";

         Files.write(f.toPath(), String.format(script, dir.getAbsolutePath(), failures).getBytes());
         return Pair.create(f.getAbsolutePath(), dir.getAbsolutePath());
     }
 }
	/*
	* 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.cassandra.utils.binlog;

	import java.io.File;
	import java.io.IOException;
	import java.nio.file.Files;
	import java.nio.file.Path;
	import java.nio.file.attribute.PosixFilePermission;
	import java.nio.file.attribute.PosixFilePermissions;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicInteger;

	import com.google.common.collect.Sets;
	import org.junit.Test;

	import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
	import org.apache.cassandra.utils.Pair;

	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertTrue;

	public class ExternalArchiverTest
	{
	@Test
	public void testArchiver() throws IOException, InterruptedException
	{
	Pair<String, String> s = createScript();
	String script = s.left;
	String dir = s.right;
	Path logdirectory = Files.createTempDirectory("logdirectory");
	File logfileToArchive = Files.createTempFile(logdirectory, "logfile", "xyz").toFile();
	Files.write(logfileToArchive.toPath(), "content".getBytes());

	ExternalArchiver ea = new ExternalArchiver(script+" %path", null, 10);
	ea.onReleased(1, logfileToArchive);
	while (logfileToArchive.exists())
	{
	Thread.sleep(100);
	}

	File movedFile = new File(dir, logfileToArchive.getName());
	assertTrue(movedFile.exists());
	movedFile.deleteOnExit();
	ea.stop();
	assertEquals(0, logdirectory.toFile().listFiles().length);
	}

	@Test
	public void testArchiveExisting() throws IOException, InterruptedException
	{
	Pair<String, String> s = createScript();
	String script = s.left;
	String moveDir = s.right;
	List<File> existingFiles = new ArrayList<>();
	Path dir = Files.createTempDirectory("archive");
	for (int i = 0; i < 10; i++)
	{
	File logfileToArchive = Files.createTempFile(dir, "logfile", SingleChronicleQueue.SUFFIX).toFile();
	logfileToArchive.deleteOnExit();
	Files.write(logfileToArchive.toPath(), ("content"+i).getBytes());
	existingFiles.add(logfileToArchive);
	}

	ExternalArchiver ea = new ExternalArchiver(script + " %path", dir, 10);
	boolean allGone = false;
	while (!allGone)
	{
	allGone = true;
	for (File f : existingFiles)
	{
	if (f.exists())
	{
	allGone = false;
	Thread.sleep(100);
	break;
	}
	File movedFile = new File(moveDir, f.getName());
	assertTrue(movedFile.exists());
	movedFile.deleteOnExit();
	}
	}
	ea.stop();
	assertEquals(0, dir.toFile().listFiles().length);
	}

	@Test
	public void testArchiveOnShutdown() throws IOException, InterruptedException
	{
	Pair<String, String> s = createScript();
	String script = s.left;
	String moveDir = s.right;
	Path dir = Files.createTempDirectory("archive");
	ExternalArchiver ea = new ExternalArchiver(script + " %path", dir, 10);
	List<File> existingFiles = new ArrayList<>();
	for (int i = 0; i < 10; i++)
	{
	File logfileToArchive = Files.createTempFile(dir, "logfile", SingleChronicleQueue.SUFFIX).toFile();
	logfileToArchive.deleteOnExit();
	Files.write(logfileToArchive.toPath(), ("content"+i).getBytes());
	existingFiles.add(logfileToArchive);
	}
	// ea.stop will archive all .cq4 files in the directory
	ea.stop();
	for (File f : existingFiles)
	{
	assertFalse(f.exists());
	File movedFile = new File(moveDir, f.getName());
	assertTrue(movedFile.exists());
	movedFile.deleteOnExit();
	}
	}

	/**
	* Make sure retries work
	* 1. create a script that will fail two times before executing the command
	* 2. create an ExternalArchiver that retries two times (this means we execute the script 3 times, meaning the last one will be successful)
	* 3. make sure the file is on disk until the script has been executed 3 times
	* 4. make sure the file is gone and that the command was executed successfully
	*/
	@Test
	public void testRetries() throws IOException, InterruptedException
	{
	Pair<String, String> s = createFailingScript(2);
	String script = s.left;
	String moveDir = s.right;
	Path logdirectory = Files.createTempDirectory("logdirectory");
	File logfileToArchive = Files.createTempFile(logdirectory, "logfile", "xyz").toFile();
	Files.write(logfileToArchive.toPath(), "content".getBytes());
	AtomicInteger tryCounter = new AtomicInteger();
	AtomicBoolean success = new AtomicBoolean();
	ExternalArchiver ea = new ExternalArchiver(script + " %path", null, 1000, 2, (cmd) ->
	{
	tryCounter.incrementAndGet();
	ExternalArchiver.exec(cmd);
	success.set(true);
	});
	ea.onReleased(0, logfileToArchive);
	while (tryCounter.get() < 2) // while we have only executed this 0 or 1 times, the file should still be on disk
	{
	Thread.sleep(100);
	assertTrue(logfileToArchive.exists());
	}

	while (!success.get())
	Thread.sleep(100);

	// there will be 3 attempts in total, 2 failing ones, then the successful one:
	assertEquals(3, tryCounter.get());
	assertFalse(logfileToArchive.exists());
	File movedFile = new File(moveDir, logfileToArchive.getName());
	assertTrue(movedFile.exists());
	ea.stop();
	}


	/**
	* Makes sure that max retries is honored
	*
	* 1. create a script that will fail 3 times before actually executing the command
	* 2. create an external archiver that retries 2 times (this means that the script will get executed 3 times)
	* 3. make sure the file is still on disk and that we have not successfully executed the script
	*
	*/
	@Test
	public void testMaxRetries() throws IOException, InterruptedException
	{
	Pair<String, String> s = createFailingScript(3);
	String script = s.left;
	String moveDir = s.right;
	Path logdirectory = Files.createTempDirectory("logdirectory");
	File logfileToArchive = Files.createTempFile(logdirectory, "logfile", "xyz").toFile();
	Files.write(logfileToArchive.toPath(), "content".getBytes());

	AtomicInteger tryCounter = new AtomicInteger();
	AtomicBoolean success = new AtomicBoolean();
	ExternalArchiver ea = new ExternalArchiver(script + " %path", null, 1000, 2, (cmd) ->
	{
	try
	{
	ExternalArchiver.exec(cmd);
	success.set(true);
	}
	catch (Throwable t)
	{
	tryCounter.incrementAndGet();
	throw t;
	}
	});
	ea.onReleased(0, logfileToArchive);
	while (tryCounter.get() < 3)
	Thread.sleep(500);
	assertTrue(logfileToArchive.exists());
	// and the file should not get moved:
	Thread.sleep(5000);
	assertTrue(logfileToArchive.exists());
	assertFalse(success.get());
	File [] fs = new File(moveDir).listFiles(f ->
	{
	if (f.getName().startsWith("file."))
	{
	f.deleteOnExit();
	return true;
	}
	throw new AssertionError("There should be no other files in the directory");
	});
	assertEquals(3, fs.length); // maxRetries + the first try
	ea.stop();
	}


	private Pair<String, String> createScript() throws IOException
	{
	File f = Files.createTempFile("script", "", PosixFilePermissions.asFileAttribute(Sets.newHashSet(PosixFilePermission.OWNER_WRITE,
	PosixFilePermission.OWNER_READ,
	PosixFilePermission.OWNER_EXECUTE))).toFile();
	f.deleteOnExit();
	File dir = Files.createTempDirectory("archive").toFile();
	dir.deleteOnExit();
	String script = "#!/bin/sh\nmv $1 "+dir.getAbsolutePath();
	Files.write(f.toPath(), script.getBytes());
	return Pair.create(f.getAbsolutePath(), dir.getAbsolutePath());
	}

	private Pair<String, String> createFailingScript(int failures) throws IOException
	{
	File f = Files.createTempFile("script", "", PosixFilePermissions.asFileAttribute(Sets.newHashSet(PosixFilePermission.OWNER_WRITE,
	PosixFilePermission.OWNER_READ,
	PosixFilePermission.OWNER_EXECUTE))).toFile();
	f.deleteOnExit();
	File dir = Files.createTempDirectory("archive").toFile();
	dir.deleteOnExit();
	// this script counts files in dir.getAbsolutePath, then if there are more than failures files in there, it moves the actual file
	String script = "#!/bin/bash%n" +
	"DIR=%s%n" +
	"shopt -s nullglob%n" +
	"numfiles=($DIR/*)%n" +
	"numfiles=${#numfiles[@]}%n" +
	"if (( $numfiles < %d )); then%n" +
	" mktemp $DIR/file.XXXXX%n" +
	" exit 1%n" +
	"else%n" +
	" mv $1 $DIR%n"+
	"fi%n";

	Files.write(f.toPath(), String.format(script, dir.getAbsolutePath(), failures).getBytes());
	return Pair.create(f.getAbsolutePath(), dir.getAbsolutePath());
	}
	}