github/arthas
4
0
Fork 0
mirror of https://github.com/alibaba/arthas.git synced 2026-10-21 05:51:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

parse json to pb

Browse Source
This commit is contained in:
gongdewei
2020-09-14 10:50:07 +08:00
parent 5322278416
commit 3b00b0dbdf
11 changed files with 100 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files
channel/channel-server/pom.xml
+6
View File
@@ -97,6 +97,12 @@
<scope>test</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiAction.java
-48
View File
@@ -1,48 +0,0 @@
package com.alibaba.arthas.channel.server.api;
/**
* Http api action enums
*
* @author gongdewei 2020-03-25
*/
public enum ApiAction {
/**
* Execute command synchronized
*/
EXEC,
/**
* Execute command async
*/
ASYNC_EXEC,
/**
* Interrupt executing job
*/
INTERRUPT_JOB,
/**
* Create a new session
*/
INIT_SESSION,
/**
* Terminate the session
*/
CLOSE_SESSION,
/**
* Join a exist session
*/
JOIN_SESSION,
/**
* Open new WebConsole and create new session
*/
OPEN_CONSOLE,
/**
* WebConsole input
*/
CONSOLE_INPUT
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiException.java
-16
View File
@@ -1,16 +0,0 @@
package com.alibaba.arthas.channel.server.api;
/**
* Http Api exception
* @author gongdewei 2020-03-19
*/
public class ApiException extends Exception {
public ApiException(String message) {
super(message);
}
public ApiException(String message, Throwable cause) {
super(message, cause);
}
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiRequest.java
-67
View File
@@ -1,67 +0,0 @@
package com.alibaba.arthas.channel.server.api;
import java.util.Map;
/**
* Http Api request
*
* @author gongdewei 2020-03-19
*/
public class ApiRequest {
private String action;
private String command;
private String sessionId;
private String consumerId;
private Integer execTimeout;
@Override
public String toString() {
return "ApiRequest{" +
"action='" + action + '\'' +
", command='" + command + '\'' +
", sessionId='" + sessionId + '\'' +
", consumerId='" + consumerId + '\'' +
", timeout=" + execTimeout +
'}';
}
public String getAction() {
return action;
}
public void setAction(String action) {
this.action = action;
}
public String getCommand() {
return command;
}
public void setCommand(String command) {
this.command = command;
}
public String getSessionId() {
return sessionId;
}
public void setSessionId(String sessionId) {
this.sessionId = sessionId;
}
public String getConsumerId() {
return consumerId;
}
public void setConsumerId(String consumerId) {
this.consumerId = consumerId;
}
public Integer getExecTimeout() {
return execTimeout;
}
public void setExecTimeout(Integer execTimeout) {
this.execTimeout = execTimeout;
}
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiResponse.java
-78
View File
@@ -1,78 +0,0 @@
package com.alibaba.arthas.channel.server.api;
/**
* Http Api exception
* @author gongdewei 2020-03-19
*/
public class ApiResponse<T> {
private String agentId;
private String requestId;
private ApiStatus status;
private String message;
private String sessionId;
// private String consumerId;
private T result;
public String getAgentId() {
return agentId;
}
public ApiResponse<T> setAgentId(String agentId) {
this.agentId = agentId;
return this;
}
public String getRequestId() {
return requestId;
}
public ApiResponse<T> setRequestId(String requestId) {
this.requestId = requestId;
return this;
}
public ApiStatus getStatus() {
return status;
}
public ApiResponse<T> setStatus(ApiStatus status) {
this.status = status;
return this;
}
public String getMessage() {
return message;
}
public ApiResponse<T> setMessage(String message) {
this.message = message;
return this;
}
public String getSessionId() {
return sessionId;
}
public ApiResponse<T> setSessionId(String sessionId) {
this.sessionId = sessionId;
return this;
}
// public String getConsumerId() {
// return consumerId;
// }
//
// public ApiResponse<T> setConsumerId(String consumerId) {
// this.consumerId = consumerId;
// return this;
// }
public T getResult() {
return result;
}
public ApiResponse<T> setResult(T result) {
this.result = result;
return this;
}
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiStatus.java
-33
View File
@@ -1,33 +0,0 @@
package com.alibaba.arthas.channel.server.api;
/**
* Http API response state
*
* @author gongdewei 2020-03-19
*/
public enum ApiStatus {
/**
* Response is CONTINUOUS, receiving streaming data of async exec job
*/
CONTINUOUS,
/**
* Request processed successfully
*/
SUCCEEDED,
/**
* Request processing interrupt
*/
INTERRUPTED,
/**
* Request processing failed
*/
FAILED,
/**
* Request is refused
*/
REFUSED
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java
+1 -1
View File
@@ -90,7 +90,7 @@ public class MessageExchangeServiceImpl implements MessageExchangeService {
public Mono<byte[]> pollMessage(Topic topic, int timeout) {
try {
TopicData topicData = getAndCheckTopicExists(topic);
return Mono.just(topicData.messageQueue.poll(timeout, TimeUnit.MILLISECONDS));
return Mono.justOrEmpty(topicData.messageQueue.poll(timeout, TimeUnit.MILLISECONDS));
} catch (Throwable e) {
return Mono.error(new MessageExchangeException("poll message failure: "+e.getMessage(), e));
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/ApiActionDelegateServiceImpl.java
+2 -1
View File
@@ -82,8 +82,9 @@ public class ApiActionDelegateServiceImpl implements ApiActionDelegateService {
// 获取JobId？
return Mono.just(ActionResponse.newBuilder()
.setStatus(ResponseStatus.CONTINUOUS)
.setAgentId(agentId)
.setRequestId(requestId)
.setStatus(ResponseStatus.CONTINUOUS)
.build());
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/utils/PbJsonUtils.java
+23
View File
@@ -0,0 +1,23 @@
package com.alibaba.arthas.channel.server.utils;
import com.alibaba.arthas.channel.proto.ActionRequest;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.MessageOrBuilder;
import com.google.protobuf.util.JsonFormat;
/**
* @author gongdewei 2020/9/14
*/
public class PbJsonUtils {
public static String convertToJson(MessageOrBuilder message) throws InvalidProtocolBufferException {
return JsonFormat.printer().print(message);
}
public static ActionRequest parseRequest(String json) throws InvalidProtocolBufferException {
ActionRequest.Builder builder = ActionRequest.newBuilder();
JsonFormat.parser().merge(json, builder);
return builder.build();
}
}
channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java
+11 -87
View File
@@ -3,21 +3,12 @@ package com.alibaba.arthas.channel.server.web;
import com.alibaba.arthas.channel.proto.ActionRequest;
import com.alibaba.arthas.channel.proto.ActionResponse;
import com.alibaba.arthas.channel.proto.ExecuteParams;
import com.alibaba.arthas.channel.proto.RequestAction;
import com.alibaba.arthas.channel.proto.ResponseStatus;
import com.alibaba.arthas.channel.proto.ResultFormat;
import com.alibaba.arthas.channel.server.api.ApiAction;
import com.alibaba.arthas.channel.server.api.ApiException;
import com.alibaba.arthas.channel.server.api.ApiRequest;
import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
import com.alibaba.arthas.channel.server.model.AgentVO;
import com.alibaba.arthas.channel.server.service.AgentManageService;
import com.alibaba.arthas.channel.server.service.ApiActionDelegateService;
import com.alibaba.fastjson.JSON;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.MessageOrBuilder;
import com.google.protobuf.util.JsonFormat;
import org.apache.commons.lang3.StringUtils;
import com.alibaba.arthas.channel.server.utils.PbJsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -120,13 +111,14 @@ public class AgentController {
@PostMapping("/agent/{agentId}/exec")
public Mono<ActionResponse> execCommand(@PathVariable String agentId, @RequestBody String requestBody) {
try {
ApiRequest apiRequest = parseRequest(requestBody);
ActionRequest request = PbJsonUtils.parseRequest(requestBody);
long execTimeout = 30000;
if (apiRequest.getExecTimeout() != null && apiRequest.getExecTimeout() > 0) {
execTimeout = apiRequest.getExecTimeout();
ExecuteParams executeParams = request.getExecuteParams();
int execTimeoutParam = executeParams.getExecTimeout();
if (execTimeoutParam > 0) {
execTimeout = execTimeoutParam;
}
ActionRequest request = convertApiRequest(apiRequest);
return apiActionDelegateService.execCommand(agentId, request)
.timeout(Duration.ofMillis(execTimeout));
@@ -143,8 +135,7 @@ public class AgentController {
@PostMapping("/agent/{agentId}/async_exec")
public Mono<ActionResponse> asyncExecCommand(@PathVariable String agentId, @RequestBody String requestBody) {
try {
ApiRequest apiRequest = parseRequest(requestBody);
ActionRequest request = convertApiRequest(apiRequest);
ActionRequest request = PbJsonUtils.parseRequest(requestBody);
return apiActionDelegateService.asyncExecCommand(agentId, request);
} catch (Throwable e) {
logger.error("async exec command failure: " + e.getMessage(), e);
@@ -177,8 +168,7 @@ public class AgentController {
@RequestParam(value = "timeout", defaultValue = "300000") final int timeout) {
final SseEmitter emitter = new SseEmitter(Long.valueOf(timeout));
try {
ApiRequest apiRequest = parseRequest(requestBody);
ActionRequest request = convertApiRequest(apiRequest);
ActionRequest request = PbJsonUtils.parseRequest(requestBody);
ActionResponse actionResponse = apiActionDelegateService.asyncExecCommand(agentId, request).block();
String requestId = actionResponse.getRequestId();
subscribeResults(agentId, requestId, timeout, emitter);
@@ -190,9 +180,7 @@ public class AgentController {
.setMessage("async exec command failure: " + e.getMessage())
.build();
try {
// ApiResponse apiResponse = convertApiResponse(response);
// emitter.send(JSON.toJSONString(apiResponse));
emitter.send(convertToJson(response));
emitter.send(PbJsonUtils.convertToJson(response));
emitter.complete();
} catch (Exception ex) {
emitter.completeWithError(ex);
@@ -201,20 +189,14 @@ public class AgentController {
return emitter;
}
private String convertToJson(MessageOrBuilder message) throws InvalidProtocolBufferException {
return JsonFormat.printer().print(message);
}
private void subscribeResults(String agentId, String requestId, int timeout, final SseEmitter emitter) {
try {
apiActionDelegateService.subscribeResults(agentId, requestId, timeout, new ApiActionDelegateService.ResponseListener() {
@Override
public boolean onMessage(ActionResponse response) {
try {
//TODO convert pb message to json
// ApiResponse apiResponse = convertApiResponse(response);
// emitter.send(JSON.toJSONString(apiResponse));
emitter.send(convertToJson(response));
// convert pb message to json
emitter.send(PbJsonUtils.convertToJson(response));
if (!response.getStatus().equals(ResponseStatus.CONTINUOUS)) {
emitter.complete();
return false;
@@ -245,64 +227,6 @@ public class AgentController {
}
}
private static ActionRequest convertApiRequest(ApiRequest request) throws Exception {
RequestAction action = getAction(request.getAction());
ActionRequest.Builder actionRequestBuilder = ActionRequest.newBuilder()
.setAction(action);
if (request.getSessionId() != null) {
actionRequestBuilder = actionRequestBuilder.setSessionId(request.getSessionId());
}
// if (request.getConsumerId() != null) {
// actionRequestBuilder = actionRequestBuilder.setConsumerId(StringValue.of(request.getConsumerId()));
// }
if (request.getCommand() != null) {
int execTimeout = request.getExecTimeout() !=null && request.getExecTimeout() > 0 ? request.getExecTimeout() : 30000;
actionRequestBuilder = actionRequestBuilder.setExecuteParams(ExecuteParams.newBuilder()
.setResultFormat(ResultFormat.JSON)
.setCommandLine(request.getCommand())
.setExecTimeout(execTimeout)
.build());
}
return actionRequestBuilder.build();
}
private static RequestAction getAction(String action) {
ApiAction apiAction = ApiAction.valueOf(action.trim().toUpperCase());
switch (apiAction) {
case EXEC:
return RequestAction.EXECUTE;
case ASYNC_EXEC:
return RequestAction.ASYNC_EXECUTE;
// case JOIN_SESSION:
// return RequestAction.JOIN_SESSION;
case INIT_SESSION:
return RequestAction.INIT_SESSION;
case CLOSE_SESSION:
return RequestAction.CLOSE_SESSION;
case INTERRUPT_JOB:
return RequestAction.INTERRUPT_JOB;
case OPEN_CONSOLE:
return RequestAction.OPEN_CONSOLE;
case CONSOLE_INPUT:
return RequestAction.CONSOLE_INPUT;
}
throw new IllegalArgumentException("Unsupported request action: " + action);
}
private ApiRequest parseRequest(String requestBody) throws ApiException {
if (StringUtils.isBlank(requestBody)) {
throw new ApiException("parse request failed: request body is empty");
}
try {
return JSON.parseObject(requestBody, ApiRequest.class);
} catch (Exception e) {
throw new ApiException("parse request failed: " + e.getMessage(), e);
}
}
private AgentVO checkAgentExists(String agentId) {
Optional<AgentVO> optionalAgentVO = agentManageService.findAgentById(agentId).block();
if (!optionalAgentVO.isPresent()) {
channel/channel-server/src/test/java/com/alibaba/arthas/channel/server/json/PbJsonTest.java
+57
View File
@@ -0,0 +1,57 @@
package com.alibaba.arthas.channel.server.json;
import com.alibaba.arthas.channel.proto.ActionRequest;
import com.alibaba.arthas.channel.proto.ConsoleParams;
import com.alibaba.arthas.channel.proto.ExecuteParams;
import com.alibaba.arthas.channel.proto.RequestAction;
import com.alibaba.arthas.channel.proto.ResultFormat;
import com.alibaba.arthas.channel.server.utils.PbJsonUtils;
import com.google.protobuf.InvalidProtocolBufferException;
import org.junit.Assert;
import org.junit.Test;
/**
* @author gongdewei 2020/9/14
*/
public class PbJsonTest {
@Test
public void testExecuteRequestJson() throws InvalidProtocolBufferException {
ActionRequest request = ActionRequest.newBuilder()
.setAgentId("agentid-123423")
.setRequestId("requestid-12342")
.setAction(RequestAction.EXECUTE)
.setExecuteParams(ExecuteParams.newBuilder()
.setResultFormat(ResultFormat.JSON)
.setCommandLine("session")
)
.build();
String json = PbJsonUtils.convertToJson(request);
System.out.println(json);
ActionRequest request2 = PbJsonUtils.parseRequest(json);
Assert.assertEquals("ActionRequest json parse failure", request, request2);
}
@Test
public void testConsoleRequestJson() throws InvalidProtocolBufferException {
ActionRequest request = ActionRequest.newBuilder()
.setAgentId("agentid-123423")
.setRequestId("requestid-12342")
.setAction(RequestAction.CONSOLE_INPUT)
.setConsoleParams(ConsoleParams.newBuilder()
.setConsoleId("consoleid-12343")
.setInputData("session\n")
)
.build();
String json = PbJsonUtils.convertToJson(request);
System.out.println(json);
ActionRequest request2 = PbJsonUtils.parseRequest(json);
Assert.assertEquals("ActionRequest json parse failure", request, request2);
}
}