avoid duplicate sending and global notifier mode

src/main/java/jenkins/plugins/logstash/LogstashConsoleLogFilter.java

@@ -8,7 +8,6 @@
 
 import hudson.Extension;
 import hudson.console.ConsoleLogFilter;
-import hudson.model.AbstractBuild;
 import hudson.model.AbstractProject;
 import hudson.model.Run;
 
@@ -19,59 +18,100 @@ public class LogstashConsoleLogFilter extends ConsoleLogFilter implements Serial
   private static final Logger LOGGER = Logger.getLogger(LogstashConsoleLogFilter.class.getName());
 
   private transient Run<?, ?> run;
+
   public LogstashConsoleLogFilter() {}
 
   public LogstashConsoleLogFilter(Run<?, ?> run)
   {
     this.run = run;
   }
+
   private static final long serialVersionUID = 1L;
 
+  /**
+   * {@inheritDoc}
+   *
+   * Currently pipeline is not supporting global ConsoleLogFilters. So even when we have it enabled globally
+   * we would not log to an indexer without the logstash step.
+   * But when pipeline supports global ConsoleLogFilters we don't want to log twice when we have the step in
+   * the pipeline.
+   * With the LogstashMarkerRunAction we can detect if it is enabled globally and if pipeline is supporting
+   * global ConsoleLogFilters.
+   * The LogstashMarkerRunAction will be only attached to a WorkflowRun when pipeline supports global
+   * ConsoleLogFilters (JENKINS-45693). And the assumption is that the marker action is attached before the
+   * logstashStep is initialized.
+   * This marker action will also disable the notifier.
+   *
+   */
   @Override
   public OutputStream decorateLogger(Run build, OutputStream logger) throws IOException, InterruptedException
   {
+
     LogstashConfiguration configuration = LogstashConfiguration.getInstance();
     if (!configuration.isEnabled())
     {
       LOGGER.log(Level.FINE, "Logstash is disabled. Logs will not be forwarded.");
       return logger;
     }
 
-    if (build != null && build instanceof AbstractBuild<?, ?>)
+
+    // A pipeline step uses the constructor which sets run.
+    if (run != null)
     {
-      if (isLogstashEnabled(build))
-      {
-        LogstashWriter logstash = getLogStashWriter(build, logger);
-        return new LogstashOutputStream(logger, logstash);
-      }
-      else
+      if (run.getAction(LogstashMarkerAction.class) != null)
       {
+        LOGGER.log(Level.FINEST, "Logstash is enabled globally. No need to decorate the logger another time for {0}",
+            run.toString());
         return logger;
       }
+      return getLogstashOutputStream(run, logger);
     }
-    if (run != null)
-    {
-      LogstashWriter logstash = getLogStashWriter(run, logger);
-      return new LogstashOutputStream(logger, logstash);
-    }
-    else
+
+    // Not pipeline step so @{code build} should be set.
+    if (isLogstashEnabled(build))
     {
-      return logger;
+      if (build.getAction(LogstashMarkerAction.class) == null)
+      {
+        build.addAction(new LogstashMarkerAction());
+      }
+      return getLogstashOutputStream(build, logger);
     }
+
+    return logger;
+  }
+
+  private LogstashOutputStream getLogstashOutputStream(Run<?, ?> run, OutputStream logger)
+  {
+    LogstashWriter logstash = getLogStashWriter(run, logger);
+    return new LogstashOutputStream(logger, logstash);
   }
 
   LogstashWriter getLogStashWriter(Run<?, ?> build, OutputStream errorStream)
   {
     return new LogstashWriter(build, errorStream, null, build.getCharset());
   }
 
-  private boolean isLogstashEnabled(Run<?, ?> build)
+  private boolean isLogstashEnabledGlobally()
   {
     LogstashConfiguration configuration = LogstashConfiguration.getInstance();
     if (configuration.isEnableGlobally())
     {
       return true;
     }
+    return false;
+  }
+
+  private boolean isLogstashEnabled(Run<?, ?> build)
+  {
+    if (build == null)
+    {
+      return false;
+    }
+
+    if (isLogstashEnabledGlobally())
+    {
+      return true;
+    }
 
     if (build.getParent() instanceof AbstractProject)
     {
@@ -83,5 +123,4 @@ private boolean isLogstashEnabled(Run<?, ?> build)
     }
     return false;
   }
-
 }

src/main/java/jenkins/plugins/logstash/LogstashMarkerAction.java

@@ -0,0 +1,17 @@
+package jenkins.plugins.logstash;
+
+import org.kohsuke.accmod.Restricted;
+import org.kohsuke.accmod.restrictions.NoExternalUse;
+
+import hudson.Extension;
+import hudson.model.InvisibleAction;
+
+/*
+ * A marker for Runs if logstash is enabled globally or set via JobProperty
+ * When enabled globally or per JobProperty it doesn't make sense to send the data another time via a logstash step in a pipeline or
+ * the logstashSend notifier
+ */
+@Extension
+@Restricted(NoExternalUse.class)
+public class LogstashMarkerAction extends InvisibleAction
+{}

src/main/java/jenkins/plugins/logstash/LogstashNotifier.java

@@ -99,6 +99,12 @@ private boolean perform(Run<?, ?> run, TaskListener listener) {
       return true;
     }
 
+    // globally enabled and active. No need to send the data another time.
+    if (run.getAction(LogstashMarkerAction.class) != null)
+    {
+      return true;
+    }
+
     PrintStream errorPrintStream = listener.getLogger();
     LogstashWriter logstash = getLogStashWriter(run, errorPrintStream, listener);
     logstash.writeBuildLog(maxLines);

src/test/java/jenkins/plugins/logstash/LogstashIntegrationTest.java

@@ -284,4 +284,43 @@ public void disabledWillNotWrite() throws Exception
       List<JSONObject> dataLines = memoryDao.getOutput();
       assertThat(dataLines.size(), equalTo(0));
     }
+
+    public void enabledGloballyNotifierWillNotSend() throws Exception
+    {
+      LogstashConfiguration.getInstance().setEnableGlobally(true);
+      project.getPublishersList().add(new LogstashNotifier(10, false));
+
+      QueueTaskFuture<FreeStyleBuild> f = project.scheduleBuild2(0);
+
+      FreeStyleBuild build = f.get();
+      assertThat(build.getResult(), equalTo(Result.SUCCESS));
+      List<JSONObject> dataLines = memoryDao.getOutput();
+      assertThat(dataLines.size(), is(4));
+      JSONObject firstLine = dataLines.get(0);
+      JSONObject lastLine = dataLines.get(dataLines.size()-1);
+      JSONObject data = firstLine.getJSONObject("data");
+      assertThat(data.getString("buildHost"),equalTo("Jenkins"));
+      assertThat(data.getString("buildLabel"),equalTo("master"));
+      assertThat(lastLine.getJSONArray("message").get(0).toString(),equalTo("Finished: SUCCESS"));
+    }
+
+    @Test
+    public void enabledJobPropertyNotifierWillNotSend() throws Exception
+    {
+      project.addProperty(new LogstashJobProperty());
+      project.getPublishersList().add(new LogstashNotifier(10, false));
+
+      QueueTaskFuture<FreeStyleBuild> f = project.scheduleBuild2(0);
+
+      FreeStyleBuild build = f.get();
+      assertThat(build.getResult(), equalTo(Result.SUCCESS));
+      List<JSONObject> dataLines = memoryDao.getOutput();
+      assertThat(dataLines.size(), is(4));
+      JSONObject firstLine = dataLines.get(0);
+      JSONObject lastLine = dataLines.get(dataLines.size()-1);
+      JSONObject data = firstLine.getJSONObject("data");
+      assertThat(data.getString("buildHost"),equalTo("Jenkins"));
+      assertThat(data.getString("buildLabel"),equalTo("master"));
+      assertThat(lastLine.getJSONArray("message").get(0).toString(),equalTo("Finished: SUCCESS"));
+    }
 }

src/test/java/jenkins/plugins/logstash/LogstashNotifierTest.java

@@ -9,6 +9,7 @@
 import hudson.model.BuildListener;
 import hudson.model.AbstractBuild;
 import hudson.model.AbstractProject;
+import hudson.model.Action;
 import hudson.model.TaskListener;
 import hudson.model.Run;
 import hudson.model.Result;
@@ -62,6 +63,7 @@ LogstashWriter getLogStashWriter(Run<?, ?> run, OutputStream errorStream, TaskLi
   }
 
   static class MockRun<P extends AbstractProject<P,R>, R extends AbstractBuild<P,R>> extends Run<P,R> {
+
     Result result;
 
     MockRun(P job) throws IOException {
@@ -136,6 +138,7 @@ public void performSuccess() throws Exception {
     verify(mockListener).getLogger();
     verify(mockWriter).writeBuildLog(3);
     verify(mockWriter).isConnectionBroken();
+    verify(mockBuild).getAction(LogstashMarkerAction.class);
 
     assertEquals("Errors were written", "", errorBuffer.toString());
 
@@ -172,6 +175,7 @@ public void performBadWriterDoNotFailBuild() throws Exception {
     verify(mockListener).getLogger();
     verify(mockWriter).writeBuildLog(3);
     verify(mockWriter).isConnectionBroken();
+    verify(mockBuild).getAction(LogstashMarkerAction.class);
 
     assertEquals("Error was not written", "Mocked Constructor failure", errorBuffer.toString());
   }
@@ -213,6 +217,7 @@ public void performBadWriterDoFailBuild() throws Exception {
     verify(mockListener).getLogger();
     verify(mockWriter).writeBuildLog(3);
     verify(mockWriter, times(2)).isConnectionBroken();
+    verify(mockBuild).getAction(LogstashMarkerAction.class);
     assertEquals("Error was not written", "Mocked Constructor failure", errorBuffer.toString());
   }
 
@@ -264,6 +269,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
     verify(mockListener).getLogger();
     verify(mockWriter).writeBuildLog(3);
     verify(mockWriter, times(3)).isConnectionBroken();
+    verify(mockBuild).getAction(LogstashMarkerAction.class);
 
     assertThat("Wrong error message", errorBuffer.toString(), containsString(errorMsg));
   }
@@ -286,6 +292,7 @@ public void performAllLines() throws Exception {
     verify(mockListener).getLogger();
     verify(mockWriter).writeBuildLog(-1);
     verify(mockWriter, times(2)).isConnectionBroken();
+    verify(mockBuild).getAction(LogstashMarkerAction.class);
 
     assertEquals("Errors were written", "", errorBuffer.toString());
   }
@@ -308,6 +315,7 @@ public void performZeroLines() throws Exception {
     verify(mockListener).getLogger();
     verify(mockWriter).writeBuildLog(0);
     verify(mockWriter, times(2)).isConnectionBroken();
+    verify(mockBuild).getAction(LogstashMarkerAction.class);
 
     assertEquals("Errors were written", "", errorBuffer.toString());
   }