Skip to content

Commit 3a35b0f

Browse files
lwllvybwenlongwlli
lwllvyb
and
wenlongwlli
authored
[#2207] feat(dashboard): Add the write information of appinfo in Shuflle Server heartbeat (#2208)
### What changes were proposed in this pull request? Add the write information of appinfo in Shuflle Server heartbeat ### Why are the changes needed? Fix: #2207 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Locally Co-authored-by: wenlongwlli <wenlongwlli@tencent.com>
1 parent 152ea7a commit 3a35b0f

8 files changed

Lines changed: 77 additions & 10 deletions

File tree

‎coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -537,7 +537,8 @@ private ServerNode toServerNode(ShuffleServerHeartBeatRequest request) {
537537
request.getServerId().getJettyPort(),
538538
request.getStartTimeMs(),
539539
request.getVersion(),
540-
request.getGitCommitId());
540+
request.getGitCommitId(),
541+
request.getApplicationInfoList());
541542
}
542543

543544
/**

‎coordinator/src/main/java/org/apache/uniffle/coordinator/ServerNode.java‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,18 @@
1717

1818
package org.apache.uniffle.coordinator;
1919

20+
import java.util.Collections;
21+
import java.util.List;
2022
import java.util.Map;
2123
import java.util.Set;
24+
import java.util.concurrent.ConcurrentHashMap;
2225

2326
import com.google.common.collect.Maps;
2427
import com.google.common.collect.Sets;
2528

2629
import org.apache.uniffle.common.ServerStatus;
2730
import org.apache.uniffle.common.storage.StorageInfo;
31+
import org.apache.uniffle.proto.RssProtos;
2832
import org.apache.uniffle.proto.RssProtos.ShuffleServerId;
2933

3034
public class ServerNode implements Comparable<ServerNode> {
@@ -46,6 +50,7 @@ public class ServerNode implements Comparable<ServerNode> {
4650
private long startTime = -1;
4751
private String version;
4852
private String gitCommitId;
53+
Map<String, RssProtos.ApplicationInfo> appIdToInfos;
4954

5055
public ServerNode(String id) {
5156
this(id, "", 0, 0, 0, 0, 0, Sets.newHashSet(), ServerStatus.EXCLUDED);
@@ -181,7 +186,8 @@ public ServerNode(
181186
jettyPort,
182187
startTime,
183188
"",
184-
"");
189+
"",
190+
Collections.EMPTY_LIST);
185191
}
186192

187193
public ServerNode(
@@ -199,7 +205,8 @@ public ServerNode(
199205
int jettyPort,
200206
long startTime,
201207
String version,
202-
String gitCommitId) {
208+
String gitCommitId,
209+
List<RssProtos.ApplicationInfo> appInfos) {
203210
this.id = id;
204211
this.ip = ip;
205212
this.grpcPort = grpcPort;
@@ -221,6 +228,8 @@ public ServerNode(
221228
this.startTime = startTime;
222229
this.version = version;
223230
this.gitCommitId = gitCommitId;
231+
this.appIdToInfos = new ConcurrentHashMap<>();
232+
appInfos.forEach(appInfo -> appIdToInfos.put(appInfo.getAppId(), appInfo));
224233
}
225234

226235
public ShuffleServerId convertToGrpcProto() {

‎internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/CoordinatorGrpcClient.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -127,7 +127,8 @@ public ShuffleServerHeartBeatResponse doSendHeartBeat(
127127
Map<String, StorageInfo> storageInfo,
128128
int nettyPort,
129129
int jettyPort,
130-
long startTimeMs) {
130+
long startTimeMs,
131+
List<RssProtos.ApplicationInfo> appInfos) {
131132
ShuffleServerId serverId =
132133
ShuffleServerId.newBuilder()
133134
.setId(id)
@@ -149,6 +150,7 @@ public ShuffleServerHeartBeatResponse doSendHeartBeat(
149150
.setStartTimeMs(startTimeMs)
150151
.setVersion(Constants.VERSION)
151152
.setGitCommitId(Constants.REVISION_SHORT)
153+
.addAllApplicationInfo(appInfos)
152154
.build();
153155

154156
RssProtos.StatusCode status;
@@ -225,7 +227,8 @@ public RssSendHeartBeatResponse sendHeartBeat(RssSendHeartBeatRequest request) {
225227
request.getStorageInfo(),
226228
request.getNettyPort(),
227229
request.getJettyPort(),
228-
request.getStartTimeMs());
230+
request.getStartTimeMs(),
231+
request.getAppInfos());
229232

230233
RssSendHeartBeatResponse response;
231234
RssProtos.StatusCode statusCode = rpcResponse.getStatus();

‎internal-client/src/main/java/org/apache/uniffle/client/request/RssSendHeartBeatRequest.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,13 @@
1717

1818
package org.apache.uniffle.client.request;
1919

20+
import java.util.List;
2021
import java.util.Map;
2122
import java.util.Set;
2223

2324
import org.apache.uniffle.common.ServerStatus;
2425
import org.apache.uniffle.common.storage.StorageInfo;
26+
import org.apache.uniffle.proto.RssProtos;
2527

2628
public class RssSendHeartBeatRequest {
2729

@@ -39,6 +41,7 @@ public class RssSendHeartBeatRequest {
3941
private final int nettyPort;
4042
private final int jettyPort;
4143
private final long startTimeMs;
44+
private final List<RssProtos.ApplicationInfo> appInfos;
4245

4346
public RssSendHeartBeatRequest(
4447
String shuffleServerId,
@@ -54,7 +57,8 @@ public RssSendHeartBeatRequest(
5457
Map<String, StorageInfo> storageInfo,
5558
int nettyPort,
5659
int jettyPort,
57-
long startTimeMs) {
60+
long startTimeMs,
61+
List<RssProtos.ApplicationInfo> appInfos) {
5862
this.shuffleServerId = shuffleServerId;
5963
this.shuffleServerIp = shuffleServerIp;
6064
this.shuffleServerPort = shuffleServerPort;
@@ -69,6 +73,7 @@ public RssSendHeartBeatRequest(
6973
this.nettyPort = nettyPort;
7074
this.jettyPort = jettyPort;
7175
this.startTimeMs = startTimeMs;
76+
this.appInfos = appInfos;
7277
}
7378

7479
public String getShuffleServerId() {
@@ -126,4 +131,8 @@ public int getJettyPort() {
126131
public long getStartTimeMs() {
127132
return startTimeMs;
128133
}
134+
135+
public List<RssProtos.ApplicationInfo> getAppInfos() {
136+
return appInfos;
137+
}
129138
}

‎proto/src/main/proto/Rss.proto‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -273,6 +273,17 @@ enum ServerStatus {
273273
// todo: more status, such as UPGRADING
274274
}
275275

276+
message ApplicationInfo {
277+
string appId = 1;
278+
int64 partitionNum = 2;
279+
int64 memorySize = 3;
280+
int64 localFileNum = 4;
281+
int64 localTotalSize = 5;
282+
int64 hadoopFileNum = 6;
283+
int64 hadoopTotalSize = 7;
284+
int64 totalSize = 8;
285+
}
286+
276287
message ShuffleServerHeartBeatRequest {
277288
ShuffleServerId serverId = 1;
278289
int64 usedMemory = 2;
@@ -286,6 +297,7 @@ message ShuffleServerHeartBeatRequest {
286297
optional string version = 22;
287298
optional string gitCommitId = 23;
288299
optional int64 startTimeMs = 24;
300+
repeated ApplicationInfo applicationInfo = 25;
289301
}
290302

291303
message ShuffleServerHeartBeatResponse {

‎server/src/main/java/org/apache/uniffle/server/RegisterHeartBeat.java‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
package org.apache.uniffle.server;
1919

20+
import java.util.List;
2021
import java.util.Map;
2122
import java.util.Set;
2223
import java.util.concurrent.ScheduledExecutorService;
@@ -33,6 +34,7 @@
3334
import org.apache.uniffle.common.rpc.StatusCode;
3435
import org.apache.uniffle.common.storage.StorageInfo;
3536
import org.apache.uniffle.common.util.ThreadUtils;
37+
import org.apache.uniffle.proto.RssProtos;
3638

3739
public class RegisterHeartBeat {
3840

@@ -84,7 +86,8 @@ public void startHeartBeat() {
8486
shuffleServer.getStorageManager().getStorageInfo(),
8587
shuffleServer.getNettyPort(),
8688
shuffleServer.getJettyPort(),
87-
shuffleServer.getStartTimeMs());
89+
shuffleServer.getStartTimeMs(),
90+
shuffleServer.getAppInfos());
8891
} catch (Exception e) {
8992
LOG.warn("Error happened when send heart beat to coordinator");
9093
}
@@ -107,7 +110,8 @@ public boolean sendHeartBeat(
107110
Map<String, StorageInfo> localStorageInfo,
108111
int nettyPort,
109112
int jettyPort,
110-
long startTimeMs) {
113+
long startTimeMs,
114+
List<RssProtos.ApplicationInfo> appInfos) {
111115
// use `rss.server.heartbeat.interval` as the timeout option
112116
RssSendHeartBeatRequest request =
113117
new RssSendHeartBeatRequest(
@@ -124,7 +128,8 @@ public boolean sendHeartBeat(
124128
localStorageInfo,
125129
nettyPort,
126130
jettyPort,
127-
startTimeMs);
131+
startTimeMs,
132+
appInfos);
128133

129134
if (coordinatorClient.sendHeartBeat(request).getStatusCode() == StatusCode.SUCCESS) {
130135
return true;

‎server/src/main/java/org/apache/uniffle/server/ShuffleServer.java‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,10 @@
1717

1818
package org.apache.uniffle.server;
1919

20+
import java.util.ArrayList;
2021
import java.util.Collections;
2122
import java.util.List;
23+
import java.util.Map;
2224
import java.util.Set;
2325
import java.util.concurrent.ExecutorService;
2426
import java.util.concurrent.Future;
@@ -57,6 +59,7 @@
5759
import org.apache.uniffle.common.util.ThreadUtils;
5860
import org.apache.uniffle.common.web.CoalescedCollectorRegistry;
5961
import org.apache.uniffle.common.web.JettyServer;
62+
import org.apache.uniffle.proto.RssProtos;
6063
import org.apache.uniffle.server.buffer.ShuffleBufferManager;
6164
import org.apache.uniffle.server.buffer.ShuffleBufferType;
6265
import org.apache.uniffle.server.merge.ShuffleMergeManager;
@@ -584,6 +587,26 @@ public long getStartTimeMs() {
584587
return startTimeMs;
585588
}
586589

590+
public List<RssProtos.ApplicationInfo> getAppInfos() {
591+
List<RssProtos.ApplicationInfo> appInfos = new ArrayList<>();
592+
Map<String, ShuffleTaskInfo> taskInfos = getShuffleTaskManager().getShuffleTaskInfos();
593+
taskInfos.forEach(
594+
(appId, taskInfo) -> {
595+
RssProtos.ApplicationInfo applicationInfo =
596+
RssProtos.ApplicationInfo.newBuilder()
597+
.setAppId(appId)
598+
.setPartitionNum(taskInfo.getPartitionNum())
599+
.setMemorySize(taskInfo.getInMemoryDataSize())
600+
.setLocalTotalSize(taskInfo.getOnLocalFileDataSize())
601+
.setHadoopTotalSize(taskInfo.getOnHadoopDataSize())
602+
.setTotalSize(taskInfo.getTotalDataSize())
603+
.build();
604+
605+
appInfos.add(applicationInfo);
606+
});
607+
return appInfos;
608+
}
609+
587610
@VisibleForTesting
588611
public void sendHeartbeat() {
589612
ShuffleServer shuffleServer = this;
@@ -600,7 +623,8 @@ public void sendHeartbeat() {
600623
shuffleServer.getStorageManager().getStorageInfo(),
601624
shuffleServer.getNettyPort(),
602625
shuffleServer.getJettyPort(),
603-
shuffleServer.getStartTimeMs());
626+
shuffleServer.getStartTimeMs(),
627+
shuffleServer.getAppInfos());
604628
}
605629

606630
public ShuffleMergeManager getShuffleMergeManager() {

‎server/src/main/java/org/apache/uniffle/server/ShuffleTaskInfo.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -283,6 +283,10 @@ public ShuffleDetailInfo getShuffleDetailInfo(int shuffleId) {
283283
return shuffleDetailInfos.get(shuffleId);
284284
}
285285

286+
public long getPartitionNum() {
287+
return partitionDataSizes.values().stream().mapToLong(Map::size).sum();
288+
}
289+
286290
@Override
287291
public String toString() {
288292
return "ShuffleTaskInfo{"

0 commit comments

Comments
 (0)