-
Notifications
You must be signed in to change notification settings - Fork 89
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[controller] Add server-side audit logger for controller gRPC server (#…
…1446) This commit introduces a server-side audit logging interceptor for the controller gRPC server. The `ControllerGrpcAuditLoggingInterceptor` logs details about incoming and outgoing gRPC calls, including the API method name, server address, client address, cluster name, store name, request latency, and response status. Incoming requests are logged with `[AUDIT][gRPC][IN]`, while outgoing responses are logged with `[AUDIT][gRPC][OUT]`.
- Loading branch information
1 parent
3df5859
commit a792e65
Showing
12 changed files
with
245 additions
and
19 deletions.
There are no files selected for viewing
6 changes: 3 additions & 3 deletions
6
...c/integrationTest/java/com/linkedin/venice/controller/TestControllerSecureGrpcServer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 19 additions & 0 deletions
19
...controller/src/main/java/com/linkedin/venice/controller/grpc/ControllerGrpcConstants.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| package com.linkedin.venice.controller.grpc; | ||
|
|
||
| import com.linkedin.venice.controller.grpc.server.GrpcControllerClientDetails; | ||
| import io.grpc.Context; | ||
| import io.grpc.Metadata; | ||
|
|
||
|
|
||
| final public class ControllerGrpcConstants { | ||
| public static final Metadata.Key<String> CLUSTER_NAME_METADATA_KEY = | ||
| Metadata.Key.of("cluster-name", Metadata.ASCII_STRING_MARSHALLER); | ||
| public static final Metadata.Key<String> STORE_NAME_METADATA_KEY = | ||
| Metadata.Key.of("store-name", Metadata.ASCII_STRING_MARSHALLER); | ||
| public static final Context.Key<GrpcControllerClientDetails> GRPC_CONTROLLER_CLIENT_DETAILS = | ||
| Context.key("controller-client-details"); | ||
| public static final String UNKNOWN_REMOTE_ADDRESS = "unknown"; | ||
|
|
||
| private ControllerGrpcConstants() { | ||
| } | ||
| } |
2 changes: 1 addition & 1 deletion
2
...ver/grpc/GrpcControllerClientDetails.java → ...c/server/GrpcControllerClientDetails.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
77 changes: 77 additions & 0 deletions
77
...edin/venice/controller/grpc/server/interceptor/ControllerGrpcAuditLoggingInterceptor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| package com.linkedin.venice.controller.grpc.server.interceptor; | ||
|
|
||
| import com.linkedin.venice.controller.grpc.ControllerGrpcConstants; | ||
| import com.linkedin.venice.utils.LatencyUtils; | ||
| import io.grpc.ForwardingServerCall.SimpleForwardingServerCall; | ||
| import io.grpc.Grpc; | ||
| import io.grpc.Metadata; | ||
| import io.grpc.ServerCall; | ||
| import io.grpc.ServerCallHandler; | ||
| import io.grpc.ServerInterceptor; | ||
| import io.grpc.Status; | ||
| import java.net.SocketAddress; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
|
|
||
|
|
||
| /** | ||
| * A gRPC server interceptor for audit logging. | ||
| * | ||
| * <p>This interceptor logs incoming and outgoing gRPC calls, including the API method name, | ||
| * server address, client address, cluster name, store name, and request latency. It is useful | ||
| * for debugging and monitoring gRPC requests and responses.</p> | ||
| * | ||
| */ | ||
| public class ControllerGrpcAuditLoggingInterceptor implements ServerInterceptor { | ||
| private static final Logger LOGGER = LogManager.getLogger(ControllerGrpcAuditLoggingInterceptor.class); | ||
|
|
||
| @Override | ||
| public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall( | ||
| ServerCall<ReqT, RespT> serverCall, | ||
| Metadata headers, | ||
| ServerCallHandler<ReqT, RespT> next) { | ||
|
|
||
| // Extract details for logging | ||
| String apiName = serverCall.getMethodDescriptor().getBareMethodName(); | ||
| String serverAddr = getAddress(serverCall.getAttributes().get(Grpc.TRANSPORT_ATTR_LOCAL_ADDR)); | ||
| String clientAddr = getAddress(serverCall.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR)); | ||
| String clusterName = headers.get(ControllerGrpcConstants.CLUSTER_NAME_METADATA_KEY); | ||
| String storeName = headers.get(ControllerGrpcConstants.STORE_NAME_METADATA_KEY); | ||
|
|
||
| LOGGER.info( | ||
| "[AUDIT][gRPC][IN] api={}, serverAddr={}, clientAddr={}, clusterName={}, storeName={}", | ||
| apiName, | ||
| serverAddr, | ||
| clientAddr, | ||
| clusterName, | ||
| storeName); | ||
|
|
||
| // Start time for latency calculation | ||
| long startTime = System.currentTimeMillis(); | ||
|
|
||
| // Wrap the server call to log response status | ||
| SimpleForwardingServerCall<ReqT, RespT> auditingServerCall = | ||
| new SimpleForwardingServerCall<ReqT, RespT>(serverCall) { | ||
| @Override | ||
| public void close(Status status, Metadata trailers) { | ||
| LOGGER.info( | ||
| "[AUDIT][gRPC][OUT] api={}, serverAddr={}, clientAddr={}, clusterName={}, storeName={}, status={}, latencyMs={}", | ||
| apiName, | ||
| serverAddr, | ||
| clientAddr, | ||
| clusterName, | ||
| storeName, | ||
| status.getCode(), | ||
| LatencyUtils.getElapsedTimeFromMsToMs(startTime)); | ||
| super.close(status, trailers); | ||
| } | ||
| }; | ||
|
|
||
| // Proceed with the call | ||
| return next.startCall(auditingServerCall, headers); | ||
| } | ||
|
|
||
| private String getAddress(SocketAddress address) { | ||
| return address != null ? address.toString() : ControllerGrpcConstants.UNKNOWN_REMOTE_ADDRESS; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...ontrollerRegionValidationInterceptor.java → ...ontrollerRegionValidationInterceptor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
4 changes: 2 additions & 2 deletions
4
.../src/main/java/com/linkedin/venice/controller/server/VeniceControllerGrpcServiceImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...grpc/GrpcControllerClientDetailsTest.java → ...rver/GrpcControllerClientDetailsTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
127 changes: 127 additions & 0 deletions
127
.../venice/controller/grpc/server/interceptor/ControllerGrpcAuditLoggingInterceptorTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package com.linkedin.venice.controller.grpc.server.interceptor; | ||
|
|
||
| import static org.mockito.Mockito.any; | ||
| import static org.mockito.Mockito.eq; | ||
| import static org.mockito.Mockito.mock; | ||
| import static org.mockito.Mockito.times; | ||
| import static org.mockito.Mockito.verify; | ||
| import static org.mockito.Mockito.when; | ||
| import static org.testng.Assert.assertEquals; | ||
| import static org.testng.Assert.assertNotNull; | ||
|
|
||
| import com.linkedin.venice.controller.grpc.ControllerGrpcConstants; | ||
| import io.grpc.Attributes; | ||
| import io.grpc.Grpc; | ||
| import io.grpc.Metadata; | ||
| import io.grpc.MethodDescriptor; | ||
| import io.grpc.MethodDescriptor.Marshaller; | ||
| import io.grpc.ServerCall; | ||
| import io.grpc.ServerCallHandler; | ||
| import io.grpc.Status; | ||
| import java.net.InetSocketAddress; | ||
| import java.net.SocketAddress; | ||
| import org.mockito.ArgumentCaptor; | ||
| import org.testng.annotations.BeforeMethod; | ||
| import org.testng.annotations.Test; | ||
|
|
||
|
|
||
| public class ControllerGrpcAuditLoggingInterceptorTest { | ||
| private ControllerGrpcAuditLoggingInterceptor interceptor; | ||
| private ServerCall<String, String> mockServerCall; | ||
| private Metadata mockMetadata; | ||
| private ServerCallHandler<String, String> mockHandler; | ||
|
|
||
| @BeforeMethod | ||
| public void setUp() { | ||
| interceptor = new ControllerGrpcAuditLoggingInterceptor(); | ||
| mockServerCall = mock(ServerCall.class); | ||
| mockMetadata = new Metadata(); | ||
| mockHandler = mock(ServerCallHandler.class); | ||
|
|
||
| // Create a type-safe MethodDescriptor with mock marshallers | ||
| MethodDescriptor.Marshaller<String> stringMarshaller = mock(Marshaller.class); | ||
|
|
||
| MethodDescriptor<String, String> mockMethodDescriptor = MethodDescriptor.<String, String>newBuilder() | ||
| .setFullMethodName("f.q.m.TestService/TestMethod") | ||
| .setType(MethodDescriptor.MethodType.UNARY) | ||
| .setRequestMarshaller(stringMarshaller) | ||
| .setResponseMarshaller(stringMarshaller) | ||
| .build(); | ||
|
|
||
| when(mockServerCall.getMethodDescriptor()).thenReturn(mockMethodDescriptor); | ||
| } | ||
|
|
||
| @Test | ||
| public void testInterceptCallWithValidAddressesAndMetadata() { | ||
| // Set up attributes and metadata | ||
| SocketAddress serverAddr = new InetSocketAddress("127.0.0.1", 8080); | ||
| SocketAddress clientAddr = new InetSocketAddress("192.168.1.1", 12345); | ||
| Attributes attributes = Attributes.newBuilder() | ||
| .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, serverAddr) | ||
| .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, clientAddr) | ||
| .build(); | ||
| when(mockServerCall.getAttributes()).thenReturn(attributes); | ||
| mockMetadata.put(ControllerGrpcConstants.CLUSTER_NAME_METADATA_KEY, "TestCluster"); | ||
| mockMetadata.put(ControllerGrpcConstants.STORE_NAME_METADATA_KEY, "TestStore"); | ||
|
|
||
| // Invoke the interceptor | ||
| interceptor.interceptCall(mockServerCall, mockMetadata, mockHandler); | ||
|
|
||
| // Verify logging behavior | ||
| ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class); | ||
| verify(mockHandler, times(1)).startCall(any(), metadataCaptor.capture()); | ||
| assertEquals(metadataCaptor.getValue(), mockMetadata); | ||
| } | ||
|
|
||
| @Test | ||
| public void testInterceptCallWithMissingMetadata() { | ||
| // Set up attributes without metadata | ||
| SocketAddress serverAddr = new InetSocketAddress("127.0.0.1", 8080); | ||
| SocketAddress clientAddr = new InetSocketAddress("192.168.1.1", 12345); | ||
| Attributes attributes = Attributes.newBuilder() | ||
| .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, serverAddr) | ||
| .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, clientAddr) | ||
| .build(); | ||
| when(mockServerCall.getAttributes()).thenReturn(attributes); | ||
|
|
||
| // Invoke the interceptor | ||
| interceptor.interceptCall(mockServerCall, mockMetadata, mockHandler); | ||
|
|
||
| // Ensure no exceptions occur and verify log format | ||
| verify(mockHandler, times(1)).startCall(any(), any()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testResponseLogging() { | ||
| // Set up attributes and metadata | ||
| SocketAddress serverAddr = new InetSocketAddress("127.0.0.1", 8080); | ||
| SocketAddress clientAddr = new InetSocketAddress("192.168.1.1", 12345); | ||
| Attributes attributes = Attributes.newBuilder() | ||
| .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, serverAddr) | ||
| .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, clientAddr) | ||
| .build(); | ||
| when(mockServerCall.getAttributes()).thenReturn(attributes); | ||
|
|
||
| mockMetadata.put(ControllerGrpcConstants.CLUSTER_NAME_METADATA_KEY, "TestCluster"); | ||
| mockMetadata.put(ControllerGrpcConstants.STORE_NAME_METADATA_KEY, "TestStore"); | ||
|
|
||
| // Mock ServerCallHandler to return a dummy Listener | ||
| ServerCall.Listener<String> mockListener = mock(ServerCall.Listener.class); | ||
| when(mockHandler.startCall(any(), any())).thenReturn(mockListener); | ||
|
|
||
| // Capture and verify the listener | ||
| ServerCall.Listener<String> returnedListener = interceptor.interceptCall(mockServerCall, mockMetadata, mockHandler); | ||
| assertNotNull(returnedListener, "The returned listener should not be null."); | ||
| assertEquals(returnedListener, mockListener, "The returned listener should match the mock listener."); | ||
|
|
||
| // Verify response logging behavior | ||
| ArgumentCaptor<ServerCall<String, String>> callCaptor = ArgumentCaptor.forClass(ServerCall.class); | ||
| verify(mockHandler, times(1)).startCall(callCaptor.capture(), eq(mockMetadata)); | ||
|
|
||
| ServerCall<String, String> capturedCall = callCaptor.getValue(); | ||
| assertNotNull(capturedCall, "Captured ServerCall should not be null."); | ||
|
|
||
| capturedCall.close(Status.OK, new Metadata()); | ||
| verify(mockServerCall, times(1)).close(eq(Status.OK), any()); | ||
| } | ||
| } |
2 changes: 1 addition & 1 deletion
2
...trollerGrpcSslSessionInterceptorTest.java → ...trollerGrpcSslSessionInterceptorTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...ollerRegionValidationInterceptorTest.java → ...ollerRegionValidationInterceptorTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters