Move code from JRmonitor into logstash-core #6337

Java code is mostly the same as that from the JRMonitor gem with some general tidying and DRYing up
Report classes are now straight Java
Added junit tests for the Java monitor classes
Ruby code basically stays the same, other than calling out to pure Java report classes

Fixes #7167

logstash-core/lib/logstash/api/commands/hot_threads_reporter.rb

@@ -1,4 +1,5 @@
 # encoding: utf-8
+java_import 'org.logstash.instrument.reports.ThreadsReport'
 
 class HotThreadsReport
   STRING_SEPARATOR_LENGTH = 80.freeze
@@ -7,8 +8,7 @@ class HotThreadsReport
   def initialize(cmd, options)
     @cmd = cmd
     filter = { :stacktrace_size => options.fetch(:stacktrace_size, HOT_THREADS_STACK_TRACES_SIZE_DEFAULT) }
-    jr_dump = JRMonitor.threads.generate(filter)
-    @thread_dump = ::LogStash::Util::ThreadDump.new(options.merge(:dump => jr_dump))
+    @thread_dump = ::LogStash::Util::ThreadDump.new(options.merge(:dump => ThreadsReport.generate(filter)))
   end
 
   def to_s

logstash-core/lib/logstash/instrument/periodic_poller/jvm.rb

@@ -2,18 +2,19 @@
 require "logstash/instrument/periodic_poller/base"
 require "logstash/instrument/periodic_poller/load_average"
 require "logstash/environment"
-require "jrmonitor"
 require "set"
 
+java_import 'com.sun.management.UnixOperatingSystemMXBean'
 java_import 'java.lang.management.ManagementFactory'
 java_import 'java.lang.management.OperatingSystemMXBean'
 java_import 'java.lang.management.GarbageCollectorMXBean'
 java_import 'java.lang.management.RuntimeMXBean'
-java_import 'com.sun.management.UnixOperatingSystemMXBean'
 java_import 'javax.management.MBeanServer'
 java_import 'javax.management.ObjectName'
 java_import 'javax.management.AttributeList'
 java_import 'javax.naming.directory.Attribute'
+java_import 'org.logstash.instrument.reports.MemoryReport'
+java_import 'org.logstash.instrument.reports.ProcessReport'
 
 
 module LogStash module Instrument module PeriodicPoller
@@ -50,7 +51,7 @@ module LogStash module Instrument module PeriodicPoller
     end
 
     def collect
-      raw = JRMonitor.memory.generate
+      raw = MemoryReport.generate
       collect_jvm_metrics(raw)
       collect_pools_metrics(raw)
       collect_threads_metrics
@@ -81,11 +82,10 @@ module LogStash module Instrument module PeriodicPoller
     end
 
     def collect_process_metrics
-      process_metrics = JRMonitor.process.generate
+      process_metrics = ProcessReport.generate
 
       path = [:jvm, :process]
 
-
       open_fds = process_metrics["open_file_descriptors"]
       if @peak_open_fds.nil? || open_fds > @peak_open_fds
         @peak_open_fds = open_fds

logstash-core/lib/logstash/util/thread_dump.rb

@@ -1,4 +1,6 @@
 # encoding: utf-8
+java_import 'org.logstash.instrument.reports.ThreadsReport'
+
 module LogStash
   module Util
     class ThreadDump
@@ -10,7 +12,7 @@ module LogStash
 
       def initialize(options={})
         @options   = options
-        @dump = options.fetch(:dump, JRMonitor.threads.generate({}))
+        @dump = options.fetch(:dump, ThreadsReport.generate({}))
         @top_count = options.fetch(:threads, THREADS_COUNT_DEFAULT)
         @ignore    = options.fetch(:ignore_idle_threads, IGNORE_IDLE_THREADS_DEFAULT)
       end

logstash-core/logstash-core.gemspec

@@ -33,7 +33,6 @@ Gem::Specification.new do |gem|
   gem.add_runtime_dependency 'puma', '~> 2.16'
   gem.add_runtime_dependency "jruby-openssl", "0.9.16" # >= 0.9.13 Required to support TLSv1.2
   gem.add_runtime_dependency "chronic_duration", "0.10.6"
-  gem.add_runtime_dependency "jrmonitor", '~> 0.4.2'
 
   # TODO(sissel): Treetop 1.5.x doesn't seem to work well, but I haven't
   # investigated what the cause might be. -Jordan

logstash-core/src/main/java/org/logstash/instrument/monitors/HotThreadsMonitor.java

@@ -0,0 +1,199 @@
+package org.logstash.instrument.monitors;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Hot threads monitoring class. This class pulls information out of the JVM #
+ * provided beans and lest the different consumers query it.
+ * Created by purbon on 12/12/15.
+ */
+public class HotThreadsMonitor {
+
+    private static final String ORDERED_BY = "ordered_by";
+    private static final String STACKTRACE_SIZE = "stacktrace_size";
+    private final Logger logger = LogManager.getLogger(HotThreadsMonitor.class);
+
+    /**
+     * Placeholder for a given thread report
+     */
+    public static class ThreadReport {
+
+        private static final String CPU_TIME = "cpu.time";
+        private static final String BLOCKED_COUNT = "blocked.count";
+        private static final String BLOCKED_TIME = "blocked.time";
+        private static final String WAITED_COUNT = "waited.count";
+        private static final String WAITED_TIME = "waited.time";
+        private static final String THREAD_NAME = "thread.name";
+        private static final String THREAD_STATE = "thread.state";
+        private static final String THREAD_STACKTRACE = "thread.stacktrace";
+
+        private Map<String, Object> map = new HashMap<>();
+
+        ThreadReport(ThreadInfo info, long cpuTime) {
+            map.put(CPU_TIME, cpuTime);
+            map.put(BLOCKED_COUNT, info.getBlockedCount());
+            map.put(BLOCKED_TIME, info.getBlockedTime());
+            map.put(WAITED_COUNT, info.getWaitedCount());
+            map.put(WAITED_TIME, info.getWaitedTime());
+            map.put(THREAD_NAME, info.getThreadName());
+            map.put(THREAD_STATE, info.getThreadState().name().toLowerCase());
+            map.put(THREAD_STACKTRACE, stackTraceAsString(info.getStackTrace()));
+        }
+
+        private List<String> stackTraceAsString(StackTraceElement [] elements) {
+            return Arrays.stream(elements)
+                            .map(StackTraceElement::toString)
+                            .collect(Collectors.toList());
+        }
+
+        public Map<String, Object> toMap() {
+            return map;
+        }
+
+        public String getThreadState() {
+            return (String) map.get(THREAD_STATE);
+        }
+
+        public String getThreadName() {
+            return (String) map.get(THREAD_NAME);
+        }
+
+        @Override
+        public String toString() {
+            StringBuilder sb = new StringBuilder();
+            int i = 0;
+            for (Map.Entry<String, Object> mapEntry: map.entrySet()) {
+                if (i > 0) {
+                    sb.append(",");
+                }
+                sb.append(String.format("%s,%s", mapEntry.getKey(), mapEntry.getValue()));
+                i++;
+            }
+            return sb.toString();
+        }
+
+        Long getWaitedTime() {
+            return (Long)map.get(WAITED_TIME);
+        }
+
+        Long getBlockedTime() {
+            return (Long)map.get(BLOCKED_TIME);
+        }
+
+        Long getCpuTime() {
+            return (Long) map.get(CPU_TIME);
+        }
+    }
+
+    private List<String> VALID_ORDER_BY = new ArrayList<>();
+
+    public HotThreadsMonitor() {
+        VALID_ORDER_BY.add("cpu");
+        VALID_ORDER_BY.add("wait");
+        VALID_ORDER_BY.add("block");
+    }
+
+    /**
+     * Return the current hot threads information as provided by the JVM
+     *
+     * @return A list of ThreadReport including all selected threads
+     */
+    public List<ThreadReport> detect() {
+        Map<String, String> options = new HashMap<String, String>();
+        options.put(ORDERED_BY, "cpu");
+        return detect(options);
+    }
+
+    /**
+     * Return the current hot threads information as provided by the JVM
+     *
+     * @param options Map of options to narrow this method functionality:
+     *                Keys: ordered_by - can be "cpu", "wait" or "block"
+     *                      stacktrace_size - max depth of stack trace
+     * @return A list of ThreadReport including all selected threads
+     */
+    public List<ThreadReport> detect(Map<String, String> options) {
+        String type = "cpu";
+        if (options.containsKey(ORDERED_BY)) {
+            type = options.get(ORDERED_BY);
+            if (!isValidSortOrder(type))
+                throw new IllegalArgumentException("Invalid sort order");
+        }
+
+        Integer threadInfoMaxDepth = 3;
+        if (options.containsKey(STACKTRACE_SIZE)) {
+            threadInfoMaxDepth = Integer.valueOf(options.get(STACKTRACE_SIZE));
+        }
+
+        ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
+        enableCpuTime(threadMXBean);
+
+        Map<Long, ThreadReport> reports = new HashMap<>();
+
+        for (long threadId : threadMXBean.getAllThreadIds()) {
+            if (Thread.currentThread().getId() == threadId) {
+                continue;
+            }
+
+            long cpuTime = threadMXBean.getThreadCpuTime(threadId);
+            if (cpuTime == -1) {
+                continue;
+            }
+            ThreadInfo info = threadMXBean.getThreadInfo(threadId, threadInfoMaxDepth);
+            if (info != null) {
+                /*
+                 * Thread ID must exist and be alive, otherwise the threads just
+                 * died in the meanwhile and could be ignored.
+                 */
+                reports.put(threadId, new ThreadReport(info, cpuTime));
+            }
+        }
+        return sort(new ArrayList<>(reports.values()), type);
+     }
+
+    private List<ThreadReport> sort(List<ThreadReport> reports, final String type) {
+        reports.sort(comparatorForOrderType(type));
+        return reports;
+    }
+
+    private Comparator<ThreadReport> comparatorForOrderType(final String type){
+        if ("block".equals(type)){
+            return Comparator.comparingLong(ThreadReport::getBlockedTime).reversed();
+        } else if ("wait".equals(type)) {
+            return Comparator.comparingLong(ThreadReport::getWaitedTime).reversed();
+        } else{
+            return Comparator.comparingLong(ThreadReport::getCpuTime).reversed();
+        }
+    }
+
+    private boolean isValidSortOrder(String type) {
+        return VALID_ORDER_BY.indexOf(type.toLowerCase()) != -1;
+    }
+
+
+    private void enableCpuTime(ThreadMXBean threadMXBean) {
+        try {
+            if (threadMXBean.isThreadCpuTimeSupported()) {
+                if (!threadMXBean.isThreadCpuTimeEnabled()) {
+                    threadMXBean.setThreadCpuTimeEnabled(true);
+                }
+            }
+        } catch (SecurityException ex) {
+            // This should not happen - the security manager should not be enabled.
+            logger.debug("Cannot enable Thread Cpu Time", ex);
+        }
+    }
+}

logstash-core/src/main/java/org/logstash/instrument/monitors/MemoryMonitor.java

@@ -0,0 +1,80 @@
+package org.logstash.instrument.monitors;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryPoolMXBean;
+import java.lang.management.MemoryType;
+import java.lang.management.MemoryUsage;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Based on code created by purbon on 13/12/15.
+ */
+public class MemoryMonitor {
+
+    public enum Type {
+        All, Heap, NonHeap
+    }
+
+    public static class Report {
+
+        private static final String USAGE_INIT = "usage.init";
+        private static final String USAGE_COMMITTED = "usage.committed";
+        private static final String USAGE_USED = "usage.used";
+        private static final String USAGE_MAX = "usage.max";
+        private static final String PEAK_INIT = "peak.init";
+        private static final String PEAK_COMMITTED = "peak.committed";
+        private static final String PEAK_USED = "peak.used";
+        private static final String PEAK_MAX = "peak.max";
+
+        private Map<String, Map<String, Object>> heapMap = new HashMap<>();
+        private Map<String, Map<String, Object>> nonHeapMap = new HashMap<>();
+
+        private Report() {}
+
+        public Map<String, Map<String, Object>> getHeap() {
+            return heapMap;
+        }
+
+        public Map<String, Map<String, Object>> getNonHeap() {
+            return nonHeapMap;
+        }
+
+        void addMemoryBeanInfo(MemoryPoolMXBean bean){
+            Map<String, Map<String, Object>> memoryMap = bean.getType().equals(MemoryType.HEAP) ? heapMap : nonHeapMap;
+            Map<String, Object> beanMap = memoryMap.computeIfAbsent(bean.getName(), k -> new HashMap<>());
+            addUsage(beanMap, bean.getUsage());
+            addPeak(beanMap, bean.getPeakUsage());
+        }
+
+        private void addUsage(Map<String, Object> map, MemoryUsage usage){
+            map.put(USAGE_INIT, usage.getInit());
+            map.put(USAGE_COMMITTED, usage.getCommitted());
+            map.put(USAGE_USED, usage.getUsed());
+            map.put(USAGE_MAX, usage.getMax());
+        }
+
+        private void addPeak(Map<String, Object> map, MemoryUsage peak){
+            map.put(PEAK_INIT, peak.getInit());
+            map.put(PEAK_COMMITTED, peak.getCommitted());
+            map.put(PEAK_USED, peak.getUsed());
+            map.put(PEAK_MAX, peak.getMax());
+        }
+     }
+
+     public Report detect(Type selectType){
+        List<MemoryPoolMXBean> beans = ManagementFactory.getMemoryPoolMXBeans();
+        Report report = new Report();
+
+        beans.stream().filter(bean -> (selectType.equals(Type.All))
+                                       || !filterPool(bean.getType(), selectType))
+                      .forEach(report::addMemoryBeanInfo);
+        return report;
+    }
+
+    private boolean filterPool(MemoryType type, Type selectType) {
+       return ((selectType.equals(Type.NonHeap) && type.equals(MemoryType.HEAP))
+               || (selectType.equals(Type.Heap) && type.equals(MemoryType.NON_HEAP)));
+    }
+}

logstash-core/src/main/java/org/logstash/instrument/monitors/ProcessMonitor.java

@@ -0,0 +1,81 @@
+package org.logstash.instrument.monitors;
+
+import com.sun.management.UnixOperatingSystemMXBean;
+
+import javax.management.MBeanServer;
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Created by andrewvc on 5/12/16.
+ */
+public class ProcessMonitor {
+    private static final OperatingSystemMXBean osMxBean = ManagementFactory.getOperatingSystemMXBean();
+    private static final MBeanServer platformMxBean = ManagementFactory.getPlatformMBeanServer();
+
+    public static class Report {
+        private long memTotalVirtualInBytes = -1;
+        private short cpuSystemPercent = -4;
+        private short cpuProcessPercent = -3;
+        private long cpuMillisTotal = -1;
+        private boolean isUnix;
+        private long openFds = -1;
+        private long maxFds = -1;
+
+        private Map<String, Object> map = new HashMap<>();
+
+        Report() {
+            this.isUnix = osMxBean instanceof UnixOperatingSystemMXBean;
+            // Defaults are -1
+            if (this.isUnix) {
+                UnixOperatingSystemMXBean unixOsBean = (UnixOperatingSystemMXBean) osMxBean;;
+
+                this.openFds = unixOsBean.getOpenFileDescriptorCount();
+                this.maxFds =  unixOsBean.getMaxFileDescriptorCount();
+
+                this.cpuMillisTotal = unixOsBean.getProcessCpuTime();
+                this.cpuProcessPercent = scaleLoadToPercent(unixOsBean.getProcessCpuLoad());
+                this.cpuSystemPercent = scaleLoadToPercent(unixOsBean.getSystemCpuLoad());
+
+                this.memTotalVirtualInBytes = unixOsBean.getCommittedVirtualMemorySize();
+            }
+        }
+
+        public Map<String, Object> toMap() {
+            map.put("open_file_descriptors", this.openFds);
+            map.put("max_file_descriptors", this.maxFds);
+            map.put("is_unix", this.isUnix);
+
+            Map<String, Object> cpuMap = new HashMap<>();
+            map.put("cpu", cpuMap);
+            cpuMap.put("total_in_millis", this.cpuMillisTotal);
+            cpuMap.put("process_percent", this.cpuProcessPercent);
+            cpuMap.put("system_percent", this.cpuSystemPercent);
+
+            Map<String, Object> memoryMap = new HashMap<>();
+            map.put("mem", memoryMap);
+            memoryMap.put("total_virtual_in_bytes", this.memTotalVirtualInBytes);
+
+            return map;
+        }
+
+        private short scaleLoadToPercent(double load) {
+            if (osMxBean instanceof UnixOperatingSystemMXBean) {
+                if (load >= 0) {
+                    return (short) (load * 100);
+                } else {
+                    return -1;
+                }
+            } else {
+                return -1;
+            }
+        }
+    }
+
+    public Report detect() {
+        return new Report();
+    }
+}

logstash-core/src/main/java/org/logstash/instrument/monitors/SystemMonitor.java

@@ -0,0 +1,39 @@
+package org.logstash.instrument.monitors;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * System information as returned by the different JVM's MxBeans
+ * Created by purbon on 13/12/15.
+ */
+public class SystemMonitor {
+
+    public static class Report {
+
+        private static final String OS_NAME = "os.name";
+        private static final String OS_VERSION = "os.version";
+        private static final String OS_ARCH = "os.arch";
+        private static final String SYSTEM_AVAILABLE_PROCESSORS = "system.available_processors";
+        private static final String SYSTEM_LOAD_AVERAGE = "system.load_average";
+        private Map<String, Object> map = new HashMap<>();
+
+        Report(OperatingSystemMXBean osBean) {
+            map.put(OS_NAME, osBean.getName());
+            map.put(OS_VERSION, osBean.getVersion());
+            map.put(OS_ARCH, osBean.getArch());
+            map.put(SYSTEM_AVAILABLE_PROCESSORS, osBean.getAvailableProcessors());
+            map.put(SYSTEM_LOAD_AVERAGE, osBean.getSystemLoadAverage());
+        }
+
+        public Map<String, Object> toMap() {
+            return map;
+        }
+    }
+
+    public Report detect() {
+        return new Report(ManagementFactory.getOperatingSystemMXBean());
+    }
+}

logstash-core/src/main/java/org/logstash/instrument/reports/MemoryReport.java

@@ -0,0 +1,33 @@
+package org.logstash.instrument.reports;
+
+import org.jruby.*;
+import org.jruby.runtime.builtin.IRubyObject;
+import org.logstash.instrument.monitors.MemoryMonitor;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class MemoryReport {
+
+    private static final String NON_HEAP = "non_heap";
+    private static final String HEAP = "heap";
+
+    /**
+     * Build a report with current Memory information
+     * @return
+     */
+    public static Map<String, Map<String, Map<String, Object>>> generate() {
+        MemoryMonitor.Report report = generateReport(MemoryMonitor.Type.All);
+        Map<String, Map<String, Map<String, Object>>> container = new HashMap<>();
+        container.put(HEAP, report.getHeap());
+        container.put(NON_HEAP, report.getNonHeap());
+        return container;
+    }
+
+    private static MemoryMonitor.Report generateReport(MemoryMonitor.Type type) {
+        return new MemoryMonitor().detect(type);
+    }
+}
+

logstash-core/src/main/java/org/logstash/instrument/reports/ProcessReport.java

@@ -0,0 +1,17 @@
+package org.logstash.instrument.reports;
+
+import org.logstash.instrument.monitors.ProcessMonitor;
+
+import java.util.Map;
+
+public class ProcessReport {
+    private ProcessReport() { }
+
+    /**
+     * Build a report with current Process information
+     * @return a Map with the current process report
+     */
+    public static Map<String, Object> generate() {
+        return new ProcessMonitor().detect().toMap();
+    }
+}

logstash-core/src/main/java/org/logstash/instrument/reports/SystemReport.java

@@ -0,0 +1,21 @@
+package org.logstash.instrument.reports;
+
+
+import org.logstash.instrument.monitors.SystemMonitor;
+
+import java.util.Map;
+
+/**
+ * Created by purbon on 12/12/15.
+ */
+ public class SystemReport {
+
+    /**
+     * Build a report with current System information
+     * @return a Map with the current system report
+     */
+    public static Map<String, Object> generate() {
+        return new SystemMonitor().detect().toMap();
+    }
+}
+

logstash-core/src/main/java/org/logstash/instrument/reports/ThreadsReport.java

@@ -0,0 +1,45 @@
+package org.logstash.instrument.reports;
+
+import org.logstash.instrument.monitors.HotThreadsMonitor;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A ThreadsReport object used to hold the hot threads information
+ * Created by purbon on 12/12/15.
+ */
+public class ThreadsReport {
+
+
+    /**
+     * Generate a report with current Thread information
+     * @param options Map of options to narrow this method functionality:
+     *                Keys: ordered_by - can be "cpu", "wait" or "block"
+     *                      stacktrace_size - max depth of stack trace
+     * @return A Map containing hot threads information
+     */
+    public static Map<String, Object> generate(Map<String, String> options) {
+        HotThreadsMonitor reporter = new HotThreadsMonitor();
+        List<HotThreadsMonitor.ThreadReport> reports = reporter.detect(options);
+        return reports
+                .stream()
+                .collect(Collectors
+                    .toMap(HotThreadsMonitor.ThreadReport::getThreadName,
+                           HotThreadsMonitor.ThreadReport::toMap));
+    }
+
+
+    /**
+     * Generate a report with current Thread information
+     * @return A Map containing the hot threads information
+     */
+    public static Map<String, Object> generate() {
+        Map<String, String> options = new HashMap<>();
+        options.put("order_by", "cpu");
+        return generate(options);
+    }
+}
+

logstash-core/src/test/java/org/logstash/instruments/monitors/HotThreadMonitorTest.java

@@ -0,0 +1,98 @@
+package org.logstash.instruments.monitors;
+
+
+import org.junit.Test;
+import org.logstash.instrument.monitors.HotThreadsMonitor;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+
+public class HotThreadMonitorTest {
+
+    @Test
+    public void testThreadsReportsGenerated(){
+        assertThat(new HotThreadsMonitor().detect().size() > 0, is(true));
+    }
+
+    @Test
+    public void testAllThreadsHaveCpuTime(){
+        new HotThreadsMonitor().detect().forEach(x -> assertThat(x.toMap().keySet(), hasItem("cpu.time")));
+    }
+
+    @Test
+    public void testAllThreadsHaveThreadState(){
+        new HotThreadsMonitor().detect().forEach(x -> assertThat(x.toMap().keySet(), hasItem("thread.state")));
+    }
+
+    @Test
+    public void testAllThreadsHaveBlockedInformation(){
+        new HotThreadsMonitor().detect().forEach(x -> assertThat(x.toMap().keySet(), hasItems("blocked.count", "blocked.time")));
+    }
+
+    @Test
+    public void testAllThreadsHaveWaitedInformation(){
+        new HotThreadsMonitor().detect().forEach(x -> assertThat(x.toMap().keySet(), hasItems("waited.count", "waited.time")));
+    }
+
+    @Test
+    public void testAllThreadsHaveStackTraces(){
+        new HotThreadsMonitor().detect().forEach(tr -> assertThat(tr.toMap().keySet(), hasItem("thread.stacktrace")));
+    }
+
+    @Test
+    public void testStackTraceSizeOption(){
+        final String testStackSize = "4";
+        Map<String, String> options = new HashMap<>();
+        options.put("stacktrace_size", testStackSize);
+        new HotThreadsMonitor().detect(options).stream().filter(tr -> !tr.getThreadName().equals("Signal Dispatcher") &&
+                                                                      !tr.getThreadName().equals("Reference Handler"))
+                                                        .forEach(tr -> {
+            List stackTrace = (List)tr.toMap().get("thread.stacktrace");
+            assertThat(stackTrace.size(), is(Integer.valueOf(testStackSize)));
+        });
+    }
+
+    @Test
+    public void testOptionsOrderingCpu(){
+        Map<String, String> options = new HashMap<>();
+        options.put("ordered_by", "cpu");
+        // Using single element array to circumvent lambda expectation of 'effective final'
+        final long[] lastCpuTime = {Long.MAX_VALUE};
+        new HotThreadsMonitor().detect(options).forEach(tr -> {
+            Long cpuTime = (Long)tr.toMap().get("cpu.time");
+            assertThat(lastCpuTime[0] >= cpuTime, is(true));
+            lastCpuTime[0] = cpuTime;
+        });
+    }
+
+    @Test
+    public void testOptionsOrderingWait(){
+        Map<String, String> options = new HashMap<>();
+        options.put("ordered_by", "wait");
+        // Using single element array to circumvent lambda expectation of 'effectively final'
+        final long[] lastWaitTime = {Long.MAX_VALUE};
+        new HotThreadsMonitor().detect(options).forEach(tr -> {
+            Long waitTime = (Long)tr.toMap().get("waited.time");
+            assertThat(lastWaitTime[0] >= waitTime, is(true));
+            lastWaitTime[0] = waitTime;
+        });
+    }
+
+    @Test
+    public void testOptionsOrderingBlocked(){
+        Map<String, String> options = new HashMap<>();
+        options.put("ordered_by", "block");
+        // Using single element array to circumvent lambda expectation of 'effectively final'
+        final long[] lastBlockedTime = {Long.MAX_VALUE};
+        new HotThreadsMonitor().detect(options).forEach(tr -> {
+            Long blockedTime = (Long)tr.toMap().get("blocked.time");
+            assertThat(lastBlockedTime[0] >= blockedTime, is(true));
+            lastBlockedTime[0] = blockedTime;
+        });
+    }
+}

logstash-core/src/test/java/org/logstash/instruments/monitors/MemoryMonitorTest.java

@@ -0,0 +1,45 @@
+package org.logstash.instruments.monitors;
+
+import org.junit.Test;
+import org.logstash.instrument.monitors.MemoryMonitor;
+
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+
+public class MemoryMonitorTest {
+
+    @Test
+    public void testEachHeapSpaceRepresented() {
+        Map<String, Map<String, Object>> heap = new MemoryMonitor().detect(MemoryMonitor.Type.All).getHeap();
+        assertThat(heap, notNullValue());
+        assertThat(heap.keySet(), hasItems("PS Survivor Space", "PS Old Gen", "PS Eden Space"));
+    }
+
+    @Test
+    public void testAllStatsAreAvailableForHeap(){
+        testAllStatsAreAvailable(new MemoryMonitor().detect(MemoryMonitor.Type.All).getHeap());
+    }
+
+    @Test
+    public void testAllStatsAreAvailableForNonHeap(){
+        testAllStatsAreAvailable(new MemoryMonitor().detect(MemoryMonitor.Type.All).getNonHeap());
+    }
+
+    private void testAllStatsAreAvailable(Map<String, Map<String, Object>> stats){
+        String[] types = {"usage", "peak"};
+        String[] subtypes = {"used", "max", "committed", "init"};
+        for (Map<String, Object> spaceMap: stats.values()){
+            for (String type : types) {
+                for (String subtype : subtypes){
+                    String key = String.format("%s.%s", type, subtype);
+                    assertThat(key + " is missing", spaceMap.keySet(), hasItem(key));
+                }
+            }
+        }
+    }
+}

logstash-core/src/test/java/org/logstash/instruments/monitors/ProcessMonitorTest.java

@@ -0,0 +1,39 @@
+package org.logstash.instruments.monitors;
+
+import org.junit.Test;
+import org.logstash.instrument.monitors.ProcessMonitor;
+
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class ProcessMonitorTest {
+
+
+    @Test
+    public void testReportFDStats(){
+        Map<String, Object> processStats = new ProcessMonitor().detect().toMap();
+        assertThat("open_file_descriptors", (Long)processStats.get("open_file_descriptors") > 0L, is(true));
+        assertThat("max_file_descriptors", (Long)processStats.get("max_file_descriptors") > 0L, is(true));
+    }
+
+    @Test
+    public void testReportCpuStats(){
+        Map<String, Object> processStats = new ProcessMonitor().detect().toMap();
+        assertThat("cpu", processStats.get("cpu"), instanceOf(Map.class));
+        Map cpuStats = ((Map)processStats.get("cpu"));
+        assertThat("cpu.process_percent", (Short)cpuStats.get("process_percent") >= 0, is(true));
+        assertThat("cpu.system_percent", (Short)cpuStats.get("system_percent") >= -1, is(true));
+        assertThat("cpu.total_in_millis", (Long)cpuStats.get("total_in_millis") > 0L, is(true));
+    }
+
+    @Test
+    public void testReportMemStats() {
+        Map<String, Object> processStats = new ProcessMonitor().detect().toMap();
+        assertThat("mem", processStats.get("mem"), instanceOf(Map.class));
+        Map memStats = ((Map)processStats.get("mem"));
+        assertThat("mem.total_virtual_in_bytes", (Long)memStats.get("total_virtual_in_bytes") >= 0L, is(true));
+    }
+}

logstash-core/src/test/java/org/logstash/instruments/monitors/SystemMonitorTest.java

@@ -0,0 +1,23 @@
+package org.logstash.instruments.monitors;
+
+import org.junit.Test;
+import org.logstash.instrument.monitors.SystemMonitor;
+
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+
+public class SystemMonitorTest {
+
+    @Test
+    public void systemMonitorTest(){
+        Map<String, Object> map = new SystemMonitor().detect().toMap();
+        assertThat("system.load_average is missing", (Double)map.get("system.load_average") > 0, is(true));
+        assertThat("system.available_processors is missing ", ((Integer)map.get("system.available_processors")) > 0, is(true));
+        assertThat("os.version is missing", map.get("os.version"), allOf(notNullValue(), instanceOf(String.class)));
+        assertThat("os.arch is missing", map.get("os.arch"), allOf(notNullValue(), instanceOf(String.class)));
+        assertThat("os.name is missing", map.get("os.name"), allOf(notNullValue(), instanceOf(String.class)));
+    }
+}