Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Feature][Zeta][REST-API] Add REST API To Get System Monitoring Information #4315

Merged
merged 5 commits into from
Mar 24, 2023
+136 −0
Merged

[Feature][Zeta][REST-API] Add REST API To Get System Monitoring Information #4315

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
df507da
[Feature][Zeta][REST-API]Get system monitoring information
MonsterChenzhuo Mar 9, 2023
601545b
fix
MonsterChenzhuo Mar 9, 2023
9f2c093
Merge branch 'dev' into restapi
MonsterChenzhuo Mar 9, 2023
6f1774c
Merge branch 'dev' into restapi
MonsterChenzhuo Mar 22, 2023
c266205
fix
MonsterChenzhuo Mar 22, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 65 additions & 0 deletions docs/en/seatunnel-engine/rest-api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,3 +115,68 @@ network:

------------------------------------------------------------------------------------------

### Returns system monitoring information.

<details>
<summary><code>GET</code> <code><b>/hazelcast/rest/maps/system-monitoring-information</b></code> <code>(Returns system monitoring information.)</code></summary>

#### Parameters

#### Responses

```json
[
{
"processors":"8",
"physical.memory.total":"16.0G",
"physical.memory.free":"16.3M",
"swap.space.total":"0",
"swap.space.free":"0",
"heap.memory.used":"135.7M",
"heap.memory.free":"440.8M",
"heap.memory.total":"576.5M",
"heap.memory.max":"3.6G",
"heap.memory.used/total":"23.54%",
"heap.memory.used/max":"3.73%",
"minor.gc.count":"6",
"minor.gc.time":"110ms",
"major.gc.count":"2",
"major.gc.time":"73ms",
"load.process":"24.78%",
"load.system":"60.00%",
"load.systemAverage":"2.07",
"thread.count":"117",
"thread.peakCount":"118",
"cluster.timeDiff":"0",
"event.q.size":"0",
"executor.q.async.size":"0",
"executor.q.client.size":"0",
"executor.q.client.query.size":"0",
"executor.q.client.blocking.size":"0",
"executor.q.query.size":"0",
"executor.q.scheduled.size":"0",
"executor.q.io.size":"0",
"executor.q.system.size":"0",
"executor.q.operations.size":"0",
"executor.q.priorityOperation.size":"0",
"operations.completed.count":"10",
"executor.q.mapLoad.size":"0",
"executor.q.mapLoadAllKeys.size":"0",
"executor.q.cluster.size":"0",
"executor.q.response.size":"0",
"operations.running.count":"0",
"operations.pending.invocations.percentage":"0.00%",
"operations.pending.invocations.count":"0",
"proxy.count":"8",
"clientEndpoint.count":"0",
"connection.active.count":"2",
"client.connection.count":"0",
"connection.count":"0"
}
]
```

</details>

------------------------------------------------------------------------------------------

17 changes: 17 additions & 0 deletions ...connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@

import static io.restassured.RestAssured.given;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.lessThan;

