add logstash DAO

pom.xml

@@ -44,7 +44,7 @@
 
   <artifactId>logstash</artifactId>
   <packaging>hpi</packaging>
-  <version>1.2.1-SNAPSHOT</version>
+  <version>1.3.0-SNAPSHOT</version>
   <name>Logstash</name>
   <description>A Logstash agent to send Jenkins logs to a Logstash indexer.</description>
   <url>http://wiki.jenkins-ci.org/display/JENKINS/Logstash+Plugin</url>
@@ -108,6 +108,12 @@
       <artifactId>junit</artifactId>
       <version>1.10</version>
     </dependency>
+
+    <dependency>
+      <groupId>org.jenkins-ci.plugins</groupId>
+      <artifactId>structs</artifactId>
+      <version>1.2</version>
+    </dependency>
   </dependencies>
 
   <build>

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

@@ -108,7 +108,7 @@ public DescriptorImpl getDescriptor() {
 
   // Method to encapsulate calls for unit-testing
   LogstashWriter getLogStashWriter(AbstractBuild<?, ?> build, OutputStream errorStream) {
-    return new LogstashWriter(build, errorStream);
+    return new LogstashWriter(build, errorStream, null);
   }
 
   /**

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

@@ -26,26 +26,33 @@
 
 import hudson.Extension;
 import hudson.Launcher;
+import hudson.FilePath;
 import hudson.model.BuildListener;
+import hudson.model.TaskListener;
 import hudson.model.AbstractBuild;
+import hudson.model.Run;
 import hudson.model.AbstractProject;
+import hudson.model.Result;
 import hudson.tasks.BuildStepDescriptor;
 import hudson.tasks.BuildStepMonitor;
 import hudson.tasks.Notifier;
 import hudson.tasks.Publisher;
 
+import jenkins.tasks.SimpleBuildStep;
 import java.io.OutputStream;
 import java.io.PrintStream;
+import java.io.IOException;
 
 import org.kohsuke.stapler.DataBoundConstructor;
+import org.jenkinsci.Symbol;
 
 /**
  * Post-build action to push build log to Logstash.
  *
  * @author Rusty Gerard
  * @since 1.0.0
  */
-public class LogstashNotifier extends Notifier {
+public class LogstashNotifier extends Notifier implements SimpleBuildStep {
 
   public int maxLines;
   public boolean failBuild;
@@ -58,16 +65,27 @@ public LogstashNotifier(int maxLines, boolean failBuild) {
 
   @Override
   public boolean perform(AbstractBuild<?, ?> build, Launcher launcher, BuildListener listener) {
+    return perform(build, listener);
+  }
+
+  @Override
+  public void perform(Run<?, ?> run, FilePath workspace, Launcher launcher,
+    TaskListener listener) throws InterruptedException, IOException {
+    if (!perform(run, listener)) {
+      run.setResult(Result.FAILURE);
+    }
+  }
+
+  private boolean perform(Run<?, ?> run, TaskListener listener) {
     PrintStream errorPrintStream = listener.getLogger();
-    LogstashWriter logstash = getLogStashWriter(build, errorPrintStream);
+    LogstashWriter logstash = getLogStashWriter(run, errorPrintStream, listener);
     logstash.writeBuildLog(maxLines);
-
     return !(failBuild && logstash.isConnectionBroken());
   }
 
   // Method to encapsulate calls for unit-testing
-  LogstashWriter getLogStashWriter(AbstractBuild<?, ?> build, OutputStream errorStream) {
-    return new LogstashWriter(build, errorStream);
+  LogstashWriter getLogStashWriter(Run<?, ?> run, OutputStream errorStream, TaskListener listener) {
+    return new LogstashWriter(run, errorStream, listener);
   }
 
   public BuildStepMonitor getRequiredMonitorService() {
@@ -80,7 +98,7 @@ public Descriptor getDescriptor() {
     return (Descriptor) super.getDescriptor();
   }
 
-  @Extension
+  @Extension @Symbol("logstashSend")
   public static class Descriptor extends BuildStepDescriptor<Publisher> {
 
     @Override

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

@@ -26,6 +26,8 @@
 
 
 import hudson.model.AbstractBuild;
+import hudson.model.TaskListener;
+import hudson.model.Run;
 import jenkins.model.Jenkins;
 import jenkins.plugins.logstash.persistence.BuildData;
 import jenkins.plugins.logstash.persistence.IndexerDaoFactory;
@@ -52,15 +54,17 @@
 public class LogstashWriter {
 
   final OutputStream errorStream;
-  final AbstractBuild<?, ?> build;
+  final Run<?, ?> build;
+  final TaskListener listener;
   final BuildData buildData;
   final String jenkinsUrl;
   final LogstashIndexerDao dao;
   private boolean connectionBroken;
 
-  public LogstashWriter(AbstractBuild<?, ?> build, OutputStream error) {
+  public LogstashWriter(Run<?, ?> run, OutputStream error, TaskListener listener) {
     this.errorStream = error != null ? error : System.err;
-    this.build = build;
+    this.build = run;
+    this.listener = listener;
     this.dao = this.getDaoOrNull();
     if (this.dao == null) {
       this.jenkinsUrl = "";
@@ -69,7 +73,6 @@ public LogstashWriter(AbstractBuild<?, ?> build, OutputStream error) {
       this.jenkinsUrl = getJenkinsUrl();
       this.buildData = getBuildData();
     }
-
   }
 
   /**
@@ -131,7 +134,11 @@ LogstashIndexerDao getDao() throws InstantiationException {
   }
 
   BuildData getBuildData() {
-    return new BuildData(build, new Date());
+    if (build instanceof AbstractBuild) {
+      return new BuildData((AbstractBuild) build, new Date());
+    } else {
+      return new BuildData(build, new Date(), listener);
+    }
   }
 
   String getJenkinsUrl() {

src/main/java/jenkins/plugins/logstash/persistence/BuildData.java

@@ -28,7 +28,10 @@
 import hudson.model.Environment;
 import hudson.model.Result;
 import hudson.model.AbstractBuild;
+import hudson.model.TaskListener;
+import hudson.model.Run;
 import hudson.model.Node;
+import hudson.model.Executor;
 import hudson.tasks.test.AbstractTestResultAction;
 import hudson.tasks.test.TestResult;
 
@@ -42,6 +45,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.io.IOException;
 
 import net.sf.json.JSONObject;
 
@@ -112,19 +116,9 @@ public TestData(Action action) {
 
   BuildData() {}
 
+  // Freestyle project build
   public BuildData(AbstractBuild<?, ?> build, Date currentTime) {
-    result = build.getResult() == null ? null : build.getResult().toString();
-    id = build.getId();
-    projectName = build.getProject().getName();
-    displayName = build.getDisplayName();
-    fullDisplayName = build.getFullDisplayName();
-    description = build.getDescription();
-    url = build.getUrl();
-
-    Action testResultAction = build.getAction(AbstractTestResultAction.class);
-    if (testResultAction != null) {
-      testResults = new TestData(testResultAction);
-    }
+    initData(build, currentTime);
 
     Node node = build.getBuiltOn();
     if (node == null) {
@@ -135,10 +129,7 @@ public BuildData(AbstractBuild<?, ?> build, Date currentTime) {
       buildLabel = StringUtils.isBlank(node.getLabelString()) ? "master" : node.getLabelString();
     }
 
-    buildNum = build.getNumber();
     // build.getDuration() is always 0 in Notifiers
-    buildDuration = currentTime.getTime() - build.getStartTimeInMillis();
-    timestamp = DATE_FORMATTER.format(build.getTimestamp().getTime());
     rootProjectName = build.getRootBuild().getProject().getName();
     rootProjectDisplayName = build.getRootBuild().getDisplayName();
     rootBuildNum = build.getRootBuild().getNumber();
@@ -166,6 +157,48 @@ public BuildData(AbstractBuild<?, ?> build, Date currentTime) {
     }
   }
 
+  // Pipeline project build
+  public BuildData(Run<?, ?> build, Date currentTime, TaskListener listener) {
+    initData(build, currentTime);
+
+    Executor executor = build.getExecutor();
+    if (executor == null) {
+      buildHost = "master";
+    } else {
+      buildHost = StringUtils.isBlank(executor.getDisplayName()) ? "master" : executor.getDisplayName();
+    }
+
+    rootProjectName = projectName;
+    rootProjectDisplayName = displayName;
+    rootBuildNum = buildNum;
+
+    try {
+      // TODO: sensitive variables are not filtered, c.f. https://stackoverflow.com/questions/30916085
+      buildVariables = build.getEnvironment(listener);
+    } catch (Exception e) {
+      buildVariables = new HashMap<String, String>();
+    }
+  }
+
+  private void initData(Run<?, ?> build, Date currentTime) {
+    result = build.getResult() == null ? null : build.getResult().toString();
+    id = build.getId();
+    projectName = build.getParent().getName();
+    displayName = build.getDisplayName();
+    fullDisplayName = build.getFullDisplayName();
+    description = build.getDescription();
+    url = build.getUrl();
+    buildNum = build.getNumber();
+
+    Action testResultAction = build.getAction(AbstractTestResultAction.class);
+    if (testResultAction != null) {
+      testResults = new TestData(testResultAction);
+    }
+
+    buildDuration = currentTime.getTime() - build.getStartTimeInMillis();
+    timestamp = DATE_FORMATTER.format(build.getTimestamp().getTime());
+  }
+
   @Override
   public String toString() {
     Gson gson = new GsonBuilder().create();

src/main/java/jenkins/plugins/logstash/persistence/IndexerDaoFactory.java

@@ -52,6 +52,7 @@ public final class IndexerDaoFactory {
     indexerMap.put(IndexerType.RABBIT_MQ, RabbitMqDao.class);
     indexerMap.put(IndexerType.ELASTICSEARCH, ElasticSearchDao.class);
     indexerMap.put(IndexerType.SYSLOG, SyslogDao.class);
+    indexerMap.put(IndexerType.LOGSTASH, LogstashDao.class);
 
     INDEXER_MAP = Collections.unmodifiableMap(indexerMap);
   }

src/main/java/jenkins/plugins/logstash/persistence/LogstashDao.java

@@ -0,0 +1,55 @@
+package jenkins.plugins.logstash.persistence;
+
+import com.cloudbees.syslog.Facility;
+import com.cloudbees.syslog.MessageFormat;
+import com.cloudbees.syslog.Severity;
+import com.cloudbees.syslog.sender.UdpSyslogMessageSender;
+import org.apache.http.impl.client.HttpClientBuilder;
+
+import java.io.*;
+import java.net.Socket;
+
+
+public class LogstashDao extends AbstractLogstashIndexerDao {
+
+  final String logstashHost;
+  final int logstashPort;
+  Socket logstashClientSocket;
+
+  public LogstashDao(String logstashHostString, int logstashPortInt, String indexKey, String username, String password) {
+    this(null, logstashHostString, logstashPortInt, indexKey, username, password);
+  }
+
+  public LogstashDao(HttpClientBuilder factory, String logstashHostString, int logstashPortInt, String indexKey, String username, String password) {
+    super(logstashHostString, logstashPortInt, indexKey, username, password);
+    this.logstashHost = logstashHostString;
+    this.logstashPort = logstashPortInt;
+  }
+
+  @Override
+  public void push(String data) throws IOException {
+
+    try
+    {
+      logstashClientSocket = new Socket(logstashHost, logstashPort);
+      DataOutputStream outToServer = new DataOutputStream(logstashClientSocket.getOutputStream());
+      BufferedReader inFromServer = new BufferedReader(new InputStreamReader(logstashClientSocket.getInputStream()));
+      outToServer.writeBytes(data);
+      logstashClientSocket.setSoTimeout(10000);
+      logstashClientSocket.close();
+      outToServer.close();
+      inFromServer.close();
+    }
+    catch (Exception exc)
+    {
+      StringWriter sw = new StringWriter();
+      PrintWriter pw = new PrintWriter(sw);
+      exc.printStackTrace(pw);
+      throw new IOException(sw.toString());
+    }
+
+  }
+
+  @Override
+  public IndexerType getIndexerType() { return IndexerType.LOGSTASH; }
+}

src/main/java/jenkins/plugins/logstash/persistence/LogstashIndexerDao.java

@@ -40,7 +40,8 @@ static enum IndexerType {
     REDIS,
     RABBIT_MQ,
     ELASTICSEARCH,
-    SYSLOG
+    SYSLOG,
+    LOGSTASH
   }
 
   String getDescription();