Skip to content

Commit

Permalink
logs are stored using a Queue replace with Set (#7955)
Browse files Browse the repository at this point in the history
  • Loading branch information
@icankeep
icankeep committed Jun 2, 2021
1 parent 41d36a2 commit b6da41c
Show file tree
Hide file tree
Showing 2 changed files with 25 additions and 24 deletions.
44 changes: 23 additions & 21 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 @@ -19,7 +19,6 @@
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.utils.ConcurrentHashSet;
import org.apache.dubbo.common.utils.ConfigUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.rpc.Filter;
Expand All @@ -37,8 +36,9 @@
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -67,6 +67,8 @@ public class AccessLogFilter implements Filter {

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

private static final String LINE_SEPARATOR = "line.separator";

private static final int LOG_MAX_BUFFER = 5000;

private static final long LOG_OUTPUT_INTERVAL = 5000;
Expand All @@ -76,7 +78,7 @@ public class AccessLogFilter implements Filter {
// It's safe to declare it as singleton since it runs on single thread only
private static final DateFormat FILE_NAME_FORMATTER = new SimpleDateFormat(FILE_DATE_FORMAT);

private static final Map<String, Set<AccessLogData>> LOG_ENTRIES = new ConcurrentHashMap<>();
private static final Map<String, Queue<AccessLogData>> LOG_ENTRIES = new ConcurrentHashMap<>();

private static final ScheduledExecutorService LOG_SCHEDULED = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Dubbo-Access-Log", true));

Expand Down Expand Up @@ -112,31 +114,31 @@ public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
}

private void log(String accessLog, AccessLogData accessLogData) {
Set<AccessLogData> logSet = LOG_ENTRIES.computeIfAbsent(accessLog, k -> new ConcurrentHashSet<>());
Queue<AccessLogData> logQueue = LOG_ENTRIES.computeIfAbsent(accessLog, k -> new ConcurrentLinkedQueue<>());

if (logSet.size() < LOG_MAX_BUFFER) {
logSet.add(accessLogData);
if (logQueue.size() < LOG_MAX_BUFFER) {
logQueue.add(accessLogData);
} else {
logger.warn("AccessLog buffer is full. Do a force writing to file to clear buffer.");
//just write current logSet to file.
writeLogSetToFile(accessLog, logSet);
//after force writing, add accessLogData to current logSet
logSet.add(accessLogData);
//just write current logQueue to file.
writeLogQueueToFile(accessLog, logQueue);
//after force writing, add accessLogData to current logQueue
logQueue.add(accessLogData);
}
}

private void writeLogSetToFile(String accessLog, Set<AccessLogData> logSet) {
private void writeLogQueueToFile(String accessLog, Queue<AccessLogData> logQueue) {
try {
if (ConfigUtils.isDefault(accessLog)) {
processWithServiceLogger(logSet);
processWithServiceLogger(logQueue);
} else {
File file = new File(accessLog);
createIfLogDirAbsent(file);
if (logger.isDebugEnabled()) {
logger.debug("Append log to " + accessLog);
}
renameFile(file);
processWithAccessKeyLogger(logSet, file);
processWithAccessKeyLogger(logQueue, file);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
Expand All @@ -145,28 +147,28 @@ private void writeLogSetToFile(String accessLog, Set<AccessLogData> logSet) {

private void writeLogToFile() {
if (!LOG_ENTRIES.isEmpty()) {
for (Map.Entry<String, Set<AccessLogData>> entry : LOG_ENTRIES.entrySet()) {
for (Map.Entry<String, Queue<AccessLogData>> entry : LOG_ENTRIES.entrySet()) {
String accessLog = entry.getKey();
Set<AccessLogData> logSet = entry.getValue();
writeLogSetToFile(accessLog, logSet);
Queue<AccessLogData> logQueue = entry.getValue();
writeLogQueueToFile(accessLog, logQueue);
}
}
}

private void processWithAccessKeyLogger(Set<AccessLogData> logSet, File file) throws IOException {
private void processWithAccessKeyLogger(Queue<AccessLogData> logQueue, File file) throws IOException {
try (FileWriter writer = new FileWriter(file, true)) {
for (Iterator<AccessLogData> iterator = logSet.iterator();
for (Iterator<AccessLogData> iterator = logQueue.iterator();
iterator.hasNext();
iterator.remove()) {
writer.write(iterator.next().getLogMessage());
writer.write(System.getProperty("line.separator"));
writer.write(System.getProperty(LINE_SEPARATOR));
}
writer.flush();
}
}

private void processWithServiceLogger(Set<AccessLogData> logSet) {
for (Iterator<AccessLogData> iterator = logSet.iterator();
private void processWithServiceLogger(Queue<AccessLogData> logQueue) {
for (Iterator<AccessLogData> iterator = logQueue.iterator();
iterator.hasNext();
iterator.remove()) {
AccessLogData logData = iterator.next();
Expand Down
5 changes: 2 additions & 3 deletions dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/filter/AccessLogFilterTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,11 @@
import org.apache.dubbo.rpc.support.AccessLogData;
import org.apache.dubbo.rpc.support.MockInvocation;
import org.apache.dubbo.rpc.support.MyInvoker;

import org.junit.jupiter.api.Test;

import java.lang.reflect.Field;
import java.util.Map;
import java.util.Set;
import java.util.Queue;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
Expand Down Expand Up @@ -70,7 +69,7 @@ public void testDefault() throws NoSuchFieldException, IllegalAccessException {

accessLogFilter.invoke(invoker, invocation);

Map<String, Set<AccessLogData>> logs = (Map<String, Set<AccessLogData>>) field.get(AccessLogFilter.class);
Map<String, Queue<AccessLogData>> logs = (Map<String, Queue<AccessLogData>>) field.get(AccessLogFilter.class);
assertFalse(logs.isEmpty());
assertFalse(logs.get("true").isEmpty());
AccessLogData log = logs.get("true").iterator().next();
Expand Down

0 comments on commit b6da41c

Please sign in to comment.