@Slf4j
public class RestApiIT {
Expand Down Expand Up @@ -114,6 +115,22 @@ public void testGetRunningJobs() {
.body("[0].jobStatus", equalTo("RUNNING"));
}

@Test
public void testSystemMonitoringInformation() {
given().get(
HOST
+ hazelcastInstance
.getCluster()
.getLocalMember()
.getAddress()
.getPort()
+ RestConstant.SYSTEM_MONITORING_INFORMATION)
.then()
.assertThat()
.time(lessThan(5000L))
.statusCode(200);
}

@AfterAll
static void afterClass() {
if (hazelcastInstance != null) {
Expand Down
3 changes: 3 additions & 0 deletions ...nel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,4 +21,7 @@ public class RestConstant {

public static final String RUNNING_JOBS_URL = "/hazelcast/rest/maps/running-jobs";
public static final String RUNNING_JOB_URL = "/hazelcast/rest/maps/running-job";

public static final String SYSTEM_MONITORING_INFORMATION =
"/hazelcast/rest/maps/system-monitoring-information";
}
51 changes: 51 additions & 0 deletions ...er/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,12 @@
import org.apache.seatunnel.engine.core.job.JobStatus;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
import org.apache.seatunnel.engine.server.log.Log4j2HttpGetCommandProcessor;
import org.apache.seatunnel.engine.server.operation.GetClusterHealthMetricsOperation;
import org.apache.seatunnel.engine.server.utils.NodeEngineUtil;

import com.hazelcast.cluster.Address;
import com.hazelcast.cluster.Cluster;
import com.hazelcast.cluster.Member;
import com.hazelcast.internal.ascii.TextCommandService;
import com.hazelcast.internal.ascii.rest.HttpCommandProcessor;
import com.hazelcast.internal.ascii.rest.HttpGetCommand;
Expand All @@ -38,15 +43,20 @@
import com.hazelcast.internal.util.JsonUtil;
import com.hazelcast.internal.util.StringUtil;
import com.hazelcast.map.IMap;
import com.hazelcast.spi.impl.NodeEngine;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import static com.hazelcast.internal.ascii.rest.HttpStatusCode.SC_500;
import static org.apache.seatunnel.engine.server.rest.RestConstant.RUNNING_JOBS_URL;
import static org.apache.seatunnel.engine.server.rest.RestConstant.RUNNING_JOB_URL;
import static org.apache.seatunnel.engine.server.rest.RestConstant.SYSTEM_MONITORING_INFORMATION;

public class RestHttpGetCommandProcessor extends HttpCommandProcessor<HttpGetCommand> {

Expand All @@ -56,6 +66,8 @@ public class RestHttpGetCommandProcessor extends HttpCommandProcessor<HttpGetCom

private static final String SINK_WRITE_COUNT = "SinkWriteCount";

private NodeEngine nodeEngine;

public RestHttpGetCommandProcessor(TextCommandService textCommandService) {
this(textCommandService, new Log4j2HttpGetCommandProcessor(textCommandService));
}
Expand All @@ -77,6 +89,8 @@ public void handle(HttpGetCommand httpGetCommand) {
handleRunningJobsInfo(httpGetCommand);
} else if (uri.startsWith(RUNNING_JOB_URL)) {
handleJobInfoById(httpGetCommand, uri);
} else if (uri.startsWith(SYSTEM_MONITORING_INFORMATION)) {
getSystemMonitoringInformation(httpGetCommand);
} else {
original.handle(httpGetCommand);
}
Expand All @@ -95,6 +109,43 @@ public void handleRejection(HttpGetCommand httpGetCommand) {
handle(httpGetCommand);
}

private void getSystemMonitoringInformation(HttpGetCommand command) {
Cluster cluster = textCommandService.getNode().hazelcastInstance.getCluster();
nodeEngine = textCommandService.getNode().hazelcastInstance.node.nodeEngine;

Set<Member> members = cluster.getMembers();
JsonArray jsonValues =
members.stream()
.map(
member -> {
Address address = member.getAddress();
String input = null;
try {
input =
(String)
NodeEngineUtil.sendOperationToMemberNode(
nodeEngine,
new GetClusterHealthMetricsOperation(),
address)
.get();
} catch (InterruptedException | ExecutionException e) {
logger.severe("get system monitoring information fail", e);
}
assert input != null;
String[] parts = input.split(", ");
JsonObject jobInfo = new JsonObject();
Arrays.stream(parts)
.forEach(
part -> {
String[] keyValue = part.split("=");
jobInfo.add(keyValue[0], keyValue[1]);
});
return jobInfo;
})
.collect(JsonArray::new, JsonArray::add, JsonArray::add);
this.prepareResponse(command, jsonValues);
}

private void handleRunningJobsInfo(HttpGetCommand command) {
IMap<Long, JobInfo> values =
this.textCommandService
Expand Down