Skip to content

Commit

Permalink
Merge pull request #988 from weibocom/feat/print_trace_log
Browse files Browse the repository at this point in the history 
print access log according to the attachment settings
  • Loading branch information
@rayzhang0603
rayzhang0603 authored Aug 2, 2022
2 parents 15a2d38 + 4128c31 commit 4f1f473
Show file tree
Hide file tree
Showing 9 changed files with 262 additions and 100 deletions.
4 changes: 4 additions & 0 deletions motan-core/src/main/java/com/weibo/api/motan/common/MotanConstants.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,7 @@ public class MotanConstants {
public static final String M2_ERROR = "M_e";
public static final String M2_PROCESS_TIME = "M_pt";

// ------------------ request trace point constants -----------------
public static final String TRACE_INVOKE = "TRACE_INVOKE"; //client 发起请求
public static final String TRACE_CONNECTION = "TRACE_CONNECTION"; // client获取链接
public static final String TRACE_CENCODE = "TRACE_CENCODE"; //client编码
Expand All @@ -157,6 +158,9 @@ public class MotanConstants {
public static final String TRACE_CRECEIVE = "TRACE_CRECEIVE";// client端接收response
public static final String TRACE_CDECODE = "TRACE_CDECODE"; // client端解码response

// ------------------ attachment constants -----------------
public static final String ATT_PRINT_TRACE_LOG = "print_trace_log"; // 针对单请求是否打印（access）日志

private MotanConstants() {
}

Expand Down
73 changes: 56 additions & 17 deletions motan-core/src/main/java/com/weibo/api/motan/filter/AccessLogFilter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,8 @@
import com.weibo.api.motan.common.URLParamType;
import com.weibo.api.motan.core.extension.Activation;
import com.weibo.api.motan.core.extension.SpiMeta;
import com.weibo.api.motan.rpc.Caller;
import com.weibo.api.motan.rpc.Provider;
import com.weibo.api.motan.rpc.Request;
import com.weibo.api.motan.rpc.Response;
import com.weibo.api.motan.util.LoggerUtil;
import com.weibo.api.motan.util.MotanSwitcherUtil;
import com.weibo.api.motan.util.NetUtils;
import com.weibo.api.motan.util.StringTools;
import com.weibo.api.motan.rpc.*;
import com.weibo.api.motan.util.*;
import org.apache.commons.lang3.StringUtils;

/**
Expand All @@ -47,36 +41,78 @@
public class AccessLogFilter implements Filter {

public static final String ACCESS_LOG_SWITCHER_NAME = "feature.motan.filter.accessLog";
public static final String PRINT_TRACE_LOG_SWITCHER_NAME = "feature.motan.printTraceLog.enable";
private String side;
private Boolean accessLog;

static {
// init global switcher， default value is false
// init global switcher
MotanSwitcherUtil.initSwitcher(ACCESS_LOG_SWITCHER_NAME, false);
MotanSwitcherUtil.initSwitcher(PRINT_TRACE_LOG_SWITCHER_NAME, true);
}

@Override
public Response filter(Caller<?> caller, Request request) {
if (accessLog == null) {
accessLog = caller.getUrl().getBooleanParameter(URLParamType.accessLog.getName(), URLParamType.accessLog.getBooleanValue());
}
if (accessLog || MotanSwitcherUtil.isOpen(ACCESS_LOG_SWITCHER_NAME)) {
long t1 = System.currentTimeMillis();
if (accessLog || needLog(request)) {
long start = System.currentTimeMillis();
boolean success = false;
Response response = null;
try {
Response response = caller.call(request);
success = true;
response = caller.call(request);
if (response != null && response.getException() == null) {
success = true;
}
return response;
} finally {
long consumeTime = System.currentTimeMillis() - t1;
logAccess(caller, request, consumeTime, success);
processFinalLog(caller, request, response, start, success);
}
} else {
return caller.call(request);
}
}

private void logAccess(Caller<?> caller, Request request, long consumeTime, boolean success) {
private void processFinalLog(final Caller<?> caller, final Request request, final Response response, final long start, final boolean success) {
long wholeTime = System.currentTimeMillis() - start;
long segmentTime = wholeTime; // 分段耗时。server侧是内部业务耗时，client侧时server整体耗时+网络接收耗时

if (request instanceof Traceable && response instanceof Traceable) { // 可以取得细分时间点
if (caller instanceof Provider) { // server end
if (response instanceof Callbackable) {// 因server侧完整耗时包括response发送时间，需要通过callback机制异步记录日志。
long finalSegmentTime = segmentTime;
((Callbackable) response).addFinishCallback(() -> {
long responseSend = ((Traceable) response).getTraceableContext().getSendTime();
long requestReceive = ((Traceable) request).getTraceableContext().getReceiveTime();
long finalWholeTime = responseSend - requestReceive;
logAccess(caller, request, response, finalSegmentTime, finalWholeTime, success);
}, null);
return;
}
} else { // client end
long requestSend = ((Traceable) request).getTraceableContext().getSendTime();
long responseReceive = ((Traceable) response).getTraceableContext().getReceiveTime();
segmentTime = responseReceive - requestSend;
}
}
logAccess(caller, request, response, segmentTime, wholeTime, success); // 同步记录access日志
}

// 除了access log配置外，其他需要动态打印access的情况
private boolean needLog(Request request) {
if (MotanSwitcherUtil.isOpen(ACCESS_LOG_SWITCHER_NAME)) {
return true;
}

// check trace log
if (!MotanSwitcherUtil.isOpen(PRINT_TRACE_LOG_SWITCHER_NAME)) {
return false;
}
return "true".equalsIgnoreCase(request.getAttachments().get(MotanConstants.ATT_PRINT_TRACE_LOG));
}

private void logAccess(Caller<?> caller, Request request, Response response, long segmentTime, long wholeTime, boolean success) {
if (getSide() == null) {
String side = caller instanceof Provider ? MotanConstants.NODE_TYPE_SERVICE : MotanConstants.NODE_TYPE_REFERER;
setSide(side);
Expand Down Expand Up @@ -107,7 +143,10 @@ private void logAccess(Caller<?> caller, Request request, long consumeTime, bool
requestId = String.valueOf(request.getRequestId());
}
append(builder, requestId);
append(builder, consumeTime);
append(builder, request.getAttachments().get(MotanConstants.CONTENT_LENGTH));
append(builder, response.getAttachments().get(MotanConstants.CONTENT_LENGTH));
append(builder, segmentTime);
append(builder, wholeTime);

LoggerUtil.accessLog(builder.substring(0, builder.length() - 1));
}
Expand Down
1 change: 0 additions & 1 deletion motan-core/src/main/java/com/weibo/api/motan/rpc/DefaultProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,6 @@ public Response invoke(Request request) {
ExceptionUtil.setMockStackTrace(response.getException().getCause());
}
}
response.setAttachments(request.getAttachments());
return response;
}
}
33 changes: 24 additions & 9 deletions motan-core/src/main/java/com/weibo/api/motan/rpc/DefaultResponse.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@

package com.weibo.api.motan.rpc;

import com.weibo.api.motan.core.DefaultThreadFactory;
import com.weibo.api.motan.core.StandardThreadExecutor;
import com.weibo.api.motan.exception.MotanServiceException;
import com.weibo.api.motan.protocol.rpc.RpcProtocolVersion;
import com.weibo.api.motan.util.LoggerUtil;
Expand All @@ -24,8 +26,12 @@
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.weibo.api.motan.core.StandardThreadExecutor.DEFAULT_MAX_IDLE_TIME;
import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
* Response received via rpc.
*
Expand All @@ -34,6 +40,9 @@
*/
public class DefaultResponse implements Response, Traceable, Callbackable, Serializable {
private static final long serialVersionUID = 4281186647291615871L;
protected static ThreadPoolExecutor defaultCallbackExecutor = new StandardThreadExecutor(20, 200,
DEFAULT_MAX_IDLE_TIME, MILLISECONDS, 5000,
new DefaultThreadFactory("defaultResponseCallbackPool-", true), new ThreadPoolExecutor.DiscardPolicy());

private Object value;
private Exception exception;
Expand All @@ -42,7 +51,7 @@ public class DefaultResponse implements Response, Traceable, Callbackable, Seria
private int timeout;
private Map<String, String> attachments;// rpc协议版本兼容时可以回传一些额外的信息
private byte rpcProtocolVersion = RpcProtocolVersion.VERSION_1.getVersion();
private int serializeNumber = 0;// default serialization is hession2
private int serializeNumber = 0;// default serialization is hessian2
private List<Pair<Runnable, Executor>> taskList = new ArrayList<>();
private AtomicBoolean isFinished = new AtomicBoolean();
private TraceableContext traceableContext = new TraceableContext();
Expand Down Expand Up @@ -75,6 +84,7 @@ public DefaultResponse(Object value) {

public DefaultResponse(Object value, long requestId) {
this.value = value;
this.requestId = requestId;
}

@Override
Expand Down Expand Up @@ -126,7 +136,7 @@ public int getTimeout() {

@Override
public Map<String, String> getAttachments() {
return attachments != null ? attachments : Collections.<String, String>emptyMap();
return attachments != null ? attachments : Collections.emptyMap();
}

public void setAttachments(Map<String, String> attachments) {
Expand Down Expand Up @@ -162,6 +172,12 @@ public int getSerializeNumber() {
return serializeNumber;
}

/**
* 未指定线程池时，统一使用默认线程池执行。默认线程池满时采用丢弃策略，不保证任务一定会被执行。
* 如果默认线程池不满足需求时，可以自行携带executor。
* @param runnable 准备在response on finish时执行的任务
* @param executor 指定执行任务的线程池
*/
public void addFinishCallback(Runnable runnable, Executor executor) {
if (!isFinished.get()) {
taskList.add(Pair.of(runnable, executor));
Expand All @@ -177,13 +193,12 @@ public void onFinish() {
Runnable runnable = pair.getKey();
Executor executor = pair.getValue();
if (executor == null) {
runnable.run();
} else {
try {
executor.execute(runnable);
} catch (Exception e) {
LoggerUtil.error("Callbackable response exec callback task error, e: ", e);
}
executor = defaultCallbackExecutor;
}
try {
executor.execute(runnable);
} catch (Exception e) {
LoggerUtil.error("Callbackable response exec callback task error, e: ", e);
}
}
}
Expand Down
Loading

0 comments on commit 4f1f473

Please sign in to comment.