[JENKINS-54073] Buffering for FileLogStorage #81
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| /* | ||
| * The MIT License | ||
| * | ||
| * Copyright 2018 CloudBees, Inc. | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in | ||
| * all copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN | ||
| * THE SOFTWARE. | ||
| */ | ||
|
|
||
| package org.jenkinsci.plugins.workflow.log; | ||
|
|
||
| import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; | ||
| import hudson.CloseProofOutputStream; | ||
| import hudson.model.BuildListener; | ||
| import hudson.remoting.RemoteOutputStream; | ||
| import hudson.util.StreamTaskListener; | ||
| import java.io.Closeable; | ||
| import java.io.IOException; | ||
| import java.io.OutputStream; | ||
| import java.io.PrintStream; | ||
| import org.jenkinsci.remoting.SerializableOnlyOverRemoting; | ||
|
|
||
| /** | ||
| * Unlike {@link StreamTaskListener} this does not set {@code autoflush} on the reconstructed {@link PrintStream}. | ||
| * It also wraps on the remote side in {@link DelayBufferedOutputStream}. | ||
| */ | ||
| final class BufferedBuildListener implements BuildListener, Closeable, SerializableOnlyOverRemoting { | ||
|
|
||
| private final OutputStream out; | ||
| @SuppressFBWarnings(value = "SE_BAD_FIELD", justification = "using Replacement anyway, fields here are irrelevant") | ||
| private final PrintStream ps; | ||
|
|
||
| BufferedBuildListener(OutputStream out) throws IOException { | ||
| this.out = out; | ||
| ps = new PrintStream(out, false, "UTF-8"); | ||
| } | ||
|
|
||
| @Override public PrintStream getLogger() { | ||
| return ps; | ||
| } | ||
|
|
||
| @Override public void close() throws IOException { | ||
| ps.close(); | ||
| } | ||
|
|
||
| private Object writeReplace() { | ||
| return new Replacement(this); | ||
| } | ||
|
|
||
| private static final class Replacement implements SerializableOnlyOverRemoting { | ||
|
|
||
| private static final long serialVersionUID = 1; | ||
|
|
||
| private final RemoteOutputStream ros; | ||
| private final DelayBufferedOutputStream.Tuning tuning = DelayBufferedOutputStream.Tuning.DEFAULT; // load defaults on master | ||
|
|
||
| Replacement(BufferedBuildListener cbl) { | ||
| this.ros = new RemoteOutputStream(new CloseProofOutputStream(cbl.out)); | ||
| } | ||
|
|
||
| private Object readResolve() throws IOException { | ||
| return new BufferedBuildListener(new DelayBufferedOutputStream(ros, tuning)); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| /* | ||
| * The MIT License | ||
| * | ||
| * Copyright 2018 CloudBees, Inc. | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in | ||
| * all copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN | ||
| * THE SOFTWARE. | ||
| */ | ||
|
|
||
| package org.jenkinsci.plugins.workflow.log; | ||
|
|
||
| import java.io.BufferedOutputStream; | ||
| import java.io.FilterOutputStream; | ||
| import java.io.IOException; | ||
| import java.io.OutputStream; | ||
| import java.lang.ref.Reference; | ||
| import java.lang.ref.WeakReference; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.logging.Level; | ||
| import java.util.logging.Logger; | ||
| import jenkins.util.Timer; | ||
| import org.jenkinsci.remoting.SerializableOnlyOverRemoting; | ||
|
|
||
| /** | ||
| * Buffered output stream which is guaranteed to deliver content after some time even if idle and the buffer does not fill up. | ||
| * The automatic “flushing” does <em>not</em> flush the underlying stream, for example via {@code ProxyOutputStream.Flush}. | ||
| */ | ||
| final class DelayBufferedOutputStream extends BufferedOutputStream { | ||
|
|
||
| private static final Logger LOGGER = Logger.getLogger(DelayBufferedOutputStream.class.getName()); | ||
|
|
||
| static final class Tuning implements SerializableOnlyOverRemoting { | ||
|
There was a problem hiding this comment. 👍 👍 for making it configurable |
||
| private Tuning() {} | ||
| // nonfinal for Groovy scripting: | ||
| long minRecurrencePeriod = Long.getLong(DelayBufferedOutputStream.class.getName() + ".minRecurrencePeriod", 1_000); // 1s | ||
| long maxRecurrencePeriod = Long.getLong(DelayBufferedOutputStream.class.getName() + ".maxRecurrencePeriod", 10_000); // 10s | ||
| float recurrencePeriodBackoff = Float.parseFloat(System.getProperty(DelayBufferedOutputStream.class.getName() + ".recurrencePeriodBackoff", "1.05")); | ||
|
There was a problem hiding this comment. @jglick I think maybe you meant to make the backoff 1.5 rather than 1.05? With 1.05 we'll flush 75 times before we hit the maximum backoff (on run 76), whereas a backoff of 1.5 it's about 9 cycles (enough to give you just under 30s of fast-flushing logs). The 1.5 constant matches the behavior you seemed to want here, where the 1.05 would still generate a lot of log interleaving (and a big index file) with slow-writing logs. There was a problem hiding this comment. No, 1.05 was intentional. |
||
| int bufferSize = Integer.getInteger(DelayBufferedOutputStream.class.getName() + ".bufferSize", 1 << 16); // 64Kib | ||
|
There was a problem hiding this comment. Note that link layer packets are much smaller than this typically. However a larger buffer (divided into packets by TCP) means less overhead in the Remoting layer, for example in (de)serialization. |
||
| static final Tuning DEFAULT = new Tuning(); | ||
| } | ||
|
|
||
| private final Tuning tuning; | ||
| private long recurrencePeriod; | ||
|
|
||
| DelayBufferedOutputStream(OutputStream out) { | ||
| this(out, Tuning.DEFAULT); | ||
| } | ||
|
|
||
| DelayBufferedOutputStream(OutputStream out, Tuning tuning) { | ||
| super(new FlushControlledOutputStream(out), tuning.bufferSize); | ||
| this.tuning = tuning; | ||
| recurrencePeriod = tuning.minRecurrencePeriod; | ||
| reschedule(); | ||
| } | ||
|
|
||
| private void reschedule() { | ||
|
There was a problem hiding this comment. This feels quite over-complicated to me for what we're trying to do, and that's concerning when combined with our overall logging complexity. There was a problem hiding this comment. Can you be more specific please about what is more complicated than it could be? There was a problem hiding this comment. It makes more sense when we consider the user-tailing case (maybe overthinking it a bit, but okay). |
||
| Timer.get().schedule(new Flush(this), recurrencePeriod, TimeUnit.MILLISECONDS); | ||
| recurrencePeriod = Math.min((long) (recurrencePeriod * tuning.recurrencePeriodBackoff), tuning.maxRecurrencePeriod); | ||
| } | ||
|
|
||
| /** We can only call {@link BufferedOutputStream#flushBuffer} via {@link #flush}, but we do not wish to flush the underlying stream, only write out the buffer. */ | ||
| private void flushBuffer() throws IOException { | ||
| ThreadLocal<Boolean> enableFlush = ((FlushControlledOutputStream) out).enableFlush; | ||
| boolean orig = enableFlush.get(); | ||
| enableFlush.set(false); | ||
| try { | ||
| flush(); | ||
| } finally { | ||
| enableFlush.set(orig); | ||
| } | ||
| } | ||
|
|
||
| void flushAndReschedule() { | ||
| // TODO as an optimization, avoid flushing the buffer if it was recently flushed anyway due to filling up | ||
| try { | ||
| flushBuffer(); | ||
| } catch (IOException x) { | ||
| LOGGER.log(Level.FINE, null, x); | ||
| } | ||
| reschedule(); | ||
| } | ||
|
|
||
| @SuppressWarnings("FinalizeDeclaration") // not ideal, but PhantomReference is more of a hassle | ||
| @Override protected void finalize() throws Throwable { | ||
| super.finalize(); | ||
| // Odd that this is not the default behavior for BufferedOutputStream. | ||
| flush(); | ||
| } | ||
|
|
||
| private static final class Flush implements Runnable { | ||
|
|
||
| /** Since there is no explicit close event, just keep flushing periodically until the stream is collected. */ | ||
| private final Reference<DelayBufferedOutputStream> osr; | ||
|
|
||
| Flush(DelayBufferedOutputStream os) { | ||
| osr = new WeakReference<>(os); | ||
| } | ||
|
|
||
| @Override public void run() { | ||
| DelayBufferedOutputStream os = osr.get(); | ||
| if (os != null) { | ||
| os.flushAndReschedule(); | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| /** @see DelayBufferedOutputStream#flushBuffer */ | ||
| private static final class FlushControlledOutputStream extends FilterOutputStream { | ||
|
|
||
| private final ThreadLocal<Boolean> enableFlush = new ThreadLocal<Boolean>() { | ||
| @Override protected Boolean initialValue() { | ||
| return true; | ||
| } | ||
| }; | ||
|
|
||
| FlushControlledOutputStream(OutputStream out) { | ||
| super(out); | ||
| } | ||
|
|
||
| @Override public void write(byte[] b, int off, int len) throws IOException { | ||
| out.write(b, off, len); // super method writes one byte at a time! | ||
| } | ||
|
|
||
| @Override public void flush() throws IOException { | ||
| if (enableFlush.get()) { | ||
| super.flush(); | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Also important: the
false. Compare this.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For reference: jenkinsci/jenkins#3703