diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java index abc23a53a3e..a62493c75c7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java @@ -90,6 +90,16 @@ public ClientActivity(MessagingProcessor messagingProcessor, this.init(); } + static String summarizeSettings(Settings settings) { + if (settings == null) { + return "null"; + } + int publishingTopicCount = settings.hasPublishing() ? settings.getPublishing().getTopicsCount() : 0; + int subscriptionCount = settings.hasSubscription() ? settings.getSubscription().getSubscriptionsCount() : 0; + return String.format("clientType=%s, publishingTopicCount=%d, subscriptionCount=%d", + settings.getClientType(), publishingTopicCount, subscriptionCount); + } + protected void init() { this.messagingProcessor.registerConsumerListener(new ConsumerIdsChangeListenerImpl()); this.messagingProcessor.registerProducerListener(new ProducerChangeListenerImpl()); @@ -362,11 +372,39 @@ protected void processTelemetryException(TelemetryCommand request, Throwable t, } } if (exception.getStatus().getCode().equals(io.grpc.Status.Code.INTERNAL)) { - log.warn("process client telemetryCommand failed. request:{}", request, t); + log.warn("process client telemetryCommand failed. request:{}", summarizeTelemetryCommand(request), t); } responseObserver.onError(exception); } + static String summarizeTelemetryCommand(TelemetryCommand request) { + if (request == null) { + return "null"; + } + + StringBuilder builder = new StringBuilder(request.getCommandCase().name()); + if (request.hasStatus()) { + builder.append(", statusCode=").append(request.getStatus().getCode()); + } + switch (request.getCommandCase()) { + case SETTINGS: + builder.append(", clientType=").append(request.getSettings().getClientType()) + .append(", pubSubCase=").append(request.getSettings().getPubSubCase()); + break; + case THREAD_STACK_TRACE: + builder.append(", nonce=").append(request.getThreadStackTrace().getNonce()) + .append(", details omitted"); + break; + case VERIFY_MESSAGE_RESULT: + builder.append(", nonce=").append(request.getVerifyMessageResult().getNonce()); + break; + default: + // Add explicit cases for future telemetry requests that carry sensitive details. + break; + } + return builder.toString(); + } + protected void processAndWriteClientSettings(ProxyContext ctx, TelemetryCommand request, StreamObserver responseObserver) { GrpcClientChannel grpcClientChannel = null; @@ -597,7 +635,8 @@ protected void processClientRegister(String group, Object... args) { if (ChannelHelper.isRemote(channel)) { // save settings from channel sync from other proxy Settings settings = GrpcClientChannel.parseChannelExtendAttribute(channel); - log.debug("save client settings sync from other proxy. group:{}, channelInfo:{}, settings:{}", group, clientChannelInfo, settings); + log.debug("save client settings sync from other proxy. group:{}, channelInfo:{}, settingsSummary:{}", + group, clientChannelInfo, summarizeSettings(settings)); if (settings == null) { return; } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java index e215c6efaba..d042dd74002 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java @@ -71,6 +71,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; @@ -184,6 +185,42 @@ protected HeartbeatResponse sendConsumerHeartbeat(ProxyContext context) throws T .build()).get(); } + @Test + public void testSummarizeSettingsDoesNotExposeResourceNames() { + Settings producerSettings = Settings.newBuilder() + .setClientType(ClientType.PRODUCER) + .setPublishing(Publishing.newBuilder() + .addTopics(Resource.newBuilder().setName("sensitive-publish-topic").build()) + .build()) + .build(); + Settings consumerSettings = Settings.newBuilder() + .setClientType(ClientType.PUSH_CONSUMER) + .setSubscription(Subscription.newBuilder() + .setGroup(Resource.newBuilder().setName("sensitive-group").build()) + .addSubscriptions(SubscriptionEntry.newBuilder() + .setExpression(FilterExpression.newBuilder() + .setExpression("sensitive-tag") + .setType(FilterType.TAG) + .build()) + .setTopic(Resource.newBuilder().setName("sensitive-subscription-topic").build()) + .build()) + .build()) + .build(); + + String producerSummary = ClientActivity.summarizeSettings(producerSettings); + String consumerSummary = ClientActivity.summarizeSettings(consumerSettings); + + assertThat(producerSummary).contains("clientType=PRODUCER"); + assertThat(producerSummary).contains("publishingTopicCount=1"); + assertThat(producerSummary).doesNotContain("sensitive-publish-topic"); + + assertThat(consumerSummary).contains("clientType=PUSH_CONSUMER"); + assertThat(consumerSummary).contains("subscriptionCount=1"); + assertThat(consumerSummary).doesNotContain("sensitive-subscription-topic"); + assertThat(consumerSummary).doesNotContain("sensitive-group"); + assertThat(consumerSummary).doesNotContain("sensitive-tag"); + } + @Test public void testConsumerHeartbeat() throws Throwable { ProxyContext context = createContext(); @@ -416,6 +453,25 @@ public void onCompleted() { assertThat(result.getResult().getConsumeResult()).isEqualTo(CMResult.CR_SUCCESS); } + @Test + public void testSummarizeTelemetryCommandDoesNotIncludeThreadStackTrace() { + TelemetryCommand command = TelemetryCommand.newBuilder() + .setThreadStackTrace(ThreadStackTrace.newBuilder() + .setNonce("nonce-1") + .setThreadStackTrace("secret-stack-trace") + .build()) + .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name())) + .build(); + + String summary = ClientActivity.summarizeTelemetryCommand(command); + + assertTrue(summary.contains("THREAD_STACK_TRACE")); + assertTrue(summary.contains("statusCode=OK")); + assertTrue(summary.contains("nonce-1")); + assertTrue(summary.contains("details omitted")); + assertFalse(summary.contains("secret-stack-trace")); + } + protected CompletableFuture sendClientTelemetry(ProxyContext ctx, Settings settings) { when(grpcClientSettingsManager.getClientSettings(any())).thenReturn(settings);