Netstat output parsing on AIX and Linux.

Example dashboard provided.

client/src/main/java/sysmon/client/ClientRouteBuilder.java

@@ -21,7 +21,7 @@ public class ClientRouteBuilder extends RouteBuilder {
     private static final Logger log = LoggerFactory.getLogger(ClientRouteBuilder.class);
 
     @Override
-    public void configure() throws Exception {
+    public void configure() {
 
         Registry registry = getContext().getRegistry();
 
@@ -45,9 +45,13 @@ public class ClientRouteBuilder extends RouteBuilder {
                 log.info(">>> Enabling extension: " + ext.getDescription());
                 providers.add(provides);
 
+
+                // TODO: Make timer thread configurable
+
                 // Setup Camel route for this extension
                 // a unique timer name gives the timer it's own thread, otherwise it's a shared thread for other timers with same name.
-                from("timer:"+provides+"?fixedRate=true&period=30s")
+                //from("timer:"+provides+"?fixedRate=true&period=30s")
+                from("timer:extensions?fixedRate=true&period=30s")
                         .bean(ext, "getMetrics")
                         //.doTry()
                         .process(new MetricEnrichProcessor(registry))
@@ -55,14 +59,16 @@ public class ClientRouteBuilder extends RouteBuilder {
                             .log("Skipping empty measurement.")
                             .stop()
                         .otherwise()
-                            .to("seda:metrics");
+                            .to("seda:metrics?discardWhenFull=true");
             } else {
-                log.info(">>> Skipping extension: " + ext.getDescription());
+                log.info(">>> Skipping extension (not supported here): " + ext.getDescription());
             }
+
         }
 
 
-        from("seda:metrics")
+        // TODO: Make 'concurrentConsumers' configurable
+        from("seda:metrics?concurrentConsumers=1")
                 .setHeader(Exchange.HTTP_METHOD, constant("POST"))
                 //.setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
                 .doTry()

doc/AIX.md

@@ -1,13 +1,21 @@
 # AIX Notes
 
+Works on IBM Power VIO (Virtual IO) servers, as well as regular IBM Power AIX installations.
+
 ## Installation
 
+We require Java 8, which should already be installed. 
+The RPM packages are *"noarch"* Java bytecode, so we can use the **--ignoreos** option to install:
+
 ```shell
 rpm -i --ignoreos sysmon-client.rpm sysmon-plugins.rpm
 ```
 
 ## Run automatically at boot
 
+Change the *sysmon-server* URL for your environment.
+
 ```shell
 mkitab 'sysmon:2:respawn:env JAVA_HOME=/usr/java8_64 /opt/sysmon/client/bin/client -s http://10.20.30.40:9925/metrics >/tmp/sysmon.log 2>&1'
+init q
 ```

plugins/build.gradle

@@ -28,20 +28,6 @@ subprojects {
 
     }
 
-/*
-    jar {
-        manifest {
-            attributes(
-                    'Plugin-Id'         : "${pluginId}",
-                    'Plugin-Class'      : "${pluginClass}",
-                    'Plugin-Version'    : "${pluginVersion}",
-                    'Plugin-Provider'   : "${pluginProvider}",
-                    'Plugin-Description': "${pluginDescription}"
-            )
-        }
-
-    }*/
-
     task uberJar(type: Jar) {
         from sourceSets.main.output
         dependsOn configurations.runtimeClasspath
@@ -62,8 +48,8 @@ subprojects {
             attributes(
                     'Plugin-Id'         : "${pluginId}",
                     'Plugin-Class'      : "${pluginClass}",
-                    'Plugin-Version'    : "${pluginVersion}",
+                    'Plugin-Version'    : "${version}",
-                    'Plugin-Provider'   : "${pluginProvider}",
+                    'Plugin-Provider'   : "System Monitor",
                     'Plugin-Description': "${pluginDescription}"
             )
         }

plugins/os-aix/src/main/java/sysmon/plugins/os_aix/AixNetstatExtension.java

@@ -0,0 +1,73 @@
+package sysmon.plugins.os_aix;
+
+import org.pf4j.Extension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sysmon.shared.Measurement;
+import sysmon.shared.MetricExtension;
+import sysmon.shared.MetricResult;
+import sysmon.shared.PluginHelper;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+@Extension
+public class AixNetstatExtension implements MetricExtension  {
+
+    private static final Logger log = LoggerFactory.getLogger(AixNetstatExtension.class);
+
+    @Override
+    public boolean isSupported() {
+
+        if(!System.getProperty("os.name").toLowerCase().contains("aix")) {
+            log.warn("Requires AIX.");
+            return false;
+        }
+
+        if(!PluginHelper.canExecute("netstat")) {
+            log.warn("Requires the 'netstat' command.");
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public String getName() {
+        return "aix-network-netstat";
+    }
+
+    @Override
+    public String getProvides() {
+        return "network-netstat";
+    }
+
+    @Override
+    public String getDescription() {
+        return "AIX Netstat Metrics";
+    }
+
+    @Override
+    public MetricResult getMetrics() throws Exception {
+
+        Map<String, String> tagsMap = null;
+        Map<String, Object> fieldsMap = null;
+
+        try (InputStream buf = PluginHelper.executeCommand("netstat -s -f inet")) {
+            AixNetstatParser parser = processCommandOutput(buf);
+            tagsMap = parser.getTags();
+            fieldsMap = parser.getFields();
+        }
+
+        log.debug(fieldsMap.toString());
+        return new MetricResult("network_netstat", new Measurement(tagsMap, fieldsMap));
+    }
+
+
+    protected AixNetstatParser processCommandOutput(InputStream input) throws IOException {
+        return new AixNetstatParser(input);
+    }
+
+}
+

plugins/os-aix/src/main/java/sysmon/plugins/os_aix/AixNetstatParser.java

@@ -0,0 +1,156 @@
+package sysmon.plugins.os_aix;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.HashMap;
+import java.util.Map;
+
+public class AixNetstatParser {
+
+    private static final Logger log = LoggerFactory.getLogger(AixNetstatParser.class);
+
+    private long ipTotalPacketsReceived;
+    private long ipForwarded;
+
+    private long tcpConnectionsEstablished;
+    private long tcpPacketsReceved;
+    private long tcpPacketsSent;
+
+    private long udpPacketsReceived;
+    private long udpPacketsSent;
+
+
+    public AixNetstatParser(InputStream inputStream) throws IOException {
+
+        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
+        while (reader.ready()) {
+            String line = reader.readLine();
+            log.debug("AixNetstatParser() - Line: " + line);
+
+            if(line.startsWith("tcp:")) {
+                parseTcp(reader);
+            }
+
+            if(line.startsWith("udp:")) {
+                parseUdp(reader);
+            }
+
+            if(line.startsWith("ip:")) {
+                parseIp(reader);
+            }
+
+        }
+
+        inputStream.close();
+    }
+
+
+    protected void parseIp(BufferedReader reader) throws IOException {
+
+        while (reader.ready()) {
+            reader.mark(64);
+            String line = reader.readLine();
+
+            if(!line.startsWith(" ")) {
+                reader.reset();
+                return;
+            }
+
+            line = line.trim();
+
+            if(line.matches("(\\d+) total packets received")) {
+                ipTotalPacketsReceived = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) packets forwarded")) {
+                ipForwarded = getFirstLong(line);
+            }
+
+        }
+
+    }
+
+
+    protected void parseTcp(BufferedReader reader) throws IOException {
+
+        while (reader.ready()) {
+            reader.mark(64);
+            String line = reader.readLine();
+
+            if(!line.startsWith(" ")) {
+                reader.reset();
+                return;
+            }
+
+            line = line.trim();
+
+            if(line.matches("(\\d+) connections established \\(including accepts\\)")) {
+                tcpConnectionsEstablished = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) packets received")) {
+                tcpPacketsReceved = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) packets sent")) {
+                tcpPacketsSent = getFirstLong(line);
+            }
+
+        }
+
+    }
+
+    protected void parseUdp(BufferedReader reader) throws IOException {
+
+        while (reader.ready()) {
+            reader.mark(64);
+            String line = reader.readLine();
+
+            if(!line.startsWith(" ")) {
+                reader.reset();
+                return;
+            }
+
+            line = line.trim();
+
+            if(line.matches("(\\d+) datagrams received")) {
+                udpPacketsReceived = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) datagrams output")) {
+                udpPacketsSent = getFirstLong(line);
+            }
+        }
+
+    }
+
+
+    public Map<String, String> getTags() {
+        return new HashMap<>();
+    }
+
+    public Map<String, Object> getFields() {
+        Map<String, Object> fields = new HashMap<>();
+        fields.put("ip_forwarded", ipForwarded);
+        fields.put("ip_received", ipTotalPacketsReceived);
+
+        fields.put("tcp_connections", tcpConnectionsEstablished);
+        fields.put("tcp_pkts_recv", tcpPacketsReceved);
+        fields.put("tcp_pkts_sent", tcpPacketsSent);
+
+        fields.put("udp_pkts_recv", udpPacketsReceived);
+        fields.put("udp_pkts_sent", udpPacketsSent);
+
+        return fields;
+    }
+
+    private Long getFirstLong(String line) {
+        return Long.parseLong(line.substring(0, line.indexOf(" ")));
+    }
+
+}

plugins/os-aix/src/main/java/sysmon/plugins/os_aix/AixProcessorExtension.java

@@ -8,6 +8,8 @@ import sysmon.shared.MetricExtension;
 import sysmon.shared.MetricResult;
 import sysmon.shared.PluginHelper;
 
+import java.io.IOException;
+import java.io.InputStream;
 import java.util.List;
 import java.util.Map;
 
@@ -49,20 +51,23 @@ public class AixProcessorExtension implements MetricExtension {
     }
 
     @Override
-    public MetricResult getMetrics() {
+    public MetricResult getMetrics() throws Exception {
 
-        List<String> lparstat = PluginHelper.executeCommand("lparstat 1 1");
+        Map<String, String> tagsMap = null;
-        AixProcessorStat processorStat = processCommandOutput(lparstat);
+        Map<String, Object> fieldsMap = null;
 
-        Map<String, String> tagsMap = processorStat.getTags();
+        try (InputStream buf = PluginHelper.executeCommand("lparstat 1 1")) {
-        Map<String, Object> fieldsMap = processorStat.getFields();
+            AixProcessorStat processorStat = processCommandOutput(buf);
+            tagsMap = processorStat.getTags();
+            fieldsMap = processorStat.getFields();
+        }
 
         return new MetricResult("processor_lpar", new Measurement(tagsMap, fieldsMap));
     }
 
 
-    protected AixProcessorStat processCommandOutput(List<String> inputLines) {
+    protected AixProcessorStat processCommandOutput(InputStream input) throws IOException {
-        return new AixProcessorStat(inputLines);
+        return new AixProcessorStat(input);
     }
 
 

plugins/os-aix/src/main/java/sysmon/plugins/os_aix/AixProcessorStat.java

@@ -3,6 +3,10 @@ package sysmon.plugins.os_aix;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -39,9 +43,12 @@ public class AixProcessorStat {
     private final float lbusy;  // Indicates the percentage of logical processor(s) utilization that occurred while executing at the user and system level.
 
 
-    AixProcessorStat(List<String> lines) {
+    public AixProcessorStat(InputStream inputStream) throws IOException {
 
-        for (String line : lines) {
+        String lastLine = null;
+        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
+        while(reader.ready()) {
+            String line = reader.readLine();
 
             if (line.startsWith("System configuration:")) {
                 Matcher matcher = patternAixShared.matcher(line);
@@ -75,13 +82,14 @@ public class AixProcessorStat {
                 }
             }
 
+            lastLine = line;
         }
 
-        String lparstat = lines.get(lines.size() -1);
+        //String lparstat = lines.get(lines.size() -1);
-        String[] splitStr = lparstat.trim().split("\\s+");
+        String[] splitStr = lastLine.trim().split("\\s+");
         if(type.equalsIgnoreCase("shared") && splitStr.length < 9 ||
                 type.equalsIgnoreCase("dedicated") && splitStr.length < 8) {
-            throw new UnsupportedOperationException("lparstat string error: " + lparstat);
+            throw new UnsupportedOperationException("lparstat string error: " + lastLine);
         }
 
         this.user = Float.parseFloat(splitStr[0]);
@@ -96,6 +104,8 @@ public class AixProcessorStat {
             this.entc = 0f;
             this.lbusy = 0f;
         }
+
+        inputStream.close();
     }
 
     public float getUser() {

plugins/os-aix/src/test/groovy/AixNetstatTest.groovy

@@ -0,0 +1,25 @@
+import spock.lang.Specification
+import sysmon.plugins.os_aix.AixNetstatParser
+
+class AixNetstatTest extends Specification {
+
+    void "test netstat parsing"() {
+
+        setup:
+        InputStream inputStream = getClass().getResourceAsStream('/netstat-aix.txt');
+
+        when:
+        AixNetstatParser parser = new AixNetstatParser(inputStream)
+
+        then:
+        parser.getFields().size() > 0
+        parser.getFields().get('ip_received') == 76229L
+        parser.getFields().get('ip_forwarded') == 24L
+        parser.getFields().get('tcp_connections') == 85L
+        parser.getFields().get('tcp_pkts_sent') == 31274L
+        parser.getFields().get('tcp_pkts_recv') == 39830L
+        parser.getFields().get('udp_pkts_sent') == 26332L
+        parser.getFields().get('udp_pkts_recv') == 34559L
+    }
+
+}

plugins/os-aix/src/test/groovy/AixProcessorTest.groovy

@@ -7,12 +7,11 @@ class AixProcessorTest extends Specification {
     void "test AIX lparstat shared output processing"() {
 
         setup:
-        def testFile = new File(getClass().getResource('/lparstat-aix-shared.txt').toURI())
+        InputStream inputStream = getClass().getResourceAsStream('/lparstat-aix-shared.txt');
-        List<String> lines = testFile.readLines("UTF-8")
 
         when:
         AixProcessorExtension extension = new AixProcessorExtension()
-        AixProcessorStat stats = extension.processCommandOutput(lines)
+        AixProcessorStat stats = extension.processCommandOutput(inputStream)
 
         then:
         stats.getUser() == 83.7f
@@ -27,12 +26,11 @@ class AixProcessorTest extends Specification {
     void "test AIX lparstat dedicated output processing"() {
 
         setup:
-        def testFile = new File(getClass().getResource('/lparstat-aix-dedicated.txt').toURI())
+        InputStream inputStream = getClass().getResourceAsStream('/lparstat-aix-dedicated.txt');
-        List<String> lines = testFile.readLines("UTF-8")
 
         when:
         AixProcessorExtension extension = new AixProcessorExtension()
-        AixProcessorStat stats = extension.processCommandOutput(lines)
+        AixProcessorStat stats = extension.processCommandOutput(inputStream)
 
         then:
         stats.getUser() == 0.1f
@@ -47,12 +45,11 @@ class AixProcessorTest extends Specification {
     void "test Linux lparstat output processing"() {
 
         setup:
-        def testFile = new File(getClass().getResource('/lparstat-linux.txt').toURI())
+        InputStream inputStream = getClass().getResourceAsStream('/lparstat-linux.txt');
-        List<String> lines = testFile.readLines("UTF-8")
 
         when:
         AixProcessorExtension extension = new AixProcessorExtension()
-        AixProcessorStat stats = extension.processCommandOutput(lines)
+        AixProcessorStat stats = extension.processCommandOutput(inputStream)
 
         then:
         stats.getUser() == 0.03f
@@ -65,6 +62,4 @@ class AixProcessorTest extends Specification {
 
     }
 
-    // java.lang.UnsupportedOperationException: lparstat string error:   2.2   1.2    0.0   96.6  0.28  1100   132  24.23
-
 }

plugins/os-aix/src/test/resources/netstat-aix.txt

@@ -0,0 +1,157 @@
+icmp:
+        12 calls to icmp_error
+        0 errors not generated because old message was icmp
+        Output histogram:
+                destination unreachable: 12
+        0 messages with bad code fields
+        0 messages < minimum length
+        0 bad checksums
+        0 messages with bad length
+        Input histogram:
+                destination unreachable: 3
+        0 message responses generated
+igmp:
+        0 messages received
+        0 messages received with too few bytes
+        0 messages received with bad checksum
+        0 membership queries received
+        0 membership queries received with invalid field(s)
+        0 membership reports received
+        0 membership reports received with invalid field(s)
+        0 membership reports received for groups to which we belong
+        2 membership reports sent
+tcp:
+        31274 packets sent
+                27328 data packets (82928168 bytes)
+                86 data packets (108992 bytes) retransmitted
+                2938 ack-only packets (2698 delayed)
+                0 URG only packets
+                0 window probe packets
+                784 window update packets
+                138 control packets
+                3812 large sends
+                74913716 bytes sent using largesend
+                64069 bytes is the biggest largesend
+        39830 packets received
+                22701 acks (for 82928732 bytes)
+                112 duplicate acks
+                0 acks for unsent data
+                15579 packets (5876585 bytes) received in-sequence
+                62 completely duplicate packets (320 bytes)
+                57 old duplicate packets
+                0 packets with some dup. data (0 bytes duped)
+                75 out-of-order packets (6408 bytes)
+                0 packets (0 bytes) of data after window
+                0 window probes
+                1723 window update packets
+                0 packets received after close
+                0 packets with bad hardware assisted checksum
+                0 discarded for bad checksums
+                0 discarded for bad header offset fields
+                0 discarded because packet too short
+                1 discarded by listeners
+                0 discarded due to listener's queue full
+                3207 ack packet headers correctly predicted
+                15050 data packet headers correctly predicted
+        63 connection requests
+        23 connection accepts
+        85 connections established (including accepts)
+        114 connections closed (including 0 drops)
+        0 connections with ECN capability
+        0 times responded to ECN
+        0 embryonic connections dropped
+        20314 segments updated rtt (of 16791 attempts)
+        0 segments with congestion window reduced bit set
+        0 segments with congestion experienced bit set
+        0 resends due to path MTU discovery
+        2 path MTU discovery terminations due to retransmits
+        25 retransmit timeouts
+                0 connections dropped by rexmit timeout
+        4 fast retransmits
+                1 when congestion window less than 4 segments
+        28 newreno retransmits
+        4 times avoided false fast retransmits
+        0 persist timeouts
+                0 connections dropped due to persist timeout
+        0 keepalive timeouts
+                0 keepalive probes sent
+                0 connections dropped by keepalive
+        0 times SACK blocks array is extended
+        0 times SACK holes array is extended
+        0 packets dropped due to memory allocation failure
+        0 connections in timewait reused
+        0 delayed ACKs for SYN
+        0 delayed ACKs for FIN
+        0 send_and_disconnects
+        0 spliced connections
+        0 spliced connections closed
+        0 spliced connections reset
+        0 spliced connections timeout
+        0 spliced connections persist timeout
+        0 spliced connections keepalive timeout
+        0 TCP checksum offload disabled during retransmit
+        0 Connections dropped due to bad ACKs
+        0 Connections dropped due to duplicate SYN packets
+        0 fastpath loopback connections
+        0 fastpath loopback sent packets (0 bytes)
+        0 fastpath loopback received packets (0 bytes)
+        0 fake SYN segments dropped
+        0 fake RST segments dropped
+        0 data injection segments dropped
+        0 TCPTR maximum connections dropped
+        0 TCPTR connections dropped for no memory
+        0 TCPTR maximum per host connections dropped
+        0 connections dropped due to max assembly queue depth
+udp:
+        34559 datagrams received
+        0 incomplete headers
+        0 bad data length fields
+        0 bad checksums
+        1849 dropped due to no socket
+        8218 broadcast/multicast datagrams dropped due to no socket
+        0 socket buffer overflows
+        24492 delivered
+        26332 datagrams output
+ip:
+        76229 total packets received
+        0 bad header checksums
+        0 with size smaller than minimum
+        0 with data size < data length
+        0 with header length < data size
+        0 with data length < header length
+        0 with bad options
+        0 with incorrect version number
+        0 fragments received
+        0 fragments dropped (dup or out of space)
+        0 fragments dropped after timeout
+        0 packets reassembled ok
+        72552 packets for this host
+        3 packets for unknown/unsupported protocol
+        24 packets forwarded
+        0 packets not forwardable
+        0 redirects sent
+        55784 packets sent from this host
+        0 packets sent with fabricated ip header
+        0 output packets dropped due to no bufs, etc.
+        0 output packets discarded due to no route
+        0 output datagrams fragmented
+        0 fragments created
+        0 datagrams that can't be fragmented
+        0 IP Multicast packets dropped due to no receiver
+        0 successful path MTU discovery cycles
+        0 path MTU rediscovery cycles attempted
+        0 path MTU discovery no-response estimates
+        0 path MTU discovery response timeouts
+        0 path MTU discovery decreases detected
+        0 path MTU discovery packets sent
+        0 path MTU discovery memory allocation failures
+        0 ipintrq overflows
+        0 with illegal source
+        0 packets processed by threads
+        0 packets dropped by threads
+        0 packets dropped due to the full socket receive buffer
+        0 dead gateway detection packets sent
+        0 dead gateway detection packet allocation failures
+        0 dead gateway detection gateway allocation failures
+        0 incoming packets dropped due to MLS filters
+        0 packets not sent due to MLS filters

plugins/os-linux/gradle.properties

@@ -1,7 +1,5 @@
 pluginId=sysmon-linux
 pluginClass=sysmon.plugins.os_linux.LinuxPlugin
-pluginVersion=0.0.1
-pluginProvider=System Monitor
 pluginDependencies=
 pluginDescription=Linux OS Metrics.
 

plugins/os-linux/src/main/java/sysmon/plugins/os_linux/LinuxNetstatExtension.java

@@ -0,0 +1,73 @@
+package sysmon.plugins.os_linux;
+
+import org.pf4j.Extension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sysmon.shared.Measurement;
+import sysmon.shared.MetricExtension;
+import sysmon.shared.MetricResult;
+import sysmon.shared.PluginHelper;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+@Extension
+public class LinuxNetstatExtension implements MetricExtension  {
+
+    private static final Logger log = LoggerFactory.getLogger(LinuxNetstatExtension.class);
+
+    @Override
+    public boolean isSupported() {
+
+        if(!System.getProperty("os.name").toLowerCase().contains("linux")) {
+            log.warn("Requires Linux.");
+            return false;
+        }
+
+        if(!PluginHelper.canExecute("netstat")) {
+            log.warn("Requires the 'netstat' command.");
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public String getName() {
+        return "linux-network-netstat";
+    }
+
+    @Override
+    public String getProvides() {
+        return "network-netstat";
+    }
+
+    @Override
+    public String getDescription() {
+        return "Linux Netstat Metrics";
+    }
+
+    @Override
+    public MetricResult getMetrics() throws Exception {
+
+        Map<String, String> tagsMap = null;
+        Map<String, Object> fieldsMap = null;
+
+        try (InputStream inputStream = PluginHelper.executeCommand("netstat -s")) {
+            LinuxNetstatParser parser = processCommandOutput(inputStream);
+            tagsMap = parser.getTags();
+            fieldsMap = parser.getFields();
+        }
+
+        log.debug(fieldsMap.toString());
+        return new MetricResult("network_netstat", new Measurement(tagsMap, fieldsMap));
+    }
+
+
+    protected LinuxNetstatParser processCommandOutput(InputStream input) throws IOException {
+        return new LinuxNetstatParser(input);
+    }
+
+}
+

plugins/os-linux/src/main/java/sysmon/plugins/os_linux/LinuxNetstatParser.java

@@ -0,0 +1,168 @@
+package sysmon.plugins.os_linux;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.HashMap;
+import java.util.Map;
+
+public class LinuxNetstatParser {
+
+    private static final Logger log = LoggerFactory.getLogger(LinuxNetstatParser.class);
+
+    private long ipTotalPacketsReceived;
+    private long ipForwarded;
+    private long ipIncomingPacketsDiscarded;
+    private long ipOutgoingPacketsDropped;
+
+    private long tcpConnectionsEstablished;
+    private long tcpSegmentsReceived;
+    private long tcpSegmentsSent;
+
+    private long udpPacketsReceived;
+    private long udpPacketsSent;
+
+
+    public LinuxNetstatParser(InputStream inputStream) throws IOException {
+
+        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
+        while (reader.ready()) {
+            String line = reader.readLine();
+            log.debug("LinuxNetstatParser() - Line: " + line);
+
+            if(line.startsWith("Ip:")) {
+                parseIp(reader);
+            }
+
+            if(line.startsWith("Tcp:")) {
+                parseTcp(reader);
+            }
+
+            if(line.startsWith("Udp:")) {
+                parseUdp(reader);
+            }
+
+        }
+
+        inputStream.close();
+    }
+
+
+    protected void parseIp(BufferedReader reader) throws IOException {
+
+        while (reader.ready()) {
+            reader.mark(64);
+            String line = reader.readLine();
+
+            if(!line.startsWith(" ")) {
+                reader.reset();
+                return;
+            }
+
+            line = line.trim();
+
+            if(line.matches("(\\d+) total packets received")) {
+                ipTotalPacketsReceived = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) forwarded")) {
+                ipForwarded = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) incoming packets discarded")) {
+                ipIncomingPacketsDiscarded = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) outgoing packets dropped")) {
+                ipOutgoingPacketsDropped = getFirstLong(line);
+            }
+
+        }
+
+    }
+
+
+    protected void parseTcp(BufferedReader reader) throws IOException {
+
+        while (reader.ready()) {
+            reader.mark(64);
+            String line = reader.readLine();
+
+            if(!line.startsWith(" ")) {
+                reader.reset();
+                return;
+            }
+
+            line = line.trim();
+
+            if(line.matches("(\\d+) connections established")) {
+                tcpConnectionsEstablished = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) segments received")) {
+                tcpSegmentsReceived = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) segments sent out")) {
+                tcpSegmentsSent = getFirstLong(line);
+            }
+
+        }
+
+    }
+
+    protected void parseUdp(BufferedReader reader) throws IOException {
+
+        while (reader.ready()) {
+            reader.mark(64);
+            String line = reader.readLine();
+
+            if(!line.startsWith(" ")) {
+                reader.reset();
+                return;
+            }
+
+            line = line.trim();
+
+            if(line.matches("(\\d+) packets received")) {
+                udpPacketsReceived = getFirstLong(line);
+            }
+
+            if(line.matches("(\\d+) packets sent")) {
+                udpPacketsSent = getFirstLong(line);
+            }
+        }
+
+    }
+
+
+    public Map<String, String> getTags() {
+        return new HashMap<>();
+    }
+
+    public Map<String, Object> getFields() {
+        Map<String, Object> fields = new HashMap<>();
+        fields.put("ip_forwarded", ipForwarded);
+        fields.put("ip_received", ipTotalPacketsReceived);
+        fields.put("ip_dropped", ipOutgoingPacketsDropped);
+        fields.put("ip_discarded", ipIncomingPacketsDiscarded);
+
+        fields.put("tcp_connections", tcpConnectionsEstablished);
+        fields.put("tcp_pkts_recv", tcpSegmentsReceived);
+        fields.put("tcp_pkts_sent", tcpSegmentsSent);
+
+        fields.put("udp_pkts_recv", udpPacketsReceived);
+        fields.put("udp_pkts_sent", udpPacketsSent);
+
+        return fields;
+    }
+
+    private Long getFirstLong(String line) {
+        return Long.parseLong(line.substring(0, line.indexOf(" ")));
+    }
+
+}

plugins/os-linux/src/main/java/sysmon/plugins/os_linux/LinuxSockstatExtension.java

@@ -12,9 +12,9 @@ import java.util.List;
 import java.util.Map;
 
 @Extension
-public class LinuxNetworkExtension implements MetricExtension {
+public class LinuxSockstatExtension implements MetricExtension {
 
-    private static final Logger log = LoggerFactory.getLogger(LinuxNetworkExtension.class);
+    private static final Logger log = LoggerFactory.getLogger(LinuxSockstatExtension.class);
 
     @Override
     public boolean isSupported() {

plugins/os-linux/src/test/groovy/LinuxNetstatTest.groovy

@@ -0,0 +1,26 @@
+import spock.lang.Specification
+import sysmon.plugins.os_linux.LinuxNetstatParser
+
+class LinuxNetstatTest extends Specification {
+
+    void "test netstat parsing"() {
+
+        setup:
+        InputStream inputStream = getClass().getResourceAsStream('/netstat-linux.txt');
+
+        when:
+        LinuxNetstatParser parser = new LinuxNetstatParser(inputStream)
+
+        then:
+        parser.getFields().size() > 0
+        parser.getFields().get('ip_received') == 109772L
+        parser.getFields().get('ip_dropped') == 70L
+        parser.getFields().get('ip_discarded') == 0L
+        parser.getFields().get('tcp_pkts_sent') == 89891L
+        parser.getFields().get('tcp_pkts_recv') == 86167L
+        parser.getFields().get('udp_pkts_sent') == 10682L
+        parser.getFields().get('udp_pkts_recv') == 31928L
+
+    }
+
+}

plugins/os-linux/src/test/groovy/LinuxNetworkTest.groovy

@@ -1,5 +1,5 @@
 import spock.lang.Specification
-import sysmon.plugins.os_linux.LinuxNetworkExtension
+import sysmon.plugins.os_linux.LinuxSockstatExtension
 import sysmon.plugins.os_linux.LinuxNetworkSockStat
 
 class LinuxNetworkTest extends Specification {
@@ -11,7 +11,7 @@ class LinuxNetworkTest extends Specification {
         List<String> lines = testFile.readLines("UTF-8")
 
         when:
-        LinuxNetworkExtension extension = new LinuxNetworkExtension()
+        LinuxSockstatExtension extension = new LinuxSockstatExtension()
         LinuxNetworkSockStat stats = extension.processSockOutput(lines)
 
         then:

plugins/os-linux/src/test/resources/netstat-linux.txt

@@ -0,0 +1,112 @@
+Ip:
+    Forwarding: 1
+    109772 total packets received
+    1 with invalid addresses
+    0 forwarded
+    0 incoming packets discarded
+    109769 incoming packets delivered
+    103916 requests sent out
+    70 outgoing packets dropped
+    1 dropped because of missing route
+Icmp:
+    52 ICMP messages received
+    0 input ICMP message failed
+    ICMP input histogram:
+        destination unreachable: 40
+        echo requests: 12
+    108 ICMP messages sent
+    0 ICMP messages failed
+    ICMP output histogram:
+        destination unreachable: 96
+        echo replies: 12
+IcmpMsg:
+        InType3: 40
+        InType8: 12
+        OutType0: 12
+        OutType3: 96
+Tcp:
+    3142 active connection openings
+    5 passive connection openings
+    2105 failed connection attempts
+    193 connection resets received
+    70 connections established
+    86167 segments received
+    89891 segments sent out
+    184 segments retransmitted
+    3 bad segments received
+    2735 resets sent
+Udp:
+    31928 packets received
+    96 packets to unknown port received
+    0 packet receive errors
+    10682 packets sent
+    0 receive buffer errors
+    0 send buffer errors
+    IgnoredMulti: 22
+UdpLite:
+TcpExt:
+    30 packets pruned from receive queue because of socket buffer overrun
+    178 TCP sockets finished time wait in fast timer
+    426 delayed acks sent
+    1 delayed acks further delayed because of locked socket
+    Quick ack mode was activated 1059 times
+    45809 packet headers predicted
+    7293 acknowledgments not containing data payload received
+    7659 predicted acknowledgments
+    TCPSackRecovery: 3
+    Detected reordering 4 times using SACK
+    TCPDSACKUndo: 1
+    1 congestion windows recovered without slow start after partial ack
+    TCPLostRetransmit: 82
+    3 timeouts after reno fast retransmit
+    1 timeouts in loss state
+    3 fast retransmits
+    3 retransmits in slow start
+    TCPTimeouts: 129
+    TCPLossProbes: 69
+    TCPLossProbeRecovery: 10
+    TCPBacklogCoalesce: 450
+    TCPDSACKOldSent: 991
+    TCPDSACKOfoSent: 6
+    TCPDSACKRecv: 45
+    202 connections reset due to unexpected data
+    147 connections reset due to early user close
+    13 connections aborted due to timeout
+    TCPDSACKIgnoredNoUndo: 10
+    TCPSackShifted: 1
+    TCPSackMerged: 1
+    TCPSackShiftFallback: 9
+    TCPRcvCoalesce: 5338
+    TCPOFOQueue: 793
+    TCPOFOMerge: 6
+    TCPChallengeACK: 3
+    TCPSYNChallenge: 3
+    TCPSpuriousRtxHostQueues: 6
+    TCPAutoCorking: 710
+    TCPFromZeroWindowAdv: 1
+    TCPToZeroWindowAdv: 1
+    TCPWantZeroWindowAdv: 4
+    TCPSynRetrans: 98
+    TCPOrigDataSent: 19048
+    TCPHystartTrainDetect: 3
+    TCPHystartTrainCwnd: 54
+    TCPHystartDelayDetect: 1
+    TCPHystartDelayCwnd: 24
+    TCPACKSkippedSeq: 1
+    TCPKeepAlive: 2595
+    TCPDelivered: 20025
+    TCPAckCompressed: 260
+    TcpTimeoutRehash: 116
+IpExt:
+    InMcastPkts: 2257
+    OutMcastPkts: 480
+    InBcastPkts: 98
+    OutBcastPkts: 78
+    InOctets: 147193028
+    OutOctets: 14723163
+    InMcastOctets: 478599
+    OutMcastOctets: 73462
+    InBcastOctets: 10094
+    OutBcastOctets: 5580
+    InNoECTPkts: 177661
+MPTcpExt:

server/src/main/java/sysmon/server/ServerRouteBuilder.java

@@ -36,9 +36,11 @@ public class ServerRouteBuilder extends RouteBuilder {
                 .to("seda:inbound")
                 .endRest();
 
+
         //from("seda:inbound").log("Got metric from: ${header.component}").to("mock:sink");
 
-        from("seda:inbound")
+        // TODO: Make 'concurrentConsumers' configurable
+        from("seda:inbound?concurrentConsumers=5")
                 .log(">>> metric: ${header.hostname} - ${body}")
                 .doTry()
                     .process(new MetricResultToPointProcessor())

shared/src/main/java/sysmon/shared/MetricExtension.java

@@ -2,6 +2,8 @@ package sysmon.shared;
 
 import org.pf4j.ExtensionPoint;
 
+import java.io.IOException;
+
 public interface MetricExtension extends ExtensionPoint {
 
     boolean isSupported();
@@ -10,5 +12,5 @@ public interface MetricExtension extends ExtensionPoint {
     String getProvides();
     String getDescription();
 
-    MetricResult getMetrics();
+    MetricResult getMetrics() throws Exception;
 }

shared/src/main/java/sysmon/shared/PluginHelper.java

@@ -3,10 +3,7 @@ package sysmon.shared;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.BufferedReader;
+import java.io.*;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
@@ -23,10 +20,9 @@ public class PluginHelper {
             .toLowerCase().startsWith("windows");
 
 
-    public static List<String> executeCommand(String... cmd) {
+    public static InputStream executeCommand(String... cmd) {
-
-        List<String> outputLines = new ArrayList<>();
 
+        InputStream inputStream = null;
         ProcessBuilder builder = new ProcessBuilder();
         if (isWindows) {
             builder.command("cmd.exe", "/c");
@@ -40,15 +36,8 @@ public class PluginHelper {
 
         builder.directory(new File(System.getProperty("user.home")));
         try {
-
             Process process = builder.start();
-            BufferedReader reader =
+            inputStream = process.getInputStream();
-                    new BufferedReader(new InputStreamReader(process.getInputStream()));
-
-            String line;
-            while ((line = reader.readLine()) != null) {
-                outputLines.add(line);
-            }
 
             int exitCode = process.waitFor();
             if(exitCode > 0) {
@@ -59,7 +48,7 @@ public class PluginHelper {
             e.printStackTrace();
         }
 
-        return outputLines;
+        return inputStream;
     }