查看入口
image.png
代码为
@GetMapping("/rules")
@AuthAction(PrivilegeType.READ_RULE)
public Result<List<FlowRuleEntity>> apiQueryMachineRules(@RequestParam String app,
@RequestParam String ip,
@RequestParam Integer port) {
if (StringUtil.isEmpty(app)) {
return Result.ofFail(-1, "app can't be null or empty");
}
if (StringUtil.isEmpty(ip)) {
return Result.ofFail(-1, "ip can't be null or empty");
}
if (port == null) {
return Result.ofFail(-1, "port can't be null");
}
try {
List<FlowRuleEntity> rules = sentinelApiClient.fetchFlowRuleOfMachine(app, ip, port);
rules = repository.saveAll(rules);
return Result.ofSuccess(rules);
} catch (Throwable throwable) {
logger.error("Error when querying flow rules", throwable);
return Result.ofThrowable(-1, throwable);
}
}
fetchFlowRuleOfMachine
public List<FlowRuleEntity> fetchFlowRuleOfMachine(String app, String ip, int port) {
List<FlowRule> rules = fetchRules(ip, port, FLOW_RULE_TYPE, FlowRule.class);
if (rules != null) {
return rules.stream().map(rule -> FlowRuleEntity.fromFlowRule(app, ip, port, rule))
.collect(Collectors.toList());
} else {
return null;
}
}
调用链路为
com.alibaba.csp.sentinel.dashboard.client.SentinelApiClient#fetchRules
->com.alibaba.csp.sentinel.dashboard.client.SentinelApiClient#fetchItems
->com.alibaba.csp.sentinel.dashboard.client.SentinelApiClient#fetchItemsAsync
->com.alibaba.csp.sentinel.dashboard.client.SentinelApiClient#executeCommand()
private CompletableFuture<String> executeCommand(HttpUriRequest request) {
CompletableFuture<String> future = new CompletableFuture<>();
httpClient.execute(request, new FutureCallback<HttpResponse>() {
@Override
public void completed(final HttpResponse response) {
int statusCode = response.getStatusLine().getStatusCode();
try {
String value = getBody(response);
if (isSuccess(statusCode)) {
future.complete(value);
} else {
if (isCommandNotFound(statusCode, value)) {
future.completeExceptionally(new CommandNotFoundException(request.getURI().getPath()));
} else {
future.completeExceptionally(new CommandFailedException(value));
}
}
} catch (Exception ex) {
future.completeExceptionally(ex);
logger.error("HTTP request failed: {}", request.getURI().toString(), ex);
}
}
@Override
public void failed(final Exception ex) {
future.completeExceptionally(ex);
logger.error("HTTP request failed: {}", request.getURI().toString(), ex);
}
@Override
public void cancelled() {
future.complete(null);
}
});
return future;
}
从代码中可以看到，是通过一个异步的 httpClient 再结合 CountDownLatch 等待 5 秒的超时时间去获取结果的。
获取数据的请求从 dashboard 中发出去了，那 sentinel-core 中是怎么进行相应处理的呢？
sentinel-core 在启动的时候，执行了一个 InitExecutor.init 的方法，该方法会触发所有 InitFunc 实现类的 init 方法，其中就包括两个最重要的实现类：
- HeartbeatSenderInitFunc
- CommandCenterInitFunc
CommandCenterInitFunc 则会启动一个 CommandCenter 对外提供 sentinel-core 的数据服务，而这些数据服务是通过一个一个的 CommandHandler 来提供的，如下图所示：
image.png
Sentinel-core的启动流程
com.alibaba.csp.sentinel.SphU#entry()
->com.alibaba.csp.sentinel.Env#sph
->com.alibaba.csp.sentinel.init.InitExecutor#doInit
->com.alibaba.csp.sentinel.transport.init.HeartbeatSenderInitFunc#init
->com.alibaba.csp.sentinel.transport.init.CommandCenterInitFunc#init
CommandCenterInitFunc的初始化
public void init() throws Exception {
CommandCenter commandCenter = CommandCenterProvider.getCommandCenter();
if (commandCenter == null) {
RecordLog.warn("[CommandCenterInitFunc] Cannot resolve CommandCenter");
return;
}
commandCenter.beforeStart();
commandCenter.start();
RecordLog.info("[CommandCenterInit] Starting command center: "
+ commandCenter.getClass().getCanonicalName());
}
com.alibaba.csp.sentinel.transport.command.SimpleHttpCommandCenter#start
->com.alibaba.csp.sentinel.transport.command.SimpleHttpCommandCenter.ServerThread#run
->com.alibaba.csp.sentinel.transport.command.http.HttpEventTask#run
->com.alibaba.csp.sentinel.transport.command.SimpleHttpCommandCenter#getHandler
关键代码为
// Find the matching command handler.
CommandHandler<?> commandHandler = SimpleHttpCommandCenter.getHandler(commandName);
if (commandHandler != null) {
CommandResponse<?> response = commandHandler.handle(request);
handleResponse(response, printWriter);
} else {
// No matching command handler.
writeResponse(printWriter, StatusCode.BAD_REQUEST, "Unknown command `" + commandName + '`');
}
心跳发送流程
HeartbeatSenderInitFunc 会启动一个 HeartbeatSender 来定时的向 dashboard 发送自己的心跳包
public void init() {
HeartbeatSender sender = HeartbeatSenderProvider.getHeartbeatSender();
if (sender == null) {
RecordLog.warn("[HeartbeatSenderInitFunc] WARN: No HeartbeatSender loaded");
return;
}
initSchedulerIfNeeded();
long interval = retrieveInterval(sender);
setIntervalIfNotExists(interval);
scheduleHeartbeatTask(sender, interval);
}
scheduleHeartbeatTask
private void scheduleHeartbeatTask(/*@NonNull*/ final HeartbeatSender sender, /*@Valid*/ long interval) {
pool.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
sender.sendHeartbeat();
} catch (Throwable e) {
RecordLog.warn("[HeartbeatSender] Send heartbeat error", e);
}
}
}, 5000, interval, TimeUnit.MILLISECONDS);
RecordLog.info("[HeartbeatSenderInit] HeartbeatSender started: "
+ sender.getClass().getCanonicalName());
}
com.alibaba.csp.sentinel.transport.heartbeat.SimpleHttpHeartbeatSender#sendHeartbeat
public boolean sendHeartbeat() throws Exception {
if (TransportConfig.getRuntimePort() <= 0) {
RecordLog.info("[SimpleHttpHeartbeatSender] Command server port not initialized, won't send heartbeat");
return false;
}
Endpoint addrInfo = getAvailableAddress();
if (addrInfo == null) {
return false;
}
SimpleHttpRequest request = new SimpleHttpRequest(addrInfo, TransportConfig.getHeartbeatApiPath());
request.setParams(heartBeat.generateCurrentMessage());
try {
SimpleHttpResponse response = httpClient.post(request);
if (response.getStatusCode() == OK_STATUS) {
return true;
} else if (clientErrorCode(response.getStatusCode()) || serverErrorCode(response.getStatusCode())) {
RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addrInfo
+ ", http status code: " + response.getStatusCode());
}
} catch (Exception e) {
RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addrInfo, e);
}
return false;
}
继续深入
com.alibaba.csp.sentinel.transport.heartbeat.SimpleHttpHeartbeatSender#sendHeartbeat
->com.alibaba.csp.sentinel.transport.heartbeat.SimpleHttpHeartbeatSender#getAvailableAddress
->com.alibaba.csp.sentinel.transport.heartbeat.SimpleHttpHeartbeatSender#SimpleHttpHeartbeatSender
-> com.alibaba.csp.sentinel.transport.config.TransportConfig#getConsoleServerList
image.png
com.alibaba.csp.sentinel.dashboard.controller.MachineRegistryController#receiveHeartBeat代码如下
@ResponseBody
@RequestMapping("/machine")
public Result<?> receiveHeartBeat(String app, @RequestParam(value = "app_type", required = false, defaultValue = "0") Integer appType, Long version, String v, String hostname, String ip, Integer port) {
if (app == null) {
app = MachineDiscovery.UNKNOWN_APP_NAME;
}
if (ip == null) {
return Result.ofFail(-1, "ip can't be null");
}
if (port == null) {
return Result.ofFail(-1, "port can't be null");
}
if (port == -1) {
logger.info("Receive heartbeat from " + ip + " but port not set yet");
return Result.ofFail(-1, "your port not set yet");
}
String sentinelVersion = StringUtil.isEmpty(v) ? "unknown" : v;
version = version == null ? System.currentTimeMillis() : version;
try {
MachineInfo machineInfo = new MachineInfo();
machineInfo.setApp(app);
machineInfo.setAppType(appType);
machineInfo.setHostname(hostname);
machineInfo.setIp(ip);
machineInfo.setPort(port);
machineInfo.setHeartbeatVersion(version);
machineInfo.setLastHeartbeat(System.currentTimeMillis());
machineInfo.setVersion(sentinelVersion);
appManagement.addMachine(machineInfo);
return Result.ofSuccessMsg("success");
} catch (Exception e) {
logger.error("Receive heartbeat error", e);
return Result.ofFail(-1, e.getMessage());
}
}
小结
- sentinel-core 在初始化的时候，通过 JVM 参数中指定的 dashboard 的 ip 和 port，会主动向 dashboard 发起连接的请求，该请求是通过 HeartbeatSender 接口以心跳的方式发送的，并将自己的 ip 和 port 告知 dashboard。这里 sentinel-core 上报给 dashboard 的端口是 sentinel 对外暴露的自己的 CommandCenter 的端口。
-
dashboard 在接收到 sentinel-core 的连接之后，就会与 sentinel-core 建立连接，并将 sentinel-core 上报的 ip 和 port 的信息包装成一个 MachineInfo 对象，然后通过 SimpleMachineDiscovery 将该对象保存在一个 map 中，如下图所示：
image.png
dashboard 获取到实时数据完整流程
首先 sentinel-core 向 dashboard 发送心跳包
dashboard 将 sentinel-core 的机器信息保存在内存中
dashboard 根据 sentinel-core 的机器信息通过 httpClient 获取实时的数据
sentinel-core 接收到请求之后，会找到具体的 CommandHandler 来处理
sentinel-core 将处理好的结果返回给 dashboard
