Skip to content

Commit

Permalink
incorporated review comment of PR apache#3027
Browse files Browse the repository at this point in the history
  • Loading branch information
@khanimteyaz
khanimteyaz committed Dec 28, 2018
1 parent 09cbf8f commit b1ef5b7
Show file tree
Hide file tree
Showing 3 changed files with 389 additions and 112 deletions.
232 changes: 120 additions & 112 deletions dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/AccessLogFilter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadlocal.InternalThreadLocal;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.ConfigUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
Expand All @@ -29,11 +30,11 @@
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;

import com.alibaba.fastjson.JSON;
import org.apache.dubbo.rpc.support.AccessLogData;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
Expand All @@ -43,7 +44,6 @@
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
Expand All @@ -67,139 +67,147 @@ public class AccessLogFilter implements Filter {

private static final String ACCESS_LOG_KEY = "dubbo.accesslog";

private static final String FILE_DATE_FORMAT = "yyyyMMdd";

private static final String MESSAGE_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";

private static final int LOG_MAX_BUFFER = 5000;

private static final long LOG_OUTPUT_INTERVAL = 5000;

private final ConcurrentMap<String, Set<String>> logQueue = new ConcurrentHashMap<String, Set<String>>();
private static final SimpleDateFormat FILE_DATE_FORMATTER = new SimpleDateFormat("yyyyMMdd");

private final ScheduledExecutorService logScheduled = Executors.newScheduledThreadPool(2, new NamedThreadFactory("Dubbo-Access-Log", true));
private static final SimpleDateFormat MESSAGE_DATE_FORMATTER = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

private volatile ScheduledFuture<?> logFuture = null;
private final ConcurrentMap<String, Set<AccessLogData>> logQueue = new ConcurrentHashMap<String, Set<AccessLogData>>();

private void init() {
if (logFuture == null) {
synchronized (logScheduled) {
if (logFuture == null) {
logFuture = logScheduled.scheduleWithFixedDelay(new LogTask(), LOG_OUTPUT_INTERVAL, LOG_OUTPUT_INTERVAL, TimeUnit.MILLISECONDS);
}
}
}
}
private static final ScheduledExecutorService logScheduled = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-Access-Log", true));

private void log(String accesslog, String logmessage) {
init();
Set<String> logSet = logQueue.get(accesslog);
if (logSet == null) {
logQueue.putIfAbsent(accesslog, new ConcurrentHashSet<String>());
logSet = logQueue.get(accesslog);
}
if (logSet.size() < LOG_MAX_BUFFER) {
logSet.add(logmessage);
private static final InternalThreadLocal<SimpleDateFormat> INLINE_MESSAGE_DATE_FORMATTER = new InternalThreadLocal<SimpleDateFormat>() {
@Override
protected SimpleDateFormat initialValue() {
return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
}
};

/**
* Default constructor initialize demon thread for writing into access log file with names with access log key
* defined in url <b>accesslog</b>
*/
public AccessLogFilter() {
logScheduled.scheduleWithFixedDelay(this::writeLogToFile, LOG_OUTPUT_INTERVAL, LOG_OUTPUT_INTERVAL, TimeUnit.MILLISECONDS);
}

/**
* This method logs the access log for service method invocation call.
*
* @param invoker service
* @param inv Invocation service method.
* @return Result from service method.
* @throws RpcException
*/

@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
try {
String accesslog = invoker.getUrl().getParameter(Constants.ACCESS_LOG_KEY);
if (ConfigUtils.isNotEmpty(accesslog)) {
RpcContext context = RpcContext.getContext();
String serviceName = invoker.getInterface().getName();
String version = invoker.getUrl().getParameter(Constants.VERSION_KEY);
String group = invoker.getUrl().getParameter(Constants.GROUP_KEY);
StringBuilder sn = new StringBuilder();
sn.append("[").append(new SimpleDateFormat(MESSAGE_DATE_FORMAT).format(new Date())).append("] ").append(context.getRemoteHost()).append(":").append(context.getRemotePort())
.append(" -> ").append(context.getLocalHost()).append(":").append(context.getLocalPort())
.append(" - ");
if (null != group && group.length() > 0) {
sn.append(group).append("/");
}
sn.append(serviceName);
if (null != version && version.length() > 0) {
sn.append(":").append(version);
}
sn.append(" ");
sn.append(inv.getMethodName());
sn.append("(");
Class<?>[] types = inv.getParameterTypes();
if (types != null && types.length > 0) {
boolean first = true;
for (Class<?> type : types) {
if (first) {
first = false;
} else {
sn.append(",");
}
sn.append(type.getName());
}
}
sn.append(") ");
Object[] args = inv.getArguments();
if (args != null && args.length > 0) {
sn.append(JSON.toJSONString(args));
}
String msg = sn.toString();
if (ConfigUtils.isDefault(accesslog)) {
LoggerFactory.getLogger(ACCESS_LOG_KEY + "." + invoker.getInterface().getName()).info(msg);
} else {
log(accesslog, msg);
}
String accessLogKey = invoker.getUrl().getParameter(Constants.ACCESS_LOG_KEY);
if (ConfigUtils.isNotEmpty(accessLogKey)) {
AccessLogData logData = buildAccessLogData(invoker, inv);
log(accessLogKey, logData);
}
} catch (Throwable t) {
logger.warn("Exception in AcessLogFilter of service(" + invoker + " -> " + inv + ")", t);
}
return invoker.invoke(inv);
}

private class LogTask implements Runnable {
@Override
public void run() {
try {
if (logQueue != null && logQueue.size() > 0) {
for (Map.Entry<String, Set<String>> entry : logQueue.entrySet()) {
try {
String accesslog = entry.getKey();
Set<String> logSet = entry.getValue();
File file = new File(accesslog);
File dir = file.getParentFile();
if (null != dir && !dir.exists()) {
dir.mkdirs();
}
if (logger.isDebugEnabled()) {
logger.debug("Append log to " + accesslog);
}
if (file.exists()) {
String now = new SimpleDateFormat(FILE_DATE_FORMAT).format(new Date());
String last = new SimpleDateFormat(FILE_DATE_FORMAT).format(new Date(file.lastModified()));
if (!now.equals(last)) {
File archive = new File(file.getAbsolutePath() + "." + last);
file.renameTo(archive);
}
}
FileWriter writer = new FileWriter(file, true);
try {
for (Iterator<String> iterator = logSet.iterator();
iterator.hasNext();
iterator.remove()) {
writer.write(iterator.next());
writer.write("\r\n");
}
writer.flush();
} finally {
writer.close();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
private void log(String accessLog, AccessLogData accessLogData) {
Set<AccessLogData> logSet = logQueue.get(accessLog);
if (logSet == null) {
logQueue.putIfAbsent(accessLog, new ConcurrentHashSet<AccessLogData>());
logSet = logQueue.get(accessLog);
}
if (logSet.size() < LOG_MAX_BUFFER) {
logSet.add(accessLogData);
}
}

private void writeLogToFile() {
if (!logQueue.isEmpty()) {
for (Map.Entry<String, Set<AccessLogData>> entry : logQueue.entrySet()) {
try {
String accessLog = entry.getKey();
Set<AccessLogData> logSet = entry.getValue();
if (ConfigUtils.isDefault(accessLog)) {
processWithServiceLogger(logSet);
} else {
File file = new File(accessLog);
createIfLogDirAbsent(file);
if (logger.isDebugEnabled()) {
logger.debug("Append log to " + accessLog);
}
renameFile(file);
processWithAccessKeyLogger(logSet, file);
}

} catch (Exception e) {
logger.error(e.getMessage(), e);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}

private void processWithAccessKeyLogger(Set<AccessLogData> logSet, File file) throws IOException {
FileWriter writer = new FileWriter(file, true);
try {
for (Iterator<AccessLogData> iterator = logSet.iterator();
iterator.hasNext();
iterator.remove()) {
writer.write(iterator.next().getLogMessage(MESSAGE_DATE_FORMATTER));
writer.write("\r\n");
}
writer.flush();
} finally {
writer.close();
}
}

private AccessLogData buildAccessLogData(Invoker<?> invoker, Invocation inv) {
RpcContext context = RpcContext.getContext();
AccessLogData logData = AccessLogData.newLogData();
logData.setServiceName(invoker.getInterface().getName());
logData.setMethodName(inv.getMethodName());
logData.setVersion(invoker.getUrl().getParameter(Constants.VERSION_KEY));
logData.setGroup(invoker.getUrl().getParameter(Constants.GROUP_KEY));
logData.setInvocationTime(new Date());
logData.setLocalHost(context.getLocalHost());
logData.setLocalPort(context.getLocalPort());
logData.setRemoteHost(context.getRemoteHost());
logData.setRemotePort(context.getRemotePort());
logData.setTypes(inv.getParameterTypes());
logData.setArguments(inv.getArguments());
return logData;
}

private void processWithServiceLogger(Set<AccessLogData> logSet) {
for (Iterator<AccessLogData> iterator = logSet.iterator();
iterator.hasNext();
iterator.remove()) {
AccessLogData logData = iterator.next();
LoggerFactory.getLogger(ACCESS_LOG_KEY + "." + logData.getServiceName()).info(logData.getLogMessage(MESSAGE_DATE_FORMATTER));
}
}

private void createIfLogDirAbsent(File file) {
File dir = file.getParentFile();
if (null != dir && !dir.exists()) {
dir.mkdirs();
}
}

private void renameFile(File file) {
if (file.exists()) {
String now = FILE_DATE_FORMATTER.format(new Date());
String last = FILE_DATE_FORMATTER.format(new Date(file.lastModified()));
if (!now.equals(last)) {
File archive = new File(file.getAbsolutePath() + "." + last);
file.renameTo(archive);
}
}
}
Expand Down
Loading

0 comments on commit b1ef5b7

Please sign in to comment.