Add Logz.io support

README.md

@@ -4,7 +4,7 @@ Jenkins Logstash Plugin
 Travis: [![Build Status](https://travis-ci.org/jenkinsci/logstash-plugin.svg?branch=master)](https://travis-ci.org/jenkinsci/logstash-plugin)
 Jenkins: [![Build Status](https://ci.jenkins.io/job/Plugins/job/logstash-plugin/job/master/badge/icon)](https://ci.jenkins.io/job/Plugins/job/logstash-plugin/job/master/)
 
-This plugin adds support for sending a job's console log to Logstash indexers such as [Elastic Search](https://www.elastic.co/products/elasticsearch), [Logstash](https://www.elastic.co/de/products/logstash), [RabbitMQ](https://www.rabbitmq.com), [Redis](https://redis.io/) or to Syslog.
+This plugin adds support for sending a job's console log to Logstash indexers such as [Logzio](https://logz.io/), [Elastic Search](https://www.elastic.co/products/elasticsearch), [Logstash](https://www.elastic.co/de/products/logstash), [RabbitMQ](https://www.rabbitmq.com), [Redis](https://redis.io/) or to Syslog.
 
 * see [Jenkins wiki](https://wiki.jenkins-ci.org/display/JENKINS/Logstash+Plugin) for detailed feature descriptions
 * use [JIRA](https://issues.jenkins-ci.org) to report issues / feature requests
@@ -15,6 +15,7 @@ Install
 * Generate the `hpi` file with the command: `mvn package`
 
 * Put the `hpi` file in the directory `$JENKINS_HOME/plugins`
+
 * Restart jenkins
 
 Configure
@@ -27,6 +28,7 @@ Currently supported methods of input/output:
 * Redis {format => 'json_event'}
 * RabbitMQ {mechanism => PLAIN}
 * Syslog {format => cee/json ([RFC-5424](https://tools.ietf.org/html/rfc5424),[RFC-3164](https://tools.ietf.org/html/rfc3164)), protocol => UDP}
+* Logz.io
 
 Pipeline
 ========

pom.xml

@@ -79,6 +79,18 @@
       <scope>compile</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.logz.sender</groupId>
+      <artifactId>logzio-sender</artifactId>
+      <version>1.0.14</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.github.wnameless</groupId>
+      <artifactId>json-flattener</artifactId>
+      <version>0.5.0</version>
+    </dependency>
+
     <dependency>
       <groupId>com.google.code.gson</groupId>
       <artifactId>gson</artifactId>

src/main/java/jenkins/plugins/logstash/configuration/Logzio.java

@@ -0,0 +1,110 @@
+package jenkins.plugins.logstash.configuration;
+
+import hudson.Extension;
+import hudson.util.FormValidation;
+import hudson.util.Secret;
+
+import jenkins.plugins.logstash.persistence.LogzioDao;
+import jenkins.plugins.logstash.Messages;
+
+import org.apache.commons.lang.StringUtils;
+import org.kohsuke.stapler.DataBoundConstructor;
+import org.kohsuke.stapler.DataBoundSetter;
+import org.kohsuke.stapler.QueryParameter;
+
+import javax.annotation.Nonnull;
+
+public class Logzio extends LogstashIndexer<LogzioDao>
+{
+    private Secret key;
+    private String host;
+
+    @DataBoundConstructor
+    public Logzio(){}
+
+    public String getHost(){ return this.host; }
+
+    @DataBoundSetter
+    public void setHost(String host){ this.host = host; }
+
+    public String getKey()
+    {
+        return Secret.toString(key);
+    }
+
+    @DataBoundSetter
+    public void setKey(String key)
+    {
+        this.key = Secret.fromString(key);
+    }
+
+
+    @Override
+    public boolean equals(Object obj)
+    {
+        if (obj == null)
+            return false;
+        if (this == obj)
+            return true;
+        if (getClass() != obj.getClass())
+            return false;
+        Logzio other = (Logzio) obj;
+        if (!Secret.toString(key).equals(other.getKey()))
+        {
+            return false;
+        }
+        if (host == null)
+        {
+            return other.host == null;
+        }
+        else return host.equals(other.host);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        final int prime = 31;
+        int result = super.hashCode();
+        result = prime * result + ((host == null) ? 0 : host.hashCode());
+        result = prime * result + Secret.toString(key).hashCode();
+        return result;
+    }
+
+    @Override
+    public LogzioDao createIndexerInstance() { return new LogzioDao(host, Secret.toString(key)); }
+
+    @Extension
+    public static class LogzioDescriptor extends LogstashIndexerDescriptor
+    {
+        @Nonnull
+        @Override
+        public String getDisplayName()
+        {
+            return "Logz.io";
+        }
+
+        @Override
+        public int getDefaultPort()
+        {
+            return 0;
+        }
+
+        public FormValidation doCheckKey(@QueryParameter("value") String value)
+        {
+            if (StringUtils.isBlank(value))
+            {
+                return FormValidation.error(Messages.ValueIsRequired());
+            }
+            return FormValidation.ok();
+        }
+
+        public FormValidation doCheckHost(@QueryParameter("value") String value)
+        {
+            if (StringUtils.isBlank(value))
+            {
+                return FormValidation.error(Messages.ValueIsRequired());
+            }
+            return FormValidation.ok();
+        }
+    }
+}

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

@@ -24,7 +24,7 @@
 
 package jenkins.plugins.logstash.persistence;
 
-import static com.google.common.collect.Ranges.closedOpen;
+import static com.google.common.collect.Range.closedOpen;
 
 import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.lang.StringUtils;

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

@@ -0,0 +1,109 @@
+package jenkins.plugins.logstash.persistence;
+
+import com.github.wnameless.json.flattener.JsonFlattener;
+import com.google.gson.JsonObject;
+import io.logz.sender.FormattedLogMessage;
+import io.logz.sender.exceptions.LogzioParameterErrorException;
+import io.logz.sender.exceptions.LogzioServerErrorException;
+import net.sf.json.JSONArray;
+import net.sf.json.JSONObject;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Logz.io Data Access Object.
+ *
+ * @author Ido Halevi
+ *
+ */
+
+public class LogzioDao extends AbstractLogstashIndexerDao {
+    private final String TYPE = "jenkins_logstash_plugin";
+    private String key;
+    private String host;
+    private LogzioHttpsClient httpsClient;
+
+    //primary constructor used by indexer factory
+    public LogzioDao(String host, String key){
+        this(null, host, key);
+    }
+
+    // Factored for unit testing
+    LogzioDao(LogzioHttpsClient factory, String host, String key) {
+        this.host = host;
+        this.key = key;
+        try{
+            this.httpsClient = factory == null ? new LogzioHttpsClient(key, host, TYPE) : factory;
+        }catch (LogzioParameterErrorException e) {
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+    @Override
+    public void push(String data) throws IOException {
+        JSONObject jsonData = JSONObject.fromObject(data);
+        JSONArray logMessages = jsonData.getJSONArray("message");
+        if (!logMessages.isEmpty()) {
+            try{
+                for (Object logMsg : logMessages) {
+                    JsonObject logLine = createLogLine(jsonData, logMsg.toString());
+                    httpsClient.send(new FormattedLogMessage((logLine + "\n").getBytes(Charset.forName("UTF-8"))));
+                }
+                httpsClient.flush();
+            }catch (LogzioServerErrorException e){
+                throw new IOException(e);
+            }
+        }
+    }
+
+
+    private JsonObject createLogLine(JSONObject jsonData, String logMsg) {
+        JsonObject logLine = new JsonObject();
+
+        logLine.addProperty("message", logMsg);
+        logLine.addProperty("@timestamp", ZonedDateTime.now(ZoneOffset.UTC).format(DateTimeFormatter.ISO_OFFSET_DATE_TIME));
+
+        for (Object key : jsonData.keySet()) {
+            if (!key.equals("message")){
+                logLine.addProperty(key.toString(), jsonData.getString(key.toString()));
+            }
+        }
+
+        return logLine;
+    }
+
+    @Override
+    public JSONObject buildPayload(BuildData buildData, String jenkinsUrl, List<String> logLines) {
+        JSONObject payload = new JSONObject();
+        payload.put("message", logLines);
+        payload.put("source", "jenkins");
+        payload.put("source_host", jenkinsUrl);
+        payload.put("@buildTimestamp", buildData.getTimestamp());
+        payload.put("@version", 1);
+        // Flatten build data - so the user will be able to use fields for visualization in Kibana.
+        // In addition, it makes the query much easier.
+        Map<String, Object> flattenJson = JsonFlattener.flattenAsMap(buildData.toString());
+        for (Map.Entry<String, Object> entry : flattenJson.entrySet()) {
+            String key = entry.getKey().replace('.','_');
+            Object value = entry.getValue();
+            payload.put(key, value);
+        }
+
+        return payload;
+    }
+
+    @Override
+    public String getDescription(){ return host; }
+
+    public String getHost(){ return host; }
+
+    public String getKey(){ return key; }
+
+    public String getType(){ return TYPE; }
+}

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

@@ -0,0 +1,100 @@
+package jenkins.plugins.logstash.persistence;
+
+import io.logz.sender.FormattedLogMessage;
+import io.logz.sender.HttpsRequestConfiguration;
+import io.logz.sender.HttpsSyncSender;
+import io.logz.sender.SenderStatusReporter;
+import io.logz.sender.exceptions.LogzioParameterErrorException;
+import io.logz.sender.exceptions.LogzioServerErrorException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class LogzioHttpsClient{
+    private static final int MAX_SIZE_IN_BYTES = 8 * 1024 * 1024;  // 8 MB
+    private static final int CONNECT_TIMEOUT = 10 * 1000;
+    private static final int SOCKET_TIMEOUT = 10 * 1000;
+    private final HttpsSyncSender logzioClient;
+    private List<FormattedLogMessage> messages; //Not thread safe
+    private int size;
+
+    LogzioHttpsClient(String token, String listener, String type) throws LogzioParameterErrorException {
+        HttpsRequestConfiguration gzipHttpsRequestConfiguration = HttpsRequestConfiguration
+                .builder()
+                .setLogzioToken(token)
+                .setLogzioType(type)
+                .setLogzioListenerUrl(listener)
+                .setSocketTimeout(SOCKET_TIMEOUT)
+                .setConnectTimeout(CONNECT_TIMEOUT)
+                .setCompressRequests(true)
+                .build();
+
+        logzioClient = new HttpsSyncSender(gzipHttpsRequestConfiguration, new Reporter());
+        messages = new ArrayList<>();
+        size = 0;
+    }
+
+    public void send(FormattedLogMessage log) throws LogzioServerErrorException {
+        if (size + log.getSize() > MAX_SIZE_IN_BYTES) {
+            sendAndReset();
+        }
+        messages.add(log);
+        size += log.getSize();
+    }
+
+    private void reset(){
+        size = 0;
+        messages.clear();
+    }
+
+    void flush() throws LogzioServerErrorException {
+        if(messages.size() > 0) {
+            sendAndReset();
+        }
+    }
+
+    private void sendAndReset() throws LogzioServerErrorException {
+        logzioClient.sendToLogzio(messages);
+        reset();
+    }
+
+    private static class Reporter implements SenderStatusReporter{
+        private static final Logger LOGGER = Logger.getLogger(LogzioDao.class.getName());
+
+        private void pringLogMessage(Level level, String msg) {
+            LOGGER.log(level, msg);
+        }
+
+        @Override
+        public void error(String msg) {
+            pringLogMessage(Level.SEVERE, "[LogzioSender]ERROR: " + msg);
+        }
+
+        @Override
+        public void error(String msg, Throwable e) {
+            pringLogMessage(Level.SEVERE, "[LogzioSender]ERROR: " + msg + "\n" +e);
+        }
+
+        @Override
+        public void warning(String msg) {
+            pringLogMessage(Level.WARNING, "[LogzioSender]WARNING: " + msg);
+        }
+
+        @Override
+        public void warning(String msg, Throwable e) {
+            pringLogMessage(Level.WARNING, "[LogzioSender]WARNING: " + msg + "\n" + e);
+        }
+
+        @Override
+        public void info(String msg) {
+            pringLogMessage(Level.INFO, "[LogzioSender]INFO: " + msg);
+        }
+
+        @Override
+        public void info(String msg, Throwable e) {
+            pringLogMessage(Level.INFO, "[LogzioSender]INFO: " + msg + "\n" + e);
+        }
+    }
+}

src/main/resources/index.jelly

@@ -1,3 +1,3 @@
 <div>
-  Adds the possibility to push builds logs and build data to a Logstash indexer such as Redis, RabbitMQ, Elastic Search or to Syslog.
+  Adds the possibility to push builds logs and build data to a Logstash indexer such as Redis, RabbitMQ, Elastic Search, Logz.io or to Syslog.
 </div>

src/main/resources/jenkins/plugins/logstash/configuration/Logzio/config.jelly

@@ -0,0 +1,9 @@
+<?jelly escape-by-default='true'?>
+<j:jelly xmlns:j="jelly:core" xmlns:st="jelly:stapler" xmlns:d="jelly:define" xmlns:l="/lib/layout" xmlns:t="/lib/hudson" xmlns:f="/lib/form">
+      <f:entry title="${%Logz.io Host}" field="host">
+        <f:textbox default="https://listener.logz.io:8071"/>
+      </f:entry>
+      <f:entry title="${%Logz.io key}" field="key">
+        <f:password/>
+      </f:entry>
+</j:jelly>

src/main/resources/jenkins/plugins/logstash/configuration/Logzio/help-host.jelly

@@ -0,0 +1,5 @@
+<div>
+  <p>Logz.io listener URL.<br/>
+  If you are in the EU region insert https://listener-eu.logz.io:8071. Otherwise, use https://listener.logz.io:8071<br/>
+  You can tell which region you are in by checking your login URL</p>
+</div>