Add http api basic function (#1203)

This commit is contained in:
hengyunabc
2020-06-17 11:07:00 +08:00
committed by GitHub
78 changed files with 3878 additions and 236 deletions
@@ -5,6 +5,7 @@ import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
import com.taobao.arthas.core.command.model.CatModel;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.taobao.arthas.core.shell.cli.Completion;
@@ -24,6 +25,8 @@ public class CatCommand extends AnnotatedCommand {
private static final Logger logger = LoggerFactory.getLogger(CatCommand.class);
private List<String> files;
private String encoding;
private Integer sizeLimit = 128 * 1024;
private int maxSizeLimit = 8 * 1024 * 1024;
@Argument(argName = "files", index = 0)
@Description("files")
@@ -37,37 +40,43 @@ public class CatCommand extends AnnotatedCommand {
this.encoding = encoding;
}
@Option(shortName = "M", longName = "sizeLimit")
@Description("Upper size limit in bytes for the result (128 * 1024 by default, the maximum value is 8 * 1024 * 1024)")
public void setSizeLimit(Integer sizeLimit) {
this.sizeLimit = sizeLimit;
}
@Override
public void process(CommandProcess process) {
if (!verifyOptions(process)) {
return;
}
for (String file : files) {
File f = new File(file);
if (!f.exists()) {
process.write("cat " + file + ": No such file or directory\n");
process.end();
process.end(-1, "cat " + file + ": No such file or directory");
return;
}
if (f.isDirectory()) {
process.write("cat " + file + ": Is a directory\n");
process.end();
process.end(-1, "cat " + file + ": Is a directory");
return;
}
}
for (String file : files) {
File f = new File(file);
if (f.length() > 1024 * 1024 * 8) {
process.write("cat " + file + ": Is to large, size: " + f.length() + '\n');
process.end();
if (f.length() > sizeLimit) {
process.end(-1, "cat " + file + ": Is too large, size: " + f.length());
return;
}
try {
String fileToString = FileUtils.readFileToString(f,
encoding == null ? Charset.defaultCharset() : Charset.forName(encoding));
process.write(fileToString);
encoding == null ? Charset.defaultCharset() : Charset.forName(encoding));
process.appendResult(new CatModel(file, fileToString));
} catch (IOException e) {
logger.error("cat read file error. name: " + file, e);
process.write("cat read file error: " + e.getMessage() + '\n');
process.end(1);
process.end(1, "cat read file error: " + e.getMessage());
return;
}
}
@@ -75,6 +84,22 @@ public class CatCommand extends AnnotatedCommand {
process.end();
}
private boolean verifyOptions(CommandProcess process) {
if (sizeLimit > maxSizeLimit) {
process.end(-1, "sizeLimit cannot be large than: " + maxSizeLimit);
return false;
}
//目前不支持过滤,限制http请求执行的文件大小
int maxSizeLimitOfHttp = 128 * 1024;
boolean isHttpApiRequest = !process.session().isTty();
if (isHttpApiRequest && sizeLimit > maxSizeLimitOfHttp) {
process.end(-1, "When executing commands with http, sizeLimit cannot be large than: " + maxSizeLimitOfHttp);
return false;
}
return true;
}
@Override
public void complete(Completion completion) {
if (!CompletionUtils.completeFilePath(completion)) {
@@ -1,6 +1,8 @@
package com.taobao.arthas.core.command.basic1000;
import com.taobao.arthas.core.command.Constants;
import com.taobao.arthas.core.command.model.EchoModel;
import com.taobao.arthas.core.command.model.MessageModel;
import com.taobao.arthas.core.shell.command.AnnotatedCommand;
import com.taobao.arthas.core.shell.command.CommandProcess;
import com.taobao.middleware.cli.annotations.Argument;
@@ -30,8 +32,7 @@ public class EchoCommand extends AnnotatedCommand {
@Override
public void process(CommandProcess process) {
if (message != null) {
process.write(message);
process.write("\n");
process.appendResult(new EchoModel(message));
}
process.end();
@@ -1,5 +1,9 @@
package com.taobao.arthas.core.command.basic1000;
import com.taobao.arthas.core.command.model.ArgumentVO;
import com.taobao.arthas.core.command.model.CommandOptionVO;
import com.taobao.arthas.core.command.model.CommandVO;
import com.taobao.arthas.core.command.model.HelpModel;
import com.taobao.arthas.core.shell.cli.Completion;
import com.taobao.arthas.core.shell.cli.CompletionUtils;
import com.taobao.arthas.core.shell.command.AnnotatedCommand;
@@ -9,24 +13,15 @@ import com.taobao.arthas.core.shell.command.CommandResolver;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.util.usage.StyledUsageFormatter;
import com.taobao.middleware.cli.CLI;
import com.taobao.middleware.cli.Option;
import com.taobao.middleware.cli.annotations.Argument;
import com.taobao.middleware.cli.annotations.Description;
import com.taobao.middleware.cli.annotations.Name;
import com.taobao.middleware.cli.annotations.Summary;
import com.taobao.text.Color;
import com.taobao.text.Decoration;
import com.taobao.text.Style;
import com.taobao.text.ui.Element;
import com.taobao.text.ui.LabelElement;
import com.taobao.text.ui.TableElement;
import com.taobao.text.util.RenderUtil;
import java.util.ArrayList;
import java.util.List;
import static com.taobao.text.ui.Element.label;
import static com.taobao.text.ui.Element.row;
/**
* @author vlinux on 14/10/26.
*/
@@ -47,16 +42,91 @@ public class HelpCommand extends AnnotatedCommand {
public void process(CommandProcess process) {
List<Command> commands = allCommands(process.session());
Command targetCmd = findCommand(commands);
String message;
if (targetCmd == null) {
message = RenderUtil.render(mainHelp(commands), process.width());
process.appendResult(createHelpModel(commands));
} else {
message = commandHelp(targetCmd, process.width());
process.appendResult(createHelpDetailModel(targetCmd));
}
process.write(message);
process.end();
}
public HelpModel createHelpDetailModel(Command targetCmd) {
return new HelpModel(createCommandVO(targetCmd, true));
}
private HelpModel createHelpModel(List<Command> commands) {
HelpModel helpModel = new HelpModel();
for (Command command : commands) {
if(command.cli() == null || command.cli().isHidden()){
continue;
}
helpModel.addCommandVO(createCommandVO(command, false));
}
return helpModel;
}
private CommandVO createCommandVO(Command command, boolean withDetail) {
CLI cli = command.cli();
CommandVO commandVO = new CommandVO();
commandVO.setName(command.name());
if (cli!=null){
commandVO.setSummary(cli.getSummary());
if (withDetail){
commandVO.setCli(cli);
StyledUsageFormatter usageFormatter = new StyledUsageFormatter(null);
String usageLine = usageFormatter.computeUsageLine(null, cli);
commandVO.setUsage(usageLine);
commandVO.setDescription(cli.getDescription());
//以线程安全的方式遍历options
List<Option> options = cli.getOptions();
for (int i = 0; i < options.size(); i++) {
Option option = options.get(i);
if (option.isHidden()){
continue;
}
commandVO.addOption(createOptionVO(option));
}
//arguments
List<com.taobao.middleware.cli.Argument> arguments = cli.getArguments();
for (int i = 0; i < arguments.size(); i++) {
com.taobao.middleware.cli.Argument argument = arguments.get(i);
if (argument.isHidden()){
continue;
}
commandVO.addArgument(createArgumentVO(argument));
}
}
}
return commandVO;
}
private ArgumentVO createArgumentVO(com.taobao.middleware.cli.Argument argument) {
ArgumentVO argumentVO = new ArgumentVO();
argumentVO.setArgName(argument.getArgName());
argumentVO.setMultiValued(argument.isMultiValued());
argumentVO.setRequired(argument.isRequired());
return argumentVO;
}
private CommandOptionVO createOptionVO(Option option) {
CommandOptionVO optionVO = new CommandOptionVO();
if (!isEmptyName(option.getLongName())) {
optionVO.setLongName(option.getLongName());
}
if (!isEmptyName(option.getShortName())) {
optionVO.setShortName(option.getShortName());
}
optionVO.setDescription(option.getDescription());
optionVO.setAcceptValue(option.acceptValue());
return optionVO;
}
private boolean isEmptyName(String name) {
return name == null || name.equals(Option.NO_NAME);
}
@Override
public void complete(Completion completion) {
List<Command> commands = allCommands(completion.session());
@@ -72,24 +142,6 @@ public class HelpCommand extends AnnotatedCommand {
CompletionUtils.complete(completion, names);
}
private static String commandHelp(Command command, int width) {
return StyledUsageFormatter.styledUsage(command.cli(), width);
}
private static Element mainHelp(List<Command> commands) {
TableElement table = new TableElement().leftCellPadding(1).rightCellPadding(1);
table.row(new LabelElement("NAME").style(Style.style(Decoration.bold)), new LabelElement("DESCRIPTION"));
for (Command command : commands) {
CLI cli = command.cli();
// com.taobao.arthas.core.shell.impl.BuiltinCommandResolver doesn't have CLI instance
if (cli == null || cli.isHidden()) {
continue;
}
table.add(row().add(label(cli.getName()).style(Style.style(Color.green))).add(label(cli.getSummary())));
}
return table;
}
private List<Command> allCommands(Session session) {
List<CommandResolver> commandResolvers = session.getCommandResolvers();
List<Command> commands = new ArrayList<Command>();
@@ -4,8 +4,11 @@ import java.util.ArrayList;
import java.util.List;
import com.taobao.arthas.core.command.Constants;
import com.taobao.arthas.core.command.model.HistoryModel;
import com.taobao.arthas.core.shell.command.AnnotatedCommand;
import com.taobao.arthas.core.shell.command.CommandProcess;
import com.taobao.arthas.core.shell.history.HistoryManager;
import com.taobao.arthas.core.shell.history.impl.HistoryManagerImpl;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.term.impl.TermImpl;
import com.taobao.middleware.cli.annotations.Argument;
@@ -69,6 +72,14 @@ public class HistoryCommand extends AnnotatedCommand {
process.write(sb.toString());
}
} else {
if (clear) {
HistoryManagerImpl.getInstance().clearHistory();
} else {
//http api
List<String> history = HistoryManagerImpl.getInstance().getHistory();
process.appendResult(new HistoryModel(new ArrayList<String>(history)));
}
}
process.end();
@@ -1,9 +1,11 @@
package com.taobao.arthas.core.command.basic1000;
import com.taobao.arthas.core.advisor.Enhancer;
import com.taobao.arthas.core.server.ArthasBootstrap;
import com.taobao.arthas.core.shell.ShellServer;
import com.taobao.arthas.core.shell.command.AnnotatedCommand;
import com.taobao.arthas.core.shell.command.CommandProcess;
import com.taobao.arthas.core.shell.session.SessionManager;
import com.taobao.arthas.core.util.affect.EnhancerAffect;
import com.taobao.arthas.core.util.matcher.WildcardMatcher;
import com.taobao.middleware.cli.annotations.Hidden;
@@ -39,8 +41,15 @@ public class ShutdownCommand extends AnnotatedCommand {
// ignore
} finally {
process.end();
ShellServer server = process.session().getServer();
server.close();
ShellServer server = ArthasBootstrap.getInstance().getShellServer();
if (server != null) {
server.close();
}
SessionManager sessionManager = ArthasBootstrap.getInstance().getSessionManager();
if (sessionManager != null){
sessionManager.close();
}
}
}
}
@@ -1,6 +1,7 @@
package com.taobao.arthas.core.command.basic1000;
import com.taobao.arthas.core.command.model.VersionModel;
import com.taobao.arthas.core.shell.command.AnnotatedCommand;
import com.taobao.arthas.core.shell.command.CommandProcess;
import com.taobao.arthas.core.util.ArthasBanner;
@@ -15,8 +16,13 @@ import com.taobao.middleware.cli.annotations.Summary;
@Name("version")
@Summary("Display Arthas version")
public class VersionCommand extends AnnotatedCommand {
@Override
public void process(CommandProcess process) {
process.write(ArthasBanner.version()).write("\n").end();
VersionModel result = new VersionModel();
result.setVersion(ArthasBanner.version());
process.appendResult(result);
process.end();
}
}
@@ -0,0 +1,43 @@
package com.taobao.arthas.core.command.model;
/**
* @author gongdewei 2020/4/3
*/
public class ArgumentVO {
private String argName;
private boolean required;
private boolean multiValued;
public ArgumentVO() {
}
public ArgumentVO(String argName, boolean required, boolean multiValued) {
this.argName = argName;
this.required = required;
this.multiValued = multiValued;
}
public String getArgName() {
return argName;
}
public void setArgName(String argName) {
this.argName = argName;
}
public boolean isRequired() {
return required;
}
public void setRequired(boolean required) {
this.required = required;
}
public boolean isMultiValued() {
return multiValued;
}
public void setMultiValued(boolean multiValued) {
this.multiValued = multiValued;
}
}
@@ -0,0 +1,49 @@
package com.taobao.arthas.core.command.model;
/**
* Result model for CatCommand
* @author gongdewei 2020/5/11
*/
public class CatModel extends ResultModel implements Countable {
private String file;
private String content;
public CatModel() {
}
public CatModel(String file, String content) {
this.file = file;
this.content = content;
}
@Override
public String getType() {
return "cat";
}
public String getFile() {
return file;
}
public void setFile(String file) {
this.file = file;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
@Override
public int size() {
if (content != null) {
//粗略计算行数作为item size
return content.length()/100 + 1;
}
return 0;
}
}
@@ -0,0 +1,62 @@
package com.taobao.arthas.core.command.model;
/**
* @author gongdewei 2020/4/3
*/
public class CommandOptionVO {
/**
* the option long name.
*/
private String longName;
/**
* the option short name.
*/
private String shortName;
/**
* The option description.
*/
private String description;
/**
* whether or not the option receives a single value or multiple values.
*/
private boolean acceptValue;
public CommandOptionVO() {
}
public String getLongName() {
return longName;
}
public void setLongName(String longName) {
this.longName = longName;
}
public String getShortName() {
return shortName;
}
public void setShortName(String shortName) {
this.shortName = shortName;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public boolean isAcceptValue() {
return acceptValue;
}
public void setAcceptValue(boolean acceptValue) {
this.acceptValue = acceptValue;
}
}
@@ -0,0 +1,54 @@
package com.taobao.arthas.core.command.model;
import com.taobao.arthas.core.shell.term.impl.http.api.ApiState;
/**
* Command async exec process result, not the command exec result
* @author gongdewei 2020/4/2
*/
public class CommandRequestModel extends ResultModel {
private ApiState state;
private String command;
private String message;
public CommandRequestModel(String command, ApiState state) {
this.command = command;
this.state = state;
}
public CommandRequestModel(String command, ApiState state, String message) {
this.state = state;
this.command = command;
this.message = message;
}
public String getCommand() {
return command;
}
public void setCommand(String command) {
this.command = command;
}
public ApiState getState() {
return state;
}
public void setState(ApiState state) {
this.state = state;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
@Override
public String getType() {
return "command";
}
}
@@ -0,0 +1,94 @@
package com.taobao.arthas.core.command.model;
import com.taobao.middleware.cli.CLI;
import java.util.ArrayList;
import java.util.List;
/**
* @author gongdewei 2020/4/3
*/
public class CommandVO {
//TODO remove cli
private transient CLI cli;
private String name;
private String description;
private String usage;
private String summary;
private List<CommandOptionVO> options = new ArrayList<CommandOptionVO>();
private List<ArgumentVO> arguments = new ArrayList<ArgumentVO>();
public CommandVO() {
}
public CommandVO(String name, String description) {
this.name = name;
this.description = description;
}
public CommandVO addOption(CommandOptionVO optionVO){
this.options.add(optionVO);
return this;
}
public CommandVO addArgument(ArgumentVO argumentVO){
this.arguments.add(argumentVO);
return this;
}
public CLI cli() {
return cli;
}
public void setCli(CLI cli) {
this.cli = cli;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public String getUsage() {
return usage;
}
public void setUsage(String usage) {
this.usage = usage;
}
public String getSummary() {
return summary;
}
public void setSummary(String summary) {
this.summary = summary;
}
public List<CommandOptionVO> getOptions() {
return options;
}
public void setOptions(List<CommandOptionVO> options) {
this.options = options;
}
public List<ArgumentVO> getArguments() {
return arguments;
}
public void setArguments(List<ArgumentVO> arguments) {
this.arguments = arguments;
}
}
@@ -0,0 +1,15 @@
package com.taobao.arthas.core.command.model;
/**
* Item countable for ResultModel
* @author gongdewei 2020/6/8
*/
public interface Countable {
/**
* Get item size of this result model, the value of size is greater than or equal to 1
* @return item size of this result model
*/
int size();
}
@@ -0,0 +1,29 @@
package com.taobao.arthas.core.command.model;
/**
* @author gongdewei 2020/5/11
*/
public class EchoModel extends ResultModel {
private String content;
public EchoModel() {
}
public EchoModel(String content) {
this.content = content;
}
@Override
public String getType() {
return "echo";
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}
@@ -0,0 +1,51 @@
package com.taobao.arthas.core.command.model;
import java.util.ArrayList;
import java.util.List;
/**
* @author gongdewei 2020/4/3
*/
public class HelpModel extends ResultModel {
//list
private List<CommandVO> commands;
//details
private CommandVO detailCommand;
public HelpModel() {
}
public HelpModel(List<CommandVO> commands) {
this.commands = commands;
}
public HelpModel(CommandVO command) {
this.detailCommand = command;
}
public void addCommandVO(CommandVO commandVO){
if (commands == null) {
commands = new ArrayList<CommandVO>();
}
this.commands.add(commandVO);
}
public List<CommandVO> getCommands() {
return commands;
}
public void setCommands(List<CommandVO> commands) {
this.commands = commands;
}
public CommandVO getDetailCommand() {
return detailCommand;
}
@Override
public String getType() {
return "help";
}
}
@@ -0,0 +1,27 @@
package com.taobao.arthas.core.command.model;
import java.util.List;
/**
* @author gongdewei 2020/4/8
*/
public class HistoryModel extends ResultModel {
private List<String> history;
public HistoryModel() {
}
public HistoryModel(List<String> history) {
this.history = history;
}
public List<String> getHistory() {
return history;
}
@Override
public String getType() {
return "history";
}
}
@@ -0,0 +1,22 @@
package com.taobao.arthas.core.command.model;
/**
* Command input status for webui
* @author gongdewei 2020/4/14
*/
public enum InputStatus {
/**
* Allow input new commands
*/
ALLOW_INPUT,
/**
* Allow interrupt running job
*/
ALLOW_INTERRUPT,
/**
* Disable input and interrupt
*/
DISABLED
}
@@ -0,0 +1,28 @@
package com.taobao.arthas.core.command.model;
/**
* Input status for webui
* @author gongdewei 2020/4/14
*/
public class InputStatusModel extends ResultModel {
private InputStatus inputStatus;
public InputStatusModel(InputStatus inputStatus) {
this.inputStatus = inputStatus;
}
public InputStatus getInputStatus() {
return inputStatus;
}
public void setInputStatus(InputStatus inputStatus) {
this.inputStatus = inputStatus;
}
@Override
public String getType() {
return "input_status";
}
}
@@ -0,0 +1,24 @@
package com.taobao.arthas.core.command.model;
/**
* @author gongdewei 2020/4/2
*/
public class MessageModel extends ResultModel {
private String message;
public MessageModel() {
}
public MessageModel(String message) {
this.message = message;
}
public String getMessage() {
return message;
}
@Override
public String getType() {
return "message";
}
}
@@ -0,0 +1,27 @@
package com.taobao.arthas.core.command.model;
/**
* Command execute result
*
* @author gongdewei 2020-03-26
*/
public abstract class ResultModel {
private int jobId;
/**
* Command type (name)
*
* @return
*/
public abstract String getType();
public int getJobId() {
return jobId;
}
public void setJobId(int jobId) {
this.jobId = jobId;
}
}
@@ -0,0 +1,52 @@
package com.taobao.arthas.core.command.model;
public class StatusModel extends ResultModel {
private int statusCode;
private String message;
public StatusModel() {
}
public StatusModel(int statusCode) {
this.statusCode = statusCode;
}
public StatusModel(int statusCode, String message) {
this.statusCode = statusCode;
this.message = message;
}
public int getStatusCode() {
return statusCode;
}
public StatusModel setStatusCode(int statusCode) {
this.statusCode = statusCode;
return this;
}
public String getMessage() {
return message;
}
public StatusModel setMessage(String message) {
this.message = message;
return this;
}
public StatusModel setStatus(int statusCode, String message) {
this.statusCode = statusCode;
this.message = message;
return this;
}
public StatusModel setStatus(int statusCode) {
return this.setStatus(statusCode, null);
}
@Override
public String getType() {
return "status";
}
}
@@ -0,0 +1,20 @@
package com.taobao.arthas.core.command.model;
public class VersionModel extends ResultModel {
private String version;
@Override
public String getType() {
return "version";
}
public String getVersion() {
return version;
}
public void setVersion(String version) {
this.version = version;
}
}
@@ -0,0 +1,61 @@
package com.taobao.arthas.core.command.model;
/**
* @author gongdewei 2020/4/20
*/
public class WelcomeModel extends ResultModel {
private String pid;
private String time;
private String version;
private String wiki;
private String tutorials;
public WelcomeModel() {
}
@Override
public String getType() {
return "welcome";
}
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public String getTime() {
return time;
}
public void setTime(String time) {
this.time = time;
}
public String getVersion() {
return version;
}
public void setVersion(String version) {
this.version = version;
}
public String getWiki() {
return wiki;
}
public void setWiki(String wiki) {
this.wiki = wiki;
}
public String getTutorials() {
return tutorials;
}
public void setTutorials(String tutorials) {
this.tutorials = tutorials;
}
}
@@ -0,0 +1,17 @@
package com.taobao.arthas.core.command.view;
import com.taobao.arthas.core.command.model.CatModel;
import com.taobao.arthas.core.shell.command.CommandProcess;
/**
* Result view for CatCommand
* @author gongdewei 2020/5/11
*/
public class CatView extends ResultView<CatModel> {
@Override
public void draw(CommandProcess process, CatModel result) {
process.write(result.getContent()).write("\n");
}
}
@@ -0,0 +1,14 @@
package com.taobao.arthas.core.command.view;
import com.taobao.arthas.core.command.model.EchoModel;
import com.taobao.arthas.core.shell.command.CommandProcess;
/**
* @author gongdewei 2020/5/11
*/
public class EchoView extends ResultView<EchoModel> {
@Override
public void draw(CommandProcess process, EchoModel result) {
process.write(result.getContent()).write("\n");
}
}
@@ -0,0 +1,48 @@
package com.taobao.arthas.core.command.view;
import com.taobao.arthas.core.command.model.CommandVO;
import com.taobao.arthas.core.command.model.HelpModel;
import com.taobao.arthas.core.shell.command.CommandProcess;
import com.taobao.arthas.core.util.usage.StyledUsageFormatter;
import com.taobao.middleware.cli.CLI;
import com.taobao.text.Color;
import com.taobao.text.Decoration;
import com.taobao.text.Style;
import com.taobao.text.ui.Element;
import com.taobao.text.ui.LabelElement;
import com.taobao.text.ui.TableElement;
import com.taobao.text.util.RenderUtil;
import java.util.List;
import static com.taobao.text.ui.Element.label;
import static com.taobao.text.ui.Element.row;
/**
* @author gongdewei 2020/4/3
*/
public class HelpView extends ResultView<HelpModel> {
@Override
public void draw(CommandProcess process, HelpModel result) {
if (result.getCommands() != null) {
String message = RenderUtil.render(mainHelp(result.getCommands()), process.width());
process.write(message);
} else if (result.getDetailCommand() != null) {
process.write(commandHelp(result.getDetailCommand().cli(), process.width()));
}
}
private static Element mainHelp(List<CommandVO> commands) {
TableElement table = new TableElement().leftCellPadding(1).rightCellPadding(1);
table.row(new LabelElement("NAME").style(Style.style(Decoration.bold)), new LabelElement("DESCRIPTION"));
for (CommandVO commandVO : commands) {
table.add(row().add(label(commandVO.getName()).style(Style.style(Color.green))).add(label(commandVO.getSummary())));
}
return table;
}
private static String commandHelp(CLI command, int width) {
return StyledUsageFormatter.styledUsage(command, width);
}
}
@@ -0,0 +1,14 @@
package com.taobao.arthas.core.command.view;
import com.taobao.arthas.core.command.model.MessageModel;
import com.taobao.arthas.core.shell.command.CommandProcess;
/**
* @author gongdewei 2020/4/2
*/
public class MessageView extends ResultView<MessageModel> {
@Override
public void draw(CommandProcess process, MessageModel result) {
writeln(process, result.getMessage());
}
}
@@ -0,0 +1,30 @@
package com.taobao.arthas.core.command.view;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.shell.command.CommandProcess;
/**
* Command result view for telnet term/tty.
* Note: Result view is a reusable and stateless instance
*
* @author gongdewei 2020/3/27
*/
public abstract class ResultView<T extends ResultModel> {
/**
* formatted printing data to term/tty
*
* @param process
*/
public abstract void draw(CommandProcess process, T result);
/**
* write str and append a new line
*
* @param process
* @param str
*/
protected void writeln(CommandProcess process, String str) {
process.write(str).write("\n");
}
}
@@ -0,0 +1,105 @@
package com.taobao.arthas.core.command.view;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.taobao.arthas.core.command.model.*;
import com.taobao.arthas.core.shell.command.CommandProcess;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Result view resolver for term
*
* @author gongdewei 2020/3/27
*/
public class ResultViewResolver {
private static final Logger logger = LoggerFactory.getLogger(ResultViewResolver.class);
// modelClass -> view
private Map<Class, ResultView> resultViewMap = new ConcurrentHashMap<Class, ResultView>();
private static ResultViewResolver viewResolver;
public static ResultViewResolver getInstance() {
if (viewResolver == null) {
synchronized (ResultViewResolver.class) {
viewResolver = new ResultViewResolver();
}
}
return viewResolver;
}
static {
getInstance().registerResultViews();
}
private void registerResultViews() {
try {
registerView(StatusView.class);
registerView(VersionView.class);
registerView(MessageView.class);
registerView(HelpView.class);
//registerView(HistoryView.class);
registerView(EchoView.class);
registerView(CatView.class);
} catch (Throwable e) {
logger.error("register result view failed", e);
}
}
private ResultViewResolver() {
}
public ResultView getResultView(ResultModel model) {
return resultViewMap.get(model.getClass());
}
public void registerView(Class modelClass, ResultView view) {
//TODO 检查model的type是否重复,减少复制代码带来的bug
this.resultViewMap.put(modelClass, view);
}
public void registerView(ResultView view) {
Class modelClass = getModelClass(view);
if (modelClass == null) {
throw new NullPointerException("model class is null");
}
this.registerView(modelClass, view);
}
public void registerView(Class<? extends ResultView> viewClass) {
ResultView view = null;
try {
view = viewClass.newInstance();
} catch (Throwable e) {
throw new RuntimeException("create view instance failure, viewClass:" + viewClass, e);
}
this.registerView(view);
}
/**
* Get model class of result view
*
* @return
*/
public static <V extends ResultView> Class getModelClass(V view) {
//类反射获取子类的draw方法第二个参数的ResultModel具体类型
Class<? extends ResultView> viewClass = view.getClass();
Method[] declaredMethods = viewClass.getDeclaredMethods();
for (int i = 0; i < declaredMethods.length; i++) {
Method method = declaredMethods[i];
if (method.getName().equals("draw")) {
Class<?>[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length == 2
&& parameterTypes[0] == CommandProcess.class
&& parameterTypes[1] != ResultModel.class
&& ResultModel.class.isAssignableFrom(parameterTypes[1])) {
return parameterTypes[1];
}
}
}
return null;
}
}
@@ -0,0 +1,18 @@
package com.taobao.arthas.core.command.view;
import com.taobao.arthas.core.command.model.StatusModel;
import com.taobao.arthas.core.shell.command.CommandProcess;
/**
* @author gongdewei 2020/3/27
*/
public class StatusView extends ResultView<StatusModel> {
@Override
public void draw(CommandProcess process, StatusModel result) {
if (result.getMessage() != null) {
writeln(process, result.getMessage());
}
}
}
@@ -0,0 +1,16 @@
package com.taobao.arthas.core.command.view;
import com.taobao.arthas.core.command.model.VersionModel;
import com.taobao.arthas.core.shell.command.CommandProcess;
/**
* @author gongdewei 2020/3/27
*/
public class VersionView extends ResultView<VersionModel> {
@Override
public void draw(CommandProcess process, VersionModel result) {
writeln(process, result.getVersion());
}
}
@@ -0,0 +1,12 @@
package com.taobao.arthas.core.distribution;
/**
* 复合结果分发器,将消息同时分发给其包含的多个真实分发器
* @author gongdewei 2020/4/30
*/
public interface CompositeResultDistributor extends ResultDistributor {
void addDistributor(ResultDistributor distributor);
void removeDistributor(ResultDistributor distributor);
}
@@ -0,0 +1,14 @@
package com.taobao.arthas.core.distribution;
/**
* 命令结果分发器选项
* @author gongdewei 2020/5/18
*/
public class DistributorOptions {
/**
* ResultConsumer的结果队列长度,用于控制内存缓存的命令结果数据量
*/
public static int resultQueueSize = 50;
}
@@ -0,0 +1,14 @@
package com.taobao.arthas.core.distribution;
import com.taobao.arthas.core.command.model.ResultModel;
import java.util.List;
public interface PackingResultDistributor extends ResultDistributor {
/**
* Get results of command
*/
List<ResultModel> getResults();
}
@@ -0,0 +1,43 @@
package com.taobao.arthas.core.distribution;
import com.taobao.arthas.core.command.model.ResultModel;
import java.util.List;
/**
* Command result consumer
* @author gongdewei 2020-03-26
*/
public interface ResultConsumer {
/**
* Append the phased result to queue
* @param result a phased result of the command
* @return true means distribution success, return false means discard data
*/
boolean appendResult(ResultModel result);
/**
* Retrieves and removes a pack of results from the head
* @return a pack of results
*/
List<ResultModel> pollResults();
long getLastAccessTime();
void close();
boolean isClosed();
boolean isPolling();
String getConsumerId();
void setConsumerId(String consumerId);
/**
* Retrieves the consumer's health status
* @return
*/
boolean isHealthy();
}
@@ -0,0 +1,94 @@
package com.taobao.arthas.core.distribution;
import com.alibaba.fastjson.JSON;
import com.taobao.arthas.core.command.model.Countable;
import com.taobao.arthas.core.command.model.ResultModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Array;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 命令结果模型辅助类
*
* @author gongdewei 2020/5/18
*/
public class ResultConsumerHelper {
private static final Logger logger = LoggerFactory.getLogger(ResultConsumerHelper.class);
private static ConcurrentHashMap<String, List<Field>> modelFieldMap = new ConcurrentHashMap<String, List<Field>>();
/**
* 估算命令执行结果的item数量,目的是提供一个度量值,作为Consumer分发时进行切片的参考依据,避免单次发送大量数据。
* 注意:此方法调用频繁,避免产生内存碎片
*
* @param model
* @return
*/
public static int getItemCount(ResultModel model) {
//如果实现Countable接口,则认为model自己统计元素数量
if (model instanceof Countable) {
return ((Countable) model).size();
}
//对于普通的Model,通过类反射统计容器类字段统计元素数量
//缓存Field对象,避免产生内存碎片
Class modelClass = model.getClass();
List<Field> fields = modelFieldMap.get(modelClass.getName());
if (fields == null) {
fields = new ArrayList<Field>();
Field[] declaredFields = modelClass.getDeclaredFields();
for (int i = 0; i < declaredFields.length; i++) {
Field field = declaredFields[i];
Class<?> fieldClass = field.getType();
//如果是List/Map/Array/Countable类型的字段,则缓存起来后面统计数量
if (Collection.class.isAssignableFrom(fieldClass)
|| Map.class.isAssignableFrom(fieldClass)
|| Countable.class.isAssignableFrom(fieldClass)
|| fieldClass.isArray()) {
field.setAccessible(true);
fields.add(field);
}
}
List<Field> old_fields = modelFieldMap.putIfAbsent(modelClass.getName(), fields);
if (old_fields != null) {
fields = old_fields;
}
}
//统计Model对象的item数量
int count = 0;
try {
for (int i = 0; i < fields.size(); i++) {
Field field = fields.get(i);
if (!field.isAccessible()) {
field.setAccessible(true);
}
Object value = field.get(model);
if (value != null) {
if (value instanceof Collection) {
count += ((Collection) value).size();
} else if (value.getClass().isArray()) {
count += Array.getLength(value);
} else if (value instanceof Map) {
count += ((Map) value).size();
} else if (value instanceof Countable) {
count += ((Countable) value).size();
}
}
}
} catch (Exception e) {
logger.error("get item count of result model failed, model: {}", JSON.toJSONString(model), e);
}
return count > 0 ? count : 1;
}
}
@@ -0,0 +1,21 @@
package com.taobao.arthas.core.distribution;
import com.taobao.arthas.core.command.model.ResultModel;
/**
* Command result distributor, sending results to consumers who joins in the same session.
* @author gongdewei 2020-03-26
*/
public interface ResultDistributor {
/**
* Append the phased result to queue
* @param result a phased result of the command
*/
void appendResult(ResultModel result);
/**
* Close result distribtor, release resources
*/
void close();
}
@@ -0,0 +1,31 @@
package com.taobao.arthas.core.distribution;
import java.util.List;
public interface SharingResultDistributor extends ResultDistributor {
/**
* Add consumer to sharing session
* @param consumer
*/
void addConsumer(ResultConsumer consumer);
/**
* Remove consumer from sharing session
* @param consumer
*/
void removeConsumer(ResultConsumer consumer);
/**
* Get all consumers of session
* @return
*/
List<ResultConsumer> getConsumers();
/**
* Get consumer by id
* @param consumerId
* @return
*/
ResultConsumer getConsumer(String consumerId);
}
@@ -0,0 +1,52 @@
package com.taobao.arthas.core.distribution.impl;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.distribution.CompositeResultDistributor;
import com.taobao.arthas.core.distribution.ResultDistributor;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
/**
* 复合结果分发器,将消息同时分发给其包含的真实分发器
*
* @author gongdewei 2020/4/30
*/
public class CompositeResultDistributorImpl implements CompositeResultDistributor {
private List<ResultDistributor> distributors = Collections.synchronizedList(new ArrayList<ResultDistributor>());
public CompositeResultDistributorImpl() {
}
public CompositeResultDistributorImpl(ResultDistributor ... distributors) {
for (ResultDistributor distributor : distributors) {
this.addDistributor(distributor);
}
}
@Override
public void addDistributor(ResultDistributor distributor) {
distributors.add(distributor);
}
@Override
public void removeDistributor(ResultDistributor distributor) {
distributors.remove(distributor);
}
@Override
public void appendResult(ResultModel result) {
for (ResultDistributor distributor : distributors) {
distributor.appendResult(result);
}
}
@Override
public void close() {
for (ResultDistributor distributor : distributors) {
distributor.close();
}
}
}
@@ -0,0 +1,43 @@
package com.taobao.arthas.core.distribution.impl;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.distribution.PackingResultDistributor;
import com.taobao.arthas.core.shell.session.Session;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class PackingResultDistributorImpl implements PackingResultDistributor {
private static final Logger logger = LoggerFactory.getLogger(PackingResultDistributorImpl.class);
private BlockingQueue<ResultModel> resultQueue = new ArrayBlockingQueue<ResultModel>(500);
private final Session session;
public PackingResultDistributorImpl(Session session) {
this.session = session;
}
@Override
public void appendResult(ResultModel result) {
if (!resultQueue.offer(result)) {
logger.warn("result queue is full: {}, discard later result: {}", resultQueue.size(), JSON.toJSONString(result));
}
}
@Override
public void close() {
}
@Override
public List<ResultModel> getResults() {
ArrayList<ResultModel> results = new ArrayList<ResultModel>(resultQueue.size());
resultQueue.drainTo(results);
return results;
}
}
@@ -0,0 +1,167 @@
package com.taobao.arthas.core.distribution.impl;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.distribution.DistributorOptions;
import com.taobao.arthas.core.distribution.ResultConsumer;
import com.taobao.arthas.core.distribution.ResultConsumerHelper;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author gongdewei 2020/3/27
*/
public class ResultConsumerImpl implements ResultConsumer {
private static final Logger logger = LoggerFactory.getLogger(ResultConsumerImpl.class);
private BlockingQueue<ResultModel> resultQueue;
private volatile long lastAccessTime;
private volatile boolean polling;
private ReentrantLock lock = new ReentrantLock();
private int resultBatchSizeLimit = 20;
private int resultQueueSize = DistributorOptions.resultQueueSize;
private long pollTimeLimit = 2 * 1000;
private String consumerId;
private boolean closed;
private long sendingItemCount;
public ResultConsumerImpl() {
lastAccessTime = System.currentTimeMillis();
resultQueue = new ArrayBlockingQueue<ResultModel>(resultQueueSize);
}
@Override
public boolean appendResult(ResultModel result) {
//可能某些Consumer已经断开,不会再读取,这里不能堵塞!
boolean discard = false;
while (!resultQueue.offer(result)) {
ResultModel discardResult = resultQueue.poll();
discard = true;
}
return !discard;
}
@Override
public List<ResultModel> pollResults() {
try {
lastAccessTime = System.currentTimeMillis();
long accessTime = lastAccessTime;
if (lock.tryLock(500, TimeUnit.MILLISECONDS)) {
polling = true;
sendingItemCount = 0;
long firstResultTime = 0;
// sending delay: time elapsed after firstResultTime
long sendingDelay = 0;
// waiting time: time elapsed after access
long waitingTime = 0;
List<ResultModel> sendingResults = new ArrayList<ResultModel>(resultBatchSizeLimit);
while (!closed
&&sendingResults.size() < resultBatchSizeLimit
&& sendingDelay < 100
&& waitingTime < pollTimeLimit) {
ResultModel aResult = resultQueue.poll(100, TimeUnit.MILLISECONDS);
if (aResult != null) {
sendingResults.add(aResult);
//是否为第一次获取到数据
if (firstResultTime == 0) {
firstResultTime = System.currentTimeMillis();
}
//判断是否需要立即发送出去
if (shouldFlush(sendingResults, aResult)) {
break;
}
} else {
if (firstResultTime > 0) {
//获取到部分数据后,队列已经取完,计算发送延时时间
sendingDelay = System.currentTimeMillis() - firstResultTime;
}
//计算总共等待时间,长轮询最大等待时间
waitingTime = System.currentTimeMillis() - accessTime;
}
}
//resultQueue.drainTo(sendingResults, resultSizeLimit-sendingResults.size());
if(logger.isDebugEnabled()) {
logger.debug("pollResults: {}, results: {}", sendingResults.size(), JSON.toJSONString(sendingResults));
}
return sendingResults;
}
} catch (InterruptedException e) {
//e.printStackTrace();
} finally {
if (lock.isHeldByCurrentThread()) {
lastAccessTime = System.currentTimeMillis();
polling = false;
lock.unlock();
}
}
return Collections.emptyList();
}
/**
* 估算对象数量及大小,判断是否需要立即发送出去
* @param sendingResults
* @param last
* @return
*/
private boolean shouldFlush(List<ResultModel> sendingResults, ResultModel last) {
//TODO 引入一个估算模型,每个model自统计对象数量
sendingItemCount += ResultConsumerHelper.getItemCount(last);
return sendingItemCount >= 100;
}
@Override
public boolean isHealthy() {
return isPolling()
|| resultQueue.size() < resultQueueSize
|| System.currentTimeMillis() - lastAccessTime < 1000;
}
@Override
public long getLastAccessTime() {
return lastAccessTime;
}
@Override
public void close(){
this.closed = true;
}
@Override
public boolean isClosed() {
return closed;
}
@Override
public boolean isPolling() {
return polling;
}
public int getResultBatchSizeLimit() {
return resultBatchSizeLimit;
}
public void setResultBatchSizeLimit(int resultBatchSizeLimit) {
this.resultBatchSizeLimit = resultBatchSizeLimit;
}
@Override
public String getConsumerId() {
return consumerId;
}
@Override
public void setConsumerId(String consumerId) {
this.consumerId = consumerId;
}
}
@@ -0,0 +1,220 @@
package com.taobao.arthas.core.distribution.impl;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.taobao.arthas.core.command.model.InputStatusModel;
import com.taobao.arthas.core.command.model.MessageModel;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.distribution.DistributorOptions;
import com.taobao.arthas.core.distribution.ResultConsumer;
import com.taobao.arthas.core.distribution.SharingResultDistributor;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.system.Job;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
public class SharingResultDistributorImpl implements SharingResultDistributor {
private static final Logger logger = LoggerFactory.getLogger(SharingResultDistributorImpl.class);
private List<ResultConsumer> consumers = new CopyOnWriteArrayList<ResultConsumer>();
private BlockingQueue<ResultModel> pendingResultQueue = new ArrayBlockingQueue<ResultModel>(10);
private final Session session;
private Thread distributorThread;
private volatile boolean running;
private AtomicInteger consumerNumGenerator = new AtomicInteger(0);
private SharingResultConsumerImpl sharingResultConsumer = new SharingResultConsumerImpl();
public SharingResultDistributorImpl(Session session) {
this.session = session;
this.running = true;
distributorThread = new Thread(new DistributorTask(), "ResultDistributor");
distributorThread.start();
}
@Override
public void appendResult(ResultModel result) {
//要避免阻塞影响业务线程执行
try {
if (!pendingResultQueue.offer(result, 100, TimeUnit.MILLISECONDS)) {
ResultModel discardResult = pendingResultQueue.poll();
// 正常情况走不到这里,除非distribute 循环堵塞或异常终止
// 输出队列满,终止当前执行的命令
interruptJob("result queue is full: "+ pendingResultQueue.size());
}
} catch (InterruptedException e) {
//ignore
}
}
private void interruptJob(String message) {
Job job = session.getForegroundJob();
if (job != null) {
logger.warn(message+", current job was interrupted.", job.id());
job.interrupt();
pendingResultQueue.offer(new MessageModel(message+", current job was interrupted."));
}
}
private void distribute() {
while (running) {
try {
ResultModel result = pendingResultQueue.poll(100, TimeUnit.MILLISECONDS);
if (result != null) {
sharingResultConsumer.appendResult(result);
//判断是否有至少一个consumer是健康的
int healthCount = 0;
for (int i = 0; i < consumers.size(); i++) {
ResultConsumer consumer = consumers.get(i);
if(consumer.isHealthy()){
healthCount += 1;
}
consumer.appendResult(result);
}
//所有consumer都不是健康状态,终止当前执行的命令
if (healthCount == 0) {
interruptJob("all consumers are unhealthy");
}
}
} catch (Throwable e) {
logger.warn("distribute result failed: " + e.getMessage(), e);
}
}
}
@Override
public void close() {
this.running = false;
}
@Override
public void addConsumer(ResultConsumer consumer) {
int consumerNo = consumerNumGenerator.incrementAndGet();
String consumerId = UUID.randomUUID().toString().replaceAll("-", "") + "_" + consumerNo;
consumer.setConsumerId(consumerId);
//将队列中的消息复制给新的消费者
sharingResultConsumer.copyTo(consumer);
consumers.add(consumer);
}
@Override
public void removeConsumer(ResultConsumer consumer) {
consumers.remove(consumer);
consumer.close();
}
@Override
public List<ResultConsumer> getConsumers() {
return consumers;
}
@Override
public ResultConsumer getConsumer(String consumerId) {
for (int i = 0; i < consumers.size(); i++) {
ResultConsumer consumer = consumers.get(i);
if (consumer.getConsumerId().equals(consumerId)) {
return consumer;
}
}
return null;
}
private class DistributorTask implements Runnable {
@Override
public void run() {
distribute();
}
}
private class SharingResultConsumerImpl implements ResultConsumer {
private BlockingQueue<ResultModel> resultQueue = new ArrayBlockingQueue<ResultModel>(DistributorOptions.resultQueueSize);
private ReentrantLock queueLock = new ReentrantLock();
private InputStatusModel lastInputStatus;
@Override
public boolean appendResult(ResultModel result) {
queueLock.lock();
try {
//输入状态不入历史指令队列,复制时在最后发送
if (result instanceof InputStatusModel) {
lastInputStatus = (InputStatusModel) result;
return true;
}
while (!resultQueue.offer(result)) {
ResultModel discardResult = resultQueue.poll();
}
} finally {
if (queueLock.isHeldByCurrentThread()) {
queueLock.unlock();
}
}
return true;
}
public void copyTo(ResultConsumer consumer) {
//复制时加锁,避免消息顺序错乱,这里堵塞只影响分发线程,不会影响到业务线程
queueLock.lock();
try {
for (ResultModel result : resultQueue) {
consumer.appendResult(result);
}
//发送输入状态
if (lastInputStatus != null) {
consumer.appendResult(lastInputStatus);
}
} finally {
if (queueLock.isHeldByCurrentThread()) {
queueLock.unlock();
}
}
}
@Override
public List<ResultModel> pollResults() {
return null;
}
@Override
public long getLastAccessTime() {
return 0;
}
@Override
public void close() {
}
@Override
public boolean isClosed() {
return false;
}
@Override
public boolean isPolling() {
return false;
}
@Override
public String getConsumerId() {
return "shared-consumer";
}
@Override
public void setConsumerId(String consumerId) {
}
@Override
public boolean isHealthy() {
return true;
}
}
}
@@ -0,0 +1,36 @@
package com.taobao.arthas.core.distribution.impl;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.command.view.ResultView;
import com.taobao.arthas.core.command.view.ResultViewResolver;
import com.taobao.arthas.core.distribution.ResultDistributor;
import com.taobao.arthas.core.shell.command.CommandProcess;
/**
* Term/Tty Result Distributor
*
* @author gongdewei 2020-03-26
*/
public class TermResultDistributorImpl implements ResultDistributor {
private final CommandProcess commandProcess;
private final ResultViewResolver resultViewResolver;
public TermResultDistributorImpl(CommandProcess commandProcess) {
this.commandProcess = commandProcess;
this.resultViewResolver = ResultViewResolver.getInstance();
}
@Override
public void appendResult(ResultModel model) {
ResultView resultView = resultViewResolver.getResultView(model);
if (resultView != null) {
resultView.draw(commandProcess, model);
}
}
@Override
public void close() {
}
}
@@ -40,26 +40,30 @@ import com.taobao.arthas.core.shell.ShellServerOptions;
import com.taobao.arthas.core.shell.command.CommandResolver;
import com.taobao.arthas.core.shell.handlers.BindHandler;
import com.taobao.arthas.core.shell.impl.ShellServerImpl;
import com.taobao.arthas.core.shell.session.SessionManager;
import com.taobao.arthas.core.shell.session.impl.SessionManagerImpl;
import com.taobao.arthas.core.shell.term.impl.HttpTermServer;
import com.taobao.arthas.core.shell.term.impl.httptelnet.HttpTelnetTermServer;
import com.taobao.arthas.core.util.ArthasBanner;
import com.taobao.arthas.core.util.FileUtils;
import com.taobao.arthas.core.util.LogUtil;
import com.taobao.arthas.core.util.UserStatUtil;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.EventExecutorGroup;
/**
* @author vlinux on 15/5/2.
* @author gongdewei 2020-03-25
*/
public class ArthasBootstrap {
public static final String ARTHAS_HOME_PROPERTY = "arthas.home";
private static String ARTHAS_SHOME = null;
public static final String CONFIG_NAME_PROPERTY = "arthas.config.name";
public static final String CONFIG_NAME_PROPERTY = "arthas.config.name";
public static final String CONFIG_LOCATION_PROPERTY = "arthas.config.location";
public static final String CONFIG_OVERRIDE_ALL= "arthas.config.overrideAll";
public static final String CONFIG_OVERRIDE_ALL = "arthas.config.overrideAll";
private static ArthasBootstrap arthasBootstrap;
@@ -71,11 +75,13 @@ public class ArthasBootstrap {
private Thread shutdown;
private ShellServer shellServer;
private ScheduledExecutorService executorService;
private SessionManager sessionManager;
private TunnelClient tunnelClient;
private File arthasOutputDir;
private static LoggerContext loggerContext;
private EventExecutorGroup workerGroup;
private Timer timer = new Timer("arthas-timer", true);
@@ -262,20 +268,25 @@ public class ArthasBootstrap {
welcomeInfos.put("id", agentId);
options.setWelcomeMessage(ArthasBanner.welcome(welcomeInfos));
}
shellServer = new ShellServerImpl(options, this);
BuiltinCommandPack builtinCommands = new BuiltinCommandPack();
List<CommandResolver> resolvers = new ArrayList<CommandResolver>();
resolvers.add(builtinCommands);
//worker group
workerGroup = new NioEventLoopGroup(8);
// TODO: discover user provided command resolver
if (configure.getTelnetPort() > 0) {
shellServer.registerTermServer(new HttpTelnetTermServer(configure.getIp(), configure.getTelnetPort(),
options.getConnectionTimeout()));
options.getConnectionTimeout(), workerGroup));
} else {
logger().info("telnet port is {}, skip bind telnet server.", configure.getTelnetPort());
}
if (configure.getHttpPort() > 0) {
shellServer.registerTermServer(new HttpTermServer(configure.getIp(), configure.getHttpPort(),
options.getConnectionTimeout()));
options.getConnectionTimeout(), workerGroup));
} else {
logger().info("http port is {}, skip bind http server.", configure.getHttpPort());
}
@@ -286,8 +297,12 @@ public class ArthasBootstrap {
shellServer.listen(new BindHandler(isBindRef));
//http api session manager
sessionManager = new SessionManagerImpl(options, this, shellServer.getCommandManager(), shellServer.getJobController());
logger().info("as-server listening on network={};telnet={};http={};timeout={};", configure.getIp(),
configure.getTelnetPort(), configure.getHttpPort(), options.getConnectionTimeout());
// 异步回报启动次数
if (configure.getStatUrl() != null) {
logger().info("arthas stat url: {}", configure.getStatUrl());
@@ -295,16 +310,27 @@ public class ArthasBootstrap {
UserStatUtil.setStatUrl(configure.getStatUrl());
UserStatUtil.arthasStart();
logger().info("as-server started in {} ms", System.currentTimeMillis() - start );
logger().info("as-server started in {} ms", System.currentTimeMillis() - start);
} catch (Throwable e) {
logger().error("Error during bind to port " + configure.getTelnetPort(), e);
if (shellServer != null) {
shellServer.close();
}
if (sessionManager != null){
sessionManager.close();
}
shutdownWorkGroup();
throw e;
}
}
private void shutdownWorkGroup() {
if (workerGroup != null) {
workerGroup.shutdownGracefully(200, 200, TimeUnit.MILLISECONDS);
workerGroup = null;
}
}
/**
* 判断服务端是否已经启动
*
@@ -320,12 +346,13 @@ public class ArthasBootstrap {
try {
tunnelClient.stop();
} catch (Throwable e) {
logger().error("arthas", "stop tunnel client error", e);
logger().error("stop tunnel client error", e);
}
}
executorService.shutdownNow();
transformerManager.destroy();
UserStatUtil.destroy();
shutdownWorkGroup();
// clear the reference in Spy class.
cleanUpSpyReference();
try {
@@ -337,6 +364,8 @@ public class ArthasBootstrap {
if (loggerContext != null) {
loggerContext.stop();
}
shellServer = null;
sessionManager = null;
}
/**
@@ -352,6 +381,7 @@ public class ArthasBootstrap {
}
return arthasBootstrap;
}
/**
* @return ArthasServer单例
*/
@@ -385,6 +415,14 @@ public class ArthasBootstrap {
return tunnelClient;
}
public ShellServer getShellServer() {
return shellServer;
}
public SessionManager getSessionManager() {
return sessionManager;
}
public Timer getTimer() {
return this.timer;
}
@@ -5,6 +5,8 @@ import com.taobao.arthas.core.shell.future.Future;
import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.handlers.NoOpHandler;
import com.taobao.arthas.core.shell.impl.ShellServerImpl;
import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
import com.taobao.arthas.core.shell.system.impl.JobControllerImpl;
import com.taobao.arthas.core.shell.term.Term;
import com.taobao.arthas.core.shell.term.TermServer;
@@ -102,4 +104,14 @@ public abstract class ShellServer {
* @param completionHandler handler for getting notified when service is stopped
*/
public abstract void close(Handler<Future<Void>> completionHandler);
/**
* @return global job controller instance
*/
public abstract JobControllerImpl getJobController();
/**
* @return get command manager instance
*/
public abstract InternalCommandManager getCommandManager();
}
@@ -1,6 +1,7 @@
package com.taobao.arthas.core.shell.command;
import com.taobao.arthas.core.advisor.AdviceListener;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.session.Session;
@@ -119,6 +120,13 @@ public interface CommandProcess extends Tty {
*/
void end(int status);
/**
* End the process.
*
* @param status the exit status.
*/
void end(int status, String message);
/**
* Register listener
@@ -167,4 +175,11 @@ public interface CommandProcess extends Tty {
* Whether the process is running
*/
boolean isRunning();
/**
* Append the phased result to queue
* @param result a phased result of the command
*/
void appendResult(ResultModel result);
}
@@ -0,0 +1,21 @@
package com.taobao.arthas.core.shell.history;
import java.util.List;
/**
* @author gongdewei 2020/4/8
*/
public interface HistoryManager {
void addHistory(String commandLine);
List<String> getHistory();
void setHistory(List<String> history);
void saveHistory();
void loadHistory();
void clearHistory();
}
@@ -0,0 +1,70 @@
package com.taobao.arthas.core.shell.history.impl;
import com.taobao.arthas.core.shell.history.HistoryManager;
import com.taobao.arthas.core.util.Constants;
import com.taobao.arthas.core.util.FileUtils;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
/**
* @author gongdewei 2020/4/8
*/
public class HistoryManagerImpl implements HistoryManager {
/**
* The max number of history item that will be saved in memory.
*/
private static final int MAX_HISTORY_SIZE = 500;
private List<String> history = new ArrayList<String>();
private static HistoryManager instance;
public static HistoryManager getInstance() {
if (instance == null) {
synchronized (HistoryManagerImpl.class) {
instance = new HistoryManagerImpl();
}
}
return instance;
}
private HistoryManagerImpl() {
}
@Override
public void saveHistory() {
FileUtils.saveCommandHistoryString(history, new File(Constants.CMD_HISTORY_FILE));
}
@Override
public void loadHistory() {
history = FileUtils.loadCommandHistoryString(new File(Constants.CMD_HISTORY_FILE));
}
@Override
public void clearHistory() {
this.history.clear();
}
@Override
public void addHistory(String commandLine) {
while (history.size() >= MAX_HISTORY_SIZE) {
history.remove(0);
}
history.add(commandLine);
}
@Override
public List<String> getHistory() {
return history;
}
@Override
public void setHistory(List<String> history) {
this.history = history;
}
}
@@ -7,22 +7,21 @@ import com.taobao.arthas.core.shell.ShellServer;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.cli.CliTokens;
import com.taobao.arthas.core.shell.future.Future;
import com.taobao.arthas.core.shell.handlers.shell.CloseHandler;
import com.taobao.arthas.core.shell.handlers.shell.CommandManagerCompletionHandler;
import com.taobao.arthas.core.shell.handlers.shell.FutureHandler;
import com.taobao.arthas.core.shell.handlers.shell.InterruptHandler;
import com.taobao.arthas.core.shell.handlers.shell.ShellLineHandler;
import com.taobao.arthas.core.shell.handlers.shell.SuspendHandler;
import com.taobao.arthas.core.shell.handlers.shell.*;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.session.impl.SessionImpl;
import com.taobao.arthas.core.shell.system.ExecStatus;
import com.taobao.arthas.core.shell.system.Job;
import com.taobao.arthas.core.shell.system.JobController;
import com.taobao.arthas.core.shell.system.JobListener;
import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
import com.taobao.arthas.core.shell.system.impl.JobControllerImpl;
import com.taobao.arthas.core.shell.term.Term;
import com.taobao.arthas.core.shell.term.impl.httptelnet.HttpTelnetTermServer;
import com.taobao.arthas.core.shell.term.impl.TermImpl;
import com.taobao.arthas.core.util.Constants;
import com.taobao.arthas.core.util.FileUtils;
import java.io.File;
import java.lang.instrument.Instrumentation;
import java.util.Date;
import java.util.List;
@@ -79,7 +78,7 @@ public class ShellImpl implements Shell {
@Override
public synchronized Job createJob(List<CliToken> args) {
Job job = jobController.createJob(commandManager, args, this);
Job job = jobController.createJob(commandManager, args, session, new ShellJobHandler(this), term, null);
return job;
}
@@ -165,7 +164,7 @@ public class ShellImpl implements Shell {
// sometimes an NPE will be thrown during shutdown via web-socket,
// this ensures the shutdown process is finished properly
// https://github.com/alibaba/arthas/issues/320
logger.error("ARTHAS", "Error writing data:", t);
logger.error("Error writing data:", t);
}
term.close();
} else {
@@ -180,4 +179,53 @@ public class ShellImpl implements Shell {
public Job getForegroundJob() {
return currentForegroundJob;
}
private class ShellJobHandler implements JobListener {
ShellImpl shell;
public ShellJobHandler(ShellImpl shell) {
this.shell = shell;
}
@Override
public void onForeground(Job job) {
shell.setForegroundJob(job);
//reset stdin handler to job's origin handler
//shell.term().stdinHandler(job.process().getStdinHandler());
}
@Override
public void onBackground(Job job) {
resetAndReadLine();
}
@Override
public void onTerminated(Job job) {
if (!job.isRunInBackground()){
resetAndReadLine();
}
// save command history
Term term = shell.term();
if (term instanceof TermImpl) {
List<int[]> history = ((TermImpl) term).getReadline().getHistory();
FileUtils.saveCommandHistory(history, new File(Constants.CMD_HISTORY_FILE));
}
}
@Override
public void onSuspend(Job job) {
if (!job.isRunInBackground()){
resetAndReadLine();
}
}
private void resetAndReadLine() {
//reset stdin handler to echo handler
//shell.term().stdinHandler(null);
shell.setForegroundJob(null);
shell.readline();
}
}
}
@@ -242,4 +242,12 @@ public class ShellServerImpl extends ShellServer {
bootstrap.destroy();
}
}
public JobControllerImpl getJobController() {
return jobController;
}
public InternalCommandManager getCommandManager() {
return commandManager;
}
}
@@ -1,7 +1,8 @@
package com.taobao.arthas.core.shell.session;
import com.taobao.arthas.core.shell.ShellServer;
import com.taobao.arthas.core.distribution.SharingResultDistributor;
import com.taobao.arthas.core.shell.command.CommandResolver;
import com.taobao.arthas.core.shell.system.Job;
import java.lang.instrument.Instrumentation;
import java.util.List;
@@ -10,6 +11,7 @@ import java.util.List;
* A shell session.
*
* @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
* @author gongdewei 2020-03-23
*/
public interface Session {
String COMMAND_MANAGER = "arthas-command-manager";
@@ -22,6 +24,27 @@ public interface Session {
*/
String TTY = "tty";
/**
* Session create time
*/
String CREATE_TIME = "createTime";
/**
* Session last active time
*/
String LAST_ACCESS_TIME = "lastAccessedTime";
/**
* Command Result Distributor
*/
String RESULT_DISTRIBUTOR = "resultDistributor";
/**
* The executing foreground job
*/
String FOREGROUND_JOB = "foregroundJob";
/**
* Put some data in a session
*
@@ -80,13 +103,6 @@ public interface Session {
*/
String getSessionId();
/**
* Get shell server
*
* @return shell server
*/
ShellServer getServer();
/**
* Get Java PID
*
@@ -107,4 +123,49 @@ public interface Session {
* @return instrumentation instance
*/
Instrumentation getInstrumentation();
/**
* Update session last access time
* @param time new time
*/
void setLastAccessTime(long time);
/**
* Get session last access time
* @return session last access time
*/
long getLastAccessTime();
/**
* Get session create time
* @return session create time
*/
long getCreateTime();
/**
* Update session's command result distributor
* @param resultDistributor
*/
void setResultDistributor(SharingResultDistributor resultDistributor);
/**
* Get session's command result distributor
* @return
*/
SharingResultDistributor getResultDistributor();
/**
* Set the foreground job
*/
void setForegroundJob(Job job);
/**
* Get the foreground job
*/
Job getForegroundJob();
/**
* Whether the session is tty term
*/
boolean isTty();
}
@@ -0,0 +1,29 @@
package com.taobao.arthas.core.shell.session;
import com.taobao.arthas.core.shell.system.JobController;
import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
import java.lang.instrument.Instrumentation;
/**
* Arthas Session Manager
* @author gongdewei 2020-03-20
*/
public interface SessionManager {
Session createSession();
Session getSession(String sessionId);
Session removeSession(String sessionId);
void updateAccessTime(Session session);
void close();
InternalCommandManager getCommandManager();
Instrumentation getInstrumentation();
JobController getJobController();
}
@@ -1,8 +1,9 @@
package com.taobao.arthas.core.shell.session.impl;
import com.taobao.arthas.core.shell.ShellServer;
import com.taobao.arthas.core.distribution.SharingResultDistributor;
import com.taobao.arthas.core.shell.command.CommandResolver;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.system.Job;
import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
import java.lang.instrument.Instrumentation;
@@ -21,6 +22,12 @@ public class SessionImpl implements Session {
private Map<String, Object> data = new HashMap<String, Object>();
public SessionImpl() {
long now = System.currentTimeMillis();
data.put(CREATE_TIME, now);
this.setLastAccessTime(now);
}
@Override
public Session put(String key, Object obj) {
if (obj == null) {
@@ -69,11 +76,6 @@ public class SessionImpl implements Session {
return (String) data.get(ID);
}
@Override
public ShellServer getServer() {
return (ShellServer) data.get(SERVER);
}
@Override
public long getPid() {
return (Long) data.get(PID);
@@ -89,4 +91,45 @@ public class SessionImpl implements Session {
public Instrumentation getInstrumentation() {
return (Instrumentation) data.get(INSTRUMENTATION);
}
@Override
public void setLastAccessTime(long time) {
data.put(LAST_ACCESS_TIME, time);
}
@Override
public long getLastAccessTime() {
return (Long)data.get(LAST_ACCESS_TIME);
}
@Override
public long getCreateTime() {
return (Long)data.get(CREATE_TIME);
}
@Override
public void setResultDistributor(SharingResultDistributor resultDistributor) {
data.put(RESULT_DISTRIBUTOR, resultDistributor);
}
@Override
public SharingResultDistributor getResultDistributor() {
return (SharingResultDistributor) data.get(RESULT_DISTRIBUTOR);
}
@Override
public void setForegroundJob(Job job) {
data.put(FOREGROUND_JOB, job);
}
@Override
public Job getForegroundJob() {
return (Job) data.get(FOREGROUND_JOB);
}
@Override
public boolean isTty() {
return get(TTY) != null;
}
}
@@ -0,0 +1,193 @@
package com.taobao.arthas.core.shell.session.impl;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.taobao.arthas.core.command.model.MessageModel;
import com.taobao.arthas.core.distribution.ResultConsumer;
import com.taobao.arthas.core.distribution.SharingResultDistributor;
import com.taobao.arthas.core.distribution.impl.SharingResultDistributorImpl;
import com.taobao.arthas.core.server.ArthasBootstrap;
import com.taobao.arthas.core.shell.ShellServerOptions;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.session.SessionManager;
import com.taobao.arthas.core.shell.system.Job;
import com.taobao.arthas.core.shell.system.JobController;
import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
import java.lang.instrument.Instrumentation;
import java.util.*;
import java.util.concurrent.*;
/**
* Arthas Session Manager
*
* @author gongdewei 2020-03-20
*/
public class SessionManagerImpl implements SessionManager {
private static final Logger logger = LoggerFactory.getLogger(SessionManagerImpl.class);
private final ArthasBootstrap bootstrap;
private final InternalCommandManager commandManager;
private final Instrumentation instrumentation;
private final JobController jobController;
private final long timeoutMillis;
private final long reaperInterval;
private final Map<String, Session> sessions;
private final long pid;
private boolean closed = false;
private ScheduledExecutorService scheduledExecutorService;
public SessionManagerImpl(ShellServerOptions options, ArthasBootstrap bootstrap, InternalCommandManager commandManager,
JobController jobController) {
this.bootstrap = bootstrap;
this.commandManager = commandManager;
this.jobController = jobController;
this.sessions = new ConcurrentHashMap<String, Session>();
this.timeoutMillis = options.getSessionTimeout();
this.reaperInterval = options.getReaperInterval();
this.instrumentation = options.getInstrumentation();
this.pid = options.getPid();
//start evict session timer
this.setEvictTimer();
}
@Override
public Session createSession() {
Session session = new SessionImpl();
session.put(Session.COMMAND_MANAGER, commandManager);
session.put(Session.INSTRUMENTATION, instrumentation);
session.put(Session.PID, pid);
//session.put(Session.SERVER, server);
//session.put(Session.TTY, term);
String sessionId = UUID.randomUUID().toString();
session.put(Session.ID, sessionId);
//Result Distributor
session.setResultDistributor(new SharingResultDistributorImpl(session));
sessions.put(sessionId, session);
return session;
}
@Override
public Session getSession(String sessionId) {
return sessions.get(sessionId);
}
@Override
public Session removeSession(String sessionId) {
return sessions.remove(sessionId);
}
@Override
public void updateAccessTime(Session session) {
session.setLastAccessTime(System.currentTimeMillis());
}
@Override
public void close() {
//TODO clear resources while shutdown arthas
closed = true;
if (scheduledExecutorService != null) {
scheduledExecutorService.shutdownNow();
}
ArrayList<Session> sessions = new ArrayList<Session>(this.sessions.values());
for (Session session : sessions) {
SharingResultDistributor resultDistributor = session.getResultDistributor();
resultDistributor.appendResult(new MessageModel("arthas server is going to shutdown."));
resultDistributor.close();
logger.info("Removing session before shutdown: {}, last access time: {}", session.getSessionId(), session.getLastAccessTime());
this.removeSession(session.getSessionId());
}
jobController.close();
bootstrap.destroy();
}
private synchronized void setEvictTimer() {
if (!closed && reaperInterval > 0) {
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
final Thread t = new Thread(r, "arthas-shell-server");
return t;
}
});
scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
evictSessions();
}
}, 0, reaperInterval, TimeUnit.MILLISECONDS);
}
}
/**
* Check and remove inactive session
*/
public void evictSessions() {
long now = System.currentTimeMillis();
List<Session> toClose = new ArrayList<Session>();
for (Session session : sessions.values()) {
// 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) {
toClose.add(session);
}
evictConsumers(session);
}
for (Session session : toClose) {
//interrupt foreground job
Job job = session.getForegroundJob();
if (job != null) {
job.interrupt();
}
long timeOutInMinutes = timeoutMillis / 1000 / 60;
String reason = "session is inactive for " + timeOutInMinutes + " min(s).";
session.getResultDistributor().appendResult(new MessageModel(reason));
this.removeSession(session.getSessionId());
logger.info("Removing inactive session: {}, last access time: {}", session.getSessionId(), session.getLastAccessTime());
}
}
/**
* Check and remove inactive consumer
*/
public void evictConsumers(Session session) {
SharingResultDistributor distributor = session.getResultDistributor();
if (distributor instanceof SharingResultDistributor) {
SharingResultDistributor sharingResultDistributor = (SharingResultDistributor) distributor;
List<ResultConsumer> consumers = sharingResultDistributor.getConsumers();
//remove inactive consumer from session directly
long now = System.currentTimeMillis();
for (ResultConsumer consumer : consumers) {
long inactiveTime = now - consumer.getLastAccessTime();
if (inactiveTime > 30000) {
//inactive duration must be large than pollTimeLimit
logger.info("Removing inactive consumer from session, sessionId: {}, consumerId: {}, inactive duration: {}",
session.getSessionId(), consumer.getConsumerId(), inactiveTime);
consumer.appendResult(new MessageModel("consumer is inactive for a while, please refresh the page."));
sharingResultDistributor.removeConsumer(consumer);
}
}
}
}
@Override
public InternalCommandManager getCommandManager() {
return commandManager;
}
@Override
public Instrumentation getInstrumentation() {
return instrumentation;
}
@Override
public JobController getJobController() {
return jobController;
}
}
@@ -57,6 +57,11 @@ public interface Job {
*/
Job resume();
/**
* @return true if the job is running in background
*/
boolean isRunInBackground();
/**
* Send the job to background.
*
@@ -1,9 +1,11 @@
package com.taobao.arthas.core.shell.system;
import com.taobao.arthas.core.distribution.ResultDistributor;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.impl.ShellImpl;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
import com.taobao.arthas.core.shell.term.Term;
import java.util.List;
import java.util.Set;
@@ -33,10 +35,13 @@ public interface JobController {
*
* @param commandManager command manager
* @param tokens the command tokens
* @param shell the current shell
* @param session the current session
* @param jobHandler job event handler
* @param term telnet term
* @param resultDistributor
* @return the created job
*/
Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, ShellImpl shell);
Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, Session session, JobListener jobHandler, Term term, ResultDistributor resultDistributor);
/**
* Close the controller and terminate all the underlying jobs, a closed controller does not accept anymore jobs.
@@ -0,0 +1,16 @@
package com.taobao.arthas.core.shell.system;
/**
* Job listener
* @author gongdewei 2020-03-23
*/
public interface JobListener {
void onForeground(Job job);
void onBackground(Job job);
void onTerminated(Job job);
void onSuspend(Job job);
}
@@ -9,11 +9,15 @@ import java.util.concurrent.TimeUnit;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.taobao.arthas.core.GlobalOptions;
import com.taobao.arthas.core.distribution.ResultDistributor;
import com.taobao.arthas.core.server.ArthasBootstrap;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.impl.ShellImpl;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.system.Job;
import com.taobao.arthas.core.shell.system.JobListener;
import com.taobao.arthas.core.shell.term.Term;
/**
* 全局的Job Controller,不应该存在启停的概念,不需要在连接的断开时关闭,
@@ -49,8 +53,8 @@ public class GlobalJobControllerImpl extends JobControllerImpl {
}
@Override
public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, ShellImpl shell) {
final Job job = super.createJob(commandManager, tokens, shell);
public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, Session session, JobListener jobHandler, Term term, ResultDistributor resultDistributor) {
final Job job = super.createJob(commandManager, tokens, session, jobHandler, term, resultDistributor);
/*
* 达到超时时间将会停止job
@@ -1,6 +1,7 @@
package com.taobao.arthas.core.shell.system.impl;
import com.taobao.arthas.core.GlobalOptions;
import com.taobao.arthas.core.distribution.ResultDistributor;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.command.Command;
import com.taobao.arthas.core.shell.command.internal.RedirectHandler;
@@ -8,9 +9,10 @@ import com.taobao.arthas.core.shell.command.internal.StdoutHandler;
import com.taobao.arthas.core.shell.command.internal.TermHandler;
import com.taobao.arthas.core.shell.future.Future;
import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.impl.ShellImpl;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.system.Job;
import com.taobao.arthas.core.shell.system.JobController;
import com.taobao.arthas.core.shell.system.JobListener;
import com.taobao.arthas.core.shell.system.Process;
import com.taobao.arthas.core.shell.system.impl.ProcessImpl.ProcessOutput;
import com.taobao.arthas.core.shell.term.Term;
@@ -35,6 +37,7 @@ import java.util.concurrent.atomic.AtomicInteger;
/**
* @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
* @author hengyunabc 2019-05-14
* @author gongdewei 2020-03-23
*/
public class JobControllerImpl implements JobController {
@@ -58,16 +61,16 @@ public class JobControllerImpl implements JobController {
}
@Override
public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, ShellImpl shell) {
public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, Session session, JobListener jobHandler, Term term, ResultDistributor resultDistributor) {
int jobId = idGenerator.incrementAndGet();
StringBuilder line = new StringBuilder();
for (CliToken arg : tokens) {
line.append(arg.raw());
}
boolean runInBackground = runInBackground(tokens);
Process process = createProcess(tokens, commandManager, jobId, shell.term());
Process process = createProcess(tokens, commandManager, jobId, term, resultDistributor);
process.setJobId(jobId);
JobImpl job = new JobImpl(jobId, this, process, line.toString(), runInBackground, shell);
JobImpl job = new JobImpl(jobId, this, process, line.toString(), runInBackground, session, jobHandler);
jobs.put(jobId, job);
return job;
}
@@ -120,9 +123,10 @@ public class JobControllerImpl implements JobController {
* @param commandManager command manager
* @param jobId job id
* @param term term
* @param resultDistributor
* @return the created process
*/
private Process createProcess(List<CliToken> line, InternalCommandManager commandManager, int jobId, Term term) {
private Process createProcess(List<CliToken> line, InternalCommandManager commandManager, int jobId, Term term, ResultDistributor resultDistributor) {
try {
ListIterator<CliToken> tokens = line.listIterator();
while (tokens.hasNext()) {
@@ -130,7 +134,7 @@ public class JobControllerImpl implements JobController {
if (token.isText()) {
Command command = commandManager.getCommand(token.value());
if (command != null) {
return createCommandProcess(command, tokens, jobId, term);
return createCommandProcess(command, tokens, jobId, term, resultDistributor);
} else {
throw new IllegalArgumentException(token.value() + ": command not found");
}
@@ -152,7 +156,7 @@ public class JobControllerImpl implements JobController {
return runInBackground;
}
private Process createCommandProcess(Command command, ListIterator<CliToken> tokens, int jobId, Term term) throws IOException {
private Process createCommandProcess(Command command, ListIterator<CliToken> tokens, int jobId, Term term, ResultDistributor resultDistributor) throws IOException {
List<CliToken> remaining = new ArrayList<CliToken>();
List<CliToken> pipelineTokens = new ArrayList<CliToken>();
boolean isPipeline = false;
@@ -199,7 +203,9 @@ public class JobControllerImpl implements JobController {
}
}
ProcessOutput ProcessOutput = new ProcessOutput(stdoutHandlerChain, cacheLocation, term);
return new ProcessImpl(command, remaining, command.processHandler(), ProcessOutput);
ProcessImpl process = new ProcessImpl(command, remaining, command.processHandler(), ProcessOutput, resultDistributor);
process.setTty(term);
return process;
}
private String getRedirectFileName(ListIterator<CliToken> tokens) {
@@ -7,20 +7,16 @@ import java.util.concurrent.atomic.AtomicBoolean;
import com.taobao.arthas.core.shell.future.Future;
import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.handlers.shell.ShellForegroundUpdateHandler;
import com.taobao.arthas.core.shell.impl.ShellImpl;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.system.ExecStatus;
import com.taobao.arthas.core.shell.system.Job;
import com.taobao.arthas.core.shell.system.JobListener;
import com.taobao.arthas.core.shell.system.Process;
import com.taobao.arthas.core.shell.term.Term;
import com.taobao.arthas.core.shell.term.impl.TermImpl;
import com.taobao.arthas.core.util.Constants;
import com.taobao.arthas.core.util.FileUtils;
/**
* @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
* @author hengyunabc 2019-05-14
* @author gongdewei 2020-03-23
*/
public class JobImpl implements Job {
@@ -28,25 +24,30 @@ public class JobImpl implements Job {
final JobControllerImpl controller;
final Process process;
final String line;
private volatile Session session;
private volatile ExecStatus actualStatus; // Used internally for testing only
volatile long lastStopped; // When the job was last stopped
volatile ShellImpl shell;
volatile JobListener jobHandler;
volatile Handler<ExecStatus> statusUpdateHandler;
volatile Date timeoutDate;
final Future<Void> terminateFuture;
final AtomicBoolean runInBackground;
final Handler<Job> foregroundUpdatedHandler;
//final Handler<Job> foregroundUpdatedHandler;
JobImpl(int id, final JobControllerImpl controller, Process process, String line, boolean runInBackground,
ShellImpl shell) {
Session session, JobListener jobHandler) {
this.id = id;
this.controller = controller;
this.process = process;
this.line = line;
this.session = session;
this.terminateFuture = Future.future();
this.runInBackground = new AtomicBoolean(runInBackground);
this.shell = shell;
this.foregroundUpdatedHandler = new ShellForegroundUpdateHandler(shell);
this.jobHandler = jobHandler;
if (jobHandler == null) {
throw new IllegalArgumentException("JobListener is required");
}
//this.foregroundUpdatedHandler = new ShellForegroundUpdateHandler(shell);
process.terminatedHandler(new TerminatedHandler(controller));
}
@@ -76,7 +77,7 @@ public class JobImpl implements Job {
@Override
public Session getSession() {
return shell.session();
return session;
}
@Override
@@ -89,19 +90,21 @@ public class JobImpl implements Job {
runInBackground.set(!foreground);
if (foreground) {
if (foregroundUpdatedHandler != null) {
foregroundUpdatedHandler.handle(this);
}
}
// if (foreground) {
// if (foregroundUpdatedHandler != null) {
// foregroundUpdatedHandler.handle(this);
// }
// }
if (statusUpdateHandler != null) {
statusUpdateHandler.handle(process.status());
}
if (foreground) {
shell.setForegroundJob(this);
} else {
shell.setForegroundJob(null);
if (this.status() == ExecStatus.RUNNING) {
if (foreground) {
jobHandler.onForeground(this);
} else {
jobHandler.onBackground(this);
}
}
return this;
}
@@ -113,14 +116,15 @@ public class JobImpl implements Job {
} catch (IllegalStateException ignore) {
return this;
}
if (!runInBackground.get() && foregroundUpdatedHandler != null) {
foregroundUpdatedHandler.handle(null);
}
// if (!runInBackground.get() && foregroundUpdatedHandler != null) {
// foregroundUpdatedHandler.handle(null);
// }
if (statusUpdateHandler != null) {
statusUpdateHandler.handle(process.status());
}
shell.setForegroundJob(null);
// shell.setForegroundJob(null);
jobHandler.onSuspend(this);
return this;
}
@@ -147,6 +151,11 @@ public class JobImpl implements Job {
return line;
}
@Override
public boolean isRunInBackground() {
return runInBackground.get();
}
@Override
public Job toBackground() {
if (!this.runInBackground.get()) {
@@ -156,10 +165,12 @@ public class JobImpl implements Job {
if (statusUpdateHandler != null) {
statusUpdateHandler.handle(process.status());
}
jobHandler.onBackground(this);
}
}
shell.setForegroundJob(null);
// shell.setForegroundJob(null);
// jobHandler.onBackground(this);
return this;
}
@@ -167,15 +178,16 @@ public class JobImpl implements Job {
public Job toForeground() {
if (this.runInBackground.get()) {
if (runInBackground.compareAndSet(true, false)) {
if (foregroundUpdatedHandler != null) {
foregroundUpdatedHandler.handle(this);
}
// if (foregroundUpdatedHandler != null) {
// foregroundUpdatedHandler.handle(this);
// }
process.toForeground();
if (statusUpdateHandler != null) {
statusUpdateHandler.handle(process.status());
}
shell.setForegroundJob(this);
// shell.setForegroundJob(this);
jobHandler.onForeground(this);
}
}
@@ -194,26 +206,34 @@ public class JobImpl implements Job {
@Override
public Job run(boolean foreground) {
if (foreground && foregroundUpdatedHandler != null) {
foregroundUpdatedHandler.handle(this);
}
// if (foreground && foregroundUpdatedHandler != null) {
// foregroundUpdatedHandler.handle(this);
// }
actualStatus = ExecStatus.RUNNING;
if (statusUpdateHandler != null) {
statusUpdateHandler.handle(ExecStatus.RUNNING);
}
process.setTty(shell.term());
process.setSession(shell.session());
//set process's tty in JobControllerImpl.createCommandProcess
//process.setTty(shell.term());
process.setSession(this.session);
process.run(foreground);
if (!foreground && foregroundUpdatedHandler != null) {
foregroundUpdatedHandler.handle(null);
}
if (foreground) {
shell.setForegroundJob(this);
} else {
shell.setForegroundJob(null);
// if (!foreground && foregroundUpdatedHandler != null) {
// foregroundUpdatedHandler.handle(null);
// }
//
// if (foreground) {
// shell.setForegroundJob(this);
// } else {
// shell.setForegroundJob(null);
// }
if (this.status() == ExecStatus.RUNNING) {
if (foreground) {
jobHandler.onForeground(this);
} else {
jobHandler.onBackground(this);
}
}
return this;
}
@@ -228,24 +248,25 @@ public class JobImpl implements Job {
@Override
public void handle(Integer exitCode) {
if (!runInBackground.get() && actualStatus.equals(ExecStatus.RUNNING)) {
// if (!runInBackground.get() && actualStatus.equals(ExecStatus.RUNNING)) {
// 只有前台在运行的任务,才需要调用foregroundUpdateHandler
if (foregroundUpdatedHandler != null) {
foregroundUpdatedHandler.handle(null);
}
}
// if (foregroundUpdatedHandler != null) {
// foregroundUpdatedHandler.handle(null);
// }
// }
jobHandler.onTerminated(JobImpl.this);
controller.removeJob(JobImpl.this.id);
if (statusUpdateHandler != null) {
statusUpdateHandler.handle(ExecStatus.TERMINATED);
}
terminateFuture.complete();
// save command history
Term term = shell.term();
if (term instanceof TermImpl) {
List<int[]> history = ((TermImpl) term).getReadline().getHistory();
FileUtils.saveCommandHistory(history, new File(Constants.CMD_HISTORY_FILE));
}
// save command history (move to JobControllerImpl.ShellJobHandler.onTerminated)
// Term term = shell.term();
// if (term instanceof TermImpl) {
// List<int[]> history = ((TermImpl) term).getReadline().getHistory();
// FileUtils.saveCommandHistory(history, new File(Constants.CMD_HISTORY_FILE));
// }
}
}
@@ -4,6 +4,11 @@ import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.taobao.arthas.core.advisor.AdviceListener;
import com.taobao.arthas.core.advisor.AdviceWeaver;
import com.taobao.arthas.core.command.basic1000.HelpCommand;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.command.model.StatusModel;
import com.taobao.arthas.core.distribution.ResultDistributor;
import com.taobao.arthas.core.distribution.impl.TermResultDistributorImpl;
import com.taobao.arthas.core.server.ArthasBootstrap;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.command.Command;
@@ -16,12 +21,8 @@ import com.taobao.arthas.core.shell.system.ExecStatus;
import com.taobao.arthas.core.shell.system.Process;
import com.taobao.arthas.core.shell.system.ProcessAware;
import com.taobao.arthas.core.shell.term.Tty;
import com.taobao.arthas.core.util.usage.StyledUsageFormatter;
import com.taobao.middleware.cli.CLIException;
import com.taobao.middleware.cli.CommandLine;
import com.taobao.middleware.cli.UsageMessageFormatter;
import com.taobao.text.Color;
import io.termd.core.function.Function;
import java.lang.instrument.ClassFileTransformer;
@@ -32,6 +33,7 @@ import java.util.concurrent.atomic.AtomicInteger;
/**
* @author beiwei30 on 10/11/2016.
* @author gongdewei 2020-03-26
*/
public class ProcessImpl implements Process {
@@ -55,16 +57,18 @@ public class ProcessImpl implements Process {
private Handler<String> stdinHandler;
private Handler<Void> resizeHandler;
private Integer exitCode;
private CommandProcess process;
private CommandProcessImpl process;
private Date startTime;
private ProcessOutput processOutput;
private int jobId;
private ResultDistributor resultDistributor;
public ProcessImpl(Command commandContext, List<CliToken> args, Handler<CommandProcess> handler,
ProcessOutput processOutput) {
ProcessOutput processOutput, ResultDistributor resultDistributor) {
this.commandContext = commandContext;
this.handler = handler;
this.args = args;
this.resultDistributor = resultDistributor;
this.processStatus = ExecStatus.READY;
this.processOutput = processOutput;
}
@@ -236,13 +240,15 @@ public class ProcessImpl implements Process {
@Override
public void terminate(Handler<Void> completionHandler) {
if (!terminate(-10, completionHandler)) {
if (!terminate(-10, completionHandler, null)) {
throw new IllegalStateException("Cannot terminate terminated process");
}
}
private synchronized boolean terminate(int exitCode, Handler<Void> completionHandler) {
private synchronized boolean terminate(int exitCode, Handler<Void> completionHandler, String message) {
if (processStatus != ExecStatus.TERMINATED) {
//add status message
this.appendResult(new StatusModel(exitCode, message));
if (process != null) {
processOutput.close();
}
@@ -256,6 +262,13 @@ public class ProcessImpl implements Process {
}
}
private void appendResult(ResultModel result) {
result.setJobId(jobId);
if (resultDistributor != null) {
resultDistributor.appendResult(result);
}
}
private void updateStatus(ExecStatus statusUpdate, Integer exitCodeUpdate, boolean foregroundUpdate,
Handler<Void> handler, Handler<Integer> terminatedHandler,
Handler<Void> completionHandler) {
@@ -320,6 +333,11 @@ public class ProcessImpl implements Process {
throw new IllegalStateException("Cannot execute process without a TTY set");
}
process = new CommandProcessImpl(this, tty);
if (resultDistributor == null) {
resultDistributor = new TermResultDistributorImpl(process);
}
final List<String> args2 = new LinkedList<String>();
for (CliToken arg : args) {
if (arg.isText()) {
@@ -331,25 +349,20 @@ public class ProcessImpl implements Process {
try {
if (commandContext.cli() != null) {
if (commandContext.cli().parse(args2, false).isAskingForHelp()) {
UsageMessageFormatter formatter = new StyledUsageFormatter(Color.green);
formatter.setWidth(tty.width());
StringBuilder usage = new StringBuilder();
commandContext.cli().usage(usage, formatter);
usage.append('\n');
tty.write(usage.toString());
appendResult(new HelpCommand().createHelpDetailModel(commandContext));
terminate();
return;
}
cl = commandContext.cli().parse(args2);
process.setArgs2(args2);
process.setCommandLine(cl);
}
} catch (CLIException e) {
tty.write(e.getMessage() + "\n");
terminate();
terminate(-10, null, e.getMessage());
return;
}
process = new CommandProcessImpl(this, args2, tty, cl);
if (cacheLocation() != null) {
process.echoTips("job id : " + this.jobId + "\n");
process.echoTips("cache location : " + cacheLocation() + "\n");
@@ -372,27 +385,25 @@ public class ProcessImpl implements Process {
handler.handle(process);
} catch (Throwable t) {
logger.error("Error during processing the command:", t);
process.write("Error during processing the command, exception type: " + t.getClass().getName() + ", message:" + t.getMessage()
+ ", please check $HOME/logs/arthas/arthas.log for more details. \n");
terminate(1, null);
process.end(1, "Error during processing the command: " + t.getClass().getName() + ", message:" + t.getMessage()
+ ", please check $HOME/logs/arthas/arthas.log for more details." );
}
}
}
private class CommandProcessImpl implements CommandProcess {
private final Process process;
private final List<String> args2;
private final Tty tty;
private final CommandLine commandLine;
private List<String> args2;
private CommandLine commandLine;
private AtomicInteger times = new AtomicInteger();
private AdviceListener listener = null;
private ClassFileTransformer transformer;
public CommandProcessImpl(Process process, List<String> args2, Tty tty, CommandLine commandLine) {
public CommandProcessImpl(Process process, Tty tty) {
this.process = process;
this.args2 = args2;
this.tty = tty;
this.commandLine = commandLine;
}
@Override
@@ -440,6 +451,14 @@ public class ProcessImpl implements Process {
return times;
}
public void setArgs2(List<String> args2) {
this.args2 = args2;
}
public void setCommandLine(CommandLine commandLine) {
this.commandLine = commandLine;
}
@Override
public CommandProcess stdinHandler(Handler<String> handler) {
stdinHandler = handler;
@@ -578,13 +597,27 @@ public class ProcessImpl implements Process {
@Override
public void end(int statusCode) {
terminate(statusCode, null);
end(statusCode, null);
}
@Override
public void end(int statusCode, String message) {
terminate(statusCode, null, message);
}
@Override
public boolean isRunning() {
return processStatus == ExecStatus.RUNNING;
}
@Override
public void appendResult(ResultModel result) {
if (processStatus != ExecStatus.RUNNING) {
throw new IllegalStateException(
"Cannot write to standard output when " + status().name().toLowerCase());
}
ProcessImpl.this.appendResult(result);
}
}
static class ProcessOutput {
@@ -620,7 +653,10 @@ public class ProcessImpl implements Process {
private void write(String data) {
if (stdoutHandlerChain != null) {
for (Function<String, String> function : stdoutHandlerChain) {
//hotspot, reduce memory fragment (foreach/iterator)
int size = stdoutHandlerChain.size();
for (int i = 0; i < size; i++) {
Function<String, String> function = stdoutHandlerChain.get(i);
data = function.apply(data);
}
}
@@ -7,7 +7,7 @@ import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.term.Term;
import com.taobao.arthas.core.shell.term.TermServer;
import com.taobao.arthas.core.shell.term.impl.http.NettyWebsocketTtyBootstrap;
import com.taobao.arthas.core.shell.term.impl.httptelnet.HttpTelnetTermServer;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.tty.TtyConnection;
@@ -25,11 +25,13 @@ public class HttpTermServer extends TermServer {
private String hostIp;
private int port;
private long connectionTimeout;
private EventExecutorGroup workerGroup;
public HttpTermServer(String hostIp, int port, long connectionTimeout) {
public HttpTermServer(String hostIp, int port, long connectionTimeout, EventExecutorGroup workerGroup) {
this.hostIp = hostIp;
this.port = port;
this.connectionTimeout = connectionTimeout;
this.workerGroup = workerGroup;
}
@Override
@@ -41,7 +43,7 @@ public class HttpTermServer extends TermServer {
@Override
public TermServer listen(Handler<Future<TermServer>> listenHandler) {
// TODO: charset and inputrc from options
bootstrap = new NettyWebsocketTtyBootstrap().setHost(hostIp).setPort(port);
bootstrap = new NettyWebsocketTtyBootstrap(workerGroup).setHost(hostIp).setPort(port);
try {
bootstrap.start(new Consumer<TtyConnection>() {
@Override
@@ -1,6 +1,7 @@
package com.taobao.arthas.core.shell.term.impl.http;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URL;
@@ -8,6 +9,7 @@ import java.net.URL;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.taobao.arthas.common.IOUtils;
import com.taobao.arthas.core.shell.term.impl.http.api.HttpApiHandler;
import com.taobao.arthas.core.shell.term.impl.httptelnet.HttpTelnetTermServer;
import io.netty.channel.ChannelFuture;
@@ -15,7 +17,6 @@ import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
@@ -27,9 +28,13 @@ import io.netty.handler.codec.http.LastHttpContent;
import io.termd.core.http.HttpTtyConnection;
import io.termd.core.util.Logging;
import static com.taobao.arthas.core.util.HttpUtils.createRedirectResponse;
import static com.taobao.arthas.core.util.HttpUtils.createResponse;
/**
* @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
* @author hengyunabc 2019-11-06
* @author gongdewei 2020-03-18
*/
public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
private static final Logger logger = LoggerFactory.getLogger(HttpTelnetTermServer.class);
@@ -38,10 +43,13 @@ public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequ
private File dir;
private HttpApiHandler httpApiHandler;
public HttpRequestHandler(String wsUri, File dir) {
this.wsUri = wsUri;
this.dir = dir;
dir.mkdirs();
this.httpApiHandler = HttpApiHandler.getInstance();
}
@Override
@@ -53,67 +61,105 @@ public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequ
send100Continue(ctx);
}
HttpResponse response = new DefaultHttpResponse(request.protocolVersion(),
HttpResponseStatus.INTERNAL_SERVER_ERROR);
HttpResponse response = null;
String path = new URI(request.uri()).getPath();
if ("/".equals(path)) {
path = "/index.html";
}
InputStream in = null;
boolean isHttpApiResponse = false;
try {
DefaultFullHttpResponse fileViewResult = DirectoryBrowser.view(dir, path, request.protocolVersion());
if (fileViewResult != null) {
response = fileViewResult;
} else {
URL res = HttpTtyConnection.class.getResource("/com/taobao/arthas/core/http" + path);
if (res != null) {
DefaultFullHttpResponse fullResp = new DefaultFullHttpResponse(request.protocolVersion(),
HttpResponseStatus.OK);
in = res.openStream();
byte[] tmp = new byte[256];
for (int l = 0; l != -1; l = in.read(tmp)) {
fullResp.content().writeBytes(tmp, 0, l);
}
int li = path.lastIndexOf('.');
if (li != -1 && li != path.length() - 1) {
String ext = path.substring(li + 1, path.length());
String contentType;
if ("html".equals(ext)) {
contentType = "text/html";
} else if ("js".equals(ext)) {
contentType = "application/javascript";
} else if ("css".equals(ext)) {
contentType = "text/css";
} else {
contentType = null;
}
if (contentType != null) {
fullResp.headers().set(HttpHeaderNames.CONTENT_TYPE, contentType);
}
}
response = fullResp;
} else {
response.setStatus(HttpResponseStatus.NOT_FOUND);
}
//handle http restful api
if ("/api".equals(path)) {
response = httpApiHandler.handle(request);
isHttpApiResponse = true;
}
//handle webui requests
if (path.equals("/ui")){
response = createRedirectResponse(request, "/ui/");
}
if (path.equals("/ui/")) {
path += "index.html";
}
//try classpath resource first
if (response == null){
response = readFileFromResource(request, path);
}
//try output dir later, avoid overlay classpath resources files
if (response == null){
response = DirectoryBrowser.view(dir, path, request.protocolVersion());
}
//not found
if (response == null){
response = createResponse(request, HttpResponseStatus.NOT_FOUND, "Not found");
}
} catch (Throwable e) {
logger.error("arthas process http request error: " + request.uri(), e);
} finally {
//If it is null, an error may occur
if (response == null){
response = createResponse(request, HttpResponseStatus.INTERNAL_SERVER_ERROR, "Server error");
}
ctx.write(response);
ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
future.addListener(ChannelFutureListener.CLOSE);
IOUtils.close(in);
//reuse http api response buf
if (isHttpApiResponse && response instanceof DefaultFullHttpResponse) {
final HttpResponse finalResponse = response;
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
httpApiHandler.onCompleted((DefaultFullHttpResponse) finalResponse);
}
});
}
}
}
}
private FullHttpResponse readFileFromResource(FullHttpRequest request, String path) throws IOException {
DefaultFullHttpResponse fullResp = null;
InputStream in = null;
try {
URL res = HttpTtyConnection.class.getResource("/com/taobao/arthas/core/http" + path);
if (res != null) {
fullResp = new DefaultFullHttpResponse(request.protocolVersion(),
HttpResponseStatus.OK);
in = res.openStream();
byte[] tmp = new byte[256];
for (int l = 0; l != -1; l = in.read(tmp)) {
fullResp.content().writeBytes(tmp, 0, l);
}
int li = path.lastIndexOf('.');
if (li != -1 && li != path.length() - 1) {
String ext = path.substring(li + 1, path.length());
String contentType;
if ("html".equals(ext)) {
contentType = "text/html";
} else if ("js".equals(ext)) {
contentType = "application/javascript";
} else if ("css".equals(ext)) {
contentType = "text/css";
} else {
contentType = null;
}
if (contentType != null) {
fullResp.headers().set(HttpHeaderNames.CONTENT_TYPE, contentType);
}
}
}
} finally {
IOUtils.close(in);
}
return fullResp;
}
private static void send100Continue(ChannelHandlerContext ctx) {
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);
ctx.writeAndFlush(response);
@@ -10,6 +10,7 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ImmediateEventExecutor;
@@ -30,8 +31,10 @@ public class NettyWebsocketTtyBootstrap {
private int port;
private EventLoopGroup group;
private Channel channel;
private EventExecutorGroup workerGroup;
public NettyWebsocketTtyBootstrap() {
public NettyWebsocketTtyBootstrap(EventExecutorGroup workerGroup) {
this.workerGroup = workerGroup;
this.host = "localhost";
this.port = 8080;
}
@@ -59,7 +62,7 @@ public class NettyWebsocketTtyBootstrap {
ServerBootstrap b = new ServerBootstrap();
b.group(group).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new TtyServerInitializer(channelGroup, handler));
.childHandler(new TtyServerInitializer(channelGroup, handler, workerGroup));
final ChannelFuture f = b.bind(host, port);
f.addListener(new GenericFutureListener<Future<? super Void>>() {
@@ -1,7 +1,5 @@
package com.taobao.arthas.core.shell.term.impl.http;
import java.io.File;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.group.ChannelGroup;
@@ -10,9 +8,12 @@ import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.tty.TtyConnection;
import java.io.File;
/**
* @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
@@ -21,19 +22,22 @@ public class TtyServerInitializer extends ChannelInitializer<SocketChannel> {
private final ChannelGroup group;
private final Consumer<TtyConnection> handler;
private EventExecutorGroup workerGroup;
public TtyServerInitializer(ChannelGroup group, Consumer<TtyConnection> handler) {
public TtyServerInitializer(ChannelGroup group, Consumer<TtyConnection> handler, EventExecutorGroup workerGroup) {
this.group = group;
this.handler = handler;
this.workerGroup = workerGroup;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(64 * 1024));
pipeline.addLast(new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(workerGroup, "HttpRequestHandler", new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
pipeline.addLast(new TtyWebSocketFrameHandler(group, handler));
}
@@ -0,0 +1,48 @@
package com.taobao.arthas.core.shell.term.impl.http.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,
/**
* Pull the results from result queue of the session
*/
PULL_RESULTS,
/**
* Create a new session
*/
INIT_SESSION,
/**
* Join a exist session
*/
JOIN_SESSION,
/**
* Terminate the session
*/
CLOSE_SESSION,
/**
* Get session info
*/
SESSION_INFO
}
@@ -0,0 +1,16 @@
package com.taobao.arthas.core.shell.term.impl.http.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);
}
}
@@ -0,0 +1,87 @@
package com.taobao.arthas.core.shell.term.impl.http.api;
import java.util.Map;
/**
* Http Api request
*
* @author gongdewei 2020-03-19
*/
public class ApiRequest {
private String action;
private String command;
private String requestId;
private String sessionId;
private String consumerId;
private Integer timeout;
private Map<String, Object> options;
@Override
public String toString() {
return "ApiRequest{" +
"action='" + action + '\'' +
", command='" + command + '\'' +
", requestId='" + requestId + '\'' +
", sessionId='" + sessionId + '\'' +
", consumerId='" + consumerId + '\'' +
", timeout=" + timeout +
", options=" + options +
'}';
}
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 Map<String, Object> getOptions() {
return options;
}
public void setOptions(Map<String, Object> options) {
this.options = options;
}
public String getRequestId() {
return requestId;
}
public void setRequestId(String requestId) {
this.requestId = requestId;
}
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 getTimeout() {
return timeout;
}
public void setTimeout(Integer timeout) {
this.timeout = timeout;
}
}
@@ -0,0 +1,79 @@
package com.taobao.arthas.core.shell.term.impl.http.api;
/**
* Http Api exception
* @author gongdewei 2020-03-19
*/
public class ApiResponse<T> {
private String requestId;
private ApiState state;
private String message;
private String sessionId;
private String consumerId;
private String jobId;
private T body;
public String getRequestId() {
return requestId;
}
public ApiResponse<T> setRequestId(String requestId) {
this.requestId = requestId;
return this;
}
public ApiState getState() {
return state;
}
public ApiResponse<T> setState(ApiState state) {
this.state = state;
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 String getJobId() {
return jobId;
}
public ApiResponse<T> setJobId(String jobId) {
this.jobId = jobId;
return this;
}
public T getBody() {
return body;
}
public ApiResponse<T> setBody(T body) {
this.body = body;
return this;
}
}
@@ -0,0 +1,35 @@
package com.taobao.arthas.core.shell.term.impl.http.api;
/**
* Http API response state
*
* @author gongdewei 2020-03-19
*/
public enum ApiState {
/**
* Scheduled async exec job
*/
SCHEDULED,
// RUNNING,
/**
* Request processed successfully
*/
SUCCEEDED,
/**
* Request processing interrupt
*/
INTERRUPTED,
/**
* Request processing failed
*/
FAILED,
/**
* Request is refused
*/
REFUSED
}
@@ -0,0 +1,708 @@
package com.taobao.arthas.core.shell.term.impl.http.api;
import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.taobao.arthas.common.PidUtils;
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.impl.PackingResultDistributorImpl;
import com.taobao.arthas.core.distribution.impl.ResultConsumerImpl;
import com.taobao.arthas.core.server.ArthasBootstrap;
import com.taobao.arthas.core.shell.cli.CliToken;
import com.taobao.arthas.core.shell.cli.CliTokens;
import com.taobao.arthas.core.shell.cli.Completion;
import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.history.HistoryManager;
import com.taobao.arthas.core.shell.history.impl.HistoryManagerImpl;
import com.taobao.arthas.core.shell.session.Session;
import com.taobao.arthas.core.shell.session.SessionManager;
import com.taobao.arthas.core.shell.system.Job;
import com.taobao.arthas.core.shell.system.JobController;
import com.taobao.arthas.core.shell.system.JobListener;
import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
import com.taobao.arthas.core.shell.term.SignalHandler;
import com.taobao.arthas.core.shell.term.Term;
import com.taobao.arthas.core.util.ArthasBanner;
import com.taobao.arthas.core.util.DateUtils;
import com.taobao.arthas.core.util.JsonUtils;
import com.taobao.arthas.core.util.StringUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;
import io.termd.core.function.Function;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Http Restful Api Handler
*
* @author gongdewei 2020-03-18
*/
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;
private final HistoryManager historyManager;
private int jsonBufferSize = 1024 * 256;
private int poolSize = 8;
private ArrayBlockingQueue<ByteBuf> byteBufPool = new ArrayBlockingQueue<ByteBuf>(poolSize);
private ArrayBlockingQueue<char[]> charsBufPool = new ArrayBlockingQueue<char[]>(poolSize);
private ArrayBlockingQueue<byte[]> bytesPool = new ArrayBlockingQueue<byte[]>(poolSize);
public static HttpApiHandler getInstance() {
if (instance == null) {
synchronized (HttpApiHandler.class) {
instance = new HttpApiHandler();
}
}
return instance;
}
private HttpApiHandler() {
sessionManager = ArthasBootstrap.getInstance().getSessionManager();
commandManager = sessionManager.getCommandManager();
jobController = sessionManager.getJobController();
historyManager = HistoryManagerImpl.getInstance();
//init buf pool
JsonUtils.setSerializeWriterBufferThreshold(jsonBufferSize);
for (int i = 0; i < poolSize; i++) {
byteBufPool.offer(Unpooled.buffer(jsonBufferSize));
charsBufPool.offer(new char[jsonBufferSize]);
bytesPool.offer(new byte[jsonBufferSize]);
}
}
public HttpResponse handle(FullHttpRequest request) throws Exception {
ApiResponse result;
String requestBody = null;
String requestId = "req_" + requestIdGenerator.addAndGet(1);
try {
HttpMethod method = request.method();
if (HttpMethod.POST.equals(method)) {
requestBody = getBody(request);
ApiRequest apiRequest = parseRequest(requestBody);
apiRequest.setRequestId(requestId);
result = processRequest(apiRequest);
} else {
result = createResponse(ApiState.REFUSED, "Unsupported http method: " + method.name());
}
} catch (Throwable e) {
result = createResponse(ApiState.FAILED, "Process request error: " + e.getMessage());
logger.error("arthas process http api request error: " + request.uri() + ", request body: " + requestBody, e);
}
if (result == null) {
result = createResponse(ApiState.FAILED, "The request was not processed");
}
result.setRequestId(requestId);
//http response content
ByteBuf content = null;
//fastjson buf
char[] charsBuf = null;
byte[] bytesBuf = null;
try {
//apply response content buf first
content = byteBufPool.poll(2000, TimeUnit.MILLISECONDS);
if (content == null) {
throw new ApiException("get response content buf failure");
}
//apply fastjson buf from pool
charsBuf = charsBufPool.poll();
bytesBuf = bytesPool.poll();
if (charsBuf == null || bytesBuf == null) {
throw new ApiException("get json buf failure");
}
JsonUtils.setSerializeWriterBufThreadLocal(charsBuf, bytesBuf);
//create http response
DefaultFullHttpResponse response = new DefaultFullHttpResponse(request.protocolVersion(),
HttpResponseStatus.OK, content.retain());
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=utf-8");
writeResult(response, result);
return response;
} catch (Exception e) {
//response is discarded
if (content != null) {
content.release();
byteBufPool.offer(content);
}
throw e;
} finally {
//give back json buf to pool
JsonUtils.setSerializeWriterBufThreadLocal(null, null);
if (charsBuf != null) {
charsBufPool.offer(charsBuf);
}
if (bytesBuf != null) {
bytesPool.offer(bytesBuf);
}
}
}
public void onCompleted(DefaultFullHttpResponse httpResponse) {
ByteBuf content = httpResponse.content();
content.clear();
if (content.capacity() == jsonBufferSize) {
if (!byteBufPool.offer(content)) {
content.release();
}
} else {
//replace content ByteBuf
content.release();
if (byteBufPool.remainingCapacity() > 0) {
byteBufPool.offer(Unpooled.buffer(jsonBufferSize));
}
}
}
private void writeResult(DefaultFullHttpResponse response, Object result) throws IOException {
ByteBufOutputStream out = new ByteBufOutputStream(response.content());
try {
JSON.writeJSONString(out, result);
} catch (IOException e) {
logger.error("write json to response failed", e);
throw e;
}
}
private ApiRequest parseRequest(String requestBody) throws ApiException {
if (StringUtils.isBlank(requestBody)) {
throw new ApiException("parse request failed: request body is empty");
}
try {
//ObjectMapper objectMapper = new ObjectMapper();
//return objectMapper.readValue(requestBody, ApiRequest.class);
return JSON.parseObject(requestBody, ApiRequest.class);
} catch (Exception e) {
throw new ApiException("parse request failed: " + e.getMessage(), e);
}
}
private ApiResponse processRequest(ApiRequest apiRequest) {
String actionStr = apiRequest.getAction();
try {
if (StringUtils.isBlank(actionStr)) {
throw new ApiException("'action' is required");
}
ApiAction action;
try {
action = ApiAction.valueOf(actionStr.trim().toUpperCase());
} catch (IllegalArgumentException e) {
throw new ApiException("unknown action: " + actionStr);
}
//no session required
if (ApiAction.INIT_SESSION.equals(action)) {
return processInitSessionRequest(apiRequest);
}
//required session
String sessionId = apiRequest.getSessionId();
if (StringUtils.isBlank(sessionId)) {
throw new ApiException("'sessionId' is required");
}
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);
if (response != null) {
return response;
}
} catch (ApiException e) {
logger.info("process http api request failed: {}", e.getMessage());
return createResponse(ApiState.FAILED, e.getMessage());
} catch (Throwable e) {
logger.error("process http api request failed: " + e.getMessage(), e);
return createResponse(ApiState.FAILED, "process http api request failed: " + e.getMessage());
}
return createResponse(ApiState.REFUSED, "Unsupported action: " + actionStr);
}
private ApiResponse dispatchRequest(ApiAction action, ApiRequest apiRequest, Session session) throws ApiException {
switch (action) {
case EXEC:
return processExecRequest(apiRequest, session);
case ASYNC_EXEC:
return processAsyncExecRequest(apiRequest, session);
case INTERRUPT_JOB:
return processInterruptJob(apiRequest, session);
case PULL_RESULTS:
return processPullResultsRequest(apiRequest, session);
case SESSION_INFO:
return processSessionInfoRequest(apiRequest, session);
case JOIN_SESSION:
return processJoinSessionRequest(apiRequest, session);
case CLOSE_SESSION:
return processCloseSessionRequest(apiRequest, session);
case INIT_SESSION:
break;
}
return null;
}
private ApiResponse processInitSessionRequest(ApiRequest apiRequest) throws ApiException {
ApiResponse response = new ApiResponse();
//create session
Session session = sessionManager.createSession();
if (session != null) {
//create consumer
ResultConsumer resultConsumer = new ResultConsumerImpl();
session.getResultDistributor().addConsumer(resultConsumer);
session.getResultDistributor().appendResult(new MessageModel("Welcome to arthas!"));
//welcome message
WelcomeModel welcomeModel = new WelcomeModel();
welcomeModel.setVersion(ArthasBanner.version());
welcomeModel.setWiki(ArthasBanner.wiki());
welcomeModel.setTutorials(ArthasBanner.tutorials());
welcomeModel.setPid(PidUtils.currentPid());
welcomeModel.setTime(DateUtils.getCurrentDate());
session.getResultDistributor().appendResult(welcomeModel);
//allow input
updateSessionInputStatus(session, InputStatus.ALLOW_INPUT);
response.setSessionId(session.getSessionId())
.setConsumerId(resultConsumer.getConsumerId())
.setState(ApiState.SUCCEEDED);
} else {
throw new ApiException("create api session failed");
}
return response;
}
/**
* Update session input status for all consumer
*
* @param session
* @param inputStatus
*/
private void updateSessionInputStatus(Session session, InputStatus inputStatus) {
session.getResultDistributor().appendResult(new InputStatusModel(inputStatus));
}
private ApiResponse processJoinSessionRequest(ApiRequest apiRequest, Session session) {
//create consumer
ResultConsumer resultConsumer = new ResultConsumerImpl();
//disable input and interrupt
resultConsumer.appendResult(new InputStatusModel(InputStatus.DISABLED));
session.getResultDistributor().addConsumer(resultConsumer);
ApiResponse response = new ApiResponse();
response.setSessionId(session.getSessionId())
.setConsumerId(resultConsumer.getConsumerId())
.setState(ApiState.SUCCEEDED);
return response;
}
private ApiResponse processSessionInfoRequest(ApiRequest apiRequest, Session session) {
ApiResponse response = new ApiResponse();
Map<String, Object> body = new TreeMap<String, Object>();
body.put("pid", session.getPid());
body.put("createTime", session.getCreateTime());
body.put("lastAccessTime", session.getLastAccessTime());
response.setState(ApiState.SUCCEEDED)
.setSessionId(session.getSessionId())
//.setConsumerId(consumerId)
.setBody(body);
return response;
}
private ApiResponse processCloseSessionRequest(ApiRequest apiRequest, Session session) {
sessionManager.removeSession(session.getSessionId());
ApiResponse response = new ApiResponse();
response.setState(ApiState.SUCCEEDED);
return response;
}
/**
* Execute command sync, wait for job finish or timeout, sending results immediately
*
* @param apiRequest
* @param session
* @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;
}
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;
}
//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);
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.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;
}
/**
* Execute command async, create and schedule the job running, but no wait for the results.
*
* @param apiRequest
* @param session
* @return
*/
private ApiResponse processAsyncExecRequest(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;
}
int lock = session.getLock();
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;
}
//create job
Job job = this.createJob(commandLine, session, session.getResultDistributor());
body.put("jobId", job.id());
body.put("jobStatus", job.status());
response.setState(ApiState.SCHEDULED);
//add command before exec job
CommandRequestModel commandRequestModel = new CommandRequestModel(commandLine, response.getState());
commandRequestModel.setJobId(job.id());
session.getResultDistributor().appendResult(commandRequestModel);
session.setForegroundJob(job);
updateSessionInputStatus(session, InputStatus.ALLOW_INTERRUPT);
//run job
job.run();
return response;
} catch (Throwable e) {
logger.error("Async exec command failed:" + e.getMessage() + ", command:" + commandLine, e);
response.setState(ApiState.FAILED).setMessage("Async exec command failed:" + e.getMessage());
CommandRequestModel commandRequestModel = new CommandRequestModel(commandLine, response.getState(), response.getMessage());
session.getResultDistributor().appendResult(commandRequestModel);
return response;
} finally {
if (session.getLock() == lock) {
session.unLock();
}
}
}
private ApiResponse processInterruptJob(ApiRequest apiRequest, Session session) {
Job job = session.getForegroundJob();
if (job == null) {
return new ApiResponse().setState(ApiState.FAILED).setMessage("no foreground job is running");
}
job.interrupt();
Map<String, Object> body = new TreeMap<String, Object>();
body.put("jobId", job.id());
body.put("jobStatus", job.status());
return new ApiResponse()
.setState(ApiState.SUCCEEDED)
.setBody(body);
}
/**
* Pull results from result queue
*
* @param apiRequest
* @param session
* @return
*/
private ApiResponse processPullResultsRequest(ApiRequest apiRequest, Session session) throws ApiException {
String consumerId = apiRequest.getConsumerId();
if (StringUtils.isBlank(consumerId)) {
throw new ApiException("'consumerId' is required");
}
ResultConsumer consumer = session.getResultDistributor().getConsumer(consumerId);
if (consumer == null) {
throw new ApiException("consumer not found: " + consumerId);
}
List<ResultModel> results = consumer.pollResults();
Map<String, Object> body = new TreeMap<String, Object>();
body.put("results", results);
ApiResponse response = new ApiResponse();
response.setState(ApiState.SUCCEEDED)
.setSessionId(session.getSessionId())
.setConsumerId(consumerId)
.setBody(body);
return response;
}
private boolean waitForJob(Job job, int timeout) {
long startTime = System.currentTimeMillis();
while (true) {
switch (job.status()) {
case STOPPED:
case TERMINATED:
return true;
}
if (System.currentTimeMillis() - startTime > timeout) {
return false;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
}
}
private synchronized Job createJob(List<CliToken> args, Session session, ResultDistributor resultDistributor) {
Job job = jobController.createJob(commandManager, args, session, new ApiJobHandler(session), new ApiTerm(session), resultDistributor);
return job;
}
private Job createJob(String line, Session session, ResultDistributor resultDistributor) {
historyManager.addHistory(line);
historyManager.saveHistory();
return createJob(CliTokens.tokenize(line), session, resultDistributor);
}
private ApiResponse createResponse(ApiState apiState, String message) {
ApiResponse apiResponse = new ApiResponse();
apiResponse.setState(apiState);
apiResponse.setMessage(message);
return apiResponse;
}
private String getBody(FullHttpRequest request) {
ByteBuf buf = request.content();
return buf.toString(CharsetUtil.UTF_8);
}
private class ApiJobHandler implements JobListener {
private Session session;
public ApiJobHandler(Session session) {
this.session = session;
}
@Override
public void onForeground(Job job) {
session.setForegroundJob(job);
}
@Override
public void onBackground(Job job) {
if (session.getForegroundJob() == job) {
session.setForegroundJob(null);
updateSessionInputStatus(session, InputStatus.ALLOW_INPUT);
}
}
@Override
public void onTerminated(Job job) {
if (session.getForegroundJob() == job) {
session.setForegroundJob(null);
updateSessionInputStatus(session, InputStatus.ALLOW_INPUT);
}
}
@Override
public void onSuspend(Job job) {
if (session.getForegroundJob() == job) {
session.setForegroundJob(null);
updateSessionInputStatus(session, InputStatus.ALLOW_INPUT);
}
}
}
private class ApiTerm implements Term {
private Session session;
public ApiTerm(Session session) {
this.session = session;
}
@Override
public Term resizehandler(Handler<Void> handler) {
return this;
}
@Override
public String type() {
return "web";
}
@Override
public int width() {
return 1000;
}
@Override
public int height() {
return 200;
}
@Override
public Term stdinHandler(Handler<String> handler) {
return this;
}
@Override
public Term stdoutHandler(Function<String, String> handler) {
return this;
}
@Override
public Term write(String data) {
return this;
}
@Override
public long lastAccessedTime() {
return session.getLastAccessTime();
}
@Override
public Term echo(String text) {
return this;
}
@Override
public Term setSession(Session session) {
return this;
}
@Override
public Term interruptHandler(SignalHandler handler) {
return this;
}
@Override
public Term suspendHandler(SignalHandler handler) {
return this;
}
@Override
public void readline(String prompt, Handler<String> lineHandler) {
}
@Override
public void readline(String prompt, Handler<String> lineHandler, Handler<Completion> completionHandler) {
}
@Override
public Term closeHandler(Handler<Void> handler) {
return this;
}
@Override
public void close() {
}
}
}
@@ -11,6 +11,7 @@ import com.taobao.arthas.core.shell.term.TermServer;
import com.taobao.arthas.core.shell.term.impl.Helper;
import com.taobao.arthas.core.shell.term.impl.TermImpl;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.tty.TtyConnection;
@@ -29,11 +30,13 @@ public class HttpTelnetTermServer extends TermServer {
private String hostIp;
private int port;
private long connectionTimeout;
private EventExecutorGroup workerGroup;
public HttpTelnetTermServer(String hostIp, int port, long connectionTimeout) {
public HttpTelnetTermServer(String hostIp, int port, long connectionTimeout, EventExecutorGroup workerGroup) {
this.hostIp = hostIp;
this.port = port;
this.connectionTimeout = connectionTimeout;
this.workerGroup = workerGroup;
}
@Override
@@ -45,7 +48,7 @@ public class HttpTelnetTermServer extends TermServer {
@Override
public TermServer listen(Handler<Future<TermServer>> listenHandler) {
// TODO: charset and inputrc from options
bootstrap = new NettyHttpTelnetTtyBootstrap().setHost(hostIp).setPort(port);
bootstrap = new NettyHttpTelnetTtyBootstrap(workerGroup).setHost(hostIp).setPort(port);
try {
bootstrap.start(new Consumer<TtyConnection>() {
@Override
@@ -11,6 +11,7 @@ import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ImmediateEventExecutor;
@@ -28,8 +29,10 @@ public class NettyHttpTelnetBootstrap extends TelnetBootstrap {
private EventLoopGroup group;
private ChannelGroup channelGroup;
private EventExecutorGroup workerGroup;
public NettyHttpTelnetBootstrap() {
public NettyHttpTelnetBootstrap(EventExecutorGroup workerGroup) {
this.workerGroup = workerGroup;
this.group = new NioEventLoopGroup();
this.channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
}
@@ -56,7 +59,7 @@ public class NettyHttpTelnetBootstrap extends TelnetBootstrap {
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ProtocolDetectHandler(channelGroup, handlerFactory, factory));
ch.pipeline().addLast(new ProtocolDetectHandler(channelGroup, handlerFactory, factory, workerGroup));
}
});
@@ -2,6 +2,7 @@ package com.taobao.arthas.core.shell.term.impl.httptelnet;
import java.nio.charset.Charset;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.function.Supplier;
import io.termd.core.telnet.TelnetHandler;
@@ -21,8 +22,8 @@ public class NettyHttpTelnetTtyBootstrap {
private boolean inBinary;
private Charset charset = Charset.forName("UTF-8");
public NettyHttpTelnetTtyBootstrap() {
this.httpTelnetTtyBootstrap = new NettyHttpTelnetBootstrap();
public NettyHttpTelnetTtyBootstrap(EventExecutorGroup workerGroup) {
this.httpTelnetTtyBootstrap = new NettyHttpTelnetBootstrap(workerGroup);
}
public String getHost() {
@@ -4,8 +4,8 @@ import java.io.File;
import java.util.concurrent.TimeUnit;
import com.taobao.arthas.core.shell.term.impl.http.HttpRequestHandler;
import com.taobao.arthas.core.shell.term.impl.http.TtyWebSocketFrameHandler;
import com.taobao.arthas.core.shell.term.impl.http.TtyWebSocketFrameHandler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -15,6 +15,7 @@ import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.ScheduledFuture;
import io.termd.core.function.Consumer;
import io.termd.core.function.Supplier;
@@ -31,12 +32,14 @@ public class ProtocolDetectHandler extends ChannelInboundHandlerAdapter {
private ChannelGroup channelGroup;
private Supplier<TelnetHandler> handlerFactory;
private Consumer<TtyConnection> ttyConnectionFactory;
private EventExecutorGroup workerGroup;
public ProtocolDetectHandler(ChannelGroup channelGroup, final Supplier<TelnetHandler> handlerFactory,
Consumer<TtyConnection> ttyConnectionFactory) {
Consumer<TtyConnection> ttyConnectionFactory, EventExecutorGroup workerGroup) {
this.channelGroup = channelGroup;
this.handlerFactory = handlerFactory;
this.ttyConnectionFactory = ttyConnectionFactory;
this.workerGroup = workerGroup;
}
private ScheduledFuture<?> detectTelnetFuture;
@@ -82,7 +85,7 @@ public class ProtocolDetectHandler extends ChannelInboundHandlerAdapter {
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(64 * 1024));
pipeline.addLast(new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(workerGroup, "HttpRequestHandler", new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
pipeline.addLast(new TtyWebSocketFrameHandler(channelGroup, ttyConnectionFactory));
ctx.fireChannelActive();
@@ -148,6 +148,55 @@ public class FileUtils {
return history;
}
/**
* save the command history to the given file, data will be overridden.
* @param history the command history
* @param file the file to save the history
*/
public static void saveCommandHistoryString(List<String> history, File file) {
OutputStream out = null;
try {
out = new BufferedOutputStream(openOutputStream(file, false));
for (String command: history) {
out.write(command.getBytes("utf-8"));
out.write('\n');
}
} catch (IOException e) {
// ignore
} finally {
try {
if (out != null) {
out.close();
}
} catch (IOException ioe) {
// ignore
}
}
}
public static List<String> loadCommandHistoryString(File file) {
BufferedReader br = null;
List<String> history = new ArrayList<String>();
try {
br = new BufferedReader(new InputStreamReader(new FileInputStream(file), "utf-8"));
String line;
while ((line = br.readLine()) != null) {
history.add(line);
}
} catch (IOException e) {
// ignore
} finally {
try {
if (br != null) {
br.close();
}
} catch (IOException ioe) {
// ignore
}
}
return history;
}
public static String readFileToString(File file, Charset encoding) throws IOException {
FileInputStream stream = new FileInputStream(file);
try {
@@ -0,0 +1,60 @@
package com.taobao.arthas.core.util;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.cookie.Cookie;
import io.netty.handler.codec.http.cookie.ServerCookieEncoder;
import java.io.UnsupportedEncodingException;
import java.util.Set;
/**
* @author gongdewei 2020/3/31
*/
public class HttpUtils {
/**
* Get cookie value by name
* @param cookies request cookies
* @param cookieName the cookie name
*/
public static String getCookieValue(Set<Cookie> cookies, String cookieName) {
for (Cookie cookie : cookies) {
if(cookie.name().equals(cookieName)){
return cookie.value();
}
}
return null;
}
/**
*
* @param response
* @param name
* @param value
*/
public static void setCookie(DefaultFullHttpResponse response, String name, String value) {
response.headers().add(HttpHeaderNames.SET_COOKIE, ServerCookieEncoder.STRICT.encode(name, value));
}
/**
* Create http response with status code and content
* @param request request
* @param status response status code
* @param content response content
*/
public static DefaultHttpResponse createResponse(FullHttpRequest request, HttpResponseStatus status, String content) {
DefaultFullHttpResponse response = new DefaultFullHttpResponse(request.protocolVersion(), status);
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=utf-8");
try {
response.content().writeBytes(content.getBytes("UTF-8"));
} catch (UnsupportedEncodingException e) {
}
return response;
}
public static HttpResponse createRedirectResponse(FullHttpRequest request, String url) {
DefaultFullHttpResponse response = new DefaultFullHttpResponse(request.protocolVersion(), HttpResponseStatus.FOUND);
response.headers().set(HttpHeaderNames.LOCATION, url);
return response;
}
}
@@ -0,0 +1,93 @@
package com.taobao.arthas.core.util;
import com.alibaba.fastjson.serializer.SerializeWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Field;
/**
* @author gongdewei 2020/5/15
*/
public class JsonUtils {
private static final Logger logger = LoggerFactory.getLogger(JsonUtils.class);
private static Field serializeWriterBufLocalField;
private static Field serializeWriterBytesBufLocal;
private static Field serializeWriterBufferThreshold;
/**
* Set Fastjson SerializeWriter Buffer Threshold
* @param value
*/
public static void setSerializeWriterBufferThreshold(int value) {
Class<SerializeWriter> clazz = SerializeWriter.class;
try {
if (serializeWriterBufferThreshold == null) {
serializeWriterBufferThreshold = clazz.getDeclaredField("BUFFER_THRESHOLD");
}
serializeWriterBufferThreshold.setAccessible(true);
serializeWriterBufferThreshold.set(null, value);
} catch (Throwable e) {
logger.error("update SerializeWriter.BUFFER_THRESHOLD value failed", e);
}
}
/**
* Set Fastjson SerializeWriter ThreadLocal value
* @param bufSize
*/
public static void setSerializeWriterBufThreadLocal(int bufSize) {
Class<SerializeWriter> clazz = SerializeWriter.class;
try {
//set threadLocal value
if (serializeWriterBufLocalField == null) {
serializeWriterBufLocalField = clazz.getDeclaredField("bufLocal");
}
serializeWriterBufLocalField.setAccessible(true);
ThreadLocal<char[]> bufLocal = (ThreadLocal<char[]>) serializeWriterBufLocalField.get(null);
char[] charsLocal = bufLocal.get();
if (charsLocal == null || charsLocal.length < bufSize) {
bufLocal.set(new char[bufSize]);
}
if (serializeWriterBytesBufLocal == null) {
serializeWriterBytesBufLocal = clazz.getDeclaredField("bytesBufLocal");
}
serializeWriterBytesBufLocal.setAccessible(true);
ThreadLocal<byte[]> bytesBufLocal = (ThreadLocal<byte[]>) serializeWriterBytesBufLocal.get(null);
byte[] bytesLocal = bytesBufLocal.get();
if (bytesLocal == null || bytesLocal.length < bufSize) {
bytesBufLocal.set(new byte[bufSize]);
}
} catch (Throwable e) {
logger.error("update SerializeWriter.BUFFER_THRESHOLD value failed", e);
}
}
/**
* Set Fastjson SerializeWriter ThreadLocal value
*/
public static void setSerializeWriterBufThreadLocal(char[] charsBuf, byte[] bytesBuf) {
Class<SerializeWriter> clazz = SerializeWriter.class;
try {
//set threadLocal value
if (serializeWriterBufLocalField == null) {
serializeWriterBufLocalField = clazz.getDeclaredField("bufLocal");
}
serializeWriterBufLocalField.setAccessible(true);
ThreadLocal<char[]> bufLocal = (ThreadLocal<char[]>) serializeWriterBufLocalField.get(null);
bufLocal.set(charsBuf);
if (serializeWriterBytesBufLocal == null) {
serializeWriterBytesBufLocal = clazz.getDeclaredField("bytesBufLocal");
}
serializeWriterBytesBufLocal.setAccessible(true);
ThreadLocal<byte[]> bytesBufLocal = (ThreadLocal<byte[]>) serializeWriterBytesBufLocal.get(null);
bytesBufLocal.set(bytesBuf);
} catch (Throwable e) {
logger.error("update SerializeWriter.BUFFER_THRESHOLD value failed", e);
}
}
}
@@ -103,7 +103,7 @@ public class StyledUsageFormatter extends UsageMessageFormatter {
return Style.style(Decoration.bold, fontColor);
}
private String computeUsageLine(String prefix, CLI cli) {
public String computeUsageLine(String prefix, CLI cli) {
// initialise the string buffer
StringBuilder buff;
if (prefix == null) {