This is an automated email from the ASF dual-hosted git repository.
maobaolong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
The following commit(s) were added to refs/heads/master by this push:
new 910823d81 [MINOR] feat(server,dashboard,coordinator): Report
configured metrics of server to coordinator and display to dashboard (#2239)
910823d81 is described below
commit 910823d811c83aee3438647882435aa0efe9e9b1
Author: maobaolong <[email protected]>
AuthorDate: Tue Nov 19 17:04:14 2024 +0800
[MINOR] feat(server,dashboard,coordinator): Report configured metrics of
server to coordinator and display to dashboard (#2239)
### What changes were proposed in this pull request?
Report configured metrics of server to coordinator and display to dashboard
### Why are the changes needed?
With this feature, it can be extensible for displaying server state by
dashboard.
In the beginning, we only need to display the app number in the dashboard,
after a few days, we want to add partition with node in to dashboard, maybe no
more days later, we need another value displayed in to dashboard.
So it is not extensible if we add it one by one, this is why I invent this
feature, it can be easy to let user config the metrics what they think it is
necessary and should be displayed in the dashboard.
### Does this PR introduce _any_ user-facing change?
- new config: rss.server.displayMetricsList
### How was this patch tested?
Test Locally.
<img width="288" alt="image"
src="https://github.com/user-attachments/assets/a10fdbe4-08fa-4ccb-8614-80b2447b30f9";>
---
.../uniffle/common/metrics/MetricsManager.java | 8 +++
.../coordinator/CoordinatorGrpcService.java | 3 +-
.../org/apache/uniffle/coordinator/ServerNode.java | 12 +++-
.../webapp/src/pages/serverstatus/NodeListPage.vue | 7 ++
.../client/impl/grpc/CoordinatorGrpcClient.java | 71 ++++++-------------
.../client/request/RssSendHeartBeatRequest.java | 10 ++-
proto/src/main/proto/Rss.proto | 1 +
.../apache/uniffle/server/RegisterHeartBeat.java | 80 +++++++---------------
.../org/apache/uniffle/server/ShuffleServer.java | 47 +++++++++----
.../apache/uniffle/server/ShuffleServerConf.java | 7 ++
.../uniffle/server/ShuffleServerMetrics.java | 9 +++
11 files changed, 131 insertions(+), 124 deletions(-)
diff --git
a/common/src/main/java/org/apache/uniffle/common/metrics/MetricsManager.java
b/common/src/main/java/org/apache/uniffle/common/metrics/MetricsManager.java
index efa216e0c..2cc991640 100644
--- a/common/src/main/java/org/apache/uniffle/common/metrics/MetricsManager.java
+++ b/common/src/main/java/org/apache/uniffle/common/metrics/MetricsManager.java
@@ -150,4 +150,12 @@ public class MetricsManager {
supplierGaugeMap.remove(name);
}
}
+
+ public String[] getDefaultLabelNames() {
+ return defaultLabelNames;
+ }
+
+ public String[] getDefaultLabelValues() {
+ return defaultLabelValues;
+ }
}
diff --git
a/coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java
b/coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java
index e49430001..489199f50 100644
---
a/coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java
+++
b/coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java
@@ -538,7 +538,8 @@ public class CoordinatorGrpcService extends
CoordinatorServerGrpc.CoordinatorSer
request.getStartTimeMs(),
request.getVersion(),
request.getGitCommitId(),
- request.getApplicationInfoList());
+ request.getApplicationInfoList(),
+ request.getDisplayMetricsMap());
}
/**
diff --git
a/coordinator/src/main/java/org/apache/uniffle/coordinator/ServerNode.java
b/coordinator/src/main/java/org/apache/uniffle/coordinator/ServerNode.java
index d43808380..c8463666d 100644
--- a/coordinator/src/main/java/org/apache/uniffle/coordinator/ServerNode.java
+++ b/coordinator/src/main/java/org/apache/uniffle/coordinator/ServerNode.java
@@ -33,6 +33,7 @@ import org.apache.uniffle.proto.RssProtos.ShuffleServerId;
public class ServerNode implements Comparable<ServerNode> {
+ private final Map<String, String> displayMetrics;
private String id;
private String ip;
private int grpcPort;
@@ -187,7 +188,8 @@ public class ServerNode implements Comparable<ServerNode> {
startTime,
"",
"",
- Collections.EMPTY_LIST);
+ Collections.EMPTY_LIST,
+ Collections.EMPTY_MAP);
}
public ServerNode(
@@ -206,7 +208,8 @@ public class ServerNode implements Comparable<ServerNode> {
long startTime,
String version,
String gitCommitId,
- List<RssProtos.ApplicationInfo> appInfos) {
+ List<RssProtos.ApplicationInfo> appInfos,
+ Map<String, String> displayMetrics) {
this.id = id;
this.ip = ip;
this.grpcPort = grpcPort;
@@ -230,6 +233,7 @@ public class ServerNode implements Comparable<ServerNode> {
this.gitCommitId = gitCommitId;
this.appIdToInfos = new ConcurrentHashMap<>();
appInfos.forEach(appInfo -> appIdToInfos.put(appInfo.getAppId(), appInfo));
+ this.displayMetrics = displayMetrics;
}
public ShuffleServerId convertToGrpcProto() {
@@ -388,4 +392,8 @@ public class ServerNode implements Comparable<ServerNode> {
public Map<String, RssProtos.ApplicationInfo> getAppIdToInfos() {
return appIdToInfos;
}
+
+ public Map<String, String> getDisplayMetrics() {
+ return displayMetrics;
+ }
}
diff --git a/dashboard/src/main/webapp/src/pages/serverstatus/NodeListPage.vue
b/dashboard/src/main/webapp/src/pages/serverstatus/NodeListPage.vue
index cfcffdf04..ac92a1b8a 100644
--- a/dashboard/src/main/webapp/src/pages/serverstatus/NodeListPage.vue
+++ b/dashboard/src/main/webapp/src/pages/serverstatus/NodeListPage.vue
@@ -101,6 +101,13 @@
</el-button>
</template>
</el-table-column>
+ <el-table-column prop="displayMetrics" label="DisplayMetrics"
min-width="120">
+ <template v-slot="{ row }">
+ <div v-for="(value, key) in row.displayMetrics" :key="key">
+ <span>{{ key }}:{{ value }}</span>
+ </div>
+ </template>
+ </el-table-column>
</el-table>
</div>
</template>
diff --git
a/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/CoordinatorGrpcClient.java
b/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/CoordinatorGrpcClient.java
index fbfe57824..7f35b5b59 100644
---
a/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/CoordinatorGrpcClient.java
+++
b/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/CoordinatorGrpcClient.java
@@ -49,11 +49,9 @@ import
org.apache.uniffle.client.response.RssGetShuffleAssignmentsResponse;
import org.apache.uniffle.client.response.RssSendHeartBeatResponse;
import org.apache.uniffle.common.PartitionRange;
import org.apache.uniffle.common.RemoteStorageInfo;
-import org.apache.uniffle.common.ServerStatus;
import org.apache.uniffle.common.ShuffleServerInfo;
import org.apache.uniffle.common.exception.RssException;
import org.apache.uniffle.common.rpc.StatusCode;
-import org.apache.uniffle.common.storage.StorageInfo;
import org.apache.uniffle.common.storage.StorageInfoUtils;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.proto.CoordinatorServerGrpc;
@@ -113,51 +111,40 @@ public class CoordinatorGrpcClient extends GrpcClient
implements CoordinatorClie
return blockingStub.getShuffleServerList(Empty.newBuilder().build());
}
- public ShuffleServerHeartBeatResponse doSendHeartBeat(
- String id,
- String ip,
- int port,
- long usedMemory,
- long preAllocatedMemory,
- long availableMemory,
- int eventNumInFlush,
- long timeout,
- Set<String> tags,
- ServerStatus serverStatus,
- Map<String, StorageInfo> storageInfo,
- int nettyPort,
- int jettyPort,
- long startTimeMs,
- List<RssProtos.ApplicationInfo> appInfos) {
+ public ShuffleServerHeartBeatResponse
doSendHeartBeat(RssSendHeartBeatRequest rssRequest) {
ShuffleServerId serverId =
ShuffleServerId.newBuilder()
- .setId(id)
- .setIp(ip)
- .setPort(port)
- .setNettyPort(nettyPort)
- .setJettyPort(jettyPort)
+ .setId(rssRequest.getShuffleServerId())
+ .setIp(rssRequest.getShuffleServerIp())
+ .setPort(rssRequest.getShuffleServerPort())
+ .setNettyPort(rssRequest.getNettyPort())
+ .setJettyPort(rssRequest.getJettyPort())
.build();
ShuffleServerHeartBeatRequest request =
ShuffleServerHeartBeatRequest.newBuilder()
.setServerId(serverId)
- .setUsedMemory(usedMemory)
- .setPreAllocatedMemory(preAllocatedMemory)
- .setAvailableMemory(availableMemory)
- .setEventNumInFlush(eventNumInFlush)
- .addAllTags(tags)
- .setStatusValue(serverStatus.ordinal())
- .putAllStorageInfo(StorageInfoUtils.toProto(storageInfo))
- .setStartTimeMs(startTimeMs)
+ .setUsedMemory(rssRequest.getUsedMemory())
+ .setPreAllocatedMemory(rssRequest.getPreAllocatedMemory())
+ .setAvailableMemory(rssRequest.getAvailableMemory())
+ .setEventNumInFlush(rssRequest.getEventNumInFlush())
+ .addAllTags(rssRequest.getTags())
+ .setStatusValue(rssRequest.getServerStatus().ordinal())
+
.putAllStorageInfo(StorageInfoUtils.toProto(rssRequest.getStorageInfo()))
+ .setStartTimeMs(rssRequest.getStartTimeMs())
.setVersion(Constants.VERSION)
.setGitCommitId(Constants.REVISION_SHORT)
- .addAllApplicationInfo(appInfos)
+ .addAllApplicationInfo(rssRequest.getAppInfos())
+ .putAllDisplayMetrics(rssRequest.getDisplayMetrics())
.build();
RssProtos.StatusCode status;
ShuffleServerHeartBeatResponse response = null;
try {
- response = blockingStub.withDeadlineAfter(timeout,
TimeUnit.MILLISECONDS).heartbeat(request);
+ response =
+ blockingStub
+ .withDeadlineAfter(rssRequest.getTimeout(),
TimeUnit.MILLISECONDS)
+ .heartbeat(request);
status = response.getStatus();
} catch (StatusRuntimeException e) {
LOG.error("Failed to doSendHeartBeat, request: {}", request, e);
@@ -212,23 +199,7 @@ public class CoordinatorGrpcClient extends GrpcClient
implements CoordinatorClie
@Override
public RssSendHeartBeatResponse sendHeartBeat(RssSendHeartBeatRequest
request) {
- ShuffleServerHeartBeatResponse rpcResponse =
- doSendHeartBeat(
- request.getShuffleServerId(),
- request.getShuffleServerIp(),
- request.getShuffleServerPort(),
- request.getUsedMemory(),
- request.getPreAllocatedMemory(),
- request.getAvailableMemory(),
- request.getEventNumInFlush(),
- request.getTimeout(),
- request.getTags(),
- request.getServerStatus(),
- request.getStorageInfo(),
- request.getNettyPort(),
- request.getJettyPort(),
- request.getStartTimeMs(),
- request.getAppInfos());
+ ShuffleServerHeartBeatResponse rpcResponse = doSendHeartBeat(request);
RssSendHeartBeatResponse response;
RssProtos.StatusCode statusCode = rpcResponse.getStatus();
diff --git
a/internal-client/src/main/java/org/apache/uniffle/client/request/RssSendHeartBeatRequest.java
b/internal-client/src/main/java/org/apache/uniffle/client/request/RssSendHeartBeatRequest.java
index a4f23ba73..84391fd18 100644
---
a/internal-client/src/main/java/org/apache/uniffle/client/request/RssSendHeartBeatRequest.java
+++
b/internal-client/src/main/java/org/apache/uniffle/client/request/RssSendHeartBeatRequest.java
@@ -17,6 +17,7 @@
package org.apache.uniffle.client.request;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -42,6 +43,7 @@ public class RssSendHeartBeatRequest {
private final int jettyPort;
private final long startTimeMs;
private final List<RssProtos.ApplicationInfo> appInfos;
+ private final Map<String, String> displayMetrics;
public RssSendHeartBeatRequest(
String shuffleServerId,
@@ -58,7 +60,8 @@ public class RssSendHeartBeatRequest {
int nettyPort,
int jettyPort,
long startTimeMs,
- List<RssProtos.ApplicationInfo> appInfos) {
+ List<RssProtos.ApplicationInfo> appInfos,
+ Map<String, String> displayMetrics) {
this.shuffleServerId = shuffleServerId;
this.shuffleServerIp = shuffleServerIp;
this.shuffleServerPort = shuffleServerPort;
@@ -74,6 +77,7 @@ public class RssSendHeartBeatRequest {
this.jettyPort = jettyPort;
this.startTimeMs = startTimeMs;
this.appInfos = appInfos;
+ this.displayMetrics = displayMetrics;
}
public String getShuffleServerId() {
@@ -135,4 +139,8 @@ public class RssSendHeartBeatRequest {
public List<RssProtos.ApplicationInfo> getAppInfos() {
return appInfos;
}
+
+ public Map<String, String> getDisplayMetrics() {
+ return displayMetrics == null ? Collections.emptyMap() : displayMetrics;
+ }
}
diff --git a/proto/src/main/proto/Rss.proto b/proto/src/main/proto/Rss.proto
index e883eddc4..1312480de 100644
--- a/proto/src/main/proto/Rss.proto
+++ b/proto/src/main/proto/Rss.proto
@@ -305,6 +305,7 @@ message ShuffleServerHeartBeatRequest {
optional string gitCommitId = 23;
optional int64 startTimeMs = 24;
repeated ApplicationInfo applicationInfo = 25;
+ map<string, string> displayMetrics = 26;
}
message ShuffleServerHeartBeatResponse {
diff --git
a/server/src/main/java/org/apache/uniffle/server/RegisterHeartBeat.java
b/server/src/main/java/org/apache/uniffle/server/RegisterHeartBeat.java
index 5a44728b5..28a027e6b 100644
--- a/server/src/main/java/org/apache/uniffle/server/RegisterHeartBeat.java
+++ b/server/src/main/java/org/apache/uniffle/server/RegisterHeartBeat.java
@@ -17,9 +17,6 @@
package org.apache.uniffle.server;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -30,11 +27,8 @@ import org.slf4j.LoggerFactory;
import org.apache.uniffle.client.factory.CoordinatorClientFactory;
import org.apache.uniffle.client.impl.grpc.CoordinatorGrpcRetryableClient;
import org.apache.uniffle.client.request.RssSendHeartBeatRequest;
-import org.apache.uniffle.common.ServerStatus;
import org.apache.uniffle.common.rpc.StatusCode;
-import org.apache.uniffle.common.storage.StorageInfo;
import org.apache.uniffle.common.util.ThreadUtils;
-import org.apache.uniffle.proto.RssProtos;
public class RegisterHeartBeat {
@@ -73,21 +67,26 @@ public class RegisterHeartBeat {
Runnable runnable =
() -> {
try {
- sendHeartBeat(
- shuffleServer.getId(),
- shuffleServer.getIp(),
- shuffleServer.getGrpcPort(),
- shuffleServer.getUsedMemory(),
- shuffleServer.getPreAllocatedMemory(),
- shuffleServer.getAvailableMemory(),
- shuffleServer.getEventNumInFlush(),
- shuffleServer.getTags(),
- shuffleServer.getServerStatus(),
- shuffleServer.getStorageManager().getStorageInfo(),
- shuffleServer.getNettyPort(),
- shuffleServer.getJettyPort(),
- shuffleServer.getStartTimeMs(),
- shuffleServer.getAppInfos());
+ // use `rss.server.heartbeat.interval` as the timeout option
+ RssSendHeartBeatRequest request =
+ new RssSendHeartBeatRequest(
+ shuffleServer.getId(),
+ shuffleServer.getIp(),
+ shuffleServer.getGrpcPort(),
+ shuffleServer.getUsedMemory(),
+ shuffleServer.getPreAllocatedMemory(),
+ shuffleServer.getAvailableMemory(),
+ shuffleServer.getEventNumInFlush(),
+ heartBeatInterval,
+ shuffleServer.getTags(),
+ shuffleServer.getServerStatus(),
+ shuffleServer.getStorageManager().getStorageInfo(),
+ shuffleServer.getNettyPort(),
+ shuffleServer.getJettyPort(),
+ shuffleServer.getStartTimeMs(),
+ shuffleServer.getAppInfos(),
+ shuffleServer.getDisplayMetrics());
+ sendHeartBeat(request);
} catch (Exception e) {
LOG.warn("Error happened when send heart beat to coordinator");
}
@@ -97,46 +96,17 @@ public class RegisterHeartBeat {
}
@VisibleForTesting
- public boolean sendHeartBeat(
- String id,
- String ip,
- int grpcPort,
- long usedMemory,
- long preAllocatedMemory,
- long availableMemory,
- int eventNumInFlush,
- Set<String> tags,
- ServerStatus serverStatus,
- Map<String, StorageInfo> localStorageInfo,
- int nettyPort,
- int jettyPort,
- long startTimeMs,
- List<RssProtos.ApplicationInfo> appInfos) {
- // use `rss.server.heartbeat.interval` as the timeout option
- RssSendHeartBeatRequest request =
- new RssSendHeartBeatRequest(
- id,
- ip,
- grpcPort,
- usedMemory,
- preAllocatedMemory,
- availableMemory,
- eventNumInFlush,
- heartBeatInterval,
- tags,
- serverStatus,
- localStorageInfo,
- nettyPort,
- jettyPort,
- startTimeMs,
- appInfos);
-
+ public boolean sendHeartBeat(RssSendHeartBeatRequest request) {
if (coordinatorClient.sendHeartBeat(request).getStatusCode() ==
StatusCode.SUCCESS) {
return true;
}
return false;
}
+ public long getHeartBeatInterval() {
+ return heartBeatInterval;
+ }
+
public void shutdown() {
coordinatorClient.close();
service.shutdownNow();
diff --git a/server/src/main/java/org/apache/uniffle/server/ShuffleServer.java
b/server/src/main/java/org/apache/uniffle/server/ShuffleServer.java
index 59f53c97a..b5582b7d9 100644
--- a/server/src/main/java/org/apache/uniffle/server/ShuffleServer.java
+++ b/server/src/main/java/org/apache/uniffle/server/ShuffleServer.java
@@ -25,6 +25,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
@@ -37,6 +38,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import picocli.CommandLine;
+import org.apache.uniffle.client.request.RssSendHeartBeatRequest;
import org.apache.uniffle.common.Arguments;
import org.apache.uniffle.common.ReconfigurableConfManager;
import org.apache.uniffle.common.ServerStatus;
@@ -610,21 +612,25 @@ public class ShuffleServer {
@VisibleForTesting
public void sendHeartbeat() {
ShuffleServer shuffleServer = this;
- registerHeartBeat.sendHeartBeat(
- shuffleServer.getId(),
- shuffleServer.getIp(),
- shuffleServer.getGrpcPort(),
- shuffleServer.getUsedMemory(),
- shuffleServer.getPreAllocatedMemory(),
- shuffleServer.getAvailableMemory(),
- shuffleServer.getEventNumInFlush(),
- shuffleServer.getTags(),
- shuffleServer.getServerStatus(),
- shuffleServer.getStorageManager().getStorageInfo(),
- shuffleServer.getNettyPort(),
- shuffleServer.getJettyPort(),
- shuffleServer.getStartTimeMs(),
- shuffleServer.getAppInfos());
+ RssSendHeartBeatRequest request =
+ new RssSendHeartBeatRequest(
+ shuffleServer.getId(),
+ shuffleServer.getIp(),
+ shuffleServer.getGrpcPort(),
+ shuffleServer.getUsedMemory(),
+ shuffleServer.getPreAllocatedMemory(),
+ shuffleServer.getAvailableMemory(),
+ shuffleServer.getEventNumInFlush(),
+ registerHeartBeat.getHeartBeatInterval(),
+ shuffleServer.getTags(),
+ shuffleServer.getServerStatus(),
+ shuffleServer.getStorageManager().getStorageInfo(),
+ shuffleServer.getNettyPort(),
+ shuffleServer.getJettyPort(),
+ shuffleServer.getStartTimeMs(),
+ shuffleServer.getAppInfos(),
+ shuffleServer.getDisplayMetrics());
+ registerHeartBeat.sendHeartBeat(request);
}
public ShuffleMergeManager getShuffleMergeManager() {
@@ -634,4 +640,15 @@ public class ShuffleServer {
public boolean isRemoteMergeEnable() {
return remoteMergeEnable;
}
+
+ public Map<String, String> getDisplayMetrics() {
+ List<String> list =
shuffleServerConf.get(ShuffleServerConf.SERVER_DISPLAY_METRICS_LIST);
+ return list.stream()
+ .map(metric -> metric.split(":", 2))
+ .filter(parts -> parts.length == 2)
+ .collect(
+ Collectors.toMap(
+ parts -> parts[0], // 使用第一部分作为键
+ parts ->
String.valueOf(ShuffleServerMetrics.getMetricsValue(parts[1]))));
+ }
}
diff --git
a/server/src/main/java/org/apache/uniffle/server/ShuffleServerConf.java
b/server/src/main/java/org/apache/uniffle/server/ShuffleServerConf.java
index 8888e60b7..ca0a3d6c7 100644
--- a/server/src/main/java/org/apache/uniffle/server/ShuffleServerConf.java
+++ b/server/src/main/java/org/apache/uniffle/server/ShuffleServerConf.java
@@ -751,6 +751,13 @@ public class ShuffleServerConf extends RssBaseConf {
.withDescription(
"The block id manager class, the implementation of this
interface "
+ "to manage the shuffle block ids");
+ public static final ConfigOption<List<String>> SERVER_DISPLAY_METRICS_LIST =
+ ConfigOptions.key("rss.server.displayMetricsList")
+ .stringType()
+ .asList()
+ .defaultValues("app:app_num_with_node",
"partition:partition_num_with_node")
+ .withDescription(
+ "A list of metrics will report to coordinator and dashboard,
format in \"displayName:metricsName\", separated by ','");
public ShuffleServerConf() {}
diff --git
a/server/src/main/java/org/apache/uniffle/server/ShuffleServerMetrics.java
b/server/src/main/java/org/apache/uniffle/server/ShuffleServerMetrics.java
index 22d9b7c0a..b891ecc5e 100644
--- a/server/src/main/java/org/apache/uniffle/server/ShuffleServerMetrics.java
+++ b/server/src/main/java/org/apache/uniffle/server/ShuffleServerMetrics.java
@@ -539,4 +539,13 @@ public class ShuffleServerMetrics {
}
metricsManager.addLabeledCacheGauge(name, supplier, updateInterval);
}
+
+ public static Double getMetricsValue(String metricName) {
+ return metricsManager
+ .getCollectorRegistry()
+ .getSampleValue(
+ metricName,
+ metricsManager.getDefaultLabelNames(),
+ metricsManager.getDefaultLabelValues());
+ }
}
