ElasticSearchMonitorJob implements Job { Stats collect(ElasticSearchNodeStats.Node node) { var stats = new Stats(); double heapUsed = node.jvm.mem.heapUsedInBytes; stats.put("es_heap_used", heapUsed); double heapMax = node.jvm.mem.heapMaxInBytes; stats.put("es_heap_max", heapMax); stats.checkHighUsage(heapUsed / heapMax, highHeapUsageThreshold, "heap"); stats.put("es_non_heap_used", node.jvm.mem.nonHeapUsedInBytes); for (Map.Entry<String, ElasticSearchNodeStats.Collector> entry : node.jvm.gc.collectors.entrySet()) { ElasticSearchNodeStats.Collector collector = entry.getValue(); GCStat gcStat = gcStats.computeIfAbsent(entry.getKey(), GCStat::new); long count = gcStat.count(collector.collectionCount); long elapsed = gcStat.elapsed(collector.collectionTimeInMillis); stats.put("es_gc_" + gcStat.name + "_count", (double) count); stats.put("es_gc_" + gcStat.name + "_elapsed", (double) elapsed); } double diskUsed = node.fs.total.totalInBytes - node.fs.total.freeInBytes; stats.put("es_disk_used", diskUsed); double diskMax = node.fs.total.totalInBytes; stats.put("es_disk_max", diskMax); stats.checkHighUsage(diskUsed / diskMax, highDiskUsageThreshold, "disk"); stats.put("es_docs", node.indices.docs.count); return stats; } ElasticSearchMonitorJob(ElasticSearchClient elasticSearchClient, String app, String host, MessagePublisher<StatMessage> publisher); @Override void execute(JobContext context); public double highHeapUsageThreshold; public double highDiskUsageThreshold; }
@Test void collect() { ElasticSearchNodeStats nodeStats = JSON.fromJSON(ElasticSearchNodeStats.class, ClasspathResources.text("es-job-test/stats.json")); Stats stats = job.collect(nodeStats.nodes.get("lwOM4R6MQTOinxEVW1antA")); assertThat(stats.stats) .containsEntry("es_docs", 49029079d) .containsKeys("es_disk_used", "es_disk_max", "es_heap_used", "es_heap_max", "es_gc_young_count", "es_gc_old_elapsed"); }