You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2025/01/10 03:58:22 UTC
(skywalking) branch watermark updated: Finish watcher kernel
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch watermark
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/watermark by this push:
new 9a7e2e56e9 Finish watcher kernel
9a7e2e56e9 is described below
commit 9a7e2e56e9cc0cafa689fa3e38fe468e81dfa8b3
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Fri Jan 10 11:58:15 2025 +0800
Finish watcher kernel
---
.../skywalking/oap/server/core/CoreModule.java | 2 +
.../oap/server/core/CoreModuleConfig.java | 11 +-
.../oap/server/core/CoreModuleProvider.java | 4 +-
.../oap/server/core/watermark/WatermarkEvent.java | 3 +-
.../server/core/watermark/WatermarkListener.java | 43 ++++++-
.../server/core/watermark/WatermarkWatcher.java | 142 ++++++++++++---------
.../envoy/AccessLogServiceGRPCHandler.java | 31 +++--
.../oap/server/telemetry/api/MetricsCollector.java | 18 ++-
.../telemetry/none/MetricsCollectorNoop.java | 15 ++-
.../prometheus/PrometheusMetricsCollector.java | 68 +++++++++-
10 files changed, 237 insertions(+), 100 deletions(-)
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java
index 7e1cc8a3fa..96175ceba7 100755
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java
@@ -66,6 +66,7 @@ import org.apache.skywalking.oap.server.core.status.ServerStatusService;
import org.apache.skywalking.oap.server.core.storage.model.IModelManager;
import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
import org.apache.skywalking.oap.server.core.storage.model.ModelManipulator;
+import org.apache.skywalking.oap.server.core.watermark.WatermarkWatcher;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceGetter;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
@@ -108,6 +109,7 @@ public class CoreModule extends ModuleDefine {
classes.add(CommandService.class);
classes.add(HierarchyService.class);
+ classes.add(WatermarkWatcher.class);
return classes.toArray(new Class[]{});
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java
index 0c81b95858..db22d99c97 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java
@@ -237,18 +237,13 @@ public class CoreModuleConfig extends ModuleConfig {
*/
@Getter
private long maxHeapMemoryUsagePercent = 85;
+
/**
- * The int value of the max non-heap memory usage percent.
- * The default value is 85%.
- */
- @Getter
- private long maxNoheapMemoryUsagePercent = 85;
- /**
- * The long value of the max no heap memory usage.
+ * The long value of the max direct memory usage.
* The default max value is -1, representing no limit.
*/
@Getter
- private long maxNoheapMemoryUsage = -1;
+ private long maxDirectMemoryUsage = -1;
public CoreModuleConfig() {
this.downsampling = new ArrayList<>();
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
index e5e2257531..edbca88f2d 100755
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
@@ -384,8 +384,8 @@ public class CoreModuleProvider extends ModuleProvider {
loggingConfigWatcher = new LoggingConfigWatcher(this);
this.watermarkWatcher = new WatermarkWatcher(moduleConfig.getMaxHeapMemoryUsagePercent(),
- moduleConfig.getMaxNoheapMemoryUsagePercent(),
- moduleConfig.getMaxNoheapMemoryUsage());
+ moduleConfig.getMaxDirectMemoryUsage());
+ this.registerServiceImplementation(WatermarkWatcher.class, watermarkWatcher);
}
@Override
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkEvent.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkEvent.java
index 80ce9297f1..a15ad03ac3 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkEvent.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkEvent.java
@@ -26,7 +26,6 @@ public class WatermarkEvent {
public enum Type {
HEAP_MEMORY_USAGE_PERCENTAGE,
- NO_HEAP_MEMORY_USAGE,
- NO_HEAP_MEMORY_USAGE_PERCENTAGE,
+ DIRECT_HEAP_MEMORY_USAGE,
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java
index 86048ccdf4..1139b705fd 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkListener.java
@@ -19,33 +19,66 @@
package org.apache.skywalking.oap.server.core.watermark;
import java.util.List;
+import lombok.Getter;
/**
* WatermarkListener is the listener for receiving WatermarkEvent and react to it.
+ * The implementations of this listener has two ways to interact with the WatermarkEvent:
+ * 1. use {@link #isWatermarkExceeded()} to check if the watermark is exceeded.
+ * 2. override {@link #beAwareOf(WatermarkEvent.Type)} to react to the event.
+ *
+ * When the oap recovered from the limiting state, the listener has two ways to be aware of it:
+ * 1. use {@link #isHealthy()} to check if the watermark is recovered.
+ * 2. Be notified by calling {@link #beAwareOfRecovery()}.
*/
public abstract class WatermarkListener {
+ @Getter
+ private String name;
private List<WatermarkEvent.Type> acceptedTypes;
private volatile boolean isWatermarkExceeded = false;
- public WatermarkListener() {
- this(WatermarkEvent.Type.values());
+ /**
+ * Create a listener that accepts all types of WatermarkEvent.
+ * This should be the default way to create a listener.
+ */
+ public WatermarkListener(String name) {
+ this(name, WatermarkEvent.Type.values());
}
- public WatermarkListener(WatermarkEvent.Type... types) {
+ public WatermarkListener(String name, WatermarkEvent.Type... types) {
this.acceptedTypes = List.of(types);
}
- void notify(WatermarkEvent.Type event) {
+ boolean notify(WatermarkEvent.Type event) {
if (acceptedTypes.contains(event)) {
isWatermarkExceeded = true;
+ beAwareOf(event);
+ return true;
}
+ return false;
}
- void recovered() {
+ void isHealthy() {
isWatermarkExceeded = false;
}
public boolean isWatermarkExceeded() {
return isWatermarkExceeded;
}
+
+ protected void beAwareOf(WatermarkEvent.Type event) {
+ }
+
+ ;
+
+ protected void beAwareOfRecovery() {
+ // Do nothing by default.
+ }
+
+ public static class DefaultListener extends WatermarkListener {
+
+ public DefaultListener(final String name) {
+ super(name);
+ }
+ }
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java
index 7a28a643d1..e58e89a130 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/watermark/WatermarkWatcher.java
@@ -24,23 +24,22 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.library.module.Service;
+import org.apache.skywalking.oap.server.library.module.TerminalFriendlyTable;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCollector;
@RequiredArgsConstructor
-public class WatermarkWatcher {
+@Slf4j
+public class WatermarkWatcher implements Service {
private final long maxHeapMemoryUsagePercentThreshold;
- private final long maxNoHeapMemoryUsagePercentThreshold;
- private final long maxNoHeapMemoryUsageThreshold;
+ private final long maxDirectHeapMemoryUsageThreshold;
private MetricsCollector so11yCollector;
- /**
- * Noheap memory max, jvm_memory_max_bytes{area="nonheap"}
- */
- private long noheapMemoryMax = 0;
/**
* Noheap memory used, jvm_memory_used_bytes{area="nonheap"}
*/
- private long noheapMemoryUsed = 0;
+ private long directMemoryUsed = 0;
/**
* Heap memory max, jvm_memory_max_bytes{area="heap"}. Use the combination of all available ids of heap memory.
*/
@@ -51,6 +50,7 @@ public class WatermarkWatcher {
private long heapMemoryUsed = 0;
private ReentrantLock lock;
private List<WatermarkListener> listeners;
+ private volatile boolean isLimiting = false;
public void start(MetricsCollector so11yCollector) {
this.so11yCollector = so11yCollector;
@@ -61,64 +61,88 @@ public class WatermarkWatcher {
}
private void watch() {
- so11yCollector.find("jvm_memory_bytes_max")
- .ifPresent(metricFamily -> {
- metricFamily.samples.forEach(sample -> {
- for (int i = 0; i < sample.labelNames.size(); i++) {
- if (sample.labelNames.get(i).equals("area") && sample.labelValues.get(i)
- .equals("heap")) {
- heapMemoryMax += (long) sample.value;
- } else if (sample.labelNames.get(i).equals("area") && sample.labelValues.get(i)
- .equals(
- "nonheap")) {
- noheapMemoryMax = (long) sample.value;
- }
- }
- });
- });
- so11yCollector.find("jvm_memory_bytes_used")
- .ifPresent(metricFamily -> {
- metricFamily.samples.forEach(sample -> {
- for (int i = 0; i < sample.labelNames.size(); i++) {
- if (sample.labelNames.get(i).equals("area") && sample.labelValues.get(i)
- .equals("heap")) {
- heapMemoryUsed += (long) sample.value;
- } else if (sample.labelNames.get(i).equals("area") && sample.labelValues.get(i)
- .equals(
- "nonheap")) {
- noheapMemoryUsed = (long) sample.value;
- }
- }
- });
- });
- if (heapMemoryMax <= 0) {
- // No heap memory metrics found, skip the watermark control.
- return;
+ this.heapMemoryUsed = so11yCollector.heapMemoryUsage();
+ this.heapMemoryMax = so11yCollector.heapMemoryMax();
+ this.directMemoryUsed = so11yCollector.directMemoryUsage();
+
+ if (log.isDebugEnabled()) {
+ TerminalFriendlyTable table = new TerminalFriendlyTable("Watermark Controller Key Metrics");
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Max", String.format("%,d", heapMemoryMax)));
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Used", String.format("%,d", heapMemoryUsed)));
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Usage Percentage", heapMemoryUsagePercent() + "%"));
+ table.addRow(new TerminalFriendlyTable.Row("Direct Memory Used", String.format("%,d", directMemoryUsed)));
+ log.debug(table.toString());
+ }
+
+ boolean isLimitingTriggered = false;
+
+ if (heapMemoryUsagePercent() > maxHeapMemoryUsagePercentThreshold) {
+ this.notify(WatermarkEvent.Type.HEAP_MEMORY_USAGE_PERCENTAGE);
+ isLimitingTriggered = true;
}
+
+ if (maxDirectHeapMemoryUsageThreshold > 0 && directMemoryUsed > 0) {
+ if (directMemoryUsed > maxDirectHeapMemoryUsageThreshold) {
+ this.notify(WatermarkEvent.Type.DIRECT_HEAP_MEMORY_USAGE);
+ isLimitingTriggered = true;
+ }
+ }
+
+ if (!isLimitingTriggered && isLimiting) {
+ recovered();
+ }
+ }
+
+ private void notify(WatermarkEvent.Type event) {
+ TerminalFriendlyTable table = new TerminalFriendlyTable("Watermark Controller Key Metrics");
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Max", String.format("%,d", heapMemoryMax)));
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Used", String.format("%,d", heapMemoryUsed)));
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Usage Percentage", heapMemoryUsagePercent() + "%"));
+ table.addRow(new TerminalFriendlyTable.Row("Direct Memory Used", String.format("%,d", directMemoryUsed)));
+ table.addRow(new TerminalFriendlyTable.Row("Event", event.name()));
+
+ isLimiting = true;
+
lock.lock();
try {
- long heapMemoryUsagePercent = heapMemoryUsed * 100 / heapMemoryMax;
- if (heapMemoryUsagePercent > maxHeapMemoryUsagePercentThreshold) {
- listeners.forEach(
- watermarkListener -> watermarkListener.notify(WatermarkEvent.Type.HEAP_MEMORY_USAGE_PERCENTAGE));
- }
- if (noheapMemoryMax > 0) {
- long noheapMemoryUsagePercent = noheapMemoryUsed * 100 / noheapMemoryMax;
- if (noheapMemoryUsagePercent > maxNoHeapMemoryUsagePercentThreshold) {
- listeners.forEach(
- watermarkListener -> watermarkListener.notify(
- WatermarkEvent.Type.NO_HEAP_MEMORY_USAGE_PERCENTAGE));
- }
- }
- if (maxNoHeapMemoryUsageThreshold > 0) {
- if (noheapMemoryUsed > maxNoHeapMemoryUsageThreshold) {
- listeners.forEach(
- watermarkListener -> watermarkListener.notify(WatermarkEvent.Type.NO_HEAP_MEMORY_USAGE));
+ listeners.forEach(listener -> {
+ if (listener.notify(event)) {
+ table.addRow(new TerminalFriendlyTable.Row("Notified Listener", listener.getName()));
}
- }
+ });
} finally {
lock.unlock();
}
+ log.warn(table.toString());
+ }
+
+ private void recovered() {
+ TerminalFriendlyTable table = new TerminalFriendlyTable("Watermark Controller Key Metrics");
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Max", String.format("%,d", heapMemoryMax)));
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Used", String.format("%,d", heapMemoryUsed)));
+ table.addRow(new TerminalFriendlyTable.Row("Heap Memory Usage Percentage", heapMemoryUsagePercent() + "%"));
+ table.addRow(new TerminalFriendlyTable.Row("Direct Memory Used", String.format("%,d", directMemoryUsed)));
+ table.addRow(new TerminalFriendlyTable.Row("Event", "RECOVERED"));
+
+ isLimiting = false;
+ lock.lock();
+ try {
+ listeners.forEach(listener -> {
+ listener.beAwareOfRecovery();
+ table.addRow(new TerminalFriendlyTable.Row("Notified Listener", listener.getName()));
+ });
+ } finally {
+ lock.unlock();
+ }
+ log.info(table.toString());
+ }
+
+ private long heapMemoryUsagePercent() {
+ if (heapMemoryMax > 0) {
+ return heapMemoryUsed * 100 / heapMemoryMax;
+ } else {
+ return -1;
+ }
}
public void addListener(WatermarkListener listener) {
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java
index 2406ea1c8b..99ebc8217a 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java
@@ -18,6 +18,13 @@
package org.apache.skywalking.oap.server.receiver.envoy;
+import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
+import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
+import io.envoyproxy.envoy.service.accesslog.v2.AccessLogServiceGrpc;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsResponse;
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
import java.util.ArrayList;
import java.util.List;
import java.util.ServiceLoader;
@@ -25,6 +32,9 @@ import org.apache.skywalking.aop.server.receiver.mesh.TelemetryDataDispatcher;
import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetrics;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.watermark.WatermarkListener;
+import org.apache.skywalking.oap.server.core.watermark.WatermarkWatcher;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.receiver.envoy.als.ALSHTTPAnalysis;
@@ -38,13 +48,6 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
-import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
-import io.envoyproxy.envoy.service.accesslog.v2.AccessLogServiceGrpc;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsResponse;
-import io.grpc.Status;
-import io.grpc.stub.StreamObserver;
public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogServiceImplBase {
private static final Logger LOGGER = LoggerFactory.getLogger(AccessLogServiceGRPCHandler.class);
@@ -54,6 +57,7 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
private final CounterMetrics counter;
private final HistogramMetrics histogram;
private final CounterMetrics sourceDispatcherCounter;
+ private final WatermarkListener listener;
public AccessLogServiceGRPCHandler(ModuleManager manager,
EnvoyMetricReceiverConfig config) throws ModuleStartException {
@@ -93,11 +97,15 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
"envoy_als_source_dispatch_count", "The count of envoy ALS metric received", MetricsTag.EMPTY_KEY,
MetricsTag.EMPTY_VALUE
);
+
+ final WatermarkWatcher service = manager.find(CoreModule.NAME).provider().getService(WatermarkWatcher.class);
+ listener = new WatermarkListener.DefaultListener("envoy-als");
+ service.addListener(listener);
}
@Override
public StreamObserver<StreamAccessLogsMessage> streamAccessLogs(
- StreamObserver<StreamAccessLogsResponse> responseObserver) {
+ StreamObserver<StreamAccessLogsResponse> responseObserver) {
return streamAccessLogs(responseObserver, false);
}
@@ -112,6 +120,10 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
public void onNext(StreamAccessLogsMessage message) {
HistogramMetrics.Timer timer = histogram.createTimer();
try {
+ if (listener.isWatermarkExceeded()) {
+ return;
+ }
+
if (isFirst || alwaysAnalyzeIdentity && message.hasIdentifier()) {
identifier = message.getIdentifier();
isFirst = false;
@@ -127,7 +139,8 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
LOGGER.debug(
"Messaged is identified from Envoy[{}], role[{}] in [{}]. Received msg {}", identifier
.getNode()
- .getId(), role, logCase, message);
+ .getId(), role, logCase, message
+ );
}
final ServiceMeshMetrics.Builder sourceResult = ServiceMeshMetrics.newBuilder();
diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCollector.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCollector.java
index 671b9f137f..d1ddd8e523 100644
--- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCollector.java
+++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCollector.java
@@ -18,7 +18,6 @@
package org.apache.skywalking.oap.server.telemetry.api;
-import java.util.Optional;
import org.apache.skywalking.oap.server.library.module.Service;
/**
@@ -34,10 +33,17 @@ public interface MetricsCollector extends Service {
Iterable<MetricFamily> collect();
/**
- * Get existing metrics with the given metric name.
- *
- * @param name, the metric name
- * @return the optional metrics with the given metric name
+ * Get the heap memory usage.
+ */
+ long heapMemoryUsage();
+
+ /**
+ * Get the heap memory max.
+ */
+ long heapMemoryMax();
+
+ /**
+ * Get the direct memory usage.
*/
- Optional<MetricFamily> find(String name);
+ long directMemoryUsage();
}
diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricsCollectorNoop.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricsCollectorNoop.java
index 5405a30373..d88e2b64b9 100644
--- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricsCollectorNoop.java
+++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricsCollectorNoop.java
@@ -19,7 +19,6 @@
package org.apache.skywalking.oap.server.telemetry.none;
import java.util.Collections;
-import java.util.Optional;
import org.apache.skywalking.oap.server.telemetry.api.MetricFamily;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCollector;
@@ -33,7 +32,17 @@ public class MetricsCollectorNoop implements MetricsCollector {
}
@Override
- public Optional<MetricFamily> find(final String name) {
- return Optional.empty();
+ public long heapMemoryUsage() {
+ return 0;
+ }
+
+ @Override
+ public long heapMemoryMax() {
+ return 0;
+ }
+
+ @Override
+ public long directMemoryUsage() {
+ return 0;
}
}
diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusMetricsCollector.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusMetricsCollector.java
index 893faf21ef..5f6f17c3cf 100644
--- a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusMetricsCollector.java
+++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusMetricsCollector.java
@@ -24,29 +24,85 @@ import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
import java.util.Optional;
-import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.skywalking.oap.server.telemetry.api.MetricFamily;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCollector;
public class PrometheusMetricsCollector implements MetricsCollector {
- @Override public Iterable<MetricFamily> collect() {
+ @Override
+ public Iterable<MetricFamily> collect() {
Enumeration<Collector.MetricFamilySamples> mfs = CollectorRegistry.defaultRegistry.metricFamilySamples();
List<MetricFamily> result = new ArrayList<>();
while (mfs.hasMoreElements()) {
Collector.MetricFamilySamples metricFamilySamples = mfs.nextElement();
List<MetricFamily.Sample> samples = new ArrayList<>(metricFamilySamples.samples.size());
- MetricFamily m = new MetricFamily(metricFamilySamples.name, MetricFamily.Type.valueOf(metricFamilySamples.type
- .name()), metricFamilySamples.help, samples);
+ MetricFamily m = new MetricFamily(
+ metricFamilySamples.name, MetricFamily.Type.valueOf(metricFamilySamples.type
+ .name()), metricFamilySamples.help, samples
+ );
result.add(m);
for (Collector.MetricFamilySamples.Sample sample : metricFamilySamples.samples) {
- samples.add(new MetricFamily.Sample(sample.name, sample.labelNames, sample.labelValues, sample.value, sample.timestampMs));
+ samples.add(new MetricFamily.Sample(
+ sample.name, sample.labelNames, sample.labelValues, sample.value,
+ sample.timestampMs
+ ));
}
}
return result;
}
@Override
- public Optional<MetricFamily> find(final String name) {
+ public long heapMemoryUsage() {
+ final AtomicLong heapMemoryUsed = new AtomicLong();
+ this.find("jvm_memory_bytes_used")
+ .ifPresent(metricFamily -> {
+ metricFamily.samples.forEach(sample -> {
+ for (int i = 0; i < sample.labelNames.size(); i++) {
+ if (sample.labelNames.get(i).equals("area") && sample.labelValues.get(i)
+ .equals("heap")) {
+ heapMemoryUsed.addAndGet((long) sample.value);
+ }
+ }
+ });
+ });
+ return heapMemoryUsed.longValue();
+ }
+
+ @Override
+ public long heapMemoryMax() {
+ final AtomicLong heapMemoryMax = new AtomicLong();
+ this.find("jvm_memory_bytes_max")
+ .ifPresent(metricFamily -> {
+ metricFamily.samples.forEach(sample -> {
+ for (int i = 0; i < sample.labelNames.size(); i++) {
+ if (sample.labelNames.get(i).equals("area") && sample.labelValues.get(i)
+ .equals("heap")) {
+ heapMemoryMax.addAndGet((long) sample.value);
+ }
+ }
+ });
+ });
+ return heapMemoryMax.longValue();
+ }
+
+ @Override
+ public long directMemoryUsage() {
+ final AtomicLong directMemoryUsed = new AtomicLong();
+ this.find("jvm_buffer_pool_used_bytes")
+ .ifPresent(metricFamily -> {
+ metricFamily.samples.forEach(sample -> {
+ for (int i = 0; i < sample.labelNames.size(); i++) {
+ if (sample.labelNames.get(i).equals("pool") && sample.labelValues.get(i)
+ .equals("direct")) {
+ directMemoryUsed.addAndGet((long) sample.value);
+ }
+ }
+ });
+ });
+ return directMemoryUsed.longValue();
+ }
+
+ private Optional<MetricFamily> find(final String name) {
for (final MetricFamily metricFamily : this.collect()) {
if (metricFamily.name.equals(name)) {
return Optional.of(metricFamily);