diff --git a/channel/channel-server/pom.xml b/channel/channel-server/pom.xml index ef6ebaa5b..351dba9f7 100644 --- a/channel/channel-server/pom.xml +++ b/channel/channel-server/pom.xml @@ -97,6 +97,12 @@ test true + + + org.springframework.boot + spring-boot-starter-test + test + diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiAction.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiAction.java deleted file mode 100644 index 7d45fe235..000000000 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiAction.java +++ /dev/null @@ -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 -} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiException.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiException.java deleted file mode 100644 index 7153f0d23..000000000 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiException.java +++ /dev/null @@ -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); - } -} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiRequest.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiRequest.java deleted file mode 100644 index 3226c887b..000000000 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiRequest.java +++ /dev/null @@ -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; - } -} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiResponse.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiResponse.java deleted file mode 100644 index d0606211e..000000000 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiResponse.java +++ /dev/null @@ -1,78 +0,0 @@ -package com.alibaba.arthas.channel.server.api; - -/** - * Http Api exception - * @author gongdewei 2020-03-19 - */ -public class ApiResponse { - 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 setAgentId(String agentId) { - this.agentId = agentId; - return this; - } - - public String getRequestId() { - return requestId; - } - - public ApiResponse setRequestId(String requestId) { - this.requestId = requestId; - return this; - } - - public ApiStatus getStatus() { - return status; - } - - public ApiResponse setStatus(ApiStatus status) { - this.status = status; - return this; - } - - public String getMessage() { - return message; - } - - public ApiResponse setMessage(String message) { - this.message = message; - return this; - } - - public String getSessionId() { - return sessionId; - } - - public ApiResponse setSessionId(String sessionId) { - this.sessionId = sessionId; - return this; - } - -// public String getConsumerId() { -// return consumerId; -// } -// -// public ApiResponse setConsumerId(String consumerId) { -// this.consumerId = consumerId; -// return this; -// } - - public T getResult() { - return result; - } - - public ApiResponse setResult(T result) { - this.result = result; - return this; - } -} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiStatus.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiStatus.java deleted file mode 100644 index 43e2855c4..000000000 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/api/ApiStatus.java +++ /dev/null @@ -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 -} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java index 6f489fe96..445ec12e8 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java @@ -90,7 +90,7 @@ public class MessageExchangeServiceImpl implements MessageExchangeService { public Mono 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)); } diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/ApiActionDelegateServiceImpl.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/ApiActionDelegateServiceImpl.java index 15bfe56a9..20a2e48a6 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/ApiActionDelegateServiceImpl.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/ApiActionDelegateServiceImpl.java @@ -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()); } diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/utils/PbJsonUtils.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/utils/PbJsonUtils.java new file mode 100644 index 000000000..8d9dd6a9e --- /dev/null +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/utils/PbJsonUtils.java @@ -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(); + } + +} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java index e215ceb52..67e91cedd 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java @@ -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 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 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 optionalAgentVO = agentManageService.findAgentById(agentId).block(); if (!optionalAgentVO.isPresent()) { diff --git a/channel/channel-server/src/test/java/com/alibaba/arthas/channel/server/json/PbJsonTest.java b/channel/channel-server/src/test/java/com/alibaba/arthas/channel/server/json/PbJsonTest.java new file mode 100644 index 000000000..8f35c9de9 --- /dev/null +++ b/channel/channel-server/src/test/java/com/alibaba/arthas/channel/server/json/PbJsonTest.java @@ -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); + } + + +}