查看入口
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;
}
}
調(diào)用鏈路為
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;
}
從代碼中可以看到，是通過一個(gè)異步的 httpClient 再結(jié)合 CountDownLatch 等待 5 秒的超時(shí)時(shí)間去獲取結(jié)果的。
獲取數(shù)據(jù)的請(qǐng)求從 dashboard 中發(fā)出去了正压，那 sentinel-core 中是怎么進(jìn)行相應(yīng)處理的呢崇猫？
sentinel-core 在啟動(dòng)的時(shí)候买猖，執(zhí)行了一個(gè) InitExecutor.init 的方法婆赠，該方法會(huì)觸發(fā)所有 InitFunc 實(shí)現(xiàn)類的 init 方法，其中就包括兩個(gè)最重要的實(shí)現(xiàn)類：
- HeartbeatSenderInitFunc
- CommandCenterInitFunc
CommandCenterInitFunc 則會(huì)啟動(dòng)一個(gè) CommandCenter 對(duì)外提供 sentinel-core 的數(shù)據(jù)服務(wù)渊迁，而這些數(shù)據(jù)服務(wù)是通過一個(gè)一個(gè)的 CommandHandler 來提供的，如下圖所示：
image.png
Sentinel-core的啟動(dòng)流程
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
關(guān)鍵代碼為
// 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 + '`');
}
心跳發(fā)送流程
HeartbeatSenderInitFunc 會(huì)啟動(dòng)一個(gè) HeartbeatSender 來定時(shí)的向 dashboard 發(fā)送自己的心跳包
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;
}
繼續(xù)深入
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());
}
}
小結(jié)
- sentinel-core 在初始化的時(shí)候灶挟，通過 JVM 參數(shù)中指定的 dashboard 的 ip 和 port琉朽，會(huì)主動(dòng)向 dashboard 發(fā)起連接的請(qǐng)求，該請(qǐng)求是通過 HeartbeatSender 接口以心跳的方式發(fā)送的稚铣，并將自己的 ip 和 port 告知 dashboard箱叁。這里 sentinel-core 上報(bào)給 dashboard 的端口是 sentinel 對(duì)外暴露的自己的 CommandCenter 的端口。
-
dashboard 在接收到 sentinel-core 的連接之后惕医，就會(huì)與 sentinel-core 建立連接耕漱，并將 sentinel-core 上報(bào)的 ip 和 port 的信息包裝成一個(gè) MachineInfo 對(duì)象，然后通過 SimpleMachineDiscovery 將該對(duì)象保存在一個(gè) map 中抬伺，如下圖所示：
image.png
dashboard 獲取到實(shí)時(shí)數(shù)據(jù)完整流程
首先 sentinel-core 向 dashboard 發(fā)送心跳包
dashboard 將 sentinel-core 的機(jī)器信息保存在內(nèi)存中
dashboard 根據(jù) sentinel-core 的機(jī)器信息通過 httpClient 獲取實(shí)時(shí)的數(shù)據(jù)
sentinel-core 接收到請(qǐng)求之后螟够，會(huì)找到具體的 CommandHandler 來處理
sentinel-core 將處理好的結(jié)果返回給 dashboard
