Skip to content

fix: summarize GrpcClientChannel telemetry diagnostics #10721

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Aias00 wants to merge 4 commits into
base: develop
Choose a base branch
from
+180 −7
Open

fix: summarize GrpcClientChannel telemetry diagnostics #10721

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
47149b9
[ISSUE #10720] Avoid raw settings conversion logs
Aias00 Jul 31, 2026
b0dcea5
[ISSUE #10672] Sanitize proxy telemetry failure logs
Aias00 Jul 29, 2026
ec42486
[ISSUE #10672] Address telemetry log review feedback
Aias00 Jul 29, 2026
a9f0ab0
[ISSUE #10672] Improve telemetry summary test coverage
Aias00 Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 55 additions & 6 deletions proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import com.google.common.base.MoreObjects;
import com.google.common.collect.ComparisonChain;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.TextFormat;
import com.google.protobuf.util.JsonFormat;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
Expand Down Expand Up @@ -88,7 +87,8 @@ public String getChannelExtendAttribute() {
try {
return JsonFormat.printer().print(settings);
} catch (InvalidProtocolBufferException e) {
log.error("convert settings to json data failed. settings:{}", settings, e);
log.error("convert settings to json data failed. clientId:{}, settingsSummary:{}",
this.clientId, summarizeSettings(settings), e);
}
return null;
}
Expand All @@ -106,13 +106,27 @@ public static Settings parseChannelExtendAttribute(Channel channel) {
JsonFormat.parser().merge(attr, builder);
return builder.build();
} catch (InvalidProtocolBufferException e) {
log.error("convert settings json data to settings failed. data:{}", attr, e);
log.error("convert settings json data to settings failed. attrLength:{}", getAttributeLength(attr), e);
return null;
}
}
return null;
}

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);
}

static int getAttributeLength(String attr) {
return attr == null ? 0 : attr.length();
}

@Override
public RemoteChannel toRemoteChannel() {
return new RemoteChannel(
Expand Down Expand Up @@ -263,24 +277,59 @@ public String getClientId() {
public void writeTelemetryCommand(TelemetryCommand command) {
StreamObserver<TelemetryCommand> observer = this.telemetryCommandRef.get();
if (observer == null) {
log.warn("telemetry command observer is null when try to write data. command:{}, channel:{}", TextFormat.shortDebugString(command), this);
log.warn("telemetry command observer is null when try to write data. command:{}, channel:{}",
summarizeTelemetryCommand(command), this);
return;
}
synchronized (this.telemetryWriteLock) {
observer = this.telemetryCommandRef.get();
if (observer == null) {
log.warn("telemetry command observer is null when try to write data. command:{}, channel:{}", TextFormat.shortDebugString(command), this);
log.warn("telemetry command observer is null when try to write data. command:{}, channel:{}",
summarizeTelemetryCommand(command), this);
return;
}
try {
observer.onNext(command);
} catch (StatusRuntimeException | IllegalStateException exception) {
log.warn("write telemetry failed. command:{}", command, exception);
log.warn("write telemetry failed. command:{}, channel:{}", summarizeTelemetryCommand(command), this, exception);
this.clearClientObserver(observer);
}
}
}

static String summarizeTelemetryCommand(TelemetryCommand command) {
if (command == null) {
return "null";
}

StringBuilder builder = new StringBuilder(command.getCommandCase().name());
switch (command.getCommandCase()) {
case PRINT_THREAD_STACK_TRACE_COMMAND:
builder.append(", nonce=").append(command.getPrintThreadStackTraceCommand().getNonce());
break;
case VERIFY_MESSAGE_COMMAND:
builder.append(", nonce=").append(command.getVerifyMessageCommand().getNonce());
break;
case RECOVER_ORPHANED_TRANSACTION_COMMAND:
builder.append(", transactionId=")
.append(command.getRecoverOrphanedTransactionCommand().getTransactionId())
.append(", details omitted");
break;
case NOTIFY_UNSUBSCRIBE_LITE_COMMAND:
builder.append(", liteTopic=")
.append(command.getNotifyUnsubscribeLiteCommand().getLiteTopic());
break;
case SETTINGS:
builder.append(", clientType=").append(command.getSettings().getClientType())
.append(", pubSubCase=").append(command.getSettings().getPubSubCase());
break;
default:
// Add explicit cases for future commands that carry sensitive payloads.
break;
}
return builder.toString();
}

@Override
public String toString() {
return MoreObjects.toStringHelper(this)
Expand Down
126 changes: 125 additions & 1 deletion proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannelTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,19 @@

package org.apache.rocketmq.proxy.grpc.v2.channel;

import apache.rocketmq.v2.Message;
import apache.rocketmq.v2.ClientType;
import apache.rocketmq.v2.NotifyUnsubscribeLiteCommand;
import apache.rocketmq.v2.Publishing;
import apache.rocketmq.v2.PrintThreadStackTraceCommand;
import apache.rocketmq.v2.RecoverOrphanedTransactionCommand;
import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.Settings;
import apache.rocketmq.v2.Subscription;
import apache.rocketmq.v2.SubscriptionEntry;
import apache.rocketmq.v2.TelemetryCommand;
import apache.rocketmq.v2.VerifyMessageCommand;
import com.google.protobuf.ByteString;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.InitConfigTest;
Expand All @@ -35,7 +45,9 @@
import org.mockito.junit.MockitoJUnitRunner;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
Expand Down Expand Up @@ -79,4 +91,116 @@ public void testChannelExtendAttributeParse() {
assertEquals(clientSettings, GrpcClientChannel.parseChannelExtendAttribute(this.grpcClientChannel));
assertNull(GrpcClientChannel.parseChannelExtendAttribute(mock(RemotingChannel.class)));
}
}

@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()
.setTopic(Resource.newBuilder().setName("sensitive-subscription-topic").build())
.build())
.build())
.build();

String producerSummary = GrpcClientChannel.summarizeSettings(producerSettings);
String consumerSummary = GrpcClientChannel.summarizeSettings(consumerSettings);

assertTrue(producerSummary.contains("clientType=PRODUCER"));
assertTrue(producerSummary.contains("publishingTopicCount=1"));
assertFalse(producerSummary.contains("sensitive-publish-topic"));

assertTrue(consumerSummary.contains("clientType=PUSH_CONSUMER"));
assertTrue(consumerSummary.contains("subscriptionCount=1"));
assertFalse(consumerSummary.contains("sensitive-subscription-topic"));
assertFalse(consumerSummary.contains("sensitive-group"));
}

@Test
public void testSummarizeTelemetryCommandDoesNotIncludeMessagePayload() {
TelemetryCommand command = TelemetryCommand.newBuilder()
.setVerifyMessageCommand(VerifyMessageCommand.newBuilder()
.setNonce("nonce-1")
.setMessage(Message.newBuilder()
.setBody(ByteString.copyFromUtf8("secret-body"))
.build())
.build())
.build();

String summary = GrpcClientChannel.summarizeTelemetryCommand(command);

assertTrue(summary.contains("VERIFY_MESSAGE_COMMAND"));
assertTrue(summary.contains("nonce-1"));
assertFalse(summary.contains("secret-body"));
assertFalse(summary.contains("message"));
assertFalse(summary.contains("body"));
}

@Test
public void testSummarizeRecoverTransactionCommandDoesNotIncludeMessagePayload() {
TelemetryCommand command = TelemetryCommand.newBuilder()
.setRecoverOrphanedTransactionCommand(RecoverOrphanedTransactionCommand.newBuilder()
.setTransactionId("transaction-id")
.setMessage(Message.newBuilder()
.setBody(ByteString.copyFromUtf8("secret-body"))
.build())
.build())
.build();

String summary = GrpcClientChannel.summarizeTelemetryCommand(command);

assertTrue(summary.contains("RECOVER_ORPHANED_TRANSACTION_COMMAND"));
assertTrue(summary.contains("transaction-id"));
assertTrue(summary.contains("details omitted"));
assertFalse(summary.contains("secret-body"));
assertFalse(summary.contains("message"));
assertFalse(summary.contains("body"));
}

@Test
public void testSummarizeTelemetryCommandDiagnosticFields() {
assertEquals("null", GrpcClientChannel.summarizeTelemetryCommand(null));
assertEquals("COMMAND_NOT_SET", GrpcClientChannel.summarizeTelemetryCommand(TelemetryCommand.getDefaultInstance()));

TelemetryCommand settingsCommand = TelemetryCommand.newBuilder()
.setSettings(Settings.newBuilder()
.setPublishing(Publishing.getDefaultInstance())
.build())
.build();
String settingsSummary = GrpcClientChannel.summarizeTelemetryCommand(settingsCommand);
assertTrue(settingsSummary.contains("SETTINGS"));
assertTrue(settingsSummary.contains("clientType="));
assertTrue(settingsSummary.contains("pubSubCase=PUBLISHING"));

TelemetryCommand threadStackCommand = TelemetryCommand.newBuilder()
.setPrintThreadStackTraceCommand(PrintThreadStackTraceCommand.newBuilder()
.setNonce("stack-nonce")
.build())
.build();
String threadStackSummary = GrpcClientChannel.summarizeTelemetryCommand(threadStackCommand);
assertTrue(threadStackSummary.contains("PRINT_THREAD_STACK_TRACE_COMMAND"));
assertTrue(threadStackSummary.contains("stack-nonce"));

TelemetryCommand liteCommand = TelemetryCommand.newBuilder()
.setNotifyUnsubscribeLiteCommand(NotifyUnsubscribeLiteCommand.newBuilder()
.setLiteTopic("lite-topic")
.build())
.build();
String liteSummary = GrpcClientChannel.summarizeTelemetryCommand(liteCommand);
assertTrue(liteSummary.contains("NOTIFY_UNSUBSCRIBE_LITE_COMMAND"));
assertTrue(liteSummary.contains("lite-topic"));
}

@Test
public void testGetAttributeLength() {
assertEquals(0, GrpcClientChannel.getAttributeLength(null));
assertEquals(7, GrpcClientChannel.getAttributeLength("invalid"));
}
}
Loading