Merged in loki (pull request #5)

Loki

README.md

@@ -1,33 +1,35 @@
 # Syslog Server
 
-All received messages are written to *stdout* and/or forwarded to another syslog server.
+All received messages are written to *stdout* and/or forwarded to a remote logging solution.
 
 The syslog server is able to listen on both UDP and TCP and parses syslog messages in either RFC5424 or RFC3164 (BSD) format.
 
 The default syslog port (514) requires you to run syslogd as root / administrator.
 If you do not wish to do so, you can choose a port number (with the *-p* or *--port* flag) above 1024.
 
+Supported remote logging solutions are Syslog (RFC5424 over UDP), Graylog (GELF over UDP) and Grafana Loki.
+
 ## Usage Instructions
 
 - Install the syslogd package (*.deb* or *.rpm*) from [downloads](https://bitbucket.org/mnellemann/syslogd/downloads/) or build from source.
 - Run *bin/syslogd*, use the *-h* option for help :)
 
-````
+```text
-Usage: syslogd [-dghV] [--[no-]ansi] [--[no-]stdout] [--[no-]tcp] [--[no-]udp]
+Usage: syslogd [-dhV] [--[no-]ansi] [--[no-]stdout] [--[no-]tcp] [--[no-]udp]
-               [--rfc5424] [-f=<host>] [-p=<port>]
+               [--rfc5424] [-g=<uri>] [-l=<url>] [-p=<num>] [-s=<uri>]
-Syslog Server
+  -d, --debug          Enable debugging [default: 'false'].
-  -d, --debug            Enable debugging [default: 'false'].
+  -g, --gelf=<uri>     Forward to Graylog <udp://host:port>.
-  -f, --forward=<host>   Forward to UDP host[:port] (RFC-5424).
+  -h, --help           Show this help message and exit.
-  -g, --gelf             Forward in Graylog (GELF) JSON format.
+  -l, --loki=<url>     Forward to Grafana Loki <http://host:port>.
-  -h, --help             Show this help message and exit.
+      --[no-]ansi      Output ANSI colors [default: true].
-      --[no-]ansi        Output ANSI colors [default: true].
+      --[no-]stdout    Output messages to stdout [default: true].
-      --[no-]stdout      Output messages to stdout [default: true].
+      --[no-]tcp       Listen on TCP [default: true].
-      --[no-]tcp         Listen on TCP [default: true].
+      --[no-]udp       Listen on UDP [default: true].
-      --[no-]udp         Listen on UDP [default: true].
+  -p, --port=<num>     Listening port [default: 514].
-  -p, --port=<port>      Listening port [default: 514].
+      --rfc5424        Parse RFC-5424 messages [default: RFC-3164].
-      --rfc5424          Parse RFC-5424 messages [default: RFC-3164].
+  -s, --syslog=<uri>   Forward to Syslog <udp://host:port> (RFC-5424).
-  -V, --version          Print version information and exit.
+  -V, --version        Print version information and exit.
-````
+```
 
 ### Examples
 
@@ -46,20 +48,40 @@ or, if installed as a *deb* or *rpm* package:
 Listening on the standard syslog port (requires root privileges) and forwarding messages on to another log-system on a non-standard port.
 
 ```
-java -jar /path/to/syslogd-x.y.z-all.jar --forward remotehost:1514
+java -jar /path/to/syslogd-x.y.z-all.jar --syslog udp://remotehost:514
 ```
 
 Forwarding to a Graylog server in GELF format.
 
 ```
-java -jar /path/to/syslogd-x.y.z-all.jar --forward remotehost:12201 --gelf
+java -jar /path/to/syslogd-x.y.z-all.jar --gelf udp://remotehost:12201
 ```
 
+Forwarding to a Grafana Loki server.
+
+```
+java -jar /path/to/syslogd-x.y.z-all.jar --loki http://remotehost:3100
+```
 
 If you don't want any output locally (only forwarding), you can use the ```--no-stdout``` flag.
 
 
 ## Notes
 
+### IBM AIX and VIO Servers
+
 Syslog messages from AIX (and IBM Power Virtual I/O Servers) can be troublesome with some logging solutions. These can be received with
-syslogd and optionally forwarded on to Graylog, Splunk or other logging solutions.
+syslogd and then forwarded on to your preferred logging solution.
+
+
+## Development Notes
+
+### Test Grafana Loki
+
+Run Loki and Grafana in local containers to test.
+
+```shell
+docker run --rm -d --name=loki -p 3100:3100 grafana/loki
+docker run --rm -d --name=grafana --link loki:loki -p 3000:3000 grafana/grafana:7.1.3
+```
+

gradle.properties

@@ -1,3 +1,3 @@
 id            = syslogd
 group         = biz.nellemann.syslogd
-version       = 1.0.13
+version       = 1.0.14

src/main/java/biz/nellemann/syslogd/Application.java

@@ -16,6 +16,7 @@
 package biz.nellemann.syslogd;
 
 import biz.nellemann.syslogd.msg.SyslogMessage;
+import biz.nellemann.syslogd.net.LokiClient;
 import biz.nellemann.syslogd.net.TcpServer;
 import biz.nellemann.syslogd.net.UdpClient;
 import biz.nellemann.syslogd.net.UdpServer;
@@ -28,6 +29,10 @@ import picocli.CommandLine;
 import picocli.CommandLine.Command;
 
 import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URL;
+import java.util.Locale;
 import java.util.concurrent.Callable;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -40,9 +45,11 @@ public class Application implements Callable<Integer>, LogListener {
     private boolean doForward = false;
     private SyslogParser syslogParser;
     private UdpClient udpClient;
+    private UdpClient gelfClient;
+    private LokiClient lokiClient;
 
 
-    @CommandLine.Option(names = {"-p", "--port"}, description = "Listening port [default: 514].", defaultValue = "514")
+    @CommandLine.Option(names = {"-p", "--port"}, description = "Listening port [default: 514].", defaultValue = "514", paramLabel = "<num>")
     private int port;
 
     @CommandLine.Option(names = "--no-udp", negatable = true, description = "Listen on UDP [default: true].", defaultValue = "true")
@@ -60,11 +67,14 @@ public class Application implements Callable<Integer>, LogListener {
     @CommandLine.Option(names = "--rfc5424", description = "Parse RFC-5424 messages [default: RFC-3164].", defaultValue = "false")
     private boolean rfc5424;
 
-    @CommandLine.Option(names = { "-f", "--forward"}, description = "Forward to UDP host[:port] (RFC-5424).", paramLabel = "<host>")
+    @CommandLine.Option(names = { "-s", "--syslog"}, description = "Forward to Syslog <udp://host:port> (RFC-5424).", paramLabel = "<uri>")
-    private String forward;
+    private URI syslog;
 
-    @CommandLine.Option(names = { "-g", "--gelf"}, description = "Forward in Graylog (GELF) JSON format.", defaultValue = "false")
+    @CommandLine.Option(names = { "-g", "--gelf"}, description = "Forward to Graylog <udp://host:port>.", paramLabel = "<uri>")
-    private boolean gelf;
+    private URI gelf;
+
+    @CommandLine.Option(names = { "-l", "--loki"}, description = "Forward to Grafana Loki <http://host:port>.", paramLabel = "<url>")
+    private URL loki;
 
     @CommandLine.Option(names = { "-d", "--debug" }, description = "Enable debugging [default: 'false'].")
     private boolean enableDebug = false;
@@ -83,23 +93,27 @@ public class Application implements Callable<Integer>, LogListener {
             syslogParser = new SyslogParserRfc3164();
         }
 
-        if(forward != null && !forward.isEmpty()) {
+        if(syslog != null) {
-            String fHost, fPort;
+            if(syslog.getScheme().toLowerCase(Locale.ROOT).equals("udp")) {
-            Pattern pattern = Pattern.compile("^([^:]+)(?::([0-9]+))?$", Pattern.CASE_INSENSITIVE);
+                udpClient = new UdpClient(getInetSocketAddress(syslog));
-            Matcher matcher = pattern.matcher(forward);
+                doForward = true;
-            if(matcher.find()) {
-                fHost = matcher.group(1);
-                if(matcher.groupCount() == 2 && matcher.group(2) != null) {
-                    fPort = matcher.group(2);
-                } else {
-                    fPort = "514";
-                }
             } else {
-                fHost = "localhost";
+                throw new UnsupportedOperationException("Syslog protocol not implemented: " + syslog.getScheme());
-                fPort = "514";
             }
+        }
 
-            udpClient = new UdpClient(fHost, Integer.parseInt(fPort));
+        if(gelf != null) {
+            System.err.println(gelf.getScheme());
+            if(gelf.getScheme().toLowerCase(Locale.ROOT).equals("udp")) {
+                gelfClient = new UdpClient(getInetSocketAddress(gelf));
+                doForward = true;
+            } else {
+                throw new UnsupportedOperationException("GELF protocol not implemented: " + gelf.getScheme());
+            }
+        }
+
+        if(loki != null) {
+            lokiClient = new LokiClient(loki);
             doForward = true;
         }
 
@@ -143,11 +157,19 @@ public class Application implements Callable<Integer>, LogListener {
 
             if(doForward) {
                 try {
-                    if(gelf) {
+
-                        udpClient.send(SyslogPrinter.toGelf(msg));
+                    if(udpClient != null) {
-                    } else {
                         udpClient.send(SyslogPrinter.toRfc5424(msg));
                     }
+
+                    if(gelfClient != null) {
+                        gelfClient.send(SyslogPrinter.toGelf(msg));
+                    }
+
+                    if(lokiClient != null) {
+                        lokiClient.send(SyslogPrinter.toLoki(msg));
+                    }
+
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
@@ -157,6 +179,34 @@ public class Application implements Callable<Integer>, LogListener {
     }
 
 
+    private InetSocketAddress getInetSocketAddress(URI input) {
+        InetSocketAddress inetSocketAddress = new InetSocketAddress(input.getHost(), input.getPort());
+        return inetSocketAddress;
+    }
+
+
+    private InetSocketAddress getInetSocketAddress(String input) {
+
+        String dstHost;
+        int dstPort;
+        InetSocketAddress inetSocketAddress = null;
+
+        Pattern pattern = Pattern.compile("^([^:]+)(?::([0-9]+))?$", Pattern.CASE_INSENSITIVE);
+        Matcher matcher = pattern.matcher(input);
+        if(matcher.find()) {
+            dstHost = matcher.group(1);
+            if(matcher.groupCount() == 2 && matcher.group(2) != null) {
+                dstPort = Integer.parseInt(matcher.group(2));
+            } else {
+                dstPort = 514;
+            }
+            inetSocketAddress = new InetSocketAddress(dstHost, dstPort);
+        }
+
+        return inetSocketAddress;
+    }
+
+
     public static void main(String... args) {
         int exitCode = new CommandLine(new Application()).execute(args);
         System.exit(exitCode);

src/main/java/biz/nellemann/syslogd/SyslogPrinter.java

@@ -81,7 +81,7 @@ public class SyslogPrinter {
 
 
     /**
-     * Return a GELF formatted string of the SyslogMessage.
+     * Return a GELF JSON formatted string of the SyslogMessage.
      * https://www.graylog.org/features/gelf
      * @param msg
      * @return
@@ -104,6 +104,32 @@ public class SyslogPrinter {
     }
 
 
+    /**
+     * Return a Loki JSON formatted string of the SyslogMessage.
+     * https://grafana.com/docs/loki/latest/api/
+     * @param msg
+     * @return
+     */
+
+/*
+{ "streams": [ { "stream": { "label": "value" }, "values": [ [ "<unix epoch in nanoseconds>", "<log line>" ], [ "<unix epoch in nanoseconds>", "<log line>" ] ] } ] }
+{ "streams": [ { "stream": { "host": "hyperion", "facility": "USER", "severity": "NOTICE", "application": "mark"}, "values": [ [ "1615823598000000000", "Test 2345534343434" ] ] } ] }
+{ "streams": [ { "stream": { "host": "hyperion", "facility": "USER", "severity": "NOTICE", "application": "mark"}, "values": [ [ "1615842165000000000", "Test" ] ] } ] }
+*/
+
+    public static String toLoki(SyslogMessage msg) {
+        StringBuilder sb = new StringBuilder("{ \"streams\": [ { \"stream\": {");
+        sb.append(String.format(" \"host\": \"%s\",", msg.hostname));
+        sb.append(String.format(" \"facility\": \"%s\",", msg.facility));
+        sb.append(String.format(" \"severity\": \"%s\",", msg.severity));
+        sb.append(String.format(" \"application\": \"%s\"", msg.application));
+        sb.append("}, \"values\": [ ");
+        sb.append(String.format("[ \"%d\", \"%s\" ]", msg.timestamp.getEpochSecond() * 1000000000l, msg.message));
+        sb.append(" ] } ] }");
+        return sb.toString();
+    }
+
+
     static private String getPri(Facility facility, Severity severity) {
         int pri = (facility.toNumber() * 8) + severity.toNumber();
         return String.format("%c%d%c", '<', pri, '>');

src/main/java/biz/nellemann/syslogd/net/LokiClient.java

@@ -0,0 +1,55 @@
+package biz.nellemann.syslogd.net;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.*;
+import java.nio.charset.StandardCharsets;
+
+public class LokiClient {
+
+    private final static Logger log = LoggerFactory.getLogger(LokiClient.class);
+
+    private final URL url;
+
+
+    public LokiClient(URL url) {
+        this.url = url;
+    }
+
+
+    public void send(String msg) throws MalformedURLException {
+
+        URL pushUrl = new URL(url, "/loki/api/v1/push");
+        HttpURLConnection con = null;
+        try {
+            con = (HttpURLConnection)pushUrl.openConnection();
+            con.setRequestMethod("POST");
+            con.setRequestProperty("Content-Type", "application/json");
+            con.setDoOutput(true);
+
+            byte[] input = msg.getBytes(StandardCharsets.UTF_8);
+            try(OutputStream os = con.getOutputStream()) {
+                os.write(input, 0, input.length);
+            } catch (IOException e) {
+                log.warn(e.getMessage());
+            }
+
+            int responseCode = con.getResponseCode();
+            if(responseCode != 204) {
+                log.warn("send() - response: " + responseCode);
+            }
+
+        } catch (IOException e) {
+            log.warn("send() - " + e.getMessage());
+        } finally {
+            if(con != null) {
+                con.disconnect();
+            }
+        }
+
+    }
+
+}

src/main/java/biz/nellemann/syslogd/net/UdpClient.java

@@ -11,30 +11,17 @@ public class UdpClient {
 
     private final static Logger log = LoggerFactory.getLogger(UdpClient.class);
 
+    private InetSocketAddress inetSocketAddress;
     private DatagramSocket socket;
-    private InetAddress address;
-    private final Integer port;
 
-   public UdpClient(String host, Integer port) {
+   public UdpClient(InetSocketAddress inetSocketAddress) throws SocketException {
-
+        this.inetSocketAddress = inetSocketAddress;
-        try {
+        this.socket = new DatagramSocket();
-            this.address = InetAddress.getByName(host);
-        } catch (UnknownHostException e) {
-            log.error("UdpClient() - UnknownHostException: " + e.getMessage());
-        }
-
-        try {
-            this.socket = new DatagramSocket();
-        } catch (SocketException e) {
-            log.error("UdpClient() - Could not instantiate DatagramSocket: " + e.getMessage());
-        }
-
-        this.port = port;
     }
 
     public void send(String msg) {
         byte[] buf = msg.getBytes(StandardCharsets.US_ASCII);
-        DatagramPacket packet = new DatagramPacket(buf, buf.length, address, port);
+        DatagramPacket packet = new DatagramPacket(buf, buf.length, inetSocketAddress.getAddress(), inetSocketAddress.getPort());
         if(this.socket != null) {
             try {
                 socket.send(packet);

src/test/groovy/biz/nellemann/syslogd/SyslogPrinterTest.groovy

@@ -24,6 +24,19 @@ class SyslogPrinterTest extends Specification {
         output.contains("_structured-data")
     }
 
+    void "test toLoki"() {
+        setup:
+        SyslogParser syslogParser = new SyslogParserRfc5424();
+        String input = '<13>1 2020-09-23T08:57:30.950699+02:00 xps13 mark - - [exampleSDID@32473 iut="3" eventSource="Application" eventID="1011"] adfdfdf3432434565656'
+        SyslogMessage msg = syslogParser.parse(input)
+
+        when:
+        String output = SyslogPrinter.toLoki(msg)
+
+        then:
+        output == '{ "streams": [ { "stream": { "host": "xps13", "facility": "USER", "severity": "NOTICE", "application": "mark"}, "values": [ [ "1600845200000000000", "adfdfdf3432434565656" ] ] } ] }'
+    }
+
 }