crabapples
/
Sentinel
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Polish code of transport command centers and heartbeat senders 

Signed-off-by: Eric Zhao <sczyh16@gmail.com>
This commit is contained in:
Eric Zhao 2020-03-14 20:13:21 +08:00
parent 0536fb6846
commit 22df09b427
5 changed files with 73 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
sentinel-transport/sentinel-transport-common/src/main/java/com/alibaba/csp/sentinel/transport/config/TransportConfig.java
View File

@ -25,7 +25,8 @@ import java.util.ArrayList;
import java.util.List; import java.util.List;
/** /**
* @author leyou * @author Carpenter Lee
* @author Jason Joo
*/ */
public class TransportConfig { public class TransportConfig {
@ -66,7 +67,6 @@ public class TransportConfig {
String config = SentinelConfig.getConfig(CONSOLE_SERVER); String config = SentinelConfig.getConfig(CONSOLE_SERVER);
List<Tuple2<String, Integer>> list = new ArrayList<Tuple2<String, Integer>>(); List<Tuple2<String, Integer>> list = new ArrayList<Tuple2<String, Integer>>();
if (StringUtil.isBlank(config)) { if (StringUtil.isBlank(config)) {
RecordLog.warn("Dashboard server address is not configured");
return list; return list;
} }

33
sentinel-transport/sentinel-transport-netty-http/src/main/java/com/alibaba/csp/sentinel/transport/heartbeat/HttpHeartbeatSender.java
View File

@ -26,6 +26,7 @@ import com.alibaba.csp.sentinel.util.HostNameUtil;
import com.alibaba.csp.sentinel.util.PidUtil; import com.alibaba.csp.sentinel.util.PidUtil;
import com.alibaba.csp.sentinel.util.StringUtil; import com.alibaba.csp.sentinel.util.StringUtil;
import com.alibaba.csp.sentinel.util.function.Tuple2; import com.alibaba.csp.sentinel.util.function.Tuple2;
import org.apache.http.client.config.RequestConfig; import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet; import org.apache.http.client.methods.HttpGet;
@ -37,7 +38,7 @@ import java.util.List;
/** /**
* @author Eric Zhao * @author Eric Zhao
* @author leyou * @author Carpenter Lee
*/ */
@SpiOrder(SpiOrder.LOWEST_PRECEDENCE - 100) @SpiOrder(SpiOrder.LOWEST_PRECEDENCE - 100)
public class HttpHeartbeatSender implements HeartbeatSender { public class HttpHeartbeatSender implements HeartbeatSender {
@ -46,7 +47,6 @@ public class HttpHeartbeatSender implements HeartbeatSender {
private static final int OK_STATUS = 200; private static final int OK_STATUS = 200;
private final int timeoutMs = 3000; private final int timeoutMs = 3000;
private final RequestConfig requestConfig = RequestConfig.custom() private final RequestConfig requestConfig = RequestConfig.custom()
.setConnectionRequestTimeout(timeoutMs) .setConnectionRequestTimeout(timeoutMs)
@ -54,18 +54,21 @@ public class HttpHeartbeatSender implements HeartbeatSender {
.setSocketTimeout(timeoutMs) .setSocketTimeout(timeoutMs)
.build(); .build();
private String consoleHost; private final String consoleHost;
private int consolePort; private final int consolePort;
public HttpHeartbeatSender() { public HttpHeartbeatSender() {
this.client = HttpClients.createDefault(); this.client = HttpClients.createDefault();
List<Tuple2<String, Integer>> dashboardList = TransportConfig.getConsoleServerList(); List<Tuple2<String, Integer>> dashboardList = TransportConfig.getConsoleServerList();
if (dashboardList == null || dashboardList.isEmpty()) { if (dashboardList == null || dashboardList.isEmpty()) {
RecordLog.info("[NettyHttpHeartbeatSender] No dashboard available"); RecordLog.info("[NettyHttpHeartbeatSender] No dashboard server available");
consoleHost = null;
consolePort = -1;
} else { } else {
consoleHost = dashboardList.get(0).r1; consoleHost = dashboardList.get(0).r1;
consolePort = dashboardList.get(0).r2; consolePort = dashboardList.get(0).r2;
RecordLog.info("[NettyHttpHeartbeatSender] Dashboard address parsed: <" + consoleHost + ':' + consolePort + ">"); RecordLog.info(
"[NettyHttpHeartbeatSender] Dashboard address parsed: <" + consoleHost + ':' + consolePort + ">");
} }
} }
@ -96,12 +99,10 @@ public class HttpHeartbeatSender implements HeartbeatSender {
return true; return true;
} else if (clientErrorCode(statusCode) || serverErrorCode(statusCode)) { } else if (clientErrorCode(statusCode) || serverErrorCode(statusCode)) {
RecordLog.warn("[HttpHeartbeatSender] Failed to send heartbeat to " RecordLog.warn("[HttpHeartbeatSender] Failed to send heartbeat to "
+ consoleHost + ":" + consolePort + ", http status code: {0}", statusCode); + consoleHost + ":" + consolePort + ", http status code: " + statusCode);
} }
return false; return false;
} }
@Override @Override
@ -109,25 +110,11 @@ public class HttpHeartbeatSender implements HeartbeatSender {
return 5000; return 5000;
} }
/**
* 4XX Client Error
*
* @param code
* @return
*/
private boolean clientErrorCode(int code) { private boolean clientErrorCode(int code) {
return code > 399 && code < 500; return code > 399 && code < 500;
} }
/**
* 5XX Server Error
*
* @param code
* @return
*/
private boolean serverErrorCode(int code) { private boolean serverErrorCode(int code) {
return code > 499 && code < 600; return code > 499 && code < 600;
} }
} }

90
sentinel-transport/sentinel-transport-simple-http/src/main/java/com/alibaba/csp/sentinel/transport/command/http/HttpEventTask.java
View File

@ -20,7 +20,6 @@ import com.alibaba.csp.sentinel.command.CommandRequest;
import com.alibaba.csp.sentinel.command.CommandResponse; import com.alibaba.csp.sentinel.command.CommandResponse;
import com.alibaba.csp.sentinel.config.SentinelConfig; import com.alibaba.csp.sentinel.config.SentinelConfig;
import com.alibaba.csp.sentinel.log.CommandCenterLog; import com.alibaba.csp.sentinel.log.CommandCenterLog;
import com.alibaba.csp.sentinel.log.RecordLog;
import com.alibaba.csp.sentinel.transport.command.SimpleHttpCommandCenter; import com.alibaba.csp.sentinel.transport.command.SimpleHttpCommandCenter;
import com.alibaba.csp.sentinel.transport.command.exception.RequestException; import com.alibaba.csp.sentinel.transport.command.exception.RequestException;
import com.alibaba.csp.sentinel.transport.util.HttpCommandUtils; import com.alibaba.csp.sentinel.transport.util.HttpCommandUtils;
@ -41,15 +40,17 @@ import java.nio.charset.Charset;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
/**
/***
* The task handles incoming command request in HTTP protocol. * The task handles incoming command request in HTTP protocol.
* *
* @author youji.zj * @author youji.zj
* @author Eric Zhao * @author Eric Zhao
* @author Jason Joo
*/ */
public class HttpEventTask implements Runnable { public class HttpEventTask implements Runnable {
private static final String SERVER_ERROR_MESSAGE = "Command server error";
public static final String SERVER_ERROR_MESSAGE = "Command server error";
public static final String INVALID_COMMAND_MESSAGE = "Invalid command";
private final Socket socket; private final Socket socket;
@ -92,7 +93,7 @@ public class HttpEventTask implements Runnable {
// Validate the target command. // Validate the target command.
String commandName = HttpCommandUtils.getTarget(request); String commandName = HttpCommandUtils.getTarget(request);
if (StringUtil.isBlank(commandName)) { if (StringUtil.isBlank(commandName)) {
writeResponse(printWriter, StatusCode.BAD_REQUEST, "Invalid command"); writeResponse(printWriter, StatusCode.BAD_REQUEST, INVALID_COMMAND_MESSAGE);
return; return;
} }
@ -133,7 +134,7 @@ public class HttpEventTask implements Runnable {
closeResource(socket); closeResource(socket);
} }
} }
private static String readLine(InputStream in) throws IOException { private static String readLine(InputStream in) throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream(64); ByteArrayOutputStream bos = new ByteArrayOutputStream(64);
int data; int data;
@ -153,32 +154,32 @@ public class HttpEventTask implements Runnable {
} }
return new String(arr, SentinelConfig.charset()); return new String(arr, SentinelConfig.charset());
} }
/** /**
* Try to process the body of POST request additionally. * Try to process the body of POST request additionally.
* *
* @param in * @param in
* @param request * @param request
* @throws RequestException * @throws RequestException
* @throws IOException * @throws IOException
*/ */
protected static void processPostRequest(InputStream in, CommandRequest request) protected static void processPostRequest(InputStream in, CommandRequest request)
throws RequestException, IOException { throws RequestException, IOException {
Map<String, String> headerMap = parsePostHeaders(in); Map<String, String> headerMap = parsePostHeaders(in);
if (headerMap == null) { if (headerMap == null) {
// illegal request // illegal request
RecordLog.warn("Illegal request read"); CommandCenterLog.warn("Illegal request read: null headerMap");
throw new RequestException(StatusCode.BAD_REQUEST, ""); throw new RequestException(StatusCode.BAD_REQUEST, "");
} }
if (headerMap.containsKey("content-type") && !checkSupport(headerMap.get("content-type"))) { if (headerMap.containsKey("content-type") && !checkContentTypeSupported(headerMap.get("content-type"))) {
// not support Content-type // not supported Content-type
RecordLog.warn("Not supported Content-Type: {}", headerMap.get("content-type")); CommandCenterLog.warn("Request not supported: unsupported Content-Type: " + headerMap.get("content-type"));
throw new RequestException(StatusCode.UNSUPPORTED_MEDIA_TYPE, throw new RequestException(StatusCode.UNSUPPORTED_MEDIA_TYPE,
"Only form-encoded post request is supported"); "Only form-encoded post request is supported");
} }
int bodyLength = 0; int bodyLength = 0;
try { try {
bodyLength = Integer.parseInt(headerMap.get("content-length")); bodyLength = Integer.parseInt(headerMap.get("content-length"));
@ -186,19 +187,19 @@ public class HttpEventTask implements Runnable {
} }
if (bodyLength < 1) { if (bodyLength < 1) {
// illegal request without Content-length header // illegal request without Content-length header
RecordLog.warn("No available Content-Length in headers"); CommandCenterLog.warn("Request not supported: no available Content-Length in headers");
throw new RequestException(StatusCode.LENGTH_REQUIRED, "No legal Content-Length"); throw new RequestException(StatusCode.LENGTH_REQUIRED, "No legal Content-Length");
} }
parseParams(readBody(in, bodyLength), request); parseParams(readBody(in, bodyLength), request);
} }
/** /**
* Process header line in request * Process header line in request
* *
* @param in * @param in
* @return return headers in a Map, null for illegal request * @return return headers in a Map, null for illegal request
* @throws IOException * @throws IOException
*/ */
protected static Map<String, String> parsePostHeaders(InputStream in) throws IOException { protected static Map<String, String> parsePostHeaders(InputStream in) throws IOException {
Map<String, String> headerMap = new HashMap<String, String>(4); Map<String, String> headerMap = new HashMap<String, String>(4);
@ -221,8 +222,8 @@ public class HttpEventTask implements Runnable {
} }
} }
} }
private static boolean checkSupport(String contentType) { private static boolean checkContentTypeSupported(String contentType) {
int idx = contentType.indexOf(";"); int idx = contentType.indexOf(";");
String type; String type;
if (idx > 0) { if (idx > 0) {
@ -234,16 +235,15 @@ public class HttpEventTask implements Runnable {
// But some library do add it. So we will be compatible with that but force to // But some library do add it. So we will be compatible with that but force to
// encoding specified in configuration as legacy processing will do. // encoding specified in configuration as legacy processing will do.
if (!type.contains("application/x-www-form-urlencoded")) { if (!type.contains("application/x-www-form-urlencoded")) {
CommandCenterLog.warn("Content-Type not supported: " + contentType);
// Not supported request type // Not supported request type
// Now simple-http only support form-encoded post request. // Now simple-http only support form-encoded post request.
return false; return false;
} }
return true; return true;
} }
private static String readBody(InputStream in, int bodyLength) private static String readBody(InputStream in, int bodyLength)
throws IOException, RequestException { throws IOException, RequestException {
byte[] buf = new byte[bodyLength]; byte[] buf = new byte[bodyLength];
int pos = 0; int pos = 0;
while (pos < bodyLength) { while (pos < bodyLength) {
@ -259,10 +259,10 @@ public class HttpEventTask implements Runnable {
// Only allow partial // Only allow partial
return new String(buf, 0, pos, SentinelConfig.charset()); return new String(buf, 0, pos, SentinelConfig.charset());
} }
/** /**
* Consume all the body submitted and parse params into {@link CommandRequest} * Consume all the body submitted and parse params into {@link CommandRequest}
* *
* @param queryString * @param queryString
* @param request * @param request
*/ */
@ -270,12 +270,12 @@ public class HttpEventTask implements Runnable {
if (queryString == null || queryString.length() < 1) { if (queryString == null || queryString.length() < 1) {
return; return;
} }
int offset = 0, pos = -1; int offset = 0, pos = -1;
// check anchor // check anchor
queryString = removeAnchor(queryString); queryString = removeAnchor(queryString);
while (true) { while (true) {
offset = pos + 1; offset = pos + 1;
pos = queryString.indexOf('&', offset); pos = queryString.indexOf('&', offset);
@ -319,11 +319,11 @@ public class HttpEventTask implements Runnable {
writeResponse(printWriter, StatusCode.BAD_REQUEST, msg); writeResponse(printWriter, StatusCode.BAD_REQUEST, msg);
} }
} }
private void writeResponse(PrintWriter out, StatusCode statusCode, String message) { private void writeResponse(PrintWriter out, StatusCode statusCode, String message) {
out.print("HTTP/1.0 " + statusCode.toString() + "\r\n" out.print("HTTP/1.0 " + statusCode.toString() + "\r\n"
+ "Content-Length: " + (message == null ? 0 : message.getBytes().length) + "\r\n" + "Content-Length: " + (message == null ? 0 : message.getBytes().length) + "\r\n"
+ "Connection: close\r\n\r\n"); + "Connection: close\r\n\r\n");
if (message != null) { if (message != null) {
out.print(message); out.print(message);
} }
@ -354,10 +354,10 @@ public class HttpEventTask implements Runnable {
parseParams(parameterStr, request); parseParams(parameterStr, request);
return request; return request;
} }
/** /**
* Truncate query from "a=1&b=2#mark" to "a=1&b=2" * Truncate query from "a=1&b=2#mark" to "a=1&b=2"
* *
* @param str * @param str
* @return * @return
*/ */
@ -365,23 +365,23 @@ public class HttpEventTask implements Runnable {
if (str == null || str.length() == 0) { if (str == null || str.length() == 0) {
return str; return str;
} }
int anchor = str.indexOf('#'); int anchor = str.indexOf('#');
if (anchor == 0) { if (anchor == 0) {
return ""; return "";
} else if (anchor > 0) { } else if (anchor > 0) {
return str.substring(0, anchor); return str.substring(0, anchor);
} }
return str; return str;
} }
protected static void parseSingleParam(String single, CommandRequest request) { protected static void parseSingleParam(String single, CommandRequest request) {
if (single == null || single.length() < 3) { if (single == null || single.length() < 3) {
return; return;
} }
int index = single.indexOf('='); int index = single.indexOf('=');
if (index <= 0 || index >= single.length() - 1) { if (index <= 0 || index >= single.length() - 1) {
// empty key/val or nothing found // empty key/val or nothing found
@ -395,7 +395,7 @@ public class HttpEventTask implements Runnable {
value = URLDecoder.decode(value, SentinelConfig.charset()); value = URLDecoder.decode(value, SentinelConfig.charset());
} catch (UnsupportedEncodingException e) { } catch (UnsupportedEncodingException e) {
} }
request.addParam(key, value); request.addParam(key, value);
} }

8
sentinel-transport/sentinel-transport-simple-http/src/main/java/com/alibaba/csp/sentinel/transport/command/http/StatusCode.java
View File

@ -15,7 +15,13 @@
*/ */
package com.alibaba.csp.sentinel.transport.command.http; package com.alibaba.csp.sentinel.transport.command.http;
/**
* @author Jason Joo
*/
public enum StatusCode { public enum StatusCode {
/**
* 200 OK.
*/
OK(200, "OK"), OK(200, "OK"),
BAD_REQUEST(400, "Bad Request"), BAD_REQUEST(400, "Bad Request"),
REQUEST_TIMEOUT(408, "Request Timeout"), REQUEST_TIMEOUT(408, "Request Timeout"),
@ -27,7 +33,7 @@ public enum StatusCode {
private String desc; private String desc;
private String representation; private String representation;
private StatusCode(int code, String desc) { StatusCode(int code, String desc) {
this.code = code; this.code = code;
this.desc = desc; this.desc = desc;
this.representation = code + " " + desc; this.representation = code + " " + desc;

26
sentinel-transport/sentinel-transport-simple-http/src/main/java/com/alibaba/csp/sentinel/transport/heartbeat/SimpleHttpHeartbeatSender.java
View File

@ -31,7 +31,7 @@ import java.util.List;
* This implementation is based on a trivial HTTP client. * This implementation is based on a trivial HTTP client.
* *
* @author Eric Zhao * @author Eric Zhao
* @author leyou * @author Carpenter Lee
*/ */
public class SimpleHttpHeartbeatSender implements HeartbeatSender { public class SimpleHttpHeartbeatSender implements HeartbeatSender {
@ -49,14 +49,18 @@ public class SimpleHttpHeartbeatSender implements HeartbeatSender {
public SimpleHttpHeartbeatSender() { public SimpleHttpHeartbeatSender() {
// Retrieve the list of default addresses. // Retrieve the list of default addresses.
List<Tuple2<String, Integer>> newAddrs = TransportConfig.getConsoleServerList(); List<Tuple2<String, Integer>> newAddrs = TransportConfig.getConsoleServerList();
RecordLog.info("[SimpleHttpHeartbeatSender] Default console address list retrieved: " + newAddrs); if (newAddrs.isEmpty()) {
RecordLog.warn("[SimpleHttpHeartbeatSender] Dashboard server address not configured or not available");
} else {
RecordLog.info("[SimpleHttpHeartbeatSender] Default console address list retrieved: " + newAddrs);
}
this.addressList = newAddrs; this.addressList = newAddrs;
} }
@Override @Override
public boolean sendHeartbeat() throws Exception { public boolean sendHeartbeat() throws Exception {
if (TransportConfig.getRuntimePort() <= 0) { if (TransportConfig.getRuntimePort() <= 0) {
RecordLog.info("[SimpleHttpHeartbeatSender] Runtime port not initialized, won't send heartbeat"); RecordLog.info("[SimpleHttpHeartbeatSender] Command server port not initialized, won't send heartbeat");
return false; return false;
} }
Tuple2<String, Integer> addrInfo = getAvailableAddress(); Tuple2<String, Integer> addrInfo = getAvailableAddress();
@ -72,7 +76,8 @@ public class SimpleHttpHeartbeatSender implements HeartbeatSender {
if (response.getStatusCode() == OK_STATUS) { if (response.getStatusCode() == OK_STATUS) {
return true; return true;
} else if (clientErrorCode(response.getStatusCode()) || serverErrorCode(response.getStatusCode())) { } else if (clientErrorCode(response.getStatusCode()) || serverErrorCode(response.getStatusCode())) {
RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr + ", http status code: {0}", response.getStatusCode()); RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr
+ ", http status code: " + response.getStatusCode());
} }
} catch (Exception e) { } catch (Exception e) {
RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr, e); RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr, e);
@ -96,23 +101,10 @@ public class SimpleHttpHeartbeatSender implements HeartbeatSender {
return addressList.get(index); return addressList.get(index);
} }
/**
* 4XX Client Error
*
* @param code
* @return
*/
private boolean clientErrorCode(int code) { private boolean clientErrorCode(int code) {
return code > 399 && code < 500; return code > 399 && code < 500;
} }
/**
* 5XX Server Error
*
* @param code
* @return
*/
private boolean serverErrorCode(int code) { private boolean serverErrorCode(int code) {
return code > 499 && code < 600; return code > 499 && code < 600;
} }