diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java index c7542f7762..80805e73ca 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java @@ -72,7 +72,10 @@ protected CompletionStage doInvoke(Request request, CompletionStage getInvoker(ServiceInstance instance) { String key = toUniqKey(instance); ConsumerInvokerProxy result = invokerCache.get(key); - return Optional.ofNullable(result).orElseGet(() -> createInvoker(instance)); + if (result != null && result.isAvailable()) { + return result; + } + return createInvoker(instance); } @SuppressWarnings("rawtypes") diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/common/timer/HashedWheelTimer.java b/trpc-core/src/main/java/com/tencent/trpc/core/common/timer/HashedWheelTimer.java index dca6f2ee10..eacd9abab1 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/common/timer/HashedWheelTimer.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/common/timer/HashedWheelTimer.java @@ -261,7 +261,7 @@ public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit } if (duration < MILLISECOND_NANOS) { - logger.warn("Configured tickDuration %d smaller then %d, using 1ms.", tickDuration, + logger.warn("Configured tickDuration {} smaller then {}, using 1ms.", tickDuration, MILLISECOND_NANOS); this.tickDuration = MILLISECOND_NANOS; } else { diff --git a/trpc-core/src/main/java/com/tencent/trpc/core/transport/AbstractClientTransport.java b/trpc-core/src/main/java/com/tencent/trpc/core/transport/AbstractClientTransport.java index d0b34c010d..9716690422 100644 --- a/trpc-core/src/main/java/com/tencent/trpc/core/transport/AbstractClientTransport.java +++ b/trpc-core/src/main/java/com/tencent/trpc/core/transport/AbstractClientTransport.java @@ -178,10 +178,10 @@ public void close() { @Override public CompletionStage send(Object msg) throws TransportException { if (isClosed()) { - throw TransportException.create(String.format( + throw TransportException.create( "Client transport(transport=%s, class=%s, msg=%s) send fail, due to transport" + " is not available or close", - this, name, msg)); + this, name, msg); } return getChannel0().thenCompose(f -> f.send(msg)); } @@ -192,10 +192,10 @@ public CompletionStage send(Object msg) throws TransportException { @Override public CompletionStage getChannel() throws TransportException { if (isClosed()) { - throw TransportException.create(String.format( + throw TransportException.create( "Client transport(transport=%s, class=%s) get channel fail, due to transport " + "is not available or close", - this, name)); + this, name); } return getChannel0(); } @@ -420,8 +420,7 @@ protected void stopInternal() throws Exception { try { doClose(); } catch (Throwable ex) { - logger.error(String.format("Client transport(%s) destroy failed", getRemoteAddress(), - ex)); + logger.error(String.format("Client transport(%s) destroy failed", getRemoteAddress()), ex); } try { if (handler != null) { diff --git a/trpc-core/src/test/java/com/tencent/trpc/core/cluster/def/DefClusterInvokerTest.java b/trpc-core/src/test/java/com/tencent/trpc/core/cluster/def/DefClusterInvokerTest.java index 21c798d529..1ec8b1ad97 100644 --- a/trpc-core/src/test/java/com/tencent/trpc/core/cluster/def/DefClusterInvokerTest.java +++ b/trpc-core/src/test/java/com/tencent/trpc/core/cluster/def/DefClusterInvokerTest.java @@ -33,10 +33,12 @@ import com.tencent.trpc.core.worker.handler.TrpcThreadExceptionHandler; import com.tencent.trpc.core.worker.spi.WorkerPool; import com.tencent.trpc.core.worker.support.thread.ThreadWorkerPool; +import java.lang.reflect.Field; import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.CompletionStage; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicLong; @@ -204,4 +206,55 @@ public void testDoInvoke() { } } + @Test + @SuppressWarnings("unchecked") + public void testGetInvokerSkipsUnavailableCachedInvoker() throws Exception { + Assertions.assertFalse(consumerInvokerProxy.isAvailable()); + + // Put the unavailable invoker into invokerCache via reflection + Field cacheField = DefClusterInvoker.class.getDeclaredField("invokerCache"); + cacheField.setAccessible(true); + ConcurrentMap> cache = + (ConcurrentMap>) cacheField.get(defClusterInvoker); + + ServiceInstance instance = new ServiceInstance("127.0.0.1", 12345); + String key = "127.0.0.1:12345:null"; + cache.put(key, consumerInvokerProxy); + + // Spy on defClusterInvoker and mock createInvoker to return a new available proxy + DefClusterInvoker spy = Mockito.spy(defClusterInvoker); + ConsumerInvokerProxy newProxy = Mockito.mock(ConsumerInvokerProxy.class); + Mockito.when(newProxy.isAvailable()).thenReturn(true); + Mockito.doReturn(newProxy).when(spy).createInvoker(instance); + + ConsumerInvokerProxy result = spy.getInvoker(instance); + + // Should not return the unavailable cached invoker + Assertions.assertNotSame(consumerInvokerProxy, result); + // Should return the new available one from createInvoker + Assertions.assertSame(newProxy, result); + } + + @Test + @SuppressWarnings("unchecked") + public void testGetInvokerReturnsAvailableCachedInvoker() throws Exception { + // Put an available invoker into invokerCache via reflection + Field cacheField = DefClusterInvoker.class.getDeclaredField("invokerCache"); + cacheField.setAccessible(true); + ConcurrentMap> cache = + (ConcurrentMap>) cacheField.get(defClusterInvoker); + + ServiceInstance instance = new ServiceInstance("127.0.0.1", 12345); + String key = "127.0.0.1:12345:null"; + + // Create an available proxy (client.isAvailable() = true) + ConsumerInvokerProxy availableProxy = Mockito.mock(ConsumerInvokerProxy.class); + Mockito.when(availableProxy.isAvailable()).thenReturn(true); + cache.put(key, availableProxy); + + // getInvoker should return the cached available proxy directly + ConsumerInvokerProxy result = defClusterInvoker.getInvoker(instance); + Assertions.assertSame(availableProxy, result); + } + } diff --git a/trpc-core/src/test/java/com/tencent/trpc/core/common/timer/HashWheelTimerTest.java b/trpc-core/src/test/java/com/tencent/trpc/core/common/timer/HashWheelTimerTest.java index 20b64dc9da..cbf6e5bf56 100644 --- a/trpc-core/src/test/java/com/tencent/trpc/core/common/timer/HashWheelTimerTest.java +++ b/trpc-core/src/test/java/com/tencent/trpc/core/common/timer/HashWheelTimerTest.java @@ -54,6 +54,54 @@ public void testInit() { hashedWheelTimer.stop(); } + /** + * A tick duration smaller than one millisecond is normalized to one millisecond and a warning is logged. The + * warning uses the {@code {}} placeholder of slf4j, a printf style placeholder would leave the raw text in the + * log and lose the real values. + */ + @Test + public void testTickDurationSmallerThanOneMillisecond() { + HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(new NamedThreadFactory(), 1L, + TimeUnit.NANOSECONDS); + try { + Assertions.assertNotNull(hashedWheelTimer); + hashedWheelTimer.start(); + // the tick duration has been normalized, so the timer still works + Timeout timeout = hashedWheelTimer.newTimeout(t -> { + }, 1000, TimeUnit.MILLISECONDS); + Assertions.assertFalse(timeout.isExpired()); + timeout.cancel(); + } finally { + hashedWheelTimer.stop(); + } + } + + /** + * An illegal tick duration should be rejected. + */ + @Test + public void testIllegalTickDuration() { + try { + new HashedWheelTimer(new NamedThreadFactory(), 0L, TimeUnit.MILLISECONDS); + Assertions.fail("IllegalArgumentException is expected"); + } catch (IllegalArgumentException e) { + Assertions.assertTrue(e.getMessage().contains("tickDuration must be greater than 0"), e.getMessage()); + } + } + + /** + * A tick duration which overflows the wheel should be rejected. + */ + @Test + public void testTickDurationOverflow() { + try { + new HashedWheelTimer(new NamedThreadFactory(), Long.MAX_VALUE / 2, TimeUnit.DAYS); + Assertions.fail("IllegalArgumentException is expected"); + } catch (IllegalArgumentException e) { + Assertions.assertTrue(e.getMessage().contains("tickDuration"), e.getMessage()); + } + } + @Test public void test() { timer.toString(); diff --git a/trpc-core/src/test/java/com/tencent/trpc/core/exception/TransportExceptionTest.java b/trpc-core/src/test/java/com/tencent/trpc/core/exception/TransportExceptionTest.java new file mode 100644 index 0000000000..e02d3cabac --- /dev/null +++ b/trpc-core/src/test/java/com/tencent/trpc/core/exception/TransportExceptionTest.java @@ -0,0 +1,138 @@ +/* + * Tencent is pleased to support the open source community by making tRPC available. + * + * Copyright (C) 2023 Tencent. + * All rights reserved. + * + * If you have downloaded a copy of the tRPC source code from Tencent, + * please note that tRPC source code is licensed under the Apache 2.0 License, + * A copy of the Apache 2.0 License can be found in the LICENSE file. + */ + +package com.tencent.trpc.core.exception; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +import java.util.UnknownFormatConversionException; +import org.junit.jupiter.api.Test; + +/** + * Test {@link TransportException}, especially the {@code create} methods which format the message by themselves, so + * the callers must pass the raw format and the arguments instead of an already formatted string. + */ +public class TransportExceptionTest { + + private static final String MESSAGE = "transport error"; + + @Test + public void testConstructWithMessage() { + TransportException e = new TransportException(MESSAGE); + assertEquals(MESSAGE, e.getMessage()); + assertNull(e.getCause()); + } + + @Test + public void testConstructWithMessageAndCause() { + IllegalStateException cause = new IllegalStateException("cause"); + TransportException e = new TransportException(MESSAGE, cause); + assertEquals(MESSAGE, e.getMessage()); + assertSame(cause, e.getCause()); + } + + @Test + public void testConstructWithCause() { + IllegalStateException cause = new IllegalStateException("cause"); + TransportException e = new TransportException(cause); + assertEquals("cause", e.getMessage()); + assertSame(cause, e.getCause()); + } + + @Test + public void testCreateFormatsArguments() { + TransportException e = TransportException.create("send fail, addr=%s, port=%s", "127.0.0.1", 8080); + assertEquals("send fail, addr=127.0.0.1, port=8080", e.getMessage()); + assertNull(e.getCause()); + } + + @Test + public void testCreateWithCauseFormatsArguments() { + IllegalStateException cause = new IllegalStateException("cause"); + TransportException e = TransportException.create(cause, "send fail, addr=%s", "127.0.0.1"); + assertEquals("send fail, addr=127.0.0.1", e.getMessage()); + assertSame(cause, e.getCause()); + } + + @Test + public void testCreateWithNullCauseFormatsArguments() { + TransportException e = TransportException.create((Throwable) null, "send fail, addr=%s", "127.0.0.1"); + assertEquals("send fail, addr=127.0.0.1", e.getMessage()); + assertNull(e.getCause()); + } + + @Test + public void testCreateWithoutArguments() { + TransportException e = TransportException.create("send fail"); + assertEquals("send fail", e.getMessage()); + } + + @Test + public void testCreateWithNullArgument() { + TransportException e = TransportException.create("send fail, msg=%s", (Object) null); + assertEquals("send fail, msg=null", e.getMessage()); + } + + /** + * The argument may contain a percent sign, and it must be kept as it is, because only the format is formatted. + */ + @Test + public void testCreateKeepsPercentSignOfArgument() { + TransportException e = TransportException.create("send fail, msg=%s", "GET /a%2Fb?rate=100%"); + assertEquals("send fail, msg=GET /a%2Fb?rate=100%", e.getMessage()); + } + + /** + * The caller must never pass an already formatted message which contains a percent sign, otherwise the message + * is formatted twice and the real error is hidden by a format exception. This test pins down the behaviour so + * that the double formatting can be caught. + */ + @Test + public void testCreateWithAlreadyFormattedMessageContainingPercentFails() { + String formatted = String.format("send fail, msg=%s", "GET /a%2Fb"); + assertEquals("send fail, msg=GET /a%2Fb", formatted); + try { + TransportException.create(formatted); + fail("UnknownFormatConversionException is expected"); + } catch (UnknownFormatConversionException e) { + assertNotNull(e.getMessage()); + } + } + + @Test + public void testTransReturnsTheSameInstanceForTransportException() { + TransportException origin = new TransportException(MESSAGE); + assertSame(origin, TransportException.trans(origin)); + assertSame(origin, TransportException.trans(origin, "other message")); + } + + @Test + public void testTransWrapsOtherException() { + IllegalStateException cause = new IllegalStateException("cause"); + TransportException e = TransportException.trans(cause); + assertEquals("cause", e.getMessage()); + assertSame(cause, e.getCause()); + + TransportException withMessage = TransportException.trans(cause, MESSAGE); + assertEquals(MESSAGE, withMessage.getMessage()); + assertSame(cause, withMessage.getCause()); + } + + @Test + public void testIsRuntimeException() { + assertTrue(new TransportException(MESSAGE) instanceof RuntimeException); + } +} diff --git a/trpc-core/src/test/java/com/tencent/trpc/core/logger/LogPlaceholderConventionTest.java b/trpc-core/src/test/java/com/tencent/trpc/core/logger/LogPlaceholderConventionTest.java new file mode 100644 index 0000000000..4610f476a2 --- /dev/null +++ b/trpc-core/src/test/java/com/tencent/trpc/core/logger/LogPlaceholderConventionTest.java @@ -0,0 +1,385 @@ +/* + * Tencent is pleased to support the open source community by making tRPC available. + * + * Copyright (C) 2023 Tencent. + * All rights reserved. + * + * If you have downloaded a copy of the tRPC source code from Tencent, + * please note that tRPC source code is licensed under the Apache 2.0 License, + * A copy of the Apache 2.0 License can be found in the LICENSE file. + */ + +package com.tencent.trpc.core.logger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.junit.jupiter.api.Test; + +/** + * Guard the log placeholder style of the whole repository. + * + *

The logger of the framework delegates to slf4j, which only recognizes the {@code {}} placeholder. A printf + * style placeholder(such as {@code %s} or {@code %d}) is never replaced by slf4j, so the real values are silently + * lost from the log. Besides, the number of the placeholders must match the number of the arguments, otherwise + * either an argument is dropped or a raw {@code {}} is left in the log message.

+ * + *

This test scans the production sources so that such defects can be found before they reach the production + * environment.

+ */ +public class LogPlaceholderConventionTest { + + /** + * The pattern of a logging call of the framework logger. + */ + private static final Pattern LOG_CALL = Pattern.compile( + "\\b(?:logger|LOG|log|LOGGER)\\s*\\.\\s*(?:error|warn|info|debug|trace)\\s*\\("); + + /** + * The pattern of a printf style placeholder, {@code %%} is excluded because it is an escaped percent sign. + */ + private static final Pattern PRINTF_PLACEHOLDER = Pattern.compile( + "%(?!%)[-#+ 0,(]*\\d*(?:\\.\\d+)?[sdifxXoeEgGbBhHnc]"); + + /** + * The pattern of a string literal. + */ + private static final Pattern STRING_LITERAL = Pattern.compile("\"((?:[^\"\\\\]|\\\\.)*)\""); + + /** + * The name of the variables which are very likely to be a throwable, a throwable is allowed to be the last + * argument of a logging call without a matching placeholder. + */ + private static final Pattern THROWABLE_LIKE = Pattern.compile( + "^(?:e|ex|t|th|err|error|cause|throwable|exception|ignored|ignore)\\d*$" + + "|(?:\\.getCause\\(\\)|\\.cause\\(\\))$"); + + /** + * The max number of the lines a logging call may span. + */ + private static final int MAX_CALL_LINES = 12; + + private static final String MAIN_SOURCE_DIR = "src/main/java"; + + @Test + public void testNoPrintfPlaceholderInLoggingCall() throws IOException { + List defects = scan(LogCall::hasPrintfPlaceholder, + "printf style placeholder is not supported by slf4j, use {} instead"); + assertTrue(defects.isEmpty(), String.join("\n", defects)); + } + + @Test + public void testPlaceholderCountMatchesArgumentCount() throws IOException { + List defects = scan(LogCall::hasMismatchedPlaceholder, + "the number of the placeholders does not match the number of the arguments"); + assertTrue(defects.isEmpty(), String.join("\n", defects)); + } + + /** + * Make sure the scanner really works, otherwise the two test cases above would pass silently even if the + * scanner is broken. + */ + @Test + public void testScannerDetectsPrintfPlaceholder() { + LogCall call = LogCall.of("\"Configured tickDuration %d smaller then %d, using 1ms.\", a, b"); + assertTrue(call.hasPrintfPlaceholder()); + assertFalse(LogCall.of("\"Configured tickDuration {} smaller then {}\", a, b").hasPrintfPlaceholder()); + // a percent sign of a plain text is not a placeholder + assertFalse(LogCall.of("\"progress is 100%\"").hasPrintfPlaceholder()); + } + + @Test + public void testScannerDetectsMismatchedPlaceholder() { + assertTrue(LogCall.of("\"request {}, basePath is {}\", path").hasMismatchedPlaceholder()); + assertTrue(LogCall.of("\"request {}\", path, basePath").hasMismatchedPlaceholder()); + assertFalse(LogCall.of("\"request {}, basePath is {}\", path, basePath").hasMismatchedPlaceholder()); + // a throwable is allowed to be the last argument without a placeholder + assertFalse(LogCall.of("\"request {} failed\", path, e").hasMismatchedPlaceholder()); + assertFalse(LogCall.of("\"request failed\", e").hasMismatchedPlaceholder()); + // the string is built by concatenation, there is no placeholder at all + assertFalse(LogCall.of("\"request \" + path + \" failed\", e").hasMismatchedPlaceholder()); + } + + @Test + public void testScannerSkipsStringFormat() { + // String.format uses the printf style, it is the caller of the logger that formats the message + assertFalse(LogCall.of("String.format(\"request %s failed\", path), e").hasPrintfPlaceholder()); + } + + private List scan(java.util.function.Predicate predicate, String reason) throws IOException { + Path root = findRepositoryRoot(); + try (Stream stream = Files.walk(root)) { + List sources = stream + .filter(Files::isRegularFile) + .filter(p -> p.toString().endsWith(".java")) + .filter(p -> p.toString().replace(File.separatorChar, '/').contains(MAIN_SOURCE_DIR)) + .collect(Collectors.toList()); + return sources.stream() + .flatMap(p -> scanFile(p, predicate, reason).stream()) + .collect(Collectors.toList()); + } + } + + private List scanFile(Path path, java.util.function.Predicate predicate, String reason) { + List defects = new java.util.ArrayList<>(); + List lines; + try { + lines = Files.readAllLines(path); + } catch (IOException e) { + return defects; + } + for (int i = 0; i < lines.size(); i++) { + if (!LOG_CALL.matcher(lines.get(i)).find()) { + continue; + } + String arguments = extractArguments(lines, i); + if (arguments == null) { + continue; + } + LogCall call = LogCall.of(arguments); + if (predicate.test(call)) { + defects.add(String.format("%s:%d: %s%n %s", path, i + 1, reason, + arguments.replaceAll("\\s+", " "))); + } + } + return defects; + } + + /** + * Extract the argument list of the logging call which starts at the given line. + * + * @param lines all the lines of the source file + * @param begin the index of the line the logging call starts at + * @return the argument list without the enclosing parentheses, or null if it can not be extracted + */ + private String extractArguments(List lines, int begin) { + StringBuilder buffer = new StringBuilder(); + for (int i = begin; i < Math.min(begin + MAX_CALL_LINES, lines.size()); i++) { + buffer.append(lines.get(i)).append(' '); + Matcher matcher = LOG_CALL.matcher(buffer); + if (!matcher.find()) { + continue; + } + String arguments = readUntilClosed(buffer.substring(matcher.end())); + if (arguments != null) { + return arguments; + } + } + return null; + } + + /** + * Read the content until the parenthesis of the logging call is closed. + * + * @param text the text just after the opening parenthesis + * @return the content of the parentheses, or null if the parenthesis is not closed yet + */ + private static String readUntilClosed(String text) { + StringBuilder result = new StringBuilder(); + int depth = 1; + boolean inString = false; + boolean escaped = false; + for (char c : text.toCharArray()) { + if (escaped) { + result.append(c); + escaped = false; + continue; + } + if (c == '\\') { + result.append(c); + escaped = true; + continue; + } + if (c == '"') { + inString = !inString; + result.append(c); + continue; + } + if (!inString) { + if (c == '(') { + depth++; + } else if (c == ')') { + depth--; + if (depth == 0) { + return result.toString(); + } + } + } + result.append(c); + } + return null; + } + + private Path findRepositoryRoot() { + Path current = Paths.get("").toAbsolutePath(); + while (current != null && !Files.exists(current.resolve("trpc-core"))) { + current = current.getParent(); + } + return current == null ? Paths.get("").toAbsolutePath() : current; + } + + /** + * A parsed logging call. + */ + private static final class LogCall { + + private final String arguments; + + private final List topLevelArguments; + + private LogCall(String arguments) { + this.arguments = arguments; + this.topLevelArguments = splitTopLevel(arguments); + } + + static LogCall of(String arguments) { + return new LogCall(arguments); + } + + /** + * Whether the message contains a printf style placeholder while it is not formatted by String.format. + * + * @return true if the logging call is defective + */ + boolean hasPrintfPlaceholder() { + if (arguments.contains("String.format")) { + return false; + } + if (topLevelArguments.size() < 2) { + // without any argument the percent sign is only a part of the plain text + return false; + } + return PRINTF_PLACEHOLDER.matcher(literalOf(firstArgument())).find(); + } + + /** + * Whether the number of the placeholders does not match the number of the arguments. + * + * @return true if the logging call is defective + */ + boolean hasMismatchedPlaceholder() { + String first = firstArgument(); + if (!first.contains("\"") || arguments.contains("String.format")) { + return false; + } + if (first.contains("+")) { + // the message is built by concatenation, the braces of the literals are not placeholders, and the + // values are already embedded in the message + return false; + } + int placeholders = countOccurrences(literalOf(first)); + List rest = topLevelArguments.subList(1, topLevelArguments.size()); + int arguments = rest.size(); + boolean lastIsThrowable = !rest.isEmpty() + && THROWABLE_LIKE.matcher(rest.get(rest.size() - 1).trim()).find(); + return placeholders != arguments && placeholders != arguments - (lastIsThrowable ? 1 : 0); + } + + private String firstArgument() { + return topLevelArguments.isEmpty() ? "" : topLevelArguments.get(0); + } + + /** + * Concatenate all the string literals of the given text, the concatenated literals form the log message. + * + * @param text the text to be parsed + * @return the concatenated string literals + */ + private static String literalOf(String text) { + StringBuilder builder = new StringBuilder(); + Matcher matcher = STRING_LITERAL.matcher(text); + while (matcher.find()) { + builder.append(matcher.group(1)); + } + return builder.toString(); + } + + private static int countOccurrences(String literal) { + int count = 0; + int index = literal.indexOf("{}"); + while (index >= 0) { + count++; + index = literal.indexOf("{}", index + 2); + } + return count; + } + + /** + * Split the argument list by the top level commas, the commas inside a string, a parenthesis, a bracket or a + * brace are ignored. + * + * @param text the argument list + * @return the top level arguments + */ + private static List splitTopLevel(String text) { + List result = new java.util.ArrayList<>(); + StringBuilder current = new StringBuilder(); + int depth = 0; + boolean inString = false; + boolean inChar = false; + boolean escaped = false; + for (char c : text.toCharArray()) { + if (escaped) { + current.append(c); + escaped = false; + continue; + } + if (c == '\\') { + current.append(c); + escaped = true; + continue; + } + if (c == '\'' && !inString) { + inChar = !inChar; + current.append(c); + continue; + } + if (c == '"' && !inChar) { + inString = !inString; + current.append(c); + continue; + } + if (inString || inChar) { + current.append(c); + continue; + } + if (c == '(' || c == '[' || c == '{') { + depth++; + } else if (c == ')' || c == ']' || c == '}') { + depth--; + } + if (c == ',' && depth == 0) { + result.add(current.toString().trim()); + current.setLength(0); + continue; + } + current.append(c); + } + String last = current.toString().trim(); + if (!last.isEmpty() || !result.isEmpty()) { + result.add(last); + } + return result; + } + } + + /** + * Make sure the constant is used, it also documents the expected placeholder style. + */ + @Test + public void testPlaceholderStyle() { + assertEquals(2, LogCall.countOccurrences("a={}, b={}")); + assertEquals(0, LogCall.countOccurrences("a=%s")); + } +} diff --git a/trpc-core/src/test/java/com/tencent/trpc/core/transport/AbstractClientTransportTest.java b/trpc-core/src/test/java/com/tencent/trpc/core/transport/AbstractClientTransportTest.java index 9a3e920cfc..6a0d00c8c0 100644 --- a/trpc-core/src/test/java/com/tencent/trpc/core/transport/AbstractClientTransportTest.java +++ b/trpc-core/src/test/java/com/tencent/trpc/core/transport/AbstractClientTransportTest.java @@ -11,17 +11,28 @@ package com.tencent.trpc.core.transport; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; import com.tencent.trpc.core.common.config.ProtocolConfig; import com.tencent.trpc.core.exception.TransportException; import com.tencent.trpc.core.transport.codec.ClientCodec; +import java.net.InetSocketAddress; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import org.junit.jupiter.api.Test; public class AbstractClientTransportTest { + /** + * A message whose toString contains the format specifier characters, it is used to make sure the error message + * is formatted only once. + */ + private static final String MSG_WITH_PERCENT = "GET /a%2Fb?rate=100%"; + @Test public void testOpenException() throws Exception { ClientTransportTest test = new ClientTransportTest(TransporterTestUtils.newProtocolConfig(), @@ -51,16 +62,193 @@ public void testOpenException() throws Exception { test2.toString(); } + /** + * The error message of a closed transport should carry the transport, the class and the message, and the format + * specifiers must be replaced by the real arguments. + */ + @Test + public void testSendAfterClosedThrowsFormattedException() throws Exception { + ClientTransportTest transport = newClosedTransport(); + try { + transport.send("hello"); + fail("TransportException is expected"); + } catch (TransportException e) { + String message = e.getMessage(); + assertNotNull(message); + assertNull(e.getCause()); + assertTrue(message.contains(transport.toString()), message); + assertTrue(message.contains(ClientTransportTest.class.getName()), message); + assertTrue(message.contains("hello"), message); + assertTrue(message.contains("send fail"), message); + // all the format specifiers have been replaced + assertTrue(!message.contains("%s"), message); + } + } + + /** + * The message to be sent is untrusted, it may contain a percent sign. Formatting the error message twice would + * throw a format exception and hide the real error, so it must be formatted only once. + */ + @Test + public void testSendAfterClosedWithPercentInMessage() throws Exception { + ClientTransportTest transport = newClosedTransport(); + try { + transport.send(MSG_WITH_PERCENT); + fail("TransportException is expected"); + } catch (TransportException e) { + assertTrue(e.getMessage().contains(MSG_WITH_PERCENT), e.getMessage()); + } + } + + /** + * The transport itself may also contain a percent sign, for example the remote address of a unix domain socket. + */ + @Test + public void testSendAfterClosedWithPercentInTransport() throws Exception { + ClientTransportTest transport = newClosedTransport(); + transport.setDescription("transport-100%-desc"); + try { + transport.send("hello"); + fail("TransportException is expected"); + } catch (TransportException e) { + assertTrue(e.getMessage().contains("transport-100%-desc"), e.getMessage()); + } + } + + @Test + public void testGetChannelAfterClosedThrowsFormattedException() throws Exception { + ClientTransportTest transport = newClosedTransport(); + transport.setDescription("transport-100%-desc"); + try { + transport.getChannel(); + fail("TransportException is expected"); + } catch (TransportException e) { + String message = e.getMessage(); + assertNotNull(message); + assertNull(e.getCause()); + assertTrue(message.contains("transport-100%-desc"), message); + assertTrue(message.contains(ClientTransportTest.class.getName()), message); + assertTrue(message.contains("get channel fail"), message); + assertTrue(!message.contains("%s"), message); + } + } + + /** + * A null message must not break the error message building. + */ + @Test + public void testSendAfterClosedWithNullMessage() throws Exception { + ClientTransportTest transport = newClosedTransport(); + try { + transport.send(null); + fail("TransportException is expected"); + } catch (TransportException e) { + assertTrue(e.getMessage().contains("msg=null"), e.getMessage()); + } + } + + /** + * When a channel fails to be closed, the error should be logged with the channel item and the cause, and the + * remaining steps(doClose and the handler destroying) should still be executed. + * + *

The lifecycle only executes stopInternal when it has left the new state, so the transport is opened first. + * The open fails on purpose so that the state becomes FAILED, and then stop executes stopInternal.

+ */ + @Test + public void testCloseLogsChannelCloseFailure() throws Exception { + ClientTransportTest transport = new ClientTransportTest(TransporterTestUtils.newProtocolConfig(), + new ThrowingDestroyChannelHandler(), TransporterTestUtils.newClientCodec(), true); + transport.channels.add(new AbstractClientTransport.ChannelFutureItem( + CompletableFuture.completedFuture(new ThrowingCloseChannel()), + TransporterTestUtils.newProtocolConfig())); + try { + // the open fails, and the failed start triggers the stop which executes stopInternal + transport.open(); + fail("TransportException is expected"); + } catch (TransportException e) { + assertNotNull(e.getMessage()); + } + assertTrue(transport.isClosed()); + // the channel close, the doClose and the handler destroying all failed, but they were all attempted + assertTrue(transport.isDoCloseCalled()); + } + + private ClientTransportTest newClosedTransport() throws Exception { + ClientTransportTest transport = new ClientTransportTest(TransporterTestUtils.newProtocolConfig(), + TransporterTestUtils.newChannelHandler(), TransporterTestUtils.newClientCodec(), true); + transport.close(); + assertTrue(transport.isClosed()); + return transport; + } + + /** + * A channel whose close always fails, it is used to trigger the error log of the channel closing. + */ + private static class ThrowingCloseChannel implements Channel { + + @Override + public CompletionStage close() { + throw new IllegalStateException("close failed"); + } + + @Override + public CompletionStage send(Object message) { + return CompletableFuture.completedFuture(null); + } + + @Override + public boolean isClosed() { + return false; + } + + @Override + public boolean isConnected() { + return true; + } + + @Override + public InetSocketAddress getRemoteAddress() { + return new InetSocketAddress("127.0.0.1", 6666); + } + + @Override + public InetSocketAddress getLocalAddress() { + return new InetSocketAddress("127.0.0.1", 6667); + } + + @Override + public ProtocolConfig getProtocolConfig() { + return TransporterTestUtils.newProtocolConfig(); + } + } + private static class ClientTransportTest extends AbstractClientTransport { private boolean isTransportException; + private String description; + + private boolean doCloseCalled; + ClientTransportTest(ProtocolConfig config, ChannelHandler channelHandler, ClientCodec clientCodec, boolean isTransportException) throws TransportException { super(config, channelHandler, clientCodec); this.isTransportException = isTransportException; } + void setDescription(String description) { + this.description = description; + } + + boolean isDoCloseCalled() { + return doCloseCalled; + } + + @Override + public String toString() { + return description == null ? super.toString() : description; + } + @Override public Set getChannels() { return null; @@ -82,6 +270,7 @@ protected CompletableFuture make() throws Exception { @Override protected void doClose() { + doCloseCalled = true; throw new IllegalArgumentException(); } @@ -91,4 +280,36 @@ protected boolean useChannelPool() { } } + + /** + * A channel handler whose destroying always fails, it is used to trigger the error log of the handler + * destroying. + */ + private static class ThrowingDestroyChannelHandler implements ChannelHandler { + + @Override + public void connected(Channel channel) { + } + + @Override + public void disconnected(Channel channel) { + } + + @Override + public void send(Channel channel, Object message) { + } + + @Override + public void received(Channel channel, Object message) { + } + + @Override + public void caught(Channel channel, Throwable exception) { + } + + @Override + public void destroy() { + throw new IllegalStateException("destroy failed"); + } + } } diff --git a/trpc-proto/trpc-proto-standard/pom.xml b/trpc-proto/trpc-proto-standard/pom.xml index f6d2967e7d..8216d3b99c 100644 --- a/trpc-proto/trpc-proto-standard/pom.xml +++ b/trpc-proto/trpc-proto-standard/pom.xml @@ -42,6 +42,10 @@ + + com.github.ben-manes.caffeine + caffeine + commons-codec commons-codec diff --git a/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java b/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java index b489dc4757..c6a3d38541 100644 --- a/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java +++ b/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java @@ -13,6 +13,8 @@ import static com.tencent.trpc.core.rpc.RpcContextValueKeys.SERVER_SIGNATURE_VERIFY_RESULT_KEY; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.google.protobuf.ByteString; import com.tencent.trpc.core.common.config.ProtocolConfig; import com.tencent.trpc.core.compressor.spi.Compressor; @@ -41,9 +43,9 @@ import com.tencent.trpc.proto.standard.common.TRPCProtocol.ResponseProtocol.Builder; import com.tencent.trpc.proto.standard.common.TRPCProtocol.TrpcCallType; import com.tencent.trpc.proto.standard.common.TRPCProtocol.TrpcMessageType; -import java.util.Map; +import java.time.Duration; import java.util.Objects; -import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; import org.apache.commons.lang3.StringUtils; /** @@ -52,19 +54,100 @@ public class StandardServerCodec extends ServerCodec { /** - * Remotely call the cache of services and methods to avoid string cutting operations during decoding + * The max entry number of the decoding caches, the entries will be evicted when it is exceeded. */ - private static final Map FUNC_INFO_CACHE = new ConcurrentHashMap<>(); + private static final int CACHE_MAX_SIZE = 10000; + + /** + * The initial capacity of the decoding caches. + */ + private static final int CACHE_INITIAL_CAPACITY = 64; + + /** + * The expire time(in minutes) after the last access of a decoding cache entry. + */ + private static final int CACHE_EXPIRE_MINUTES = 60; + + /** + * The max length of the cache key, the key longer than it will never be cached, so that a few oversized + * attacker-controlled keys can not occupy too much memory. + */ + private static final int CACHE_KEY_MAX_LENGTH = 1024; + + /** + * The separator of the func of the request head, whose format is {@code /serviceName/methodName}. + */ + private static final String FUNC_SEPARATOR = "/"; + + /** + * The begin index of the serviceName in the func, the func always starts with {@link #FUNC_SEPARATOR}. + */ + private static final int FUNC_SERVICE_NAME_BEGIN_INDEX = 1; + + /** + * The index of the serviceName in the parsed func info. + */ + private static final int FUNC_SERVICE_NAME_INDEX = 0; + + /** + * The index of the methodName in the parsed func info. + */ + private static final int FUNC_METHOD_NAME_INDEX = 1; + + /** + * Remotely call the cache of services and methods to avoid string cutting operations during decoding. + * + *

Note: the cache key comes from the attacker-controllable protocol head, so it MUST be bounded, see + * {@link #newCache()} and {@link #getOrCompute(Cache, String, Function)}.

+ */ + private static final Cache FUNC_INFO_CACHE = newCache(); /** * The cache of the caller and callee information avoids the string cutting operation during decoding, which can * increase the throughput of the framework by about 4%. Considering that the mainstream of the current - * architecture is microservices, and each service has limited external interfaces, the built-in ConcurrentHashMap - * is used as a cache here. If the number of caches is too large, you can consider migrating to caffeine, - * but this will greatly offset the performance optimization here. + * architecture is microservices, and each service has limited external interfaces, a bounded cache is enough + * here. + * + *

Note: the cache key comes from the attacker-controllable protocol head, so it MUST be bounded, see + * {@link #newCache()} and {@link #getOrCompute(Cache, String, Function)}.

*/ - private static final Map CALL_INFO_CACHE = new ConcurrentHashMap<>(); + private static final Cache CALL_INFO_CACHE = newCache(); + /** + * Create a bounded cache whose entries are evicted by size and by idle time. + * + * @param the type of the cached value + * @return the bounded cache + */ + private static Cache newCache() { + return Caffeine.newBuilder() + .initialCapacity(CACHE_INITIAL_CAPACITY) + .maximumSize(CACHE_MAX_SIZE) + .expireAfterAccess(Duration.ofMinutes(CACHE_EXPIRE_MINUTES)) + .build(); + } + + /** + * Get the value from the bounded cache, compute it if absent. + * + *

The cache keys are built from the request head fields(func/caller/callee) which are fully controlled by the + * remote peer, and they are written before the service/method existence check. An unbounded cache would allow an + * attacker to write an entry per request and finally exhaust the heap(OOM). The cache is bounded in two + * dimensions here: the entry number is limited by {@link #CACHE_MAX_SIZE}, and an oversized key is never cached + * so that the memory of a single entry is limited as well.

+ * + * @param cache the cache to read and write + * @param key the cache key, which is untrusted + * @param mappingFunction the function to compute the value + * @param the type of the cached value + * @return the cached or newly computed value + */ + private static V getOrCompute(Cache cache, String key, Function mappingFunction) { + if (key.length() > CACHE_KEY_MAX_LENGTH) { + return mappingFunction.apply(key); + } + return cache.get(key, mappingFunction); + } @Override public void encode(Channel channel, ChannelBuffer channelBuffer, Object message) { @@ -236,21 +319,29 @@ private void setAttachments(RequestProtocol requestHeader, DefRequest request) { private RpcInvocation buildRpcInvocation(StandardPackage packet, RequestProtocol requestHeader) { RpcInvocation inv = new RpcInvocation(); String func = requestHeader.getFunc().toStringUtf8(); - String[] funcInfo = FUNC_INFO_CACHE.computeIfAbsent(func, s -> { - int idx = func.lastIndexOf("/"); - // func format: /serviceName/methodName - return (idx > 1 && func.length() > idx + 1) ? new String[]{func.substring(1, idx), func.substring(idx + 1)} - : new String[]{"", ""}; - }); + String[] funcInfo = getOrCompute(FUNC_INFO_CACHE, func, StandardServerCodec::parseFunc); inv.setFunc(func); - inv.setRpcServiceName(funcInfo[0]); - inv.setRpcMethodName(funcInfo[1]); + inv.setRpcServiceName(funcInfo[FUNC_SERVICE_NAME_INDEX]); + inv.setRpcMethodName(funcInfo[FUNC_METHOD_NAME_INDEX]); Object[] obj = new Object[]{new DecodableValue(requestHeader.getContentEncoding(), requestHeader.getContentType(), packet.getBodyBytes())}; inv.setArguments(obj); return inv; } + /** + * Parse the func of the request head, whose format is {@code /serviceName/methodName}. + * + * @param func the func of the request head + * @return an array of [serviceName, methodName], both of them are empty if the func is illegal + */ + private static String[] parseFunc(String func) { + int idx = func.lastIndexOf(FUNC_SEPARATOR); + return (idx > FUNC_SERVICE_NAME_BEGIN_INDEX && func.length() > idx + 1) + ? new String[]{func.substring(FUNC_SERVICE_NAME_BEGIN_INDEX, idx), func.substring(idx + 1)} + : new String[]{StringUtils.EMPTY, StringUtils.EMPTY}; + } + private void setDyeingKeyIfNonNull(RequestProtocol requestHeader, DefRequest request) { ByteString dyeingKeyByte = requestHeader.getTransInfoMap().get(TrpcTransInfoKeys.DYEING_KEY); if (dyeingKeyByte != null) { @@ -276,7 +367,7 @@ private CallInfo buildCallInfo(RpcInvocation rpcInvocation, TRPCProtocol.Request if (StringUtils.isBlank(cacheKey)) { return null; } - return CALL_INFO_CACHE.computeIfAbsent(cacheKey, s -> { + return getOrCompute(CALL_INFO_CACHE, cacheKey, s -> { CallInfo callInfo = new CallInfo(); fillCallerInfo(caller, callInfo); fillCalleeInfo(callee, rpcMethodName, callInfo); diff --git a/trpc-proto/trpc-proto-standard/src/test/java/com/tencent/trpc/proto/standard/common/StandardServerCodecCacheTest.java b/trpc-proto/trpc-proto-standard/src/test/java/com/tencent/trpc/proto/standard/common/StandardServerCodecCacheTest.java new file mode 100644 index 0000000000..6855d36946 --- /dev/null +++ b/trpc-proto/trpc-proto-standard/src/test/java/com/tencent/trpc/proto/standard/common/StandardServerCodecCacheTest.java @@ -0,0 +1,297 @@ +/* + * Tencent is pleased to support the open source community by making tRPC available. + * + * Copyright (C) 2023 Tencent. + * All rights reserved. + * + * If you have downloaded a copy of the tRPC source code from Tencent, + * please note that tRPC source code is licensed under the Apache 2.0 License, + * A copy of the Apache 2.0 License can be found in the LICENSE file. + */ + +package com.tencent.trpc.proto.standard.common; + +import com.github.benmanes.caffeine.cache.Cache; +import com.google.protobuf.ByteString; +import com.tencent.trpc.core.common.config.ProtocolConfig; +import com.tencent.trpc.core.rpc.CallInfo; +import com.tencent.trpc.core.rpc.Request; +import com.tencent.trpc.proto.standard.common.TRPCProtocol.RequestProtocol; +import com.tencent.trpc.transport.netty.NettyChannel; +import com.tencent.trpc.transport.netty.NettyChannelBuffer; +import io.netty.buffer.UnpooledByteBufAllocator; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.function.Function; +import org.apache.commons.lang3.StringUtils; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Test the bounded decoding caches of {@link StandardServerCodec}, which protect the server from the memory + * exhaustion attack: the cache keys(func/caller/callee) are fully controlled by the remote peer and are written + * before the service/method existence check. + */ +public class StandardServerCodecCacheTest { + + private static final String FUNC_INFO_CACHE_FIELD = "FUNC_INFO_CACHE"; + + private static final String CALL_INFO_CACHE_FIELD = "CALL_INFO_CACHE"; + + private static final String CACHE_MAX_SIZE_FIELD = "CACHE_MAX_SIZE"; + + private static final String CACHE_KEY_MAX_LENGTH_FIELD = "CACHE_KEY_MAX_LENGTH"; + + private static final String GET_OR_COMPUTE_METHOD = "getOrCompute"; + + private static final String PARSE_FUNC_METHOD = "parseFunc"; + + private static final String SERVICE_NAME = "helloservice"; + + private static final String METHOD_NAME = "sayHello"; + + private static final String FUNC = "/" + SERVICE_NAME + "/" + METHOD_NAME; + + private static final String CALLER = "trpc.callerApp.callerServer.callerService"; + + private static final String CALLEE = "trpc.calleeApp.calleeServer.calleeService.calleeMethod"; + + private static final int BUFFER_SIZE = 65535; + + private static final int LOCAL_PORT = 125; + + private static final String LOCAL_IP = "127.0.0.1"; + + private Cache funcInfoCache; + + private Cache callInfoCache; + + private int cacheMaxSize; + + private int cacheKeyMaxLength; + + @BeforeEach + public void before() throws Exception { + funcInfoCache = getStaticField(FUNC_INFO_CACHE_FIELD); + callInfoCache = getStaticField(CALL_INFO_CACHE_FIELD); + cacheMaxSize = getStaticField(CACHE_MAX_SIZE_FIELD); + cacheKeyMaxLength = getStaticField(CACHE_KEY_MAX_LENGTH_FIELD); + funcInfoCache.invalidateAll(); + callInfoCache.invalidateAll(); + funcInfoCache.cleanUp(); + callInfoCache.cleanUp(); + } + + @Test + public void testParseFuncWithLegalFunc() throws Exception { + String[] funcInfo = parseFunc(FUNC); + Assertions.assertEquals(SERVICE_NAME, funcInfo[0]); + Assertions.assertEquals(METHOD_NAME, funcInfo[1]); + // only the last separator is used to split the service and the method + String[] multiLevel = parseFunc("/a/b/c"); + Assertions.assertEquals("a/b", multiLevel[0]); + Assertions.assertEquals("c", multiLevel[1]); + } + + @Test + public void testParseFuncWithIllegalFunc() throws Exception { + String[] illegalFuncs = new String[]{StringUtils.EMPTY, "/", "//", "/abc", "abc", "/abc/"}; + for (String illegalFunc : illegalFuncs) { + String[] funcInfo = parseFunc(illegalFunc); + Assertions.assertEquals(StringUtils.EMPTY, funcInfo[0], illegalFunc); + Assertions.assertEquals(StringUtils.EMPTY, funcInfo[1], illegalFunc); + } + } + + @Test + public void testGetOrComputeReusesCachedValue() throws Exception { + String[] first = getOrCompute(funcInfoCache, FUNC); + String[] second = getOrCompute(funcInfoCache, FUNC); + Assertions.assertSame(first, second); + Assertions.assertEquals(1, funcInfoCache.estimatedSize()); + Assertions.assertNotNull(funcInfoCache.getIfPresent(FUNC)); + } + + @Test + public void testGetOrComputeCachesKeyOfMaxLength() throws Exception { + String key = buildFunc(cacheKeyMaxLength); + Assertions.assertEquals(cacheKeyMaxLength, key.length()); + String[] value = getOrCompute(funcInfoCache, key); + Assertions.assertEquals(METHOD_NAME, value[1]); + Assertions.assertEquals(1, funcInfoCache.estimatedSize()); + Assertions.assertSame(value, funcInfoCache.getIfPresent(key)); + } + + @Test + public void testGetOrComputeNeverCachesOversizedKey() throws Exception { + String key = buildFunc(cacheKeyMaxLength + 1); + Assertions.assertEquals(cacheKeyMaxLength + 1, key.length()); + String[] first = getOrCompute(funcInfoCache, key); + String[] second = getOrCompute(funcInfoCache, key); + // the value is still computed correctly, but it is never cached + Assertions.assertEquals(METHOD_NAME, first[1]); + Assertions.assertNotSame(first, second); + Assertions.assertNull(funcInfoCache.getIfPresent(key)); + Assertions.assertEquals(0, funcInfoCache.estimatedSize()); + } + + @Test + public void testCacheIsBoundedBySize() throws Exception { + int total = cacheMaxSize * 2; + for (int i = 0; i < total; i++) { + getOrCompute(funcInfoCache, "/" + SERVICE_NAME + i + "/" + METHOD_NAME); + } + funcInfoCache.cleanUp(); + Assertions.assertTrue(funcInfoCache.estimatedSize() <= cacheMaxSize, + "cache size should be bounded, but was " + funcInfoCache.estimatedSize()); + } + + @Test + public void testHotKeyIsKeptWhileCacheIsFlooded() throws Exception { + String[] hot = getOrCompute(funcInfoCache, FUNC); + // the hot key is accessed much more frequently than every flooding key + for (int i = 0; i < cacheMaxSize * 2; i++) { + getOrCompute(funcInfoCache, FUNC); + getOrCompute(funcInfoCache, "/" + SERVICE_NAME + i + "/" + METHOD_NAME); + } + funcInfoCache.cleanUp(); + Assertions.assertTrue(funcInfoCache.estimatedSize() <= cacheMaxSize); + Assertions.assertSame(hot, funcInfoCache.getIfPresent(FUNC)); + } + + @Test + public void testDecodeParsesAndCachesFuncInfo() { + Request request = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + Assertions.assertEquals(FUNC, request.getInvocation().getFunc()); + Assertions.assertEquals(SERVICE_NAME, request.getInvocation().getRpcServiceName()); + Assertions.assertEquals(METHOD_NAME, request.getInvocation().getRpcMethodName()); + Assertions.assertNotNull(funcInfoCache.getIfPresent(FUNC)); + Assertions.assertEquals(1, funcInfoCache.estimatedSize()); + // the second decoding of the same func reuses the cached entry + Request another = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + Assertions.assertEquals(SERVICE_NAME, another.getInvocation().getRpcServiceName()); + Assertions.assertEquals(1, funcInfoCache.estimatedSize()); + } + + @Test + public void testDecodeParsesAndCachesCallInfo() { + Request request = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + CallInfo callInfo = request.getMeta().getCallInfo(); + Assertions.assertEquals(CALLER, callInfo.getCaller()); + Assertions.assertEquals("callerApp", callInfo.getCallerApp()); + Assertions.assertEquals("callerServer", callInfo.getCallerServer()); + Assertions.assertEquals("callerService", callInfo.getCallerService()); + Assertions.assertEquals(CALLEE, callInfo.getCallee()); + Assertions.assertEquals("calleeApp", callInfo.getCalleeApp()); + Assertions.assertEquals("calleeServer", callInfo.getCalleeServer()); + Assertions.assertEquals("calleeService", callInfo.getCalleeService()); + Assertions.assertEquals("calleeMethod", callInfo.getCalleeMethod()); + Assertions.assertEquals(1, callInfoCache.estimatedSize()); + // the same caller/callee/method reuses the cached entry + Request another = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + Assertions.assertSame(callInfo, another.getMeta().getCallInfo()); + Assertions.assertEquals(1, callInfoCache.estimatedSize()); + } + + @Test + public void testDecodeWithOversizedFuncDoesNotPolluteCache() { + String func = buildFunc(cacheKeyMaxLength + 1); + Request request = decode(buildRequestHead(func, CALLER, CALLEE)); + Assertions.assertEquals(METHOD_NAME, request.getInvocation().getRpcMethodName()); + Assertions.assertNull(funcInfoCache.getIfPresent(func)); + Assertions.assertEquals(0, funcInfoCache.estimatedSize()); + } + + @Test + public void testDecodeWithOversizedCallerDoesNotPolluteCache() { + String caller = CALLER + StringUtils.repeat('x', cacheKeyMaxLength); + Request request = decode(buildRequestHead(FUNC, caller, CALLEE)); + Assertions.assertEquals(caller, request.getMeta().getCallInfo().getCaller()); + Assertions.assertEquals(0, callInfoCache.estimatedSize()); + } + + @Test + public void testDecodeWithBlankCallInfoIsNotCached() { + Request request = decode(buildRequestHead(StringUtils.EMPTY, StringUtils.EMPTY, StringUtils.EMPTY)); + Assertions.assertEquals(StringUtils.EMPTY, request.getInvocation().getRpcServiceName()); + Assertions.assertEquals(StringUtils.EMPTY, request.getInvocation().getRpcMethodName()); + Assertions.assertEquals(0, callInfoCache.estimatedSize()); + Assertions.assertNull(callInfoCache.getIfPresent(StringUtils.EMPTY)); + } + + @Test + public void testDecodeWithDistinctFuncKeepsCachesBounded() { + int total = 100; + for (int i = 0; i < total; i++) { + decode(buildRequestHead("/" + SERVICE_NAME + i + "/" + METHOD_NAME, CALLER + i, CALLEE)); + } + funcInfoCache.cleanUp(); + callInfoCache.cleanUp(); + Assertions.assertTrue(funcInfoCache.estimatedSize() <= Math.min(total, cacheMaxSize)); + Assertions.assertTrue(callInfoCache.estimatedSize() <= Math.min(total, cacheMaxSize)); + } + + private Request decode(RequestProtocol requestHead) { + byte[] headBytes = requestHead.toByteArray(); + StandardPackage pkg = new StandardPackage(); + pkg.setHeadBytes(headBytes); + pkg.getFrame().setHeadSize(headBytes.length); + pkg.getFrame().setSize(StandardFrame.FRAME_SIZE + headBytes.length); + ProtocolConfig config = new ProtocolConfig(); + config.setIp(LOCAL_IP); + config.setPort(LOCAL_PORT); + config.setDefault(); + NettyChannel channel = new NettyChannel(null, config); + NettyChannelBuffer buffer = new NettyChannelBuffer(UnpooledByteBufAllocator.DEFAULT.buffer(BUFFER_SIZE)); + pkg.write(buffer); + return (Request) new StandardServerCodec().decode(channel, buffer); + } + + private RequestProtocol buildRequestHead(String func, String caller, String callee) { + return RequestProtocol.newBuilder() + .setFunc(ByteString.copyFromUtf8(func)) + .setCaller(ByteString.copyFromUtf8(caller)) + .setCallee(ByteString.copyFromUtf8(callee)) + .build(); + } + + /** + * Build a legal func whose total length is the given length. + * + * @param length the expected length of the func + * @return the func like {@code /xxx.../sayHello} + */ + private String buildFunc(int length) { + int padding = length - METHOD_NAME.length() - 2; + return "/" + StringUtils.repeat('x', padding) + "/" + METHOD_NAME; + } + + @SuppressWarnings("unchecked") + private static T getStaticField(String name) throws Exception { + Field field = StandardServerCodec.class.getDeclaredField(name); + field.setAccessible(true); + return (T) field.get(null); + } + + @SuppressWarnings("unchecked") + private static String[] getOrCompute(Cache cache, String key) throws Exception { + Method method = StandardServerCodec.class.getDeclaredMethod(GET_OR_COMPUTE_METHOD, Cache.class, + String.class, Function.class); + method.setAccessible(true); + Function mappingFunction = func -> { + try { + return parseFunc(func); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }; + return (String[]) method.invoke(null, cache, key, mappingFunction); + } + + private static String[] parseFunc(String func) throws Exception { + Method method = StandardServerCodec.class.getDeclaredMethod(PARSE_FUNC_METHOD, String.class); + method.setAccessible(true); + return (String[]) method.invoke(null, func); + } +} diff --git a/trpc-spring-support/trpc-springmvc/src/main/java/com/tencent/trpc/springmvc/TRpcServiceHandlerMapping.java b/trpc-spring-support/trpc-springmvc/src/main/java/com/tencent/trpc/springmvc/TRpcServiceHandlerMapping.java index dbf4df98ff..126d362d5a 100644 --- a/trpc-spring-support/trpc-springmvc/src/main/java/com/tencent/trpc/springmvc/TRpcServiceHandlerMapping.java +++ b/trpc-spring-support/trpc-springmvc/src/main/java/com/tencent/trpc/springmvc/TRpcServiceHandlerMapping.java @@ -89,7 +89,7 @@ public void onApplicationEvent(ContextRefreshedEvent event) { @Override protected RpcMethodInfoAndInvoker getHandlerInternal(HttpServletRequest request) { String requestPath = request.getRequestURI(); - logger.debug("got trpc springmvc request {}, basePath is {}", requestPath); + logger.debug("got trpc springmvc request {}", requestPath); String method = request.getMethod(); if (!TRpcHttpConstants.HTTP_METHOD_GET.equals(method) && !TRpcHttpConstants.HTTP_METHOD_POST.equals(method)) { diff --git a/trpc-spring-support/trpc-springmvc/src/test/java/com/tencent/trpc/springmvc/TRpcServiceHandlerMappingTest.java b/trpc-spring-support/trpc-springmvc/src/test/java/com/tencent/trpc/springmvc/TRpcServiceHandlerMappingTest.java new file mode 100644 index 0000000000..dbdae99d50 --- /dev/null +++ b/trpc-spring-support/trpc-springmvc/src/test/java/com/tencent/trpc/springmvc/TRpcServiceHandlerMappingTest.java @@ -0,0 +1,105 @@ +/* + * Tencent is pleased to support the open source community by making tRPC available. + * + * Copyright (C) 2023 Tencent. + * All rights reserved. + * + * If you have downloaded a copy of the tRPC source code from Tencent, + * please note that tRPC source code is licensed under the Apache 2.0 License, + * A copy of the Apache 2.0 License can be found in the LICENSE file. + */ + +package com.tencent.trpc.springmvc; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +import com.tencent.trpc.core.rpc.common.RpcMethodInfoAndInvoker; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.mock.web.MockHttpServletRequest; + +/** + * Test {@link TRpcServiceHandlerMapping#getHandlerInternal}, which logs the request path before routing. The log + * message must not leave any unresolved placeholder, otherwise the log is polluted and the values are lost. + */ +public class TRpcServiceHandlerMappingTest { + + private static final String REQUEST_PATH = "/trpc.test.Greeter/sayHello"; + + private TRpcServiceHandlerMapping handlerMapping; + + @BeforeEach + public void before() { + handlerMapping = new TRpcServiceHandlerMapping(); + } + + /** + * A GET request goes through the routing, no route is registered so null is returned. The point of this case is + * that the debug log of the request path is executed without any exception. + */ + @Test + public void testGetHandlerInternalWithGet() { + MockHttpServletRequest request = new MockHttpServletRequest( + TRpcHttpConstants.HTTP_METHOD_GET, REQUEST_PATH); + request.setRequestURI(REQUEST_PATH); + assertNull(handlerMapping.getHandlerInternal(request)); + } + + /** + * A POST request also goes through the routing. + */ + @Test + public void testGetHandlerInternalWithPost() { + MockHttpServletRequest request = new MockHttpServletRequest( + TRpcHttpConstants.HTTP_METHOD_POST, REQUEST_PATH); + request.setRequestURI(REQUEST_PATH); + assertNull(handlerMapping.getHandlerInternal(request)); + } + + /** + * A request path containing a percent sign is a valid path(a percent encoded path), and it must not break the + * logging of the request path. + */ + @Test + public void testGetHandlerInternalWithPercentEncodedPath() { + String path = "/trpc.test.Greeter/sayHello%2Fabc"; + MockHttpServletRequest request = new MockHttpServletRequest( + TRpcHttpConstants.HTTP_METHOD_GET, path); + request.setRequestURI(path); + assertNull(handlerMapping.getHandlerInternal(request)); + } + + /** + * Only GET and POST are supported, the other methods are rejected before the routing. + */ + @Test + public void testGetHandlerInternalWithUnsupportedMethod() { + MockHttpServletRequest request = new MockHttpServletRequest("DELETE", REQUEST_PATH); + request.setRequestURI(REQUEST_PATH); + assertNull(handlerMapping.getHandlerInternal(request)); + } + + /** + * When the path is not registered, the service and the method can also be taken from the request parameters. + */ + @Test + public void testGetHandlerInternalWithServiceAndMethodParameter() { + MockHttpServletRequest request = new MockHttpServletRequest( + TRpcHttpConstants.HTTP_METHOD_POST, REQUEST_PATH); + request.setRequestURI(REQUEST_PATH); + request.setParameter(TRpcHttpConstants.TRPC_PARAM_SERVICE, "trpc.test.Greeter"); + request.setParameter(TRpcHttpConstants.TRPC_PARAM_METHOD, "sayHello"); + RpcMethodInfoAndInvoker route = handlerMapping.getHandlerInternal(request); + // nothing is registered, so no route is found, but both routing branches have been executed + assertNull(route); + } + + @Test + public void testHandlerMappingOrder() { + assertNotNull(handlerMapping); + // the order is set in the constructor so that the tRPC mapping takes precedence + org.junit.jupiter.api.Assertions.assertEquals(org.springframework.core.Ordered.HIGHEST_PRECEDENCE + 50000, + handlerMapping.getOrder()); + } +}