diff options
3 files changed, 107 insertions, 80 deletions
diff --git a/jdisc_http_service/src/main/java/com/yahoo/container/logging/JSONAccessLog.java b/jdisc_http_service/src/main/java/com/yahoo/container/logging/JSONAccessLog.java index 2cea7f7dd04..ece9d0d2c4a 100644 --- a/jdisc_http_service/src/main/java/com/yahoo/container/logging/JSONAccessLog.java +++ b/jdisc_http_service/src/main/java/com/yahoo/container/logging/JSONAccessLog.java @@ -1,6 +1,7 @@ // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.container.logging; +import com.yahoo.component.AbstractComponent; import com.yahoo.container.core.AccessLogConfig; /** @@ -9,7 +10,7 @@ import com.yahoo.container.core.AccessLogConfig; * @author frodelu * @author Tony Vaagenes */ -public final class JSONAccessLog implements RequestLogHandler { +public final class JSONAccessLog extends AbstractComponent implements RequestLogHandler { private final AccessLogHandler logHandler; @@ -22,8 +23,5 @@ public final class JSONAccessLog implements RequestLogHandler { logHandler.log(entry); } - // TODO: This is never called. We should have a DI provider and call this method from its deconstruct. - public void shutdown() { - logHandler.shutdown(); - } + @Override public void deconstruct() { logHandler.shutdown(); } } diff --git a/jdisc_http_service/src/main/java/com/yahoo/container/logging/LogFileHandler.java b/jdisc_http_service/src/main/java/com/yahoo/container/logging/LogFileHandler.java index 45af0f53bcc..c3c7bb64dc7 100644 --- a/jdisc_http_service/src/main/java/com/yahoo/container/logging/LogFileHandler.java +++ b/jdisc_http_service/src/main/java/com/yahoo/container/logging/LogFileHandler.java @@ -9,11 +9,14 @@ import com.yahoo.protect.Process; import com.yahoo.system.ProcessExecuter; import com.yahoo.yolean.Exceptions; +import java.io.BufferedOutputStream; import java.io.File; +import java.io.FileDescriptor; 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.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; @@ -25,7 +28,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import java.util.function.Supplier; import java.util.logging.Level; import java.util.logging.Logger; import java.util.zip.GZIPOutputStream; @@ -182,11 +184,10 @@ class LogFileHandler <LOGTYPE> { static class LogThread<LOGTYPE> extends Thread { private final Pollable<LOGTYPE> operationProvider; long lastFlush = 0; - private FileOutputStream currentOutputStream = null; + private PageCacheFriendlyFileOutputStream fileOutput = null; private long nextRotationTime = 0; private final String filePattern; // default to current directory, ms time stamp private volatile String fileName; - private long lastDropPosition = 0; private final LogWriter<LOGTYPE> logWriter; private final Compression compression; private final long[] rotationTimes; @@ -253,32 +254,25 @@ class LogFileHandler <LOGTYPE> { } } - private synchronized void internalFlush() { + private void internalFlush() { try { - FileOutputStream currentOut = this.currentOutputStream; - if (currentOut != null) { - if (compression == Compression.GZIP) { - long newPos = currentOut.getChannel().position(); - if (newPos > lastDropPosition + 102400) { - nativeIO.dropPartialFileFromCache(currentOut.getFD(), lastDropPosition, newPos, true); - lastDropPosition = newPos; - } - } else { - currentOut.flush(); - } + if (fileOutput != null) { + fileOutput.flush(); } } catch (IOException e) { - logger.warning("Failed dropping from cache : " + Exceptions.toMessageString(e)); + logger.log(Level.WARNING, "Failed to flush file output: " + Exceptions.toMessageString(e), e); } } private void internalClose() { try { - internalFlush(); - FileOutputStream currentOut = this.currentOutputStream; - if (currentOut != null) currentOut.close(); + if (fileOutput != null) { + fileOutput.flush(); + fileOutput.close(); + fileOutput = null; + } } catch (Exception e) { - logger.log(Level.WARNING, "Got error while closing log file", e); + logger.log(Level.WARNING, "Got error while closing log file: " + e.getMessage(), e); } } @@ -290,13 +284,12 @@ class LogFileHandler <LOGTYPE> { if (nextRotationTime <= 0) { nextRotationTime = getNextRotationTime(now); // lazy initialization } - if (now > nextRotationTime || currentOutputStream == null) { + if (now > nextRotationTime || fileOutput == null) { internalRotateNow(); } try { - FileOutputStream out = this.currentOutputStream; - logWriter.write(r, out); - out.write('\n'); + logWriter.write(r, fileOutput); + fileOutput.write('\n'); } catch (IOException e) { logger.warning("Failed writing log record: " + Exceptions.toMessageString(e)); } @@ -343,18 +336,14 @@ class LogFileHandler <LOGTYPE> { // isInterrupted() returns false after interruption in p.waitFor private void internalRotateNow() { // figure out new file name, then - // use super.setOutputStream to switch to a new file String oldFileName = fileName; long now = System.currentTimeMillis(); fileName = LogFormatter.insertDate(filePattern, now); - internalFlush(); - + internalClose(); try { checkAndCreateDir(fileName); - FileOutputStream os = new FileOutputStream(fileName, true); // append mode, for safety - currentOutputStream = os; - lastDropPosition = 0; + fileOutput = new PageCacheFriendlyFileOutputStream(nativeIO, Paths.get(fileName), 4 * 1024 * 1024); LogFileDb.nowLoggingTo(fileName); } catch (IOException e) { throw new RuntimeException("Couldn't open log file '" + fileName + "'", e); @@ -364,72 +353,85 @@ class LogFileHandler <LOGTYPE> { nextRotationTime = 0; //figure it out later (lazy evaluation) if ((oldFileName != null)) { - File oldFile = new File(oldFileName); - if (oldFile.exists()) { - if (compression != Compression.NONE) { - executor.execute(() -> runCompression(oldFile, compression)); - } else { - nativeIO.dropFileFromCache(oldFile); - } + Path oldFile = Paths.get(oldFileName); + if (Files.exists(oldFile)) { + executor.execute(() -> runCompression(nativeIO, oldFile, compression)); } } } - private static void runCompression(File oldFile, Compression compression) { + private static void runCompression(NativeIO nativeIO, Path oldFile, Compression compression) { switch (compression) { case ZSTD: - runCompressionZstd(oldFile.toPath()); + runCompressionZstd(nativeIO, oldFile); break; case GZIP: - runCompressionGzip(oldFile); + runCompressionGzip(nativeIO, oldFile); + break; + case NONE: + runCompressionNone(nativeIO, oldFile); break; default: throw new IllegalArgumentException("Unknown compression " + compression); } } - private static void runCompressionZstd(Path oldFile) { + private static void runCompressionNone(NativeIO nativeIO, Path oldFile) { + nativeIO.dropFileFromCache(oldFile.toFile()); + } + + private static void runCompressionZstd(NativeIO nativeIO, Path oldFile) { try { Path compressedFile = Paths.get(oldFile.toString() + ".zst"); Files.createFile(compressedFile); int bufferSize = 0x400000; // 4M - byte[] buffer = new byte[bufferSize]; - try (ZstdOuputStream out = new ZstdOuputStream(Files.newOutputStream(compressedFile), bufferSize); - InputStream in = Files.newInputStream(oldFile)) { - int read; - while ((read = in.read(buffer)) >= 0) { - out.write(buffer, 0, read); - } + try (FileOutputStream fileOut = new FileOutputStream(compressedFile.toFile()); + ZstdOuputStream out = new ZstdOuputStream(fileOut, bufferSize); + FileInputStream in = new FileInputStream(oldFile.toFile())) { + pageFriendlyTransfer(nativeIO, out, fileOut.getFD(), in, bufferSize); out.flush(); } Files.delete(oldFile); + nativeIO.dropFileFromCache(compressedFile.toFile()); } catch (IOException e) { logger.log(Level.WARNING, "Failed to compress log file with zstd: " + oldFile, e); + } finally { + nativeIO.dropFileFromCache(oldFile.toFile()); } } - private static void runCompressionGzip(File oldFile) { - File gzippedFile = new File(oldFile.getPath() + ".gz"); - NativeIO nativeIO = new NativeIO(); - try (GZIPOutputStream compressor = new GZIPOutputStream(new FileOutputStream(gzippedFile), 0x100000); - FileInputStream inputStream = new FileInputStream(oldFile)) { - byte[] buffer = new byte[0x400000]; // 4M buffer - - long totalBytesRead = 0; - for (int read = inputStream.read(buffer); read > 0; read = inputStream.read(buffer)) { - compressor.write(buffer, 0, read); - nativeIO.dropPartialFileFromCache(inputStream.getFD(), totalBytesRead, read, false); - totalBytesRead += read; + private static void runCompressionGzip(NativeIO nativeIO, Path oldFile) { + try { + Path gzippedFile = Paths.get(oldFile.toString() + ".gz"); + try (FileOutputStream fileOut = new FileOutputStream(gzippedFile.toFile()); + GZIPOutputStream compressor = new GZIPOutputStream(fileOut, 0x100000); + FileInputStream inputStream = new FileInputStream(oldFile.toFile())) { + pageFriendlyTransfer(nativeIO, compressor, fileOut.getFD(), inputStream, 0x400000); + compressor.finish(); + compressor.flush(); } - compressor.finish(); - compressor.flush(); - + Files.delete(oldFile); + nativeIO.dropFileFromCache(gzippedFile.toFile()); } catch (IOException e) { - logger.warning("Got '" + e + "' while compressing '" + oldFile.getPath() + "'."); + logger.log(Level.WARNING, "Failed to compress log file with gzip: " + oldFile, e); + } finally { + nativeIO.dropFileFromCache(oldFile.toFile()); + } + } + + private static void pageFriendlyTransfer(NativeIO nativeIO, OutputStream out, FileDescriptor outDescriptor, FileInputStream in, int bufferSize) throws IOException { + int read; + long totalBytesRead = 0; + byte[] buffer = new byte[bufferSize]; + while ((read = in.read(buffer)) >= 0) { + out.write(buffer, 0, read); + if (read > 0) { + nativeIO.dropPartialFileFromCache(in.getFD(), totalBytesRead, read, false); + nativeIO.dropPartialFileFromCache(outDescriptor, totalBytesRead, read, false); + } + totalBytesRead += read; } - oldFile.delete(); - nativeIO.dropFileFromCache(gzippedFile); } /** @@ -482,5 +484,37 @@ class LogFileHandler <LOGTYPE> { this.log = log; } } + + /** File output stream that signals to kernel to drop previous pages after write */ + private static class PageCacheFriendlyFileOutputStream extends OutputStream { + + private final NativeIO nativeIO; + private final FileOutputStream fileOut; + private final BufferedOutputStream bufferedOut; + private final int bufferSize; + private long lastDropPosition = 0; + + PageCacheFriendlyFileOutputStream(NativeIO nativeIO, Path file, int bufferSize) throws FileNotFoundException { + this.nativeIO = nativeIO; + this.fileOut = new FileOutputStream(file.toFile(), true); + this.bufferedOut = new BufferedOutputStream(fileOut, bufferSize); + this.bufferSize = bufferSize; + } + + @Override public void write(byte[] b) throws IOException { bufferedOut.write(b); } + @Override public void write(byte[] b, int off, int len) throws IOException { bufferedOut.write(b, off, len); } + @Override public void write(int b) throws IOException { bufferedOut.write(b); } + @Override public void close() throws IOException { bufferedOut.close(); } + + @Override + public void flush() throws IOException { + bufferedOut.flush(); + long newPos = fileOut.getChannel().position(); + if (newPos >= lastDropPosition + bufferSize) { + nativeIO.dropPartialFileFromCache(fileOut.getFD(), lastDropPosition, newPos, true); + lastDropPosition = newPos; + } + } + } } diff --git a/jdisc_http_service/src/main/java/com/yahoo/container/logging/VespaAccessLog.java b/jdisc_http_service/src/main/java/com/yahoo/container/logging/VespaAccessLog.java index 33ec736443c..254b7fe5385 100644 --- a/jdisc_http_service/src/main/java/com/yahoo/container/logging/VespaAccessLog.java +++ b/jdisc_http_service/src/main/java/com/yahoo/container/logging/VespaAccessLog.java @@ -1,6 +1,7 @@ // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root. package com.yahoo.container.logging; +import com.yahoo.component.AbstractComponent; import com.yahoo.container.core.AccessLogConfig; import java.io.IOException; @@ -14,7 +15,7 @@ import java.util.TimeZone; * @author Bjorn Borud * @author Oyvind Bakksjo */ -public final class VespaAccessLog implements RequestLogHandler, LogWriter<RequestLogEntry> { +public final class VespaAccessLog extends AbstractComponent implements RequestLogHandler, LogWriter<RequestLogEntry> { private static final ThreadLocal<SimpleDateFormat> dateFormat = ThreadLocal.withInitial(VespaAccessLog::createDateFormat); @@ -84,13 +85,7 @@ public final class VespaAccessLog implements RequestLogHandler, LogWriter<Reques sb.append(numbers); } - /** - * TODO: This is never called. We should have a DI provider and call this method from its deconstruct. - */ - public void shutdown() { - if (logHandler!=null) - logHandler.shutdown(); - } + @Override public void deconstruct() { logHandler.shutdown(); } @Override public void log(RequestLogEntry entry) { |