diff --git a/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java b/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java index c548c5f0edf..6b9080bb2a6 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java +++ b/spark-common/src/main/java/me/lucko/spark/common/SparkPlatform.java @@ -28,6 +28,8 @@ import me.lucko.spark.common.command.sender.CommandSender; import me.lucko.spark.common.monitor.cpu.CpuMonitor; import me.lucko.spark.common.monitor.memory.GarbageCollectorStatistics; +import me.lucko.spark.common.monitor.memory.MemoryAllocationInfo; +import me.lucko.spark.common.monitor.memory.MemoryMonitor; import me.lucko.spark.common.monitor.net.NetworkMonitor; import me.lucko.spark.common.monitor.ping.PingStatistics; import me.lucko.spark.common.monitor.ping.PlayerPingProvider; @@ -35,6 +37,7 @@ import me.lucko.spark.common.monitor.tick.TickStatistics; import me.lucko.spark.common.platform.PlatformInfo; import me.lucko.spark.common.platform.PlatformStatisticsProvider; +import me.lucko.spark.common.platform.WorldMetricsCollector; import me.lucko.spark.common.sampler.BackgroundSamplerManager; import me.lucko.spark.common.sampler.SamplerContainer; import me.lucko.spark.common.sampler.source.ClassSourceLookup; @@ -83,6 +86,7 @@ public class SparkPlatform { private final TickStatistics tickStatistics; private final PingStatistics pingStatistics; private final PlatformStatisticsProvider statisticsProvider; + private final WorldMetricsCollector worldMetricsCollector; private final CommandManager commandManager; private final AtomicBoolean enabled = new AtomicBoolean(false); private Map startupGcStatistics = ImmutableMap.of(); @@ -128,6 +132,7 @@ public SparkPlatform(SparkPlugin plugin) { this.pingStatistics = pingProvider != null ? new PingStatistics(pingProvider) : null; this.statisticsProvider = new PlatformStatisticsProvider(this); + this.worldMetricsCollector = new WorldMetricsCollector(this); this.commandManager = new CommandManager(this, this.configuration); } @@ -148,7 +153,12 @@ public void enable() { if (this.pingStatistics != null) { this.pingStatistics.start(); } + + this.worldMetricsCollector.start(); + CpuMonitor.ensureMonitoring(); + MemoryMonitor.ensureMonitoring(); + MemoryAllocationInfo.ensureMonitoring(); NetworkMonitor.ensureMonitoring(); // poll startup GC statistics after plugins & the world have loaded @@ -176,6 +186,7 @@ public void disable() { if (this.pingStatistics != null) { this.pingStatistics.close(); } + this.worldMetricsCollector.close(); this.samplerContainer.close(); diff --git a/spark-common/src/main/java/me/lucko/spark/common/command/modules/HealthModule.java b/spark-common/src/main/java/me/lucko/spark/common/command/modules/HealthModule.java index 9298e9d6e8d..b0d21b88b0e 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/command/modules/HealthModule.java +++ b/spark-common/src/main/java/me/lucko/spark/common/command/modules/HealthModule.java @@ -20,6 +20,7 @@ package me.lucko.spark.common.command.modules; +import me.lucko.bytesocks.client.BytesocksClient; import me.lucko.spark.common.SparkPlatform; import me.lucko.spark.common.activitylog.Activity; import me.lucko.spark.common.command.Arguments; @@ -27,6 +28,7 @@ import me.lucko.spark.common.command.CommandModule; import me.lucko.spark.common.command.CommandResponseHandler; import me.lucko.spark.common.command.sender.CommandSender; +import me.lucko.spark.common.command.tabcomplete.CompletionSupplier; import me.lucko.spark.common.command.tabcomplete.TabCompleter; import me.lucko.spark.common.monitor.cpu.CpuMonitor; import me.lucko.spark.common.monitor.disk.DiskUsage; @@ -42,6 +44,7 @@ import me.lucko.spark.common.util.MediaTypes; import me.lucko.spark.common.util.RollingAverage; import me.lucko.spark.common.util.StatisticFormatter; +import me.lucko.spark.common.ws.HealthReportViewerSocket; import me.lucko.spark.proto.SparkProtos; import net.kyori.adventure.text.Component; import net.kyori.adventure.text.event.ClickEvent; @@ -51,6 +54,9 @@ import java.lang.management.MemoryPoolMXBean; import java.lang.management.MemoryType; import java.lang.management.MemoryUsage; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.LinkedList; import java.util.List; import java.util.Locale; @@ -74,6 +80,31 @@ public class HealthModule implements CommandModule { @Override public void registerCommands(Consumer consumer) { + consumer.accept(Command.builder() + .aliases("health", "healthreport") + .allowSubCommand(true) + .argumentUsage("dashboard", "", null) + .argumentUsage("upload", "", null) + .argumentUsage("show", "memory", null) + .argumentUsage("show", "network", null) + .executor(HealthModule::healthReport) + .tabCompleter((platform, sender, arguments) -> { + List opts = Collections.emptyList(); + if (!arguments.isEmpty()) { + String subCommand = arguments.get(0); + if (subCommand.equals("show")) { + opts = new ArrayList<>(Arrays.asList("--memory", "--network")); + opts.removeAll(arguments); + } + } + return TabCompleter.create() + .at(0, CompletionSupplier.startsWith(Arrays.asList("dashboard", "upload", "show"))) + .from(1, CompletionSupplier.startsWith(opts)) + .complete(arguments); + }) + .build() + ); + consumer.accept(Command.builder() .aliases("tps", "cpu") .executor(HealthModule::tps) @@ -88,16 +119,6 @@ public void registerCommands(Consumer consumer) { .tabCompleter((platform, sender, arguments) -> TabCompleter.completeForOpts(arguments, "--player")) .build() ); - - consumer.accept(Command.builder() - .aliases("healthreport", "health", "ht") - .argumentUsage("upload", null) - .argumentUsage("memory", null) - .argumentUsage("network", null) - .executor(HealthModule::healthReport) - .tabCompleter((platform, sender, arguments) -> TabCompleter.completeForOpts(arguments, "--upload", "--memory", "--network")) - .build() - ); } private static void tps(SparkPlatform platform, CommandSender sender, CommandResponseHandler resp, Arguments arguments) { @@ -192,12 +213,17 @@ private static void ping(SparkPlatform platform, CommandSender sender, CommandRe } private static void healthReport(SparkPlatform platform, CommandSender sender, CommandResponseHandler resp, Arguments arguments) { - resp.replyPrefixed(text("Generating server health report...")); + String subCommand = arguments.subCommand() == null ? "" : arguments.subCommand(); - if (arguments.boolFlag("upload")) { - uploadHealthReport(platform, sender, resp, arguments); - return; + if (subCommand.equals("show")) { + healthReportShow(platform, sender, resp, arguments); + } else { + healthReportUpload(platform, sender, resp, !subCommand.equals("upload")); } + } + + private static void healthReportShow(SparkPlatform platform, CommandSender sender, CommandResponseHandler resp, Arguments arguments) { + resp.replyPrefixed(text("Generating server health report...")); List report = new LinkedList<>(); report.add(empty()); @@ -223,7 +249,9 @@ private static void healthReport(SparkPlatform platform, CommandSender sender, C resp.reply(report); } - private static void uploadHealthReport(SparkPlatform platform, CommandSender sender, CommandResponseHandler resp, Arguments arguments) { + private static void healthReportUpload(SparkPlatform platform, CommandSender sender, CommandResponseHandler resp, boolean dashboard) { + resp.replyPrefixed(text("Generating server health report...")); + SparkProtos.HealthMetadata.Builder metadata = SparkProtos.HealthMetadata.newBuilder(); SparkMetadata.gather(platform, sender.toData(), platform.getStartupGcStatistics()).writeTo(metadata); @@ -235,18 +263,33 @@ private static void uploadHealthReport(SparkPlatform platform, CommandSender sen data.putAllTimeWindowStatistics(activeSampler.exportWindowStatistics()); } + if (dashboard) { + BytesocksClient bytesocksClient = platform.getBytesocksClient(); + if (bytesocksClient == null) { + resp.replyPrefixed(text("The live viewer is not supported.", RED)); + return; + } + + try { + HealthReportViewerSocket socket = new HealthReportViewerSocket(platform, bytesocksClient); + data.setChannelInfo(socket.getPayload()); + } catch (Exception e) { + resp.replyPrefixed(text("An error occurred whilst opening the live viewer connection.", RED)); + platform.getPlugin().log(Level.WARNING, "Error whilst opening live viewer connection", e); + } + } + try { String key = platform.getBytebinClient().postContent(data.build(), MediaTypes.SPARK_HEALTH_MEDIA_TYPE).key(); String url = platform.getViewerUrl() + key; - resp.broadcastPrefixed(text("Health report:", GOLD)); + resp.broadcastPrefixed(text("Health Report:", GOLD)); resp.broadcast(text() .content(url) .color(GRAY) .clickEvent(ClickEvent.openUrl(url)) .build() ); - platform.getActivityLog().addToLog(Activity.urlActivity(resp.senderData(), System.currentTimeMillis(), "Health report", url)); } catch (Exception e) { resp.broadcastPrefixed(text("An error occurred whilst uploading the data.", RED)); diff --git a/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java b/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java index 07eabfbaf28..72b3100c744 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java +++ b/spark-common/src/main/java/me/lucko/spark/common/command/modules/SamplerModule.java @@ -43,6 +43,7 @@ import me.lucko.spark.common.util.FormatUtil; import me.lucko.spark.common.util.MediaTypes; import me.lucko.spark.common.util.TimeUtil; +import me.lucko.spark.common.ws.SamplerViewerSocket; import me.lucko.spark.common.ws.ViewerSocket; import me.lucko.spark.proto.SparkSamplerProtos; import net.kyori.adventure.text.Component; @@ -463,7 +464,7 @@ private void handleUpload(SparkPlatform platform, CommandResponseHandler resp, S private void handleOpen(SparkPlatform platform, BytesocksClient bytesocksClient, CommandResponseHandler resp, Sampler sampler, Sampler.ExportProps exportProps) { try { - ViewerSocket socket = new ViewerSocket(platform, bytesocksClient, exportProps); + SamplerViewerSocket socket = new SamplerViewerSocket(platform, bytesocksClient, exportProps); sampler.attachSocket(socket); exportProps.channelInfo(socket.getPayload()); diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/Metrics.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/Metrics.java new file mode 100644 index 00000000000..489b87840e5 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/Metrics.java @@ -0,0 +1,107 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.monitor; + +import me.lucko.spark.common.sampler.window.WindowStatisticsCollector; +import me.lucko.spark.common.util.MetricSeries; +import me.lucko.spark.common.util.TimeUtil; +import me.lucko.spark.proto.SparkProtos; + +import java.time.Duration; + +/** + * A collection of metrics series used for monitoring the server. + * + *

The metric series have a fixed retention period and an interval between recordings. + * The retention period determines how long the recorded metrics are kept, + * while the interval determines how often new metrics are recorded.

+ * + *

These metrics are recorded at a higher interval than those collected by {@link WindowStatisticsCollector}.

+ */ +public enum Metrics { + ; + + /** The retention period of the metrics series. */ + private static final Duration RETENTION = Duration.ofHours(1); + + /** The interval between metric recordings. */ + public static final int INTERVAL_MILLIS = (int) Duration.ofSeconds(10).toMillis(); + + /** The estimated capacity of the metrics series. */ + private static final int INITIAL_CAPACITY = Math.toIntExact(RETENTION.toMillis() / INTERVAL_MILLIS) + 1; + + /** + * The timestamp after which metrics recording should start. + * + *

A delay avoids recording incomplete values when the server first starts

+ */ + private static final long START_RECORDING_MILLIS = TimeUtil.monotonicCurrentTimeMillis() + INTERVAL_MILLIS; + + public static final MetricSeries.Doubles TPS = new MetricSeries.Doubles(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.Averages TICK_DURATION = new MetricSeries.Averages(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.Doubles CPU_USAGE_PROCESS = new MetricSeries.Doubles(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.Doubles CPU_USAGE_SYSTEM = new MetricSeries.Doubles(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.MemoryUsages MEMORY_USAGE_HEAP = new MetricSeries.MemoryUsages(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.MemoryUsages MEMORY_USAGE_NON_HEAP = new MetricSeries.MemoryUsages(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.Doubles MEMORY_ALLOCATION = new MetricSeries.Doubles(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.WorldInfo WORLD_INFO = new MetricSeries.WorldInfo(RETENTION, INITIAL_CAPACITY); + public static final MetricSeries.Averages PLAYER_PING = new MetricSeries.Averages(RETENTION, INITIAL_CAPACITY); + + public static boolean shouldRecordTps() { + return shouldRecord(TPS, TimeUtil.monotonicCurrentTimeMillis()); + } + + public static boolean shouldRecordTickDuration() { + return shouldRecord(TICK_DURATION, TimeUtil.monotonicCurrentTimeMillis()); + } + + public static boolean shouldRecordCpuUsageProcess(long timeNow) { + return shouldRecord(CPU_USAGE_PROCESS, timeNow); + } + + public static boolean shouldRecordCpuUsageSystem(long timeNow) { + return shouldRecord(CPU_USAGE_SYSTEM, timeNow); + } + + private static boolean shouldRecord(MetricSeries series, long timeNow) { + if (timeNow < START_RECORDING_MILLIS) { + return false; + } + + long newestTimestamp = series.newestTimestamp(); + return newestTimestamp == 0 || newestTimestamp < timeNow - INTERVAL_MILLIS; + } + + public static SparkProtos.Metrics exportProto() { + SparkProtos.Metrics.Builder builder = SparkProtos.Metrics.newBuilder(); + if (!TPS.isEmpty()) builder.setTps(TPS.toProto()); + if (!TICK_DURATION.isEmpty()) builder.setTickDuration(TICK_DURATION.toProto()); + if (!CPU_USAGE_PROCESS.isEmpty()) builder.setCpuUsageProcess(CPU_USAGE_PROCESS.toProto()); + if (!CPU_USAGE_SYSTEM.isEmpty()) builder.setCpuUsageSystem(CPU_USAGE_SYSTEM.toProto()); + if (!MEMORY_USAGE_HEAP.isEmpty()) builder.setMemoryUsageHeap(MEMORY_USAGE_HEAP.toProto()); + if (!MEMORY_USAGE_NON_HEAP.isEmpty()) builder.setMemoryUsageNonHeap(MEMORY_USAGE_NON_HEAP.toProto()); + if (!MEMORY_ALLOCATION.isEmpty()) builder.setMemoryAllocation(MEMORY_ALLOCATION.toProto()); + if (!WORLD_INFO.isEmpty()) builder.setWorldInfo(WORLD_INFO.toProto()); + if (!PLAYER_PING.isEmpty()) builder.setPlayerPing(PLAYER_PING.toProto()); + return builder.build(); + } + +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/MonitoringExecutor.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/MonitoringExecutor.java index ba0659df205..d840ea49794 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/monitor/MonitoringExecutor.java +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/MonitoringExecutor.java @@ -24,10 +24,21 @@ import me.lucko.spark.common.util.SparkThreadFactory; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ThreadLocalRandom; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; public enum MonitoringExecutor { ; /** The executor used to monitor & calculate rolling averages. */ - public static final ScheduledExecutorService INSTANCE = new SparkScheduledThreadPoolExecutor(1, new SparkThreadFactory("spark-monitoring", true)); + public static final ScheduledExecutorService INSTANCE = new SparkScheduledThreadPoolExecutor(4, new SparkThreadFactory("spark-monitoring", true)); + + public static ScheduledFuture scheduleAtFixedRateMillis(Runnable command, long periodMillis) { + // schedule the task with a random initial delay to avoid all fixed rate tasks running at the same time + long delay = ThreadLocalRandom.current().nextLong(Math.min(periodMillis, 10_000L)); + return INSTANCE.scheduleAtFixedRate(command, delay, periodMillis, MILLISECONDS); + } + } diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/cpu/CpuMonitor.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/cpu/CpuMonitor.java index 987af7b5b6e..c97fb220bac 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/monitor/cpu/CpuMonitor.java +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/cpu/CpuMonitor.java @@ -20,15 +20,16 @@ package me.lucko.spark.common.monitor.cpu; +import me.lucko.spark.common.monitor.Metrics; import me.lucko.spark.common.monitor.MonitoringExecutor; import me.lucko.spark.common.util.RollingAverage; +import me.lucko.spark.common.util.TimeUtil; import javax.management.JMX; import javax.management.MBeanServer; import javax.management.ObjectName; import java.lang.management.ManagementFactory; import java.math.BigDecimal; -import java.util.concurrent.TimeUnit; /** * Exposes and monitors the system/process CPU usage. @@ -58,8 +59,7 @@ public enum CpuMonitor { throw new UnsupportedOperationException("OperatingSystemMXBean is not supported by the system", e); } - // schedule rolling average calculations. - MonitoringExecutor.INSTANCE.scheduleAtFixedRate(new RollingAverageCollectionTask(), 1, 1, TimeUnit.SECONDS); + MonitoringExecutor.scheduleAtFixedRateMillis(new PollingTask(), 1000 /* 1 second */); } /** @@ -133,7 +133,7 @@ public static double processLoad15MinAvg() { /** * Task to poll CPU loads and add to the rolling averages in the enclosing class. */ - private static final class RollingAverageCollectionTask implements Runnable { + private static final class PollingTask implements Runnable { private final RollingAverage[] systemAverages = new RollingAverage[]{ SYSTEM_AVERAGE_10_SEC, SYSTEM_AVERAGE_1_MIN, @@ -147,18 +147,29 @@ private static final class RollingAverageCollectionTask implements Runnable { @Override public void run() { - BigDecimal systemCpuLoad = new BigDecimal(systemLoad()); - BigDecimal processCpuLoad = new BigDecimal(processLoad()); + double systemLoad = systemLoad(); + double processLoad = processLoad(); + long timeMillis = TimeUtil.monotonicCurrentTimeMillis(); - if (systemCpuLoad.signum() != -1) { // if value is not negative + if (systemLoad >= 0) { + BigDecimal value = new BigDecimal(systemLoad); for (RollingAverage average : this.systemAverages) { - average.add(systemCpuLoad); + average.add(value); + } + + if (Metrics.shouldRecordCpuUsageSystem(timeMillis)) { + Metrics.CPU_USAGE_SYSTEM.record(timeMillis, systemLoad); } } - if (processCpuLoad.signum() != -1) { // if value is not negative + if (processLoad >= 0) { + BigDecimal value = new BigDecimal(processLoad); for (RollingAverage average : this.processAverages) { - average.add(processCpuLoad); + average.add(value); + } + + if (Metrics.shouldRecordCpuUsageProcess(timeMillis)) { + Metrics.CPU_USAGE_PROCESS.record(timeMillis, processLoad); } } } diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryAllocationInfo.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryAllocationInfo.java new file mode 100644 index 00000000000..c8822dc243b --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryAllocationInfo.java @@ -0,0 +1,150 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.monitor.memory; + +import com.sun.management.ThreadMXBean; +import me.lucko.spark.common.monitor.Metrics; +import me.lucko.spark.common.monitor.MonitoringExecutor; +import me.lucko.spark.common.util.RollingAverage; +import me.lucko.spark.common.util.TimeUtil; + +import java.lang.management.ManagementFactory; +import java.lang.reflect.Method; +import java.math.BigDecimal; + +/** + * A utility for accessing memory allocation information from the JVM. + */ +public enum MemoryAllocationInfo { + ; + + /** If the allocation info is supported */ + public static final boolean SUPPORTED; + + private static final ThreadMXBean BEAN; + private static final Method GET_TOTAL_THREAD_ALLOCATED_BYTES_METHOD; + + /* Bytes per second - rolling averages */ + public static final RollingAverage BPS_AVERAGE_1_MIN = new RollingAverage(60); + public static final RollingAverage BPS_AVERAGE_5_MIN = new RollingAverage(60 * 5); + public static final RollingAverage BPS_AVERAGE_15_MIN = new RollingAverage(60 * 15); + + static { + java.lang.management.ThreadMXBean bean = ManagementFactory.getThreadMXBean(); + BEAN = bean instanceof ThreadMXBean ? (ThreadMXBean) bean : null; + SUPPORTED = BEAN != null && BEAN.isThreadAllocatedMemorySupported(); + + if (SUPPORTED) { + BEAN.setThreadAllocatedMemoryEnabled(true); + } + + Method getTotalThreadAllocatedBytesMethod = null; + if (BEAN != null) { + try { + // Java 21+ + getTotalThreadAllocatedBytesMethod = ThreadMXBean.class.getMethod("getTotalThreadAllocatedBytes"); + } catch (NoSuchMethodException e) { + // ignore + } + } + GET_TOTAL_THREAD_ALLOCATED_BYTES_METHOD = getTotalThreadAllocatedBytesMethod; + + if (SUPPORTED) { + MonitoringExecutor.scheduleAtFixedRateMillis(new PollingTask(), Metrics.INTERVAL_MILLIS); + } + } + + /** + * Ensures that the static initializer has been called. + */ + @SuppressWarnings("EmptyMethod") + public static void ensureMonitoring() { + // intentionally empty + } + + /** + * Returns an approximation of the total amount of memory, in bytes, allocated + * in heap memory by all threads since the Java virtual machine started. + * The returned value is an approximation because some Java virtual machine + * implementations may use object allocation mechanisms that result in a + * delay between the time an object is allocated and the time its size is + * recorded. + * + * @return an approximation of the total memory allocated, in bytes, in + * heap memory since the Java virtual machine was started + */ + public static long getTotalThreadAllocatedBytes() { + if (!SUPPORTED) { + throw new UnsupportedOperationException("Memory allocation info is not supported"); + } + + if (GET_TOTAL_THREAD_ALLOCATED_BYTES_METHOD != null) { + try { + return (long) GET_TOTAL_THREAD_ALLOCATED_BYTES_METHOD.invoke(BEAN); + } catch (ReflectiveOperationException e) { + throw new RuntimeException(e); + } + } + + long[] threadIds = BEAN.getAllThreadIds(); + long[] allocatedBytes = BEAN.getThreadAllocatedBytes(threadIds); + + long total = 0; + for (long bytes : allocatedBytes) { + if (bytes > 0) { + total += bytes; + } + } + return total; + } + + /** + * Task to poll memory allocations. + */ + private static final class PollingTask implements Runnable { + private long previousAllocatedBytes = -1; + private long previousTimeMillis = -1; + + @Override + public void run() { + long timeMillis = TimeUtil.monotonicCurrentTimeMillis(); + long totalAllocatedBytes = getTotalThreadAllocatedBytes(); + + if (this.previousAllocatedBytes != -1) { + long allocatedBytes = totalAllocatedBytes - this.previousAllocatedBytes; + long elapsedMillis = timeMillis - this.previousTimeMillis; + + if (allocatedBytes >= 0) { + double allocatedBytesPerSecond = allocatedBytes / (elapsedMillis / 1000.0); + Metrics.MEMORY_ALLOCATION.record(timeMillis, allocatedBytesPerSecond); + + BigDecimal allocatedBytesPerSecondDecimal = new BigDecimal(allocatedBytesPerSecond); + BPS_AVERAGE_1_MIN.add(allocatedBytesPerSecondDecimal); + BPS_AVERAGE_5_MIN.add(allocatedBytesPerSecondDecimal); + BPS_AVERAGE_15_MIN.add(allocatedBytesPerSecondDecimal); + } + } + + this.previousAllocatedBytes = totalAllocatedBytes; + this.previousTimeMillis = timeMillis; + } + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryMonitor.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryMonitor.java new file mode 100644 index 00000000000..e1b959359d2 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryMonitor.java @@ -0,0 +1,64 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.monitor.memory; + +import me.lucko.spark.common.monitor.Metrics; +import me.lucko.spark.common.monitor.MonitoringExecutor; +import me.lucko.spark.common.util.TimeUtil; + +import java.lang.management.ManagementFactory; +import java.lang.management.MemoryMXBean; + +/** + * Monitors the /process memory usage. + */ +public enum MemoryMonitor { + ; + + /** The MemoryMXBean instance */ + private static final MemoryMXBean BEAN = ManagementFactory.getMemoryMXBean(); + + static { + MonitoringExecutor.scheduleAtFixedRateMillis(new PollingTask(), Metrics.INTERVAL_MILLIS); + } + + /** + * Ensures that the static initializer has been called. + */ + @SuppressWarnings("EmptyMethod") + public static void ensureMonitoring() { + // intentionally empty + } + + /** + * Task to poll memory usage. + */ + private static final class PollingTask implements Runnable { + + @Override + public void run() { + long timeMillis = TimeUtil.monotonicCurrentTimeMillis(); + Metrics.MEMORY_USAGE_HEAP.record(timeMillis, BEAN.getHeapMemoryUsage()); + Metrics.MEMORY_USAGE_NON_HEAP.record(timeMillis, BEAN.getNonHeapMemoryUsage()); + } + } + +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryInfo.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/SystemMemoryInfo.java similarity index 99% rename from spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryInfo.java rename to spark-common/src/main/java/me/lucko/spark/common/monitor/memory/SystemMemoryInfo.java index b260d7ee3af..8ccf65a9285 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/MemoryInfo.java +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/memory/SystemMemoryInfo.java @@ -32,7 +32,7 @@ /** * Utility to query information about system memory usage. */ -public enum MemoryInfo { +public enum SystemMemoryInfo { ; /** The object name of the com.sun.management.OperatingSystemMXBean */ diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/net/NetworkMonitor.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/net/NetworkMonitor.java index 79ab8d9430e..b9ac0b9f329 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/monitor/net/NetworkMonitor.java +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/net/NetworkMonitor.java @@ -49,13 +49,13 @@ public enum NetworkMonitor { private static final Map SYSTEM_AVERAGES = new ConcurrentHashMap<>(); // poll every minute, keep rolling averages for 15 mins - private static final int POLL_INTERVAL = 60; + private static final int POLL_INTERVAL_SECONDS = 60; private static final int WINDOW_SIZE_SECONDS = (int) TimeUnit.MINUTES.toSeconds(15); - private static final int WINDOW_SIZE = WINDOW_SIZE_SECONDS / POLL_INTERVAL; // 15 + private static final int WINDOW_SIZE = WINDOW_SIZE_SECONDS / POLL_INTERVAL_SECONDS; // 15 static { // schedule rolling average calculations. - MonitoringExecutor.INSTANCE.scheduleAtFixedRate(new RollingAverageCollectionTask(), 1, POLL_INTERVAL, TimeUnit.SECONDS); + MonitoringExecutor.scheduleAtFixedRateMillis(new RollingAverageCollectionTask(), POLL_INTERVAL_SECONDS * 1000L); } /** @@ -79,7 +79,7 @@ public static Map systemAverages() { * Task to poll network activity and add to the rolling averages in the enclosing class. */ private static final class RollingAverageCollectionTask implements Runnable { - private static final BigDecimal POLL_INTERVAL_DECIMAL = BigDecimal.valueOf(POLL_INTERVAL); + private static final BigDecimal POLL_INTERVAL_DECIMAL = BigDecimal.valueOf(POLL_INTERVAL_SECONDS); @Override public void run() { diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingStatistics.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingStatistics.java index 7d222c6102d..e17b11f7f28 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingStatistics.java +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingStatistics.java @@ -20,6 +20,7 @@ package me.lucko.spark.common.monitor.ping; +import me.lucko.spark.common.monitor.Metrics; import me.lucko.spark.common.monitor.MonitoringExecutor; import me.lucko.spark.common.util.RollingAverage; import org.jspecify.annotations.Nullable; @@ -56,7 +57,7 @@ public void start() { if (this.future != null) { throw new IllegalStateException(); } - this.future = MonitoringExecutor.INSTANCE.scheduleAtFixedRate(this, QUERY_RATE_SECONDS, QUERY_RATE_SECONDS, TimeUnit.SECONDS); + this.future = MonitoringExecutor.scheduleAtFixedRateMillis(this, QUERY_RATE_SECONDS * 1000L); } @Override @@ -75,6 +76,7 @@ public void run() { } this.rollingAverage.add(BigDecimal.valueOf(summary.median())); + Metrics.PLAYER_PING.record(summary.toDoubleAverage()); } /** diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingSummary.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingSummary.java index 024d27d9c7f..dcc9e0ed2d3 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingSummary.java +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/ping/PingSummary.java @@ -20,6 +20,8 @@ package me.lucko.spark.common.monitor.ping; +import me.lucko.spark.common.util.ImmutableDoubleAverageInfo; + import java.util.Arrays; public final class PingSummary { @@ -78,4 +80,8 @@ public double percentile95th() { return percentile(0.95d); } + public ImmutableDoubleAverageInfo toDoubleAverage() { + return new ImmutableDoubleAverageInfo(this.mean(), this.max(), this.min(), this.median(), this.percentile95th()); + } + } diff --git a/spark-common/src/main/java/me/lucko/spark/common/monitor/tick/SparkTickStatistics.java b/spark-common/src/main/java/me/lucko/spark/common/monitor/tick/SparkTickStatistics.java index db080c31a1d..d0811fcba9b 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/monitor/tick/SparkTickStatistics.java +++ b/spark-common/src/main/java/me/lucko/spark/common/monitor/tick/SparkTickStatistics.java @@ -21,6 +21,8 @@ package me.lucko.spark.common.monitor.tick; import me.lucko.spark.api.statistic.misc.DoubleAverageInfo; +import me.lucko.spark.common.monitor.Metrics; +import me.lucko.spark.common.monitor.MonitoringExecutor; import me.lucko.spark.common.tick.TickHook; import me.lucko.spark.common.tick.TickReporter; import me.lucko.spark.common.util.RollingAverage; @@ -93,6 +95,10 @@ public void onTick(int currentTick) { rollingAverage.add(currentTps, diff, total); } + if (Metrics.shouldRecordTps()) { + Metrics.TPS.record(this.tps10Sec.getAverage()); + } + this.last = now; } @@ -103,6 +109,11 @@ public void onTick(double duration) { for (RollingAverage rollingAverage : this.tickDurationAverages) { rollingAverage.add(decimal); } + + if (Metrics.shouldRecordTickDuration() && this.tickDuration1Min.getSamples() > 0) { + // mean/max/min/median/95th are expensive to calculate, so do that async to avoid blocking main thread + MonitoringExecutor.INSTANCE.execute(() -> Metrics.TICK_DURATION.record(this.tickDuration1Min.toImmutable())); + } } @Override diff --git a/spark-common/src/main/java/me/lucko/spark/common/platform/PlatformStatisticsProvider.java b/spark-common/src/main/java/me/lucko/spark/common/platform/PlatformStatisticsProvider.java index 95fc8330267..a5d4b6a5714 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/platform/PlatformStatisticsProvider.java +++ b/spark-common/src/main/java/me/lucko/spark/common/platform/PlatformStatisticsProvider.java @@ -27,7 +27,8 @@ import me.lucko.spark.common.monitor.cpu.CpuMonitor; import me.lucko.spark.common.monitor.disk.DiskUsage; import me.lucko.spark.common.monitor.memory.GarbageCollectorStatistics; -import me.lucko.spark.common.monitor.memory.MemoryInfo; +import me.lucko.spark.common.monitor.memory.MemoryAllocationInfo; +import me.lucko.spark.common.monitor.memory.SystemMemoryInfo; import me.lucko.spark.common.monitor.net.NetworkInterfaceAverages; import me.lucko.spark.common.monitor.net.NetworkMonitor; import me.lucko.spark.common.monitor.os.OperatingSystemInfo; @@ -84,13 +85,13 @@ public SystemStatistics getSystemStatistics() { ) .setMemory(SystemStatistics.Memory.newBuilder() .setPhysical(SystemStatistics.Memory.MemoryPool.newBuilder() - .setUsed(MemoryInfo.getUsedPhysicalMemory()) - .setTotal(MemoryInfo.getTotalPhysicalMemory()) + .setUsed(SystemMemoryInfo.getUsedPhysicalMemory()) + .setTotal(SystemMemoryInfo.getTotalPhysicalMemory()) .build() ) .setSwap(SystemStatistics.Memory.MemoryPool.newBuilder() - .setUsed(MemoryInfo.getUsedSwap()) - .setTotal(MemoryInfo.getTotalSwap()) + .setUsed(SystemMemoryInfo.getUsedSwap()) + .setTotal(SystemMemoryInfo.getTotalSwap()) .build() ) .build() @@ -152,7 +153,10 @@ public PlatformStatistics getPlatformStatistics(Map memoryPoolMXBeans = ManagementFactory.getMemoryPoolMXBeans(); for (MemoryPoolMXBean memoryPool : memoryPoolMXBeans) { @@ -268,8 +272,8 @@ public static SparkProtos.RollingAverageValues rollingAvgProto(DoubleAverageInfo .build(); } - public static PlatformStatistics.Memory.MemoryUsage memoryUsageProto(MemoryUsage usage) { - return PlatformStatistics.Memory.MemoryUsage.newBuilder() + public static SparkProtos.MemoryUsage memoryUsageProto(MemoryUsage usage) { + return SparkProtos.MemoryUsage.newBuilder() .setUsed(usage.getUsed()) .setCommitted(usage.getCommitted()) .setInit(usage.getInit()) diff --git a/spark-common/src/main/java/me/lucko/spark/common/platform/SparkMetadata.java b/spark-common/src/main/java/me/lucko/spark/common/platform/SparkMetadata.java index 36f942ce563..1ca8ccc656e 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/platform/SparkMetadata.java +++ b/spark-common/src/main/java/me/lucko/spark/common/platform/SparkMetadata.java @@ -22,10 +22,12 @@ import me.lucko.spark.common.SparkPlatform; import me.lucko.spark.common.command.sender.CommandSender; +import me.lucko.spark.common.monitor.Metrics; import me.lucko.spark.common.monitor.memory.GarbageCollectorStatistics; import me.lucko.spark.common.platform.serverconfig.ServerConfigProvider; import me.lucko.spark.common.sampler.source.SourceMetadata; import me.lucko.spark.proto.SparkHeapProtos.HeapMetadata; +import me.lucko.spark.proto.SparkProtos; import me.lucko.spark.proto.SparkProtos.HealthMetadata; import me.lucko.spark.proto.SparkProtos.PlatformMetadata; import me.lucko.spark.proto.SparkProtos.PlatformStatistics; @@ -80,7 +82,9 @@ public static SparkMetadata gather(SparkPlatform platform, CommandSender.Data cr platform.getPlugin().log(Level.WARNING, "Failed to gather extra platform metadata", e); } - return new SparkMetadata(creator, platformMetadata, platformStatistics, systemStatistics, generatedTime, serverConfigurations, sources, extraPlatformMetadata); + SparkProtos.Metrics metrics = Metrics.exportProto(); + + return new SparkMetadata(creator, platformMetadata, platformStatistics, systemStatistics, generatedTime, serverConfigurations, sources, extraPlatformMetadata, metrics); } private final CommandSender.Data creator; @@ -91,8 +95,9 @@ public static SparkMetadata gather(SparkPlatform platform, CommandSender.Data cr private final Map serverConfigurations; private final Collection sources; private final Map extraPlatformMetadata; + private final SparkProtos.Metrics metrics; - public SparkMetadata(CommandSender.Data creator, PlatformMetadata platformMetadata, PlatformStatistics platformStatistics, SystemStatistics systemStatistics, long generatedTime, Map serverConfigurations, Collection sources, Map extraPlatformMetadata) { + public SparkMetadata(CommandSender.Data creator, PlatformMetadata platformMetadata, PlatformStatistics platformStatistics, SystemStatistics systemStatistics, long generatedTime, Map serverConfigurations, Collection sources, Map extraPlatformMetadata, SparkProtos.Metrics metrics) { this.creator = creator; this.platformMetadata = platformMetadata; this.platformStatistics = platformStatistics; @@ -101,6 +106,7 @@ public SparkMetadata(CommandSender.Data creator, PlatformMetadata platformMetada this.serverConfigurations = serverConfigurations; this.sources = sources; this.extraPlatformMetadata = extraPlatformMetadata; + this.metrics = metrics; } @SuppressWarnings("DuplicatedCode") @@ -117,6 +123,7 @@ public void writeTo(HealthMetadata.Builder builder) { } } if (this.extraPlatformMetadata != null) builder.putAllExtraPlatformMetadata(this.extraPlatformMetadata); + if (this.metrics != null) builder.setMetrics(this.metrics); } @SuppressWarnings("DuplicatedCode") @@ -133,6 +140,7 @@ public void writeTo(SamplerMetadata.Builder builder) { } } if (this.extraPlatformMetadata != null) builder.putAllExtraPlatformMetadata(this.extraPlatformMetadata); + if (this.metrics != null) builder.setMetrics(this.metrics); } @SuppressWarnings("DuplicatedCode") @@ -149,6 +157,7 @@ public void writeTo(HeapMetadata.Builder builder) { } } if (this.extraPlatformMetadata != null) builder.putAllExtraPlatformMetadata(this.extraPlatformMetadata); + if (this.metrics != null) builder.setMetrics(this.metrics); } } diff --git a/spark-common/src/main/java/me/lucko/spark/common/platform/WorldMetricsCollector.java b/spark-common/src/main/java/me/lucko/spark/common/platform/WorldMetricsCollector.java new file mode 100644 index 00000000000..dc4cc1cd789 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/platform/WorldMetricsCollector.java @@ -0,0 +1,61 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.platform; + +import me.lucko.spark.common.SparkPlatform; +import me.lucko.spark.common.monitor.Metrics; +import me.lucko.spark.common.monitor.MonitoringExecutor; +import me.lucko.spark.common.platform.world.AsyncWorldInfoProvider; +import me.lucko.spark.common.platform.world.WorldInfoProvider; + +import java.util.concurrent.ScheduledFuture; + +public class WorldMetricsCollector implements Runnable, AutoCloseable { + private final AsyncWorldInfoProvider infoProvider; + private ScheduledFuture task; + + public WorldMetricsCollector(SparkPlatform platform) { + WorldInfoProvider worldInfoProvider = platform.getPlugin().createWorldInfoProvider(); + this.infoProvider = worldInfoProvider == WorldInfoProvider.NO_OP ? null : new AsyncWorldInfoProvider(platform, worldInfoProvider); + } + + public void start() { + if (this.infoProvider == null) { + return; + } + this.task = MonitoringExecutor.scheduleAtFixedRateMillis(this, Metrics.INTERVAL_MILLIS); + } + + @Override + public void run() { + WorldInfoProvider.CountsResult counts = this.infoProvider.getCounts(); + if (counts != null) { + Metrics.WORLD_INFO.record(counts); + } + } + + @Override + public void close() { + if (this.task != null) { + this.task.cancel(false); + } + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/AbstractSampler.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/AbstractSampler.java index 4f0c008a378..f2b007149cd 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/AbstractSampler.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/AbstractSampler.java @@ -22,6 +22,7 @@ import me.lucko.spark.common.SparkPlatform; import me.lucko.spark.common.command.sender.CommandSender; +import me.lucko.spark.common.monitor.Metrics; import me.lucko.spark.common.monitor.memory.GarbageCollectorStatistics; import me.lucko.spark.common.platform.SparkMetadata; import me.lucko.spark.common.sampler.aggregator.DataAggregator; @@ -32,6 +33,7 @@ import me.lucko.spark.common.sampler.window.WindowStatisticsCollector; import me.lucko.spark.common.util.TimeUtil; import me.lucko.spark.common.util.classfinder.ClassFinder; +import me.lucko.spark.common.ws.SamplerViewerSocket; import me.lucko.spark.common.ws.ViewerSocket; import me.lucko.spark.proto.SparkProtos; import me.lucko.spark.proto.SparkSamplerProtos.SamplerData; @@ -80,7 +82,7 @@ public abstract class AbstractSampler implements Sampler { protected Map initialGcStats; /** A set of viewer sockets linked to the sampler */ - protected List viewerSockets = new CopyOnWriteArrayList<>(); + protected List viewerSockets = new CopyOnWriteArrayList<>(); protected AbstractSampler(SparkPlatform platform, SamplerSettings settings) { this.platform = platform; @@ -135,18 +137,18 @@ public void start() { @Override public void stop(boolean cancelled) { this.windowStatisticsCollector.stop(); - for (ViewerSocket viewerSocket : this.viewerSockets) { + for (SamplerViewerSocket viewerSocket : this.viewerSockets) { viewerSocket.processSamplerStopped(this); } } @Override - public void attachSocket(ViewerSocket socket) { + public void attachSocket(SamplerViewerSocket socket) { this.viewerSockets.add(socket); } @Override - public Collection getAttachedSockets() { + public Collection getAttachedSockets() { return this.viewerSockets; } @@ -170,9 +172,10 @@ protected void sendStatisticsToSocket() { SparkProtos.PlatformStatistics platform = this.platform.getStatisticsProvider().getPlatformStatistics(getInitialGcStats(), false); SparkProtos.SystemStatistics system = this.platform.getStatisticsProvider().getSystemStatistics(); + SparkProtos.Metrics metrics = Metrics.exportProto(); for (ViewerSocket viewerSocket : this.viewerSockets) { - viewerSocket.sendUpdatedStatistics(platform, system); + viewerSocket.sendUpdatedStatistics(platform, system, metrics); } } catch (Exception e) { this.platform.getPlugin().log(Level.WARNING, "Exception occurred while sending statistics to viewer", e); diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java index 50e6c58f281..85991ed1f76 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/Sampler.java @@ -24,10 +24,10 @@ import me.lucko.spark.common.command.sender.CommandSender; import me.lucko.spark.common.sampler.java.MergeStrategy; import me.lucko.spark.common.sampler.source.ClassSourceLookup; -import me.lucko.spark.common.ws.ViewerSocket; +import me.lucko.spark.common.ws.SamplerViewerSocket; import me.lucko.spark.proto.SparkProtos; +import me.lucko.spark.proto.SparkProtos.SocketChannelInfo; import me.lucko.spark.proto.SparkSamplerProtos.SamplerData; -import me.lucko.spark.proto.SparkSamplerProtos.SocketChannelInfo; import java.util.Collection; import java.util.Map; @@ -54,14 +54,14 @@ public interface Sampler { * * @param socket the socket */ - void attachSocket(ViewerSocket socket); + void attachSocket(SamplerViewerSocket socket); /** * Gets the sockets attached to this sampler. * * @return the attached sockets */ - Collection getAttachedSockets(); + Collection getAttachedSockets(); /** * Gets the time when the sampler started (unix timestamp in millis) diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java index 7eab25a75af..5055e2d7448 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/async/AsyncSampler.java @@ -31,7 +31,7 @@ import me.lucko.spark.common.util.SparkScheduledThreadPoolExecutor; import me.lucko.spark.common.util.SparkThreadFactory; import me.lucko.spark.common.util.TimeUtil; -import me.lucko.spark.common.ws.ViewerSocket; +import me.lucko.spark.common.ws.SamplerViewerSocket; import me.lucko.spark.proto.SparkSamplerProtos.SamplerData; import java.util.Locale; @@ -241,7 +241,7 @@ public void stop(boolean cancelled) { } @Override - public void attachSocket(ViewerSocket socket) { + public void attachSocket(SamplerViewerSocket socket) { super.attachSocket(socket); if (this.socketStatisticsTask == null) { diff --git a/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaSampler.java b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaSampler.java index 6b0fda0931d..bc8ae41aab2 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaSampler.java +++ b/spark-common/src/main/java/me/lucko/spark/common/sampler/java/JavaSampler.java @@ -32,7 +32,7 @@ import me.lucko.spark.common.util.SparkScheduledThreadPoolExecutor; import me.lucko.spark.common.util.SparkThreadFactory; import me.lucko.spark.common.util.TimeUtil; -import me.lucko.spark.common.ws.ViewerSocket; +import me.lucko.spark.common.ws.SamplerViewerSocket; import me.lucko.spark.proto.SparkSamplerProtos.SamplerData; import java.lang.management.ManagementFactory; @@ -136,7 +136,7 @@ public void run() { } @Override - public void attachSocket(ViewerSocket socket) { + public void attachSocket(SamplerViewerSocket socket) { super.attachSocket(socket); if (this.socketStatisticsTask == null) { diff --git a/spark-common/src/main/java/me/lucko/spark/common/util/ImmutableDoubleAverageInfo.java b/spark-common/src/main/java/me/lucko/spark/common/util/ImmutableDoubleAverageInfo.java new file mode 100644 index 00000000000..ce8a596f477 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/util/ImmutableDoubleAverageInfo.java @@ -0,0 +1,77 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.util; + +import me.lucko.spark.api.statistic.misc.DoubleAverageInfo; + +public class ImmutableDoubleAverageInfo implements DoubleAverageInfo { + private final double mean; + private final double max; + private final double min; + private final double median; + private final double percentile95th; + + public ImmutableDoubleAverageInfo(double mean, double max, double min, double median, double percentile95th) { + this.mean = mean; + this.max = max; + this.min = min; + this.median = median; + this.percentile95th = percentile95th; + } + + public ImmutableDoubleAverageInfo(DoubleAverageInfo other) { + this.mean = other.mean(); + this.max = other.max(); + this.min = other.min(); + this.median = other.median(); + this.percentile95th = other.percentile95th(); + } + + @Override + public double mean() { + return this.mean; + } + + @Override + public double max() { + return this.max; + } + + @Override + public double min() { + return this.min; + } + + @Override + public double median() { + return this.median; + } + + @Override + public double percentile95th() { + return this.percentile95th; + } + + @Override + public double percentile(double percentile) { + throw new UnsupportedOperationException(); + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/util/MetricSeries.java b/spark-common/src/main/java/me/lucko/spark/common/util/MetricSeries.java new file mode 100644 index 00000000000..0940fac40b2 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/util/MetricSeries.java @@ -0,0 +1,398 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.util; + +import com.google.common.primitives.Ints; +import me.lucko.spark.api.statistic.misc.DoubleAverageInfo; +import me.lucko.spark.common.platform.PlatformStatisticsProvider; +import me.lucko.spark.common.platform.world.WorldInfoProvider; +import me.lucko.spark.proto.SparkProtos; + +import java.lang.management.MemoryUsage; +import java.time.Duration; +import java.util.concurrent.locks.ReentrantLock; + +/** + * A metric time series backed by an array ring buffer. + */ +public class MetricSeries { + + /** The retention period of the series, in milliseconds */ + private final long retentionMillis; + + /** Lock for synchronizing access to the series */ + private final ReentrantLock lock; + + /* + * The ring buffer of samples, stored in chronological order. + * + *

The oldest samples are at index {@code head}, + * and the newest samples are at index {@code (head + size - 1) % timestamps.length}.

+ */ + private long[] timestamps; + private Object[] values; + + /** Index of the oldest sample */ + private int head; + + /** Number of valid samples currently stored */ + private int size; + + /** Timestamp of the newest sample, or 0 if the series is empty */ + private volatile long newestTimestamp; + + public MetricSeries(Duration retention, int initialCapacity) { + long retentionMillis = retention.toMillis(); + if (retentionMillis <= 0) { + throw new IllegalArgumentException("retention must be > 0"); + } + if (initialCapacity <= 0) { + throw new IllegalArgumentException("initialCapacity must be > 0"); + } + + this.retentionMillis = retentionMillis; + this.lock = new ReentrantLock(); + this.timestamps = new long[initialCapacity]; + this.values = new Object[initialCapacity]; + } + + /** + * Record a sample. + * + *

Samples must be supplied in chronological order.

+ * + * @param timestampMillis monotonic timestamp + * @param value metric value + * @see TimeUtil#monotonicCurrentTimeMillis() for getting a current monotonic timestamp + */ + public void record(long timestampMillis, T value) { + if (timestampMillis <= 0) { + throw new IllegalArgumentException("timestampMillis must be > 0"); + } + + this.lock.lock(); + try { + prune(timestampMillis - this.retentionMillis); + ensureCapacity(); + + int index = (this.head + this.size) % this.timestamps.length; + this.timestamps[index] = timestampMillis; + this.values[index] = value; + + this.size++; + this.newestTimestamp = timestampMillis; + } finally { + this.lock.unlock(); + } + } + + /** + * Record a sample. + * + * @param value metric value + */ + public void record(T value) { + record(TimeUtil.monotonicCurrentTimeMillis(), value); + } + + /** + * Removes all samples older than {@code cutoff}. + * + *

Must be called with lock held.

+ */ + private void prune(long cutoff) { + while (this.size > 0 && this.timestamps[this.head] < cutoff) { + this.values[this.head] = null; + this.head++; + + if (this.head == this.timestamps.length) { + this.head = 0; + } + + this.size--; + } + } + + /** + * Calculates the index in the ring buffer for the given logical index. + * + *

Must be called with lock held.

+ * + * @param i the logical index (0 = oldest sample, size-1 = newest sample) + * @return the index in the ring buffer + */ + private int indexFor(int i) { + int index = this.head + i; + if (index >= this.timestamps.length) { + index -= this.timestamps.length; + } + return index; + } + + /** + * Returns the value at the given index in the ring buffer. + * + *

Must be called with lock held.

+ * + * @param index the index in the ring buffer + * @return the value at the given index + */ + @SuppressWarnings("unchecked") + private T valueAt(int index) { + return (T) this.values[index]; + } + + /** + * Grows the ring buffer if it is full. + * + *

Must be called with lock held.

+ */ + private void ensureCapacity() { + if (this.size < this.timestamps.length) { + return; + } + + int oldCapacity = this.timestamps.length; + if (oldCapacity > Integer.MAX_VALUE / 2) { + throw new IllegalStateException("Metric series is too large"); + } + + int newCapacity = oldCapacity * 2; + + long[] newTimestamps = new long[newCapacity]; + Object[] newValues = new Object[newCapacity]; + + // copy samples from the old ring buffer to the new one, in chronological order + for (int i = 0; i < this.size; i++) { + int oldIndex = (this.head + i) % oldCapacity; + + newTimestamps[i] = this.timestamps[oldIndex]; + newValues[i] = this.values[oldIndex]; + } + + this.timestamps = newTimestamps; + this.values = newValues; + this.head = 0; + } + + public int size() { + this.lock.lock(); + try { + return this.size; + } finally { + this.lock.unlock(); + } + } + + public boolean isEmpty() { + this.lock.lock(); + try { + return this.size == 0; + } finally { + this.lock.unlock(); + } + } + + /** + * Returns the timestamp of the oldest sample in the series, or 0 if the series is empty. + * + * @return the timestamp of the oldest sample, or 0 if empty + */ + public long oldestTimestamp() { + this.lock.lock(); + try { + if (this.size == 0) { + return 0; + } + return this.timestamps[this.head]; + } finally { + this.lock.unlock(); + } + } + + /** + * Returns the timestamp of the newest sample in the series, or 0 if the series is empty. + * + * @return the timestamp of the newest sample, or 0 if empty + */ + public long newestTimestamp() { + return this.newestTimestamp; + } + + /** + * Iterates through samples in chronological order. + * + *

The callback should not call record() on this same series.

+ */ + public void forEach(Consumer consumer) { + this.lock.lock(); + try { + for (int i = 0; i < this.size; i++) { + int index = indexFor(i); + consumer.accept(this.timestamps[index], valueAt(index)); + } + } finally { + this.lock.unlock(); + } + } + + public interface Consumer { + void accept(long timestampMillis, T value); + } + + /** + * Exports the series as a compact representation. + * + * @return an export of the series + */ + public Export export() { + this.lock.lock(); + try { + Export export = new Export(0, new int[this.size], new Object[this.size]); + + long lastTimestamp = 0; + for (int i = 0; i < this.size; i++) { + int index = indexFor(i); + + if (i == 0) { + // first value - set the start timestamp and record a delta of 0 + export.startTimestampMs = this.timestamps[index]; + export.timestampDeltasMs[i] = 0; + } else { + long delta = this.timestamps[index] - lastTimestamp; + if (delta < 0 || delta > 0xFFFFFFFFL) { + throw new IllegalStateException("Timestamp delta cannot be represented as uint32: " + delta); + } + export.timestampDeltasMs[i] = (int) delta; + } + + lastTimestamp = this.timestamps[index]; + export.values[i] = this.values[index]; + } + + return export; + } finally { + this.lock.unlock(); + } + } + + public static final class Export { + private long startTimestampMs; + private final int[] timestampDeltasMs; + private final Object[] values; + + Export(long startTimestampMs, int[] timestampDeltasMs, Object[] values) { + this.startTimestampMs = startTimestampMs; + this.timestampDeltasMs = timestampDeltasMs; + this.values = values; + } + + public long startTimestampMs() { + return this.startTimestampMs; + } + + public int[] timestampDeltasMs() { + return this.timestampDeltasMs; + } + + public Object[] values() { + return this.values; + } + } + + public static class Doubles extends MetricSeries { + public Doubles(Duration retention, int initialCapacity) { + super(retention, initialCapacity); + } + + public SparkProtos.DoubleMetricSeries toProto() { + Export export = export(); + SparkProtos.DoubleMetricSeries.Builder builder = SparkProtos.DoubleMetricSeries.newBuilder() + .setStartTimestampMs(export.startTimestampMs()) + .addAllTimestampDeltasMs(Ints.asList(export.timestampDeltasMs())); + for (Object value : export.values()) { + builder.addValues((double) value); + } + return builder.build(); + } + } + + public static class Averages extends MetricSeries { + public Averages(Duration retention, int initialCapacity) { + super(retention, initialCapacity); + } + + public SparkProtos.AveragesMetricSeries toProto() { + Export export = export(); + SparkProtos.AveragesMetricSeries.Builder builder = SparkProtos.AveragesMetricSeries.newBuilder() + .setStartTimestampMs(export.startTimestampMs()) + .addAllTimestampDeltasMs(Ints.asList(export.timestampDeltasMs())); + for (Object value : export.values()) { + DoubleAverageInfo avgInfo = (DoubleAverageInfo) value; + builder.addValues(PlatformStatisticsProvider.rollingAvgProto(avgInfo)); + } + return builder.build(); + } + } + + public static class MemoryUsages extends MetricSeries { + public MemoryUsages(Duration retention, int initialCapacity) { + super(retention, initialCapacity); + } + + public SparkProtos.MemoryUsageMetricSeries toProto() { + Export export = export(); + SparkProtos.MemoryUsageMetricSeries.Builder builder = SparkProtos.MemoryUsageMetricSeries.newBuilder() + .setStartTimestampMs(export.startTimestampMs()) + .addAllTimestampDeltasMs(Ints.asList(export.timestampDeltasMs())); + for (Object value : export.values()) { + MemoryUsage memoryUsage = (MemoryUsage) value; + builder.addValues(PlatformStatisticsProvider.memoryUsageProto(memoryUsage)); + } + return builder.build(); + } + } + + public static class WorldInfo extends MetricSeries { + public WorldInfo(Duration retention, int initialCapacity) { + super(retention, initialCapacity); + } + + public SparkProtos.WorldInfoMetricSeries toProto() { + Export export = export(); + SparkProtos.WorldInfoMetricSeries.Builder builder = SparkProtos.WorldInfoMetricSeries.newBuilder() + .setStartTimestampMs(export.startTimestampMs()) + .addAllTimestampDeltasMs(Ints.asList(export.timestampDeltasMs())); + for (Object value : export.values()) { + WorldInfoProvider.CountsResult countsResult = (WorldInfoProvider.CountsResult) value; + builder.addValues(SparkProtos.WorldInfoMetricSeries.Values.newBuilder() + .setPlayers(countsResult.players()) + .setEntities(countsResult.entities()) + .setTileEntities(countsResult.tileEntities()) + .setChunks(countsResult.chunks()) + .build() + ); + } + return builder.build(); + } + } + +} \ No newline at end of file diff --git a/spark-common/src/main/java/me/lucko/spark/common/util/RollingAverage.java b/spark-common/src/main/java/me/lucko/spark/common/util/RollingAverage.java index 57dfdfff83c..69d1692e6a9 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/util/RollingAverage.java +++ b/spark-common/src/main/java/me/lucko/spark/common/util/RollingAverage.java @@ -111,4 +111,10 @@ public double percentile(double percentile) { return sortedSamples[rank].doubleValue(); } + public ImmutableDoubleAverageInfo toImmutable() { + synchronized (this) { + return new ImmutableDoubleAverageInfo(this); + } + } + } diff --git a/spark-common/src/main/java/me/lucko/spark/common/ws/HealthReportViewerSocket.java b/spark-common/src/main/java/me/lucko/spark/common/ws/HealthReportViewerSocket.java new file mode 100644 index 00000000000..da2cde96800 --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/ws/HealthReportViewerSocket.java @@ -0,0 +1,71 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.ws; + +import me.lucko.bytesocks.client.BytesocksClient; +import me.lucko.spark.common.SparkPlatform; +import me.lucko.spark.common.monitor.Metrics; +import me.lucko.spark.common.util.SparkScheduledThreadPoolExecutor; +import me.lucko.spark.common.util.SparkThreadFactory; +import me.lucko.spark.proto.SparkProtos; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; + +/** + * Represents a 'health report' connection with the spark viewer. + */ +public class HealthReportViewerSocket extends ViewerSocket { + + private final ScheduledExecutorService scheduler = new SparkScheduledThreadPoolExecutor(1, new SparkThreadFactory("spark-heath-report-socket-worker", false)); + + public HealthReportViewerSocket(SparkPlatform platform, BytesocksClient client) throws Exception { + super(platform, client); + this.scheduler.scheduleAtFixedRate(this::tryTick, 10, 10, TimeUnit.SECONDS); + } + + public void tryTick() { + try { + tick(); + } catch (Exception e) { + this.platform.getPlugin().log(Level.WARNING, "Error whilst sending updated statistics to the socket", e); + } + } + + public void tick() { + if (checkShouldClose()) { + return; + } + + SparkProtos.PlatformStatistics platform = this.platform.getStatisticsProvider().getPlatformStatistics(this.platform.getStartupGcStatistics(), false); + SparkProtos.SystemStatistics system = this.platform.getStatisticsProvider().getSystemStatistics(); + SparkProtos.Metrics metrics = Metrics.exportProto(); + + sendUpdatedStatistics(platform, system, metrics); + } + + @Override + public void close() { + this.scheduler.shutdownNow(); + super.close(); + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/ws/SamplerViewerSocket.java b/spark-common/src/main/java/me/lucko/spark/common/ws/SamplerViewerSocket.java new file mode 100644 index 00000000000..b64d3a6a12c --- /dev/null +++ b/spark-common/src/main/java/me/lucko/spark/common/ws/SamplerViewerSocket.java @@ -0,0 +1,90 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.ws; + +import me.lucko.bytesocks.client.BytesocksClient; +import me.lucko.spark.common.SparkPlatform; +import me.lucko.spark.common.sampler.AbstractSampler; +import me.lucko.spark.common.sampler.Sampler; +import me.lucko.spark.common.util.MediaTypes; +import me.lucko.spark.proto.SparkSamplerProtos; +import me.lucko.spark.proto.SparkWebSocketProtos; + +import java.util.logging.Level; + +/** + * Represents a 'sampler' connection with the spark viewer. + */ +public class SamplerViewerSocket extends ViewerSocket { + + /** The export props to use when exporting the sampler data */ + private final Sampler.ExportProps exportProps; + + public SamplerViewerSocket(SparkPlatform platform, BytesocksClient client, Sampler.ExportProps exportProps) throws Exception { + super(platform, client); + this.exportProps = exportProps; + } + + /** + * Called each time the sampler rotates to a new window. + * + * @param sampler the sampler + */ + public void processWindowRotate(AbstractSampler sampler) { + if (checkShouldClose()) { + return; + } + + try { + SparkSamplerProtos.SamplerData samplerData = sampler.toProto(this.platform, this.exportProps); + String key = this.platform.getBytebinClient().postContent(samplerData, MediaTypes.SPARK_SAMPLER_MEDIA_TYPE, "live").key(); + sendUpdatedSamplerData(key); + } catch (Exception e) { + this.platform.getPlugin().log(Level.WARNING, "Error whilst sending updated sampler data to the socket", e); + } + } + + /** + * Called when the sampler stops. + * + * @param sampler the sampler + */ + public void processSamplerStopped(AbstractSampler sampler) { + if (isClosed()) { + return; + } + + close(); + } + + /** + * Sends a message to the socket to indicate that updated sampler data is available + * + * @param payloadId the payload id of the updated data + */ + public void sendUpdatedSamplerData(String payloadId) { + this.socket.sendPacket(builder -> builder.setServerUpdateSampler(SparkWebSocketProtos.ServerUpdateSamplerData.newBuilder() + .setPayloadId(payloadId) + .build() + )); + setLastPayloadId(payloadId); + } +} diff --git a/spark-common/src/main/java/me/lucko/spark/common/ws/ViewerSocket.java b/spark-common/src/main/java/me/lucko/spark/common/ws/ViewerSocket.java index 12a9046dd85..edcf2f10d64 100644 --- a/spark-common/src/main/java/me/lucko/spark/common/ws/ViewerSocket.java +++ b/spark-common/src/main/java/me/lucko/spark/common/ws/ViewerSocket.java @@ -23,19 +23,14 @@ import com.google.protobuf.ByteString; import me.lucko.bytesocks.client.BytesocksClient; import me.lucko.spark.common.SparkPlatform; -import me.lucko.spark.common.sampler.AbstractSampler; -import me.lucko.spark.common.sampler.Sampler; import me.lucko.spark.common.sampler.window.ProfilingWindowUtils; -import me.lucko.spark.common.util.MediaTypes; import me.lucko.spark.common.util.TimeUtil; import me.lucko.spark.proto.SparkProtos; -import me.lucko.spark.proto.SparkSamplerProtos; import me.lucko.spark.proto.SparkWebSocketProtos.ClientConnect; import me.lucko.spark.proto.SparkWebSocketProtos.ClientPing; import me.lucko.spark.proto.SparkWebSocketProtos.PacketWrapper; import me.lucko.spark.proto.SparkWebSocketProtos.ServerConnectResponse; import me.lucko.spark.proto.SparkWebSocketProtos.ServerPong; -import me.lucko.spark.proto.SparkWebSocketProtos.ServerUpdateSamplerData; import me.lucko.spark.proto.SparkWebSocketProtos.ServerUpdateStatistics; import java.security.PublicKey; @@ -45,7 +40,7 @@ /** * Represents a connection with the spark viewer. */ -public class ViewerSocket implements ViewerSocketConnection.Listener, AutoCloseable { +public abstract class ViewerSocket implements ViewerSocketConnection.Listener, AutoCloseable { /** Allow 60 seconds for the first client to connect */ private static final long SOCKET_INITIAL_TIMEOUT = TimeUnit.SECONDS.toMillis(60); @@ -54,24 +49,21 @@ public class ViewerSocket implements ViewerSocketConnection.Listener, AutoClosea private static final long SOCKET_ESTABLISHED_TIMEOUT = TimeUnit.SECONDS.toMillis(30); /** The spark platform */ - private final SparkPlatform platform; - /** The export props to use when exporting the sampler data */ - private final Sampler.ExportProps exportProps; + protected final SparkPlatform platform; /** The underlying connection */ - private final ViewerSocketConnection socket; + protected final ViewerSocketConnection socket; private boolean closed = false; private final long socketOpenTime = TimeUtil.monotonicCurrentTimeMillis(); private long lastPing = 0; private String lastPayloadId = null; - public ViewerSocket(SparkPlatform platform, BytesocksClient client, Sampler.ExportProps exportProps) throws Exception { + protected ViewerSocket(SparkPlatform platform, BytesocksClient client) throws Exception { this.platform = platform; - this.exportProps = exportProps; this.socket = new ViewerSocketConnection(platform, client, this); } - private void log(String message) { + protected void log(String message) { this.platform.getPlugin().log(Level.INFO, "[Viewer - " + this.socket.getChannelId() + "] " + message); } @@ -80,8 +72,8 @@ private void log(String message) { * * @return the payload */ - public SparkSamplerProtos.SocketChannelInfo getPayload() { - return SparkSamplerProtos.SocketChannelInfo.newBuilder() + public SparkProtos.SocketChannelInfo getPayload() { + return SparkProtos.SocketChannelInfo.newBuilder() .setChannelId(this.socket.getChannelId()) .setPublicKey(ByteString.copyFrom(this.platform.getTrustedKeyStore().getLocalPublicKey().getEncoded())) .build(); @@ -91,48 +83,28 @@ public boolean isOpen() { return !this.closed && this.socket.isOpen(); } - /** - * Called each time the sampler rotates to a new window. - * - * @param sampler the sampler - */ - public void processWindowRotate(AbstractSampler sampler) { + protected boolean isClosed() { + return this.closed; + } + + public boolean checkShouldClose() { if (this.closed) { - return; + return true; } long time = TimeUtil.monotonicCurrentTimeMillis(); if ((time - this.socketOpenTime) > SOCKET_INITIAL_TIMEOUT && (time - this.lastPing) > SOCKET_ESTABLISHED_TIMEOUT) { log("No clients have pinged for 30s, closing socket"); close(); - return; + return true; } // no clients connected yet! if (this.lastPing == 0) { - return; + return true; } - try { - SparkSamplerProtos.SamplerData samplerData = sampler.toProto(this.platform, this.exportProps); - String key = this.platform.getBytebinClient().postContent(samplerData, MediaTypes.SPARK_SAMPLER_MEDIA_TYPE, "live").key(); - sendUpdatedSamplerData(key); - } catch (Exception e) { - this.platform.getPlugin().log(Level.WARNING, "Error whilst sending updated sampler data to the socket", e); - } - } - - /** - * Called when the sampler stops. - * - * @param sampler the sampler - */ - public void processSamplerStopped(AbstractSampler sampler) { - if (this.closed) { - return; - } - - close(); + return false; } @Override @@ -145,6 +117,14 @@ public void close() { this.closed = true; } + public String getLastPayloadId() { + return this.lastPayloadId; + } + + public void setLastPayloadId(String lastPayloadId) { + this.lastPayloadId = lastPayloadId; + } + @Override public boolean isKeyTrusted(PublicKey publicKey) { return this.platform.getTrustedKeyStore().isKeyTrusted(publicKey); @@ -163,29 +143,18 @@ public void sendClientTrustedMessage(String clientId) { )); } - /** - * Sends a message to the socket to indicate that updated sampler data is available - * - * @param payloadId the payload id of the updated data - */ - public void sendUpdatedSamplerData(String payloadId) { - this.socket.sendPacket(builder -> builder.setServerUpdateSampler(ServerUpdateSamplerData.newBuilder() - .setPayloadId(payloadId) - .build() - )); - this.lastPayloadId = payloadId; - } - /** * Sends a message to the socket with updated statistics * * @param platform the platform statistics * @param system the system statistics + * @param metrics the metrics */ - public void sendUpdatedStatistics(SparkProtos.PlatformStatistics platform, SparkProtos.SystemStatistics system) { + public void sendUpdatedStatistics(SparkProtos.PlatformStatistics platform, SparkProtos.SystemStatistics system, SparkProtos.Metrics metrics) { this.socket.sendPacket(builder -> builder.setServerUpdateStatistics(ServerUpdateStatistics.newBuilder() .setPlatform(platform) .setSystem(system) + .setMetrics(metrics) .build() )); } diff --git a/spark-common/src/main/proto/spark/spark.proto b/spark-common/src/main/proto/spark/spark.proto index deb42f909bc..1636d9dccc6 100644 --- a/spark-common/src/main/proto/spark/spark.proto +++ b/spark-common/src/main/proto/spark/spark.proto @@ -111,19 +111,15 @@ message PlatformStatistics { MemoryUsage heap = 1; MemoryUsage non_heap = 2; repeated MemoryPool pools = 3; + RollingAverageValues alloc_bps_last1m = 4; + RollingAverageValues alloc_bps_last5m = 5; + RollingAverageValues alloc_bps_last15m = 6; message MemoryPool { string name = 1; MemoryUsage usage = 2; MemoryUsage collection_usage = 3; } - - message MemoryUsage { - int64 used = 1; - int64 committed = 2; // previously called 'total' - int64 init = 3; // optional - int64 max = 4; // optional - } } message Gc { @@ -223,6 +219,56 @@ message RollingAverageValues { double percentile95 = 5; } +message MemoryUsage { + int64 used = 1; + int64 committed = 2; + int64 init = 3; + int64 max = 4; +} + +message DoubleMetricSeries { + int64 start_timestamp_ms = 1; + repeated uint32 timestamp_deltas_ms = 2; + repeated double values = 3; +} + +message AveragesMetricSeries { + int64 start_timestamp_ms = 1; + repeated uint32 timestamp_deltas_ms = 2; + repeated RollingAverageValues values = 3; +} + +message MemoryUsageMetricSeries { + int64 start_timestamp_ms = 1; + repeated uint32 timestamp_deltas_ms = 2; + repeated MemoryUsage values = 3; +} + +message WorldInfoMetricSeries { + int64 start_timestamp_ms = 1; + repeated uint32 timestamp_deltas_ms = 2; + repeated Values values = 3; + + message Values { + int32 players = 1; + int32 entities = 2; + int32 tile_entities = 3; + int32 chunks = 4; + } +} + +message Metrics { + DoubleMetricSeries tps = 1; + AveragesMetricSeries tick_duration = 2; + DoubleMetricSeries cpu_usage_process = 3; + DoubleMetricSeries cpu_usage_system = 4; + MemoryUsageMetricSeries memory_usage_heap = 5; + MemoryUsageMetricSeries memory_usage_non_heap = 6; + DoubleMetricSeries memory_allocation = 7; + WorldInfoMetricSeries world_info = 8; + AveragesMetricSeries player_ping = 9; +} + message CommandSenderMetadata { Type type = 1; string name = 2; @@ -245,6 +291,7 @@ message PluginOrModMetadata { message HealthData { HealthMetadata metadata = 1; map time_window_statistics = 2; + SocketChannelInfo channel_info = 3; } message HealthMetadata { @@ -256,4 +303,10 @@ message HealthMetadata { map server_configurations = 6; map sources = 7; map extra_platform_metadata = 8; + Metrics metrics = 9; +} + +message SocketChannelInfo { + string channel_id = 1; + bytes public_key = 2; } diff --git a/spark-common/src/main/proto/spark/spark_heap.proto b/spark-common/src/main/proto/spark/spark_heap.proto index aef7888bb18..836b6b959fb 100644 --- a/spark-common/src/main/proto/spark/spark_heap.proto +++ b/spark-common/src/main/proto/spark/spark_heap.proto @@ -21,6 +21,7 @@ message HeapMetadata { map server_configurations = 6; map sources = 7; map extra_platform_metadata = 8; + Metrics metrics = 9; } message HeapEntry { diff --git a/spark-common/src/main/proto/spark/spark_sampler.proto b/spark-common/src/main/proto/spark/spark_sampler.proto index 445e541cad4..3a003a605a5 100644 --- a/spark-common/src/main/proto/spark/spark_sampler.proto +++ b/spark-common/src/main/proto/spark/spark_sampler.proto @@ -36,6 +36,7 @@ message SamplerMetadata { SamplerMode sampler_mode = 15; SamplerEngine sampler_engine = 16; string sampler_engine_version = 17; + Metrics metrics = 18; message ThreadDumper { Type type = 1; @@ -103,8 +104,3 @@ message StackTraceNode { repeated double times = 8; repeated int32 children_refs = 9; } - -message SocketChannelInfo { - string channel_id = 1; - bytes public_key = 2; -} diff --git a/spark-common/src/main/proto/spark/spark_ws.proto b/spark-common/src/main/proto/spark/spark_ws.proto index 97b54807559..39fc1d0b519 100644 --- a/spark-common/src/main/proto/spark/spark_ws.proto +++ b/spark-common/src/main/proto/spark/spark_ws.proto @@ -63,6 +63,7 @@ message ServerUpdateSamplerData { message ServerUpdateStatistics { PlatformStatistics platform = 1; SystemStatistics system = 2; + Metrics metrics = 3; } // (unsigned) Sent from the client -> server on initial connection diff --git a/spark-common/src/test/java/me/lucko/spark/common/SparkPlatformTest.java b/spark-common/src/test/java/me/lucko/spark/common/SparkPlatformTest.java index 23f9ea1fbe4..63bf68d7500 100644 --- a/spark-common/src/test/java/me/lucko/spark/common/SparkPlatformTest.java +++ b/spark-common/src/test/java/me/lucko/spark/common/SparkPlatformTest.java @@ -57,7 +57,7 @@ public void testPermissions(@TempDir Path directory) { "spark.profiler", "spark.tps", "spark.ping", - "spark.healthreport", + "spark.health", "spark.gc", "spark.gcmonitor", "spark.heapsummary", diff --git a/spark-common/src/test/java/me/lucko/spark/common/monitor/memory/MemoryAllocationInfoTest.java b/spark-common/src/test/java/me/lucko/spark/common/monitor/memory/MemoryAllocationInfoTest.java new file mode 100644 index 00000000000..08ef8eb23d8 --- /dev/null +++ b/spark-common/src/test/java/me/lucko/spark/common/monitor/memory/MemoryAllocationInfoTest.java @@ -0,0 +1,34 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.monitor.memory; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class MemoryAllocationInfoTest { + + @Test + public void testMemoryAllocationInfo() { + long bytes = MemoryAllocationInfo.getTotalThreadAllocatedBytes(); + assertTrue(bytes >= 0); + } +} diff --git a/spark-common/src/test/java/me/lucko/spark/common/monitor/memory/MemoryInfoTest.java b/spark-common/src/test/java/me/lucko/spark/common/monitor/memory/SystemMemoryInfoTest.java similarity index 80% rename from spark-common/src/test/java/me/lucko/spark/common/monitor/memory/MemoryInfoTest.java rename to spark-common/src/test/java/me/lucko/spark/common/monitor/memory/SystemMemoryInfoTest.java index 5ae8fdc0aea..b619b908543 100644 --- a/spark-common/src/test/java/me/lucko/spark/common/monitor/memory/MemoryInfoTest.java +++ b/spark-common/src/test/java/me/lucko/spark/common/monitor/memory/SystemMemoryInfoTest.java @@ -24,13 +24,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue; -public class MemoryInfoTest { +public class SystemMemoryInfoTest { @Test public void testMemoryInfo() { - assertTrue(MemoryInfo.getUsedPhysicalMemory() > 0); - assertTrue(MemoryInfo.getTotalPhysicalMemory() > 0); - assertTrue(MemoryInfo.getAvailablePhysicalMemory() > 0); + assertTrue(SystemMemoryInfo.getUsedPhysicalMemory() > 0); + assertTrue(SystemMemoryInfo.getTotalPhysicalMemory() > 0); + assertTrue(SystemMemoryInfo.getAvailablePhysicalMemory() > 0); } } diff --git a/spark-common/src/test/java/me/lucko/spark/common/util/MetricSeriesTest.java b/spark-common/src/test/java/me/lucko/spark/common/util/MetricSeriesTest.java new file mode 100644 index 00000000000..58514bd44ae --- /dev/null +++ b/spark-common/src/test/java/me/lucko/spark/common/util/MetricSeriesTest.java @@ -0,0 +1,95 @@ +/* + * This file is part of spark. + * + * Copyright (c) lucko (Luck) + * Copyright (c) contributors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package me.lucko.spark.common.util; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class MetricSeriesTest { + + @Test + public void testEmpty() { + MetricSeries series = new MetricSeries<>(Duration.ofMillis(10), 1); + assertTrue(series.isEmpty()); + assertEquals(0, series.size()); + assertEquals(0, series.newestTimestamp()); + assertEquals(0, series.oldestTimestamp()); + series.forEach((timestamp, value) -> { + throw new AssertionError("Should not be called"); + }); + + MetricSeries.Export export = series.export(); + assertEquals(0, export.startTimestampMs()); + assertEquals(0, export.timestampDeltasMs().length); + assertEquals(0, export.values().length); + } + + @Test + public void testAppend() { + MetricSeries series = new MetricSeries<>(Duration.ofMillis(10), 1); + series.record(1, 1.0); + series.record(2, 2.0); + + assertEquals(2, series.size()); + assertFalse(series.isEmpty()); + assertEquals(2, series.newestTimestamp()); + assertEquals(1, series.oldestTimestamp()); + + series.forEach((timestamp, value) -> { + if (timestamp == 1) { + assertEquals(1.0, value); + } else if (timestamp == 2) { + assertEquals(2.0, value); + } else { + throw new AssertionError("Unexpected timestamp: " + timestamp); + } + }); + + MetricSeries.Export export = series.export(); + assertEquals(1, export.startTimestampMs()); + assertArrayEquals(new int[]{0, 1}, export.timestampDeltasMs()); + assertArrayEquals(new Object[]{1.0, 2.0}, export.values()); + } + + @Test + public void testRetention() { + MetricSeries series = new MetricSeries<>(Duration.ofMillis(10), 1); + series.record(10, 1.0); + series.record(20, 2.0); + series.record(30, 3.0); + + assertEquals(30, series.newestTimestamp()); + assertEquals(20, series.oldestTimestamp()); + assertEquals(2, series.size()); + + MetricSeries.Export export = series.export(); + assertEquals(20, export.startTimestampMs()); + assertArrayEquals(new int[]{0, 10}, export.timestampDeltasMs()); + assertArrayEquals(new Object[]{2.0, 3.0}, export.values()); + } + +}