[JENKINS-52165] Stabilize and better test USE_WATCHING

pom.xml

@@ -67,6 +67,7 @@
     <properties>
         <changelist>999999-SNAPSHOT</changelist>
         <jenkins.version>2.361.4</jenkins.version>
+        <jenkins-test-harness.version>2040.v39c760a_432f9</jenkins-test-harness.version> <!-- TODO https://github.com/jenkinsci/jenkins-test-harness/pull/621 -->
         <useBeta>true</useBeta>
         <gitHubRepo>jenkinsci/${project.artifactId}-plugin</gitHubRepo>
         <hpi.compatibleSinceVersion>2.40</hpi.compatibleSinceVersion>
@@ -76,10 +77,21 @@
             <dependency>
                 <groupId>io.jenkins.tools.bom</groupId>
                 <artifactId>bom-2.361.x</artifactId>
-                <version>1654.vcb_69d035fa_20</version>
+                <version>2081.v85885a_d2e5c5</version>
                 <scope>import</scope>
                 <type>pom</type>
             </dependency>
+            <!-- TODO until in BOM -->
+            <dependency>
+                <groupId>org.jenkins-ci.plugins</groupId>
+                <artifactId>durable-task</artifactId>
+                <version>510.v324450f8dca_4</version>
+            </dependency>
+            <dependency>
+                <groupId>org.jenkins-ci.plugins.workflow</groupId>
+                <artifactId>workflow-api</artifactId>
+                <version>1235.v9b_7f964a_a_d70</version> <!-- TODO https://github.com/jenkinsci/workflow-api-plugin/pull/294 -->
+            </dependency>
         </dependencies>
     </dependencyManagement>
     <dependencies>

src/main/java/org/jenkinsci/plugins/workflow/steps/durable_task/DurableTaskStep.java

@@ -30,17 +30,21 @@
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import hudson.AbortException;
 import hudson.EnvVars;
+import hudson.Extension;
 import hudson.FilePath;
 import hudson.Functions;
 import hudson.Launcher;
 import hudson.Util;
 import hudson.init.Terminator;
+import hudson.model.Computer;
 import hudson.model.Node;
 import hudson.model.Result;
 import hudson.model.Run;
 import hudson.model.TaskListener;
 import hudson.remoting.Channel;
 import hudson.remoting.ChannelClosedException;
+import hudson.slaves.ComputerListener;
+import hudson.slaves.OfflineCause;
 import hudson.util.DaemonThreadFactory;
 import hudson.util.FormValidation;
 import hudson.util.LogTaskListener;
@@ -51,8 +55,6 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.PrintStream;
-import java.io.UnsupportedEncodingException;
-import java.lang.reflect.Field;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.util.Set;
@@ -64,6 +66,7 @@
 import java.util.logging.Logger;
 import jenkins.model.Jenkins;
 import jenkins.tasks.filters.EnvVarsFilterableBuilder;
+import jenkins.util.SystemProperties;
 import jenkins.util.Timer;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.StringUtils;
@@ -72,7 +75,9 @@
 import org.jenkinsci.plugins.durabletask.Handler;
 import org.jenkinsci.plugins.workflow.FilePathUtils;
 import org.jenkinsci.plugins.workflow.actions.LabelAction;
+import org.jenkinsci.plugins.workflow.flow.FlowExecutionList;
 import org.jenkinsci.plugins.workflow.graph.FlowNode;
+import org.jenkinsci.plugins.workflow.log.OutputStreamTaskListener;
 import org.jenkinsci.plugins.workflow.steps.AbstractStepExecutionImpl;
 import org.jenkinsci.plugins.workflow.steps.FlowInterruptedException;
 import org.jenkinsci.plugins.workflow.steps.Step;
@@ -203,7 +208,7 @@ interface ExecutionRemotable {
     /** If set to false, disables {@link Execution#watching} mode. */
     @SuppressFBWarnings(value = "MS_SHOULD_BE_FINAL", justification = "public & mutable only for tests")
     @Restricted(NoExternalUse.class)
-    public static boolean USE_WATCHING = Boolean.getBoolean(DurableTaskStep.class.getName() + ".USE_WATCHING"); // JENKINS-52165: turn back on by default
+    public static boolean USE_WATCHING = SystemProperties.getBoolean(DurableTaskStep.class.getName() + ".USE_WATCHING", true);
 
     /** How many seconds to wait before interrupting remote calls and before forcing cleanup when the step is stopped */
     @SuppressFBWarnings(value = "MS_SHOULD_BE_FINAL", justification = "public & mutable for script console access")
@@ -437,26 +442,23 @@ private void getWorkspaceProblem(Exception x) {
         }
 
         /**
-         * Interprets {@link PrintStream#close} as a signal to end a final newline if necessary.
+         * Interprets {@link OutputStream#close} as a signal to end a final newline if necessary.
          */
-        private static final class NewlineSafeTaskListener implements TaskListener {
+        private static final class NewlineSafeTaskListener extends OutputStreamTaskListener.Default {
 
             private static final long serialVersionUID = 1;
 
             private final TaskListener delegate;
-            private transient PrintStream logger;
+            private transient OutputStream out;
 
             NewlineSafeTaskListener(TaskListener delegate) {
                 this.delegate = delegate;
             }
 
             // Similar to DecoratedTaskListener:
-            @NonNull
-            @Override public synchronized PrintStream getLogger() {
-                if (logger == null) {
-                    LOGGER.fine("creating filtered stream");
-                    OutputStream base = delegate.getLogger();
-                    OutputStream filtered = new FilterOutputStream(base) {
+            @Override public synchronized OutputStream getOutputStream() {
+                if (out == null) {
+                    out = new FilterOutputStream(OutputStreamTaskListener.getOutputStream(delegate)) {
                         boolean nl = true; // empty string does not need a newline
                         @Override public void write(int b) throws IOException {
                             super.write(b);
@@ -475,14 +477,12 @@ private static final class NewlineSafeTaskListener implements TaskListener {
                             }
                             flush(); // do *not* call base.close() here, unlike super.close()
                         }
+                        @Override public String toString() {
+                            return "NewlineSafeTaskListener.output[" + out + "]";
+                        }
                     };
-                    try {
-                        logger = new PrintStream(filtered, false, "UTF-8");
-                    } catch (UnsupportedEncodingException x) {
-                        throw new AssertionError(x);
-                    }
                 }
-                return logger;
+                return out;
             }
 
         }
@@ -589,9 +589,12 @@ private void check() {
                     Integer exitCode = controller.exitStatus(workspace, launcher(), listener);
                     if (exitCode == null) {
                         LOGGER.log(Level.FINE, "still running in {0} on {1}", new Object[] {remote, node});
+                    } else if (recurrencePeriod == 0) {
+                        LOGGER.fine(() -> "late check in " + remote + " on " + node + " ignored");
                     } else if (awaitingAsynchExit) {
                         recurrencePeriod = 0;
-                        getContext().onFailure(new AbortException("script apparently exited with code " + exitCode + " but asynchronous notification was lost"));
+                        listener.getLogger().println("script apparently exited with code " + exitCode + " but asynchronous notification was lost");
+                        handleExit(exitCode, () -> controller.getOutput(workspace, launcher()));
                     } else {
                         LOGGER.log(Level.FINE, "exited with {0} in {1} on {2}; expect asynchronous exit soon", new Object[] {exitCode, remote, node});
                         awaitingAsynchExit = true;
@@ -627,6 +630,7 @@ private void check() {
 
         // called remotely from HandlerImpl
         @Override public void exited(int exitCode, byte[] output) throws Exception {
+            recurrencePeriod = 0;
             try {
                 getContext().get(TaskListener.class);
             } catch (IOException | InterruptedException x) {
@@ -643,7 +647,6 @@ private void check() {
                 getContext().onFailure(new IllegalStateException("did not expect output but got some"));
                 return;
             }
-            recurrencePeriod = 0;
             handleExit(exitCode, () -> output);
         }
 
@@ -657,11 +660,11 @@ private void handleExit(int exitCode, OutputSupplier output) throws IOException,
                 getContext().onSuccess(returnStatus ? exitCode : returnStdout ? new String(output.produce(), StandardCharsets.UTF_8) : null);
             } else {
                 if (returnStdout) {
-                    listener().getLogger().write(output.produce()); // diagnostic
+                    _listener().getLogger().write(output.produce()); // diagnostic
                 }
                 if (originalCause != null) {
                     // JENKINS-28822: Use the previous cause instead of throwing a new AbortException
-                    listener().getLogger().println("script returned exit code " + exitCode);
+                    _listener().getLogger().println("script returned exit code " + exitCode);
                     getContext().onFailure(originalCause);
                 } else {
                     getContext().onFailure(new AbortException("script returned exit code " + exitCode));
@@ -694,21 +697,6 @@ private void setupTimer(long initialRecurrencePeriod) {
 
     private static class HandlerImpl extends Handler {
 
-        private static final Field printStreamDelegate;
-        static {
-            try {
-                printStreamDelegate = FilterOutputStream.class.getDeclaredField("out");
-            } catch (NoSuchFieldException x) {
-                // Defined in Java Platform and protected, so should not happen.
-                throw new ExceptionInInitializerError(x);
-            }
-            try {
-                printStreamDelegate.setAccessible(true);
-            } catch (/* TODO Java 11+ InaccessibleObjectException */RuntimeException x) {
-                LOGGER.log(Level.WARNING, "On Java 17 error handling is degraded unless `--add-opens java.base/java.io=ALL-UNNAMED` is passed to the agent", x);
-            }
-        }
-
         private static final long serialVersionUID = 1L;
 
         private final ExecutionRemotable execution;
@@ -721,34 +709,24 @@ private static class HandlerImpl extends Handler {
 
         @Override public void output(@NonNull InputStream stream) throws Exception {
             PrintStream ps = listener.getLogger();
+            OutputStream os = OutputStreamTaskListener.getOutputStream(listener);
             try {
-                if (ps.getClass() == PrintStream.class) {
-                    // Try to extract the underlying stream, since swallowing exceptions is undesirable and PrintStream.checkError is useless.
-                    OutputStream os = ps;
-                    try {
-                        os = (OutputStream) printStreamDelegate.get(ps);
-                    } catch (IllegalAccessException x) {
-                        LOGGER.log(Level.FINE, "using PrintStream rather than underlying FilterOutputStream.out", x);
-                    }
-                    if (os == null) { // like PrintStream.ensureOpen
-                        throw new IOException("Stream closed");
-                    }
-                    synchronized (ps) { // like PrintStream.write overloads do
-                        IOUtils.copy(stream, os);
-                    }
-                } else {
-                    // A subclass. Who knows why, but trust any write(…) overrides it may have.
-                    IOUtils.copy(stream, ps);
+                synchronized (ps) { // like PrintStream.write overloads do
+                    IOUtils.copy(stream, os);
                 }
+                LOGGER.finest(() -> "print to " + os + " succeeded");
             } catch (ChannelClosedException x) {
+                LOGGER.log(Level.FINE, null, x);
                 // We are giving up on this watch. Wait for some call to getWorkspace to rewatch.
                 throw x;
             } catch (Exception x) {
+                LOGGER.log(Level.FINE, null, x);
                 // Try to report it to the controller.
                 try {
                     execution.problem(x);
                     // OK, printed to log on controller side, we may have lost some text but could continue.
                 } catch (Exception x2) { // e.g., RemotingSystemException
+                    LOGGER.log(Level.FINE, null, x2);
                     // No, channel seems to be broken, give up on this watch.
                     throw x;
                 }
@@ -762,4 +740,35 @@ private static class HandlerImpl extends Handler {
 
     }
 
+    @Extension public static final class AgentReconnectionListener extends ComputerListener {
+
+        @Override public void onOffline(Computer c, OfflineCause cause) {
+            if (Jenkins.get().isTerminating()) {
+                LOGGER.fine(() -> "Skipping check on " + c.getName() + " during shutdown");
+                return;
+            }
+            check(c);
+        }
+
+        @Override public void onOnline(Computer c, TaskListener listener) throws IOException, InterruptedException {
+            if (!FlowExecutionList.get().isResumptionComplete()) {
+                LOGGER.fine(() -> "Skipping check on " + c.getName() + " before builds are ready");
+                return;
+            }
+            check(c);
+        }
+
+        private void check(Computer c) {
+            String name = c.getName();
+            StepExecution.applyAll(Execution.class, exec -> {
+                if (exec.watching && exec.node.equals(name)) {
+                    LOGGER.fine(() -> "Online/offline event on " + name + ", checking current status of " + exec.remote + " soon");
+                    threadPool().schedule(exec::check, 15, TimeUnit.SECONDS);
+                }
+                return null;
+            });
+        }
+
+    }
+
 }

src/test/java/org/jenkinsci/plugins/workflow/steps/durable_task/RealShellStepTest.java

@@ -0,0 +1,114 @@
+/*
+ * The MIT License
+ *
+ * Copyright 2023 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.steps.durable_task;
+
+import hudson.ExtensionList;
+import hudson.Functions;
+import hudson.model.ParametersAction;
+import hudson.model.ParametersDefinitionProperty;
+import hudson.model.StringParameterDefinition;
+import hudson.model.StringParameterValue;
+import hudson.slaves.AbstractCloudSlave;
+import hudson.slaves.ComputerListener;
+import java.io.File;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import jenkins.slaves.restarter.JnlpSlaveRestarterInstaller;
+import static org.awaitility.Awaitility.await;
+import org.jenkinsci.plugins.durabletask.FileMonitoringTask;
+import org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition;
+import org.jenkinsci.plugins.workflow.job.WorkflowJob;
+import org.jenkinsci.plugins.workflow.log.FileLogStorage;
+import org.junit.Assume;
+import org.junit.Rule;
+import org.junit.Test;
+import org.jvnet.hudson.test.InboundAgentRule;
+import org.jvnet.hudson.test.JenkinsRule;
+import org.jvnet.hudson.test.PrefixedOutputStream;
+import org.jvnet.hudson.test.RealJenkinsRule;
+import org.jvnet.hudson.test.TailLog;
+
+public final class RealShellStepTest {
+
+    private static final Logger LOGGER = Logger.getLogger(RealShellStepTest.class.getName());
+
+    @Rule public RealJenkinsRule rr = new RealJenkinsRule().
+        withColor(PrefixedOutputStream.Color.BLUE).
+        withLogger(DurableTaskStep.class, Level.FINE).
+        withLogger(FileMonitoringTask.class, Level.FINE);
+
+    @Rule public InboundAgentRule inboundAgents = new InboundAgentRule();
+
+    @Test public void shellScriptExitingAcrossRestart() throws Throwable {
+        Assume.assumeFalse("TODO translate to batch script", Functions.isWindows());
+        rr.startJenkins();
+        rr.runRemotely(RealShellStepTest::disableJnlpSlaveRestarterInstaller);
+        inboundAgents.createAgent(rr, InboundAgentRule.Options.newBuilder().
+            color(PrefixedOutputStream.Color.MAGENTA).
+            label("remote").
+            withLogger(FileMonitoringTask.class, Level.FINER).
+            withLogger(DurableTaskStep.class, Level.FINEST).
+            withPackageLogger(FileLogStorage.class, Level.FINE).
+            build());
+        try (var tailLog = new TailLog(rr, "p", 1).withColor(PrefixedOutputStream.Color.YELLOW)) {
+            rr.runRemotely(RealShellStepTest::shellScriptExitingAcrossRestart1);
+            rr.stopJenkins();
+            var f = new File(rr.getHome(), "f");
+            LOGGER.info(() -> "Waiting for " + f + " to be written…");
+            await().until(f::isFile);
+            LOGGER.info("…done.");
+            rr.startJenkins();
+            rr.runRemotely(RealShellStepTest::shellScriptExitingAcrossRestart2);
+            tailLog.waitForCompletion();
+        }
+    }
+
+    /**
+     * Simulate {@link AbstractCloudSlave} as in https://github.com/jenkinsci/jenkins/pull/7693.
+     * Most users should be using cloud agents,
+     * and this lets us preserve {@link InboundAgentRule.Options.Builder#withLogger(Class, Level)}.
+     */
+    private static void disableJnlpSlaveRestarterInstaller(JenkinsRule r) throws Throwable {
+        ComputerListener.all().remove(ExtensionList.lookupSingleton(JnlpSlaveRestarterInstaller.class));
+    }
+
+    private static void shellScriptExitingAcrossRestart1(JenkinsRule r) throws Throwable {
+        var p = r.createProject(WorkflowJob.class, "p");
+        var f = new File(r.jenkins.getRootDir(), "f");
+        p.addProperty(new ParametersDefinitionProperty(new StringParameterDefinition("F")));
+        p.setDefinition(new CpsFlowDefinition("node('remote') {sh 'sleep 5 && touch \"$F\"'}", true));
+        var b = p.scheduleBuild2(0, new ParametersAction(new StringParameterValue("F", f.getAbsolutePath()))).waitForStart();
+        r.waitForMessage("+ sleep 5", b);
+        r.jenkins.doQuietDown(true, 0, null);
+    }
+
+    private static void shellScriptExitingAcrossRestart2(JenkinsRule r) throws Throwable {
+        var p = (WorkflowJob) r.jenkins.getItem("p");
+        var b = p.getLastBuild();
+        r.assertBuildStatusSuccess(r.waitForCompletion(b));
+        r.assertLogContains("+ touch " + new File(r.jenkins.getRootDir(), "f"), b);
+    }
+
+}