github/arthas
4
0
Fork 0
mirror of https://github.com/alibaba/arthas.git synced 2024-04-21 10:21:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

HTTP API support one-time command execution and add HTTP API docs (#1408)

Browse Source
This commit is contained in:
gongdewei
2020-08-07 16:09:19 +08:00
committed by GitHub
parent c7dd99e8cd
commit 7a01f23a73
7 changed files with 1452 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/com/taobao/arthas/core/shell/session/impl/SessionManagerImpl.java
+7 -8
View File
@@ -29,7 +29,8 @@ public class SessionManagerImpl implements SessionManager {
private final InternalCommandManager commandManager;
private final Instrumentation instrumentation;
private final JobController jobController;
private final long timeoutMillis;
private final long sessionTimeoutMillis;
private final int consumerTimeoutMillis;
private final long reaperInterval;
private final Map<String, Session> sessions;
private final long pid;
@@ -42,7 +43,8 @@ public class SessionManagerImpl implements SessionManager {
this.commandManager = commandManager;
this.jobController = jobController;
this.sessions = new ConcurrentHashMap<String, Session>();
this.timeoutMillis = options.getSessionTimeout();
this.sessionTimeoutMillis = options.getSessionTimeout();
this.consumerTimeoutMillis = 5 * 60 * 1000; // 5 minutes
this.reaperInterval = options.getReaperInterval();
this.instrumentation = options.getInstrumentation();
this.pid = options.getPid();
@@ -62,9 +64,6 @@ public class SessionManagerImpl implements SessionManager {
String sessionId = UUID.randomUUID().toString();
session.put(Session.ID, sessionId);
//Result Distributor
session.setResultDistributor(new SharingResultDistributorImpl(session));
sessions.put(sessionId, session);
return session;
}
@@ -135,7 +134,7 @@ public class SessionManagerImpl implements SessionManager {
// do not close if there is still job running,
// e.g. trace command might wait for a long time before condition is met
//TODO check background job size
if (now - session.getLastAccessTime() > timeoutMillis && session.getForegroundJob() == null) {
if (now - session.getLastAccessTime() > sessionTimeoutMillis && session.getForegroundJob() == null) {
toClose.add(session);
}
evictConsumers(session);
@@ -146,7 +145,7 @@ public class SessionManagerImpl implements SessionManager {
if (job != null) {
job.interrupt();
}
long timeOutInMinutes = timeoutMillis / 1000 / 60;
long timeOutInMinutes = sessionTimeoutMillis / 1000 / 60;
String reason = "session is inactive for " + timeOutInMinutes + " min(s).";
session.getResultDistributor().appendResult(new MessageModel(reason));
this.removeSession(session.getSessionId());
@@ -166,7 +165,7 @@ public class SessionManagerImpl implements SessionManager {
long now = System.currentTimeMillis();
for (ResultConsumer consumer : consumers) {
long inactiveTime = now - consumer.getLastAccessTime();
if (inactiveTime > 30000) {
if (inactiveTime > consumerTimeoutMillis) {
//inactive duration must be large than pollTimeLimit
logger.info("Removing inactive consumer from session, sessionId: {}, consumerId: {}, inactive duration: {}",
session.getSessionId(), consumer.getConsumerId(), inactiveTime);
core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/api/ApiRequest.java
+6 -16
View File
@@ -13,8 +13,7 @@ public class ApiRequest {
private String requestId;
private String sessionId;
private String consumerId;
private Integer timeout;
private Map<String, Object> options;
private Integer execTimeout;
@Override
public String toString() {
@@ -24,8 +23,7 @@ public class ApiRequest {
", requestId='" + requestId + '\'' +
", sessionId='" + sessionId + '\'' +
", consumerId='" + consumerId + '\'' +
", timeout=" + timeout +
", options=" + options +
", execTimeout=" + execTimeout +
'}';
}
@@ -45,14 +43,6 @@ public class ApiRequest {
this.command = command;
}
public Map<String, Object> getOptions() {
return options;
}
public void setOptions(Map<String, Object> options) {
this.options = options;
}
public String getRequestId() {
return requestId;
}
@@ -77,11 +67,11 @@ public class ApiRequest {
this.consumerId = consumerId;
}
public Integer getTimeout() {
return timeout;
public Integer getExecTimeout() {
return execTimeout;
}
public void setTimeout(Integer timeout) {
this.timeout = timeout;
public void setExecTimeout(Integer execTimeout) {
this.execTimeout = execTimeout;
}
}
core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/api/HttpApiHandler.java
+96 -72
View File
@@ -8,8 +8,10 @@ import com.taobao.arthas.core.command.model.*;
import com.taobao.arthas.core.distribution.PackingResultDistributor;
import com.taobao.arthas.core.distribution.ResultConsumer;
import com.taobao.arthas.core.distribution.ResultDistributor;
import com.taobao.arthas.core.distribution.SharingResultDistributor;
import com.taobao.arthas.core.distribution.impl.PackingResultDistributorImpl;
import com.taobao.arthas.core.distribution.impl.ResultConsumerImpl;
import com.taobao.arthas.core.distribution.impl.SharingResultDistributorImpl;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.cli.CliTokens;
import com.taobao.arthas.core.shell.cli.Completion;
@@ -53,7 +55,6 @@ public class HttpApiHandler {
private static final Logger logger = LoggerFactory.getLogger(HttpApiHandler.class);
public static final int DEFAULT_EXEC_TIMEOUT = 30000;
private final SessionManager sessionManager;
private final AtomicInteger requestIdGenerator = new AtomicInteger(0);
private static HttpApiHandler instance;
private final InternalCommandManager commandManager;
private final JobController jobController;
@@ -84,13 +85,13 @@ public class HttpApiHandler {
ApiResponse result;
String requestBody = null;
String requestId = "req_" + requestIdGenerator.addAndGet(1);
String requestId = null;
try {
HttpMethod method = request.method();
if (HttpMethod.POST.equals(method)) {
requestBody = getBody(request);
ApiRequest apiRequest = parseRequest(requestBody);
apiRequest.setRequestId(requestId);
requestId = apiRequest.getRequestId();
result = processRequest(apiRequest);
} else {
result = createResponse(ApiState.REFUSED, "Unsupported http method: " + method.name());
@@ -210,15 +211,20 @@ public class HttpApiHandler {
}
//required session
Session session = null;
boolean allowNullSession = ApiAction.EXEC.equals(action);
String sessionId = apiRequest.getSessionId();
if (StringUtils.isBlank(sessionId)) {
throw new ApiException("'sessionId' is required");
if (!allowNullSession) {
throw new ApiException("'sessionId' is required");
}
} else {
session = sessionManager.getSession(sessionId);
if (session == null) {
throw new ApiException("session not found: " + sessionId);
}
sessionManager.updateAccessTime(session);
}
Session session = sessionManager.getSession(sessionId);
if (session == null) {
throw new ApiException("session not found: " + sessionId);
}
sessionManager.updateAccessTime(session);
//dispatch requests
ApiResponse response = dispatchRequest(action, apiRequest, session);
@@ -266,11 +272,14 @@ public class HttpApiHandler {
Session session = sessionManager.createSession();
if (session != null) {
//Result Distributor
SharingResultDistributorImpl resultDistributor = new SharingResultDistributorImpl(session);
//create consumer
ResultConsumer resultConsumer = new ResultConsumerImpl();
session.getResultDistributor().addConsumer(resultConsumer);
resultDistributor.addConsumer(resultConsumer);
session.setResultDistributor(resultDistributor);
session.getResultDistributor().appendResult(new MessageModel("Welcome to arthas!"));
resultDistributor.appendResult(new MessageModel("Welcome to arthas!"));
//welcome message
WelcomeModel welcomeModel = new WelcomeModel();
@@ -279,7 +288,7 @@ public class HttpApiHandler {
welcomeModel.setTutorials(ArthasBanner.tutorials());
welcomeModel.setPid(PidUtils.currentPid());
welcomeModel.setTime(DateUtils.getCurrentDate());
session.getResultDistributor().appendResult(welcomeModel);
resultDistributor.appendResult(welcomeModel);
//allow input
updateSessionInputStatus(session, InputStatus.ALLOW_INPUT);
@@ -300,7 +309,10 @@ public class HttpApiHandler {
* @param inputStatus
*/
private void updateSessionInputStatus(Session session, InputStatus inputStatus) {
session.getResultDistributor().appendResult(new InputStatusModel(inputStatus));
SharingResultDistributor resultDistributor = session.getResultDistributor();
if (resultDistributor != null) {
resultDistributor.appendResult(new InputStatusModel(inputStatus));
}
}
private ApiResponse processJoinSessionRequest(ApiRequest apiRequest, Session session) {
@@ -347,78 +359,90 @@ public class HttpApiHandler {
* @return
*/
private ApiResponse processExecRequest(ApiRequest apiRequest, Session session) {
String commandLine = apiRequest.getCommand();
Map<String, Object> body = new TreeMap<String, Object>();
body.put("command", commandLine);
ApiResponse response = new ApiResponse();
response.setSessionId(session.getSessionId())
.setBody(body);
if (!session.tryLock()) {
response.setState(ApiState.REFUSED)
.setMessage("Another command is executing.");
return response;
boolean oneTimeAccess = false;
if (session == null) {
oneTimeAccess = true;
session = sessionManager.createSession();
}
int lock = session.getLock();
PackingResultDistributor packingResultDistributor = null;
Job job = null;
try {
Job foregroundJob = session.getForegroundJob();
if (foregroundJob != null) {
String commandLine = apiRequest.getCommand();
Map<String, Object> body = new TreeMap<String, Object>();
body.put("command", commandLine);
ApiResponse response = new ApiResponse();
response.setSessionId(session.getSessionId())
.setBody(body);
if (!session.tryLock()) {
response.setState(ApiState.REFUSED)
.setMessage("Another job is running.");
logger.info("Another job is running, jobId: {}", foregroundJob.id());
.setMessage("Another command is executing.");
return response;
}
//distribute result message both to origin session channel and request channel by CompositeResultDistributor
packingResultDistributor = new PackingResultDistributorImpl(session);
//ResultDistributor resultDistributor = new CompositeResultDistributorImpl(packingResultDistributor, session.getResultDistributor());
job = this.createJob(commandLine, session, packingResultDistributor);
session.setForegroundJob(job);
updateSessionInputStatus(session, InputStatus.ALLOW_INTERRUPT);
int lock = session.getLock();
PackingResultDistributor packingResultDistributor = null;
Job job = null;
try {
Job foregroundJob = session.getForegroundJob();
if (foregroundJob != null) {
response.setState(ApiState.REFUSED)
.setMessage("Another job is running.");
logger.info("Another job is running, jobId: {}", foregroundJob.id());
return response;
}
job.run();
packingResultDistributor = new PackingResultDistributorImpl(session);
//distribute result message both to origin session channel and request channel by CompositeResultDistributor
//ResultDistributor resultDistributor = new CompositeResultDistributorImpl(packingResultDistributor, session.getResultDistributor());
job = this.createJob(commandLine, session, packingResultDistributor);
session.setForegroundJob(job);
updateSessionInputStatus(session, InputStatus.ALLOW_INTERRUPT);
} catch (Throwable e) {
logger.error("Exec command failed:" + e.getMessage() + ", command:" + commandLine, e);
response.setState(ApiState.FAILED).setMessage("Exec command failed:" + e.getMessage());
job.run();
} catch (Throwable e) {
logger.error("Exec command failed:" + e.getMessage() + ", command:" + commandLine, e);
response.setState(ApiState.FAILED).setMessage("Exec command failed:" + e.getMessage());
return response;
} finally {
if (session.getLock() == lock) {
session.unLock();
}
}
//wait for job completed or timeout
Integer timeout = apiRequest.getExecTimeout();
if (timeout == null || timeout <= 0) {
timeout = DEFAULT_EXEC_TIMEOUT;
}
boolean timeExpired = !waitForJob(job, timeout);
if (timeExpired) {
logger.warn("Job is exceeded time limit, force interrupt it, jobId: {}", job.id());
job.interrupt();
response.setState(ApiState.INTERRUPTED).setMessage("The job is exceeded time limit, force interrupt");
} else {
response.setState(ApiState.SUCCEEDED);
}
//packing results
body.put("jobId", job.id());
body.put("jobStatus", job.status());
body.put("timeExpired", timeExpired);
if (timeExpired) {
body.put("timeout", timeout);
}
body.put("results", packingResultDistributor.getResults());
response.setSessionId(session.getSessionId())
//.setConsumerId(consumerId)
.setBody(body);
return response;
} finally {
if (session.getLock() == lock) {
session.unLock();
if (oneTimeAccess) {
sessionManager.removeSession(session.getSessionId());
}
}
//wait for job completed or timeout
Integer timeout = apiRequest.getTimeout();
if (timeout == null || timeout <= 0) {
timeout = DEFAULT_EXEC_TIMEOUT;
}
boolean timeExpired = !waitForJob(job, timeout);
if (timeExpired) {
logger.warn("Job is exceeded time limit, force interrupt it, jobId: {}", job.id());
job.interrupt();
response.setState(ApiState.INTERRUPTED).setMessage("The job is exceeded time limit, force interrupt");
} else {
response.setState(ApiState.SUCCEEDED);
}
//packing results
body.put("jobId", job.id());
body.put("jobStatus", job.status());
body.put("timeExpired", timeExpired);
if (timeExpired) {
body.put("timeout", timeout);
}
body.put("results", packingResultDistributor.getResults());
response.setSessionId(session.getSessionId())
//.setConsumerId(consumerId)
.setBody(body);
return response;
}
/**