fix(logs): synchronise log file rotation and compression.

Foundation/include/Poco/ArchiveStrategy.h

@@ -23,6 +23,7 @@
 #include "Poco/File.h"
 #include "Poco/DateTimeFormatter.h"
 #include "Poco/NumberFormatter.h"
+#include "Poco/Mutex.h"
 #include <atomic>
 
 
@@ -58,10 +59,17 @@ class Foundation_API ArchiveStrategy
 	void moveFile(const std::string& oldName, const std::string& newName);
 	bool exists(const std::string& name);
 
+	Poco::FastMutex _rotateMutex;
+
 private:
+
+	friend class ArchiveCompressor;
+
 	ArchiveStrategy(const ArchiveStrategy&);
 	ArchiveStrategy& operator = (const ArchiveStrategy&);
 
+	void compressFile(const std::string& path);
+
 	std::atomic<bool> _compress;
 	std::atomic<ArchiveCompressor*> _pCompressor;
 };

Foundation/src/ArchiveStrategy.cpp

@@ -45,35 +45,18 @@ class ArchiveCompressor: public ActiveDispatcher
 	{
 	}
 
-	ActiveMethod<void, std::string, ArchiveCompressor, ActiveStarter<ActiveDispatcher>> compress;
+	struct ArchiveToCompress
+	{
+		ArchiveStrategy* as;
+		std::string path;
+	};
+
+	ActiveMethod<void, ArchiveToCompress, ArchiveCompressor, ActiveStarter<ActiveDispatcher>> compress;
 
 protected:
-	void compressImpl(const std::string& path)
+	void compressImpl(const ArchiveToCompress& ac)
 	{
-		std::string gzPath(path);
-		gzPath.append(".gz");
-		FileInputStream istr(path);
-		FileOutputStream ostr(gzPath);
-		try
-		{
-			DeflatingOutputStream deflater(ostr, DeflatingStreamBuf::STREAM_GZIP);
-			StreamCopier::copyStream(istr, deflater);
-			if (!deflater.good() || !ostr.good()) throw WriteFileException(gzPath);
-			deflater.close();
-			ostr.close();
-			istr.close();
-		}
-		catch (Poco::Exception&)
-		{
-			// deflating failed - remove gz file and leave uncompressed log file
-			ostr.close();
-			Poco::File gzf(gzPath);
-			gzf.remove();
-			return;
-		}
-		File f(path);
-		f.remove();
-		return;
+		ac.as->compressFile(ac.path);
 	}
 };
 
@@ -82,6 +65,10 @@ class ArchiveCompressor: public ActiveDispatcher
 // ArchiveStrategy
 //
 
+// Prefix that is added to the file being compressed to be skipped by the
+// purge strategy.
+static const char compressFilePrefix = '_';
+
 
 ArchiveStrategy::ArchiveStrategy():
 	_compress(false),
@@ -105,7 +92,7 @@ void ArchiveStrategy::compress(bool flag)
 void ArchiveStrategy::moveFile(const std::string& oldPath, const std::string& newPath)
 {
 	bool compressed = false;
-	Path p(oldPath);
+	const Path p(oldPath);
 	File f(oldPath);
 	if (!f.exists())
 	{
@@ -115,15 +102,22 @@ void ArchiveStrategy::moveFile(const std::string& oldPath, const std::string& ne
 	std::string mvPath(newPath);
 	if (_compress || compressed)
 		mvPath.append(".gz");
+
 	if (!_compress || compressed)
 	{
 		f.renameTo(mvPath);
 	}
 	else
 	{
-		f.renameTo(newPath);
-		if (!_pCompressor) _pCompressor = new ArchiveCompressor;
-		_pCompressor.load()->compress(newPath);
+		Path logdir { newPath };
+		logdir.makeParent();
+		const auto logfile { Path(newPath).getFileName() };
+		const auto compressPath = logdir.append(std::string(2, compressFilePrefix) + logfile).toString();
+		f.renameTo(compressPath);
+		if (!_pCompressor)
+			_pCompressor = new ArchiveCompressor;
+
+		_pCompressor.load()->compress( {this, compressPath} );
 	}
 }
 
@@ -146,6 +140,58 @@ bool ArchiveStrategy::exists(const std::string& name)
 }
 
 
+void ArchiveStrategy::compressFile(const std::string& path)
+{
+	FastMutex::ScopedLock l(_rotateMutex);
+
+	Path logdir { path };
+	logdir.makeParent();
+
+	auto removeFilePrefix = [&logdir](const std::string& path, char prefix) -> std::string
+	{
+		auto fname { Path(path).getFileName() };
+		auto p = fname.find_first_not_of(prefix);
+		if (p != fname.npos)
+			return Path(logdir, fname.substr(p)).toString();
+
+		return path;
+	};
+
+	std::string gzPath(path);
+	gzPath.append(".gz");
+	FileInputStream istr(path);
+	FileOutputStream ostr(gzPath);
+	try
+	{
+		DeflatingOutputStream deflater(ostr, DeflatingStreamBuf::STREAM_GZIP);
+		StreamCopier::copyStream(istr, deflater);
+		if (!deflater.good() || !ostr.good())
+			throw WriteFileException(gzPath);
+
+		deflater.close();
+		ostr.close();
+		istr.close();
+
+		// Remove temporary prefix
+		File f(gzPath);
+		f.renameTo(removeFilePrefix(gzPath, compressFilePrefix));
+	}
+	catch (const Poco::Exception&)
+	{
+		// deflating failed - remove gz file and leave uncompressed log file
+		ostr.close();
+		Poco::File gzf(gzPath);
+		gzf.remove();
+
+		File f(path);
+		f.renameTo(removeFilePrefix(path, compressFilePrefix));
+	}
+	File f(path);
+	f.remove();
+	return;
+}
+
+
 //
 // ArchiveByNumberStrategy
 //
@@ -169,6 +215,8 @@ LogFile* ArchiveByNumberStrategy::open(LogFile* pFile)
 
 LogFile* ArchiveByNumberStrategy::archive(LogFile* pFile)
 {
+	FastMutex::ScopedLock l(_rotateMutex);
+
 	std::string basePath = pFile->path();
 	delete pFile;
 	int n = -1;

Foundation/testsuite/src/FileChannelTest.cpp

@@ -506,6 +506,63 @@ void FileChannelTest::testCompress()
 }
 
 
+void FileChannelTest::testCompressedRotation()
+{
+	static const uint32_t MAX_ROLLOVER_TIMES = 8;
+	static const uint32_t LONG_MESSAGE_LENGTH = 1024;
+	static const uint32_t LONG_MAX_FILESIZE = 1024;
+
+	std::vector<uint8_t> longMessage(LONG_MESSAGE_LENGTH, '&');
+	longMessage.push_back(0);
+
+	Poco::Path logsPath(Poco::Path::current(), "logs");
+	Poco::File logsDir(logsPath.toString());
+	logsDir.createDirectory();
+	logsPath.append("test.log");
+
+	Poco::AutoPtr<Poco::FileChannel> fileChannel = new Poco::FileChannel("ABC");
+	fileChannel->setProperty(Poco::FileChannel::PROP_PATH, logsPath.toString());
+	fileChannel->setProperty(Poco::FileChannel::PROP_FLUSH, "false");
+	fileChannel->setProperty(Poco::FileChannel::PROP_ROTATION, "1 M");
+	fileChannel->setProperty(Poco::FileChannel::PROP_PURGECOUNT, "5");
+	fileChannel->setProperty(Poco::FileChannel::PROP_ARCHIVE, "number");
+	fileChannel->setProperty(Poco::FileChannel::PROP_TIMES, "local");
+	fileChannel->setProperty(Poco::FileChannel::PROP_COMPRESS, "true");
+
+	fileChannel->open();
+
+	std::string text(longMessage.begin(), longMessage.end());
+
+	for (uint32_t i = 1; i <= MAX_ROLLOVER_TIMES; ++i)
+	{
+	    for (uint32_t j = 0; j < LONG_MAX_FILESIZE; ++j)
+	    {
+	        Poco::Message message("ABC", text, Poco::Message::PRIO_INFORMATION);
+	        fileChannel->log(message);
+	    }
+	}
+
+	fileChannel->close();
+
+	Poco::Thread::sleep(200);
+
+	std::vector<std::string> files;
+	logsDir.list(files);
+	std::sort(files.begin(), files.end());
+
+	assertEqual(5+1+1, files.size()); // 5+1 rotated files, current file
+	assertEqual(files[0], "test.log");
+	assertEqual(files[1], "test.log.0.gz");
+	assertEqual(files[2], "test.log.1.gz");
+	assertEqual(files[3], "test.log.2.gz");
+	assertEqual(files[4], "test.log.3.gz");
+	assertEqual(files[5], "test.log.4.gz");
+	assertEqual(files[6], "test.log.5.gz");
+
+	logsDir.remove(true);
+}
+
+
 void FileChannelTest::purgeAge(const std::string& pa)
 {
 	std::string name = filename();
@@ -844,6 +901,7 @@ CppUnit::Test* FileChannelTest::suite()
 	CppUnit_addTest(pSuite, FileChannelTest, testArchive);
 	CppUnit_addTest(pSuite, FileChannelTest, testArchiveByStrategy);
 	CppUnit_addTest(pSuite, FileChannelTest, testCompress);
+	CppUnit_addTest(pSuite, FileChannelTest, testCompressedRotation);
 	CppUnit_addLongTest(pSuite, FileChannelTest, testPurgeAge);
 	CppUnit_addTest(pSuite, FileChannelTest, testPurgeCount);
 	CppUnit_addTest(pSuite, FileChannelTest, testWrongPurgeOption);

Foundation/testsuite/src/FileChannelTest.h

@@ -44,6 +44,7 @@ class FileChannelTest: public CppUnit::TestCase
 	void testArchive();
 	void testArchiveByStrategy();
 	void testCompress();
+	void testCompressedRotation();
 	void testPurgeAge();
 	void testPurgeCount();
 	void testWrongPurgeOption();