diff --git a/common/src/main/java/feast/common/logging/interceptors/GrpcMessageInterceptor.java b/common/src/main/java/feast/common/logging/interceptors/GrpcMessageInterceptor.java index da43a372d73..abfca86ecde 100644 --- a/common/src/main/java/feast/common/logging/interceptors/GrpcMessageInterceptor.java +++ b/common/src/main/java/feast/common/logging/interceptors/GrpcMessageInterceptor.java @@ -61,7 +61,12 @@ public GrpcMessageInterceptor(@Nullable SecurityProperties securityProperties) { public Listener interceptCall( ServerCall call, Metadata headers, ServerCallHandler next) { MessageAuditLogEntry.Builder entryBuilder = MessageAuditLogEntry.newBuilder(); - // default response message to empty proto in log entry. + // default response/request message to empty proto in log entry. + // request could be empty when the client closes the connection before sending a request + // message. + // response could be unset when the service encounters an error when processsing the service + // call. + entryBuilder.setRequest(Empty.newBuilder().build()); entryBuilder.setResponse(Empty.newBuilder().build()); // Unpack service & method name from call diff --git a/core/src/test/java/feast/core/logging/CoreLoggingIT.java b/core/src/test/java/feast/core/logging/CoreLoggingIT.java new file mode 100644 index 00000000000..53e55ebe695 --- /dev/null +++ b/core/src/test/java/feast/core/logging/CoreLoggingIT.java @@ -0,0 +1,232 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.core.logging; + +import static org.hamcrest.CoreMatchers.*; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import com.google.common.collect.Streams; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import com.google.protobuf.InvalidProtocolBufferException; +import com.google.protobuf.util.JsonFormat; +import feast.common.it.BaseIT; +import feast.common.it.DataGenerator; +import feast.common.logging.entry.AuditLogEntryKind; +import feast.proto.core.CoreServiceGrpc; +import feast.proto.core.CoreServiceGrpc.CoreServiceBlockingStub; +import feast.proto.core.CoreServiceGrpc.CoreServiceFutureStub; +import feast.proto.core.CoreServiceProto.GetFeastCoreVersionRequest; +import feast.proto.core.CoreServiceProto.ListFeatureSetsRequest; +import feast.proto.core.CoreServiceProto.ListStoresRequest; +import feast.proto.core.CoreServiceProto.ListStoresResponse; +import feast.proto.core.CoreServiceProto.UpdateStoreRequest; +import feast.proto.core.CoreServiceProto.UpdateStoreResponse; +import io.grpc.Channel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.Status.Code; +import io.grpc.StatusRuntimeException; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.stream.Collectors; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LoggerContext; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.test.context.SpringBootTest; + +@SpringBootTest( + properties = { + "feast.logging.audit.enabled=true", + "feast.logging.audit.messageLoggingEnabled=true", + }) +public class CoreLoggingIT extends BaseIT { + private static TestLogAppender testAuditLogAppender; + private static CoreServiceBlockingStub coreService; + private static CoreServiceFutureStub asyncCoreService; + + @BeforeAll + public static void globalSetUp(@Value("${grpc.server.port}") int coreGrpcPort) + throws InterruptedException, ExecutionException { + LoggerContext logContext = (LoggerContext) LogManager.getContext(false); + // NOTE: As log appender state is shared across tests use a different method + // for each test and filter by method name to ensure that you only get logs + // for a specific test. + testAuditLogAppender = logContext.getConfiguration().getAppender("TestAuditLogAppender"); + + // Connect to core service. + Channel channel = + ManagedChannelBuilder.forAddress("localhost", coreGrpcPort).usePlaintext().build(); + coreService = CoreServiceGrpc.newBlockingStub(channel); + asyncCoreService = CoreServiceGrpc.newFutureStub(channel); + + // Preflight a request to core service stubs to verify connection + coreService.getFeastCoreVersion(GetFeastCoreVersionRequest.getDefaultInstance()); + asyncCoreService.getFeastCoreVersion(GetFeastCoreVersionRequest.getDefaultInstance()).get(); + } + + /** Check that messsage audit log are produced on service call */ + @Test + public void shouldProduceMessageAuditLogsOnCall() + throws InterruptedException, InvalidProtocolBufferException { + // Generate artifical load on feast core. + UpdateStoreRequest request = + UpdateStoreRequest.newBuilder().setStore(DataGenerator.getDefaultStore()).build(); + UpdateStoreResponse response = coreService.updateStore(request); + + // Wait required to ensure audit logs are flushed into test audit log appender + Thread.sleep(1000); + // Check message audit logs are produced for each audit log. + JsonFormat.Parser protoJSONParser = JsonFormat.parser(); + // Pull message audit logs logs from test log appender + List logJsonObjects = + parseMessageJsonLogObjects(testAuditLogAppender.getLogs(), "UpdateStore"); + assertEquals(1, logJsonObjects.size()); + JsonObject logObj = logJsonObjects.get(0); + + // Extract & Check that request/response are returned correctly + String requestJson = logObj.getAsJsonObject("request").toString(); + UpdateStoreRequest.Builder gotRequest = UpdateStoreRequest.newBuilder(); + protoJSONParser.merge(requestJson, gotRequest); + + String responseJson = logObj.getAsJsonObject("response").toString(); + UpdateStoreResponse.Builder gotResponse = UpdateStoreResponse.newBuilder(); + protoJSONParser.merge(responseJson, gotResponse); + + assertThat(gotRequest.build(), equalTo(request)); + assertThat(gotResponse.build(), equalTo(response)); + } + + /** Check that message audit logs are produced when server encounters an error */ + @Test + public void shouldProduceMessageAuditLogsOnError() throws InterruptedException { + // Send a bad request which should cause Core to error + ListFeatureSetsRequest request = + ListFeatureSetsRequest.newBuilder() + .setFilter( + ListFeatureSetsRequest.Filter.newBuilder() + .setProject("*") + .setFeatureSetName("nop") + .build()) + .build(); + + boolean hasExpectedException = false; + Code statusCode = null; + try { + coreService.listFeatureSets(request); + } catch (StatusRuntimeException e) { + hasExpectedException = true; + statusCode = e.getStatus().getCode(); + } + assertTrue(hasExpectedException); + + // Wait required to ensure audit logs are flushed into test audit log appender + Thread.sleep(1000); + // Pull message audit logs logs from test log appender + List logJsonObjects = + parseMessageJsonLogObjects(testAuditLogAppender.getLogs(), "ListFeatureSets"); + + assertEquals(1, logJsonObjects.size()); + JsonObject logJsonObject = logJsonObjects.get(0); + // Check correct status code is tracked on error. + assertEquals(logJsonObject.get("statusCode").getAsString(), statusCode.toString()); + } + + /** Check that expected message audit logs are produced when under load. */ + @Test + public void shouldProduceExpectedAuditLogsUnderLoad() + throws InterruptedException, ExecutionException { + // Generate artifical requests on core to simulate load. + int LOAD_SIZE = 40; // Total number of requests to send. + int BURST_SIZE = 5; // Number of requests to send at once. + + ListStoresRequest request = ListStoresRequest.getDefaultInstance(); + List responses = new LinkedList<>(); + for (int i = 0; i < LOAD_SIZE; i += 5) { + List> futures = new LinkedList<>(); + for (int j = 0; j < BURST_SIZE; j++) { + futures.add(asyncCoreService.listStores(request)); + } + + responses.addAll(Futures.allAsList(futures).get()); + } + // Wait required to ensure audit logs are flushed into test audit log appender + Thread.sleep(1000); + + // Pull message audit logs from test log appender + List logJsonObjects = + parseMessageJsonLogObjects(testAuditLogAppender.getLogs(), "ListStores"); + assertEquals(responses.size(), logJsonObjects.size()); + + // Extract & Check that request/response are returned correctly + JsonFormat.Parser protoJSONParser = JsonFormat.parser(); + Streams.zip( + responses.stream(), + logJsonObjects.stream(), + (response, logObj) -> Pair.of(response, logObj)) + .forEach( + responseLogJsonPair -> { + ListStoresResponse response = responseLogJsonPair.getLeft(); + JsonObject logObj = responseLogJsonPair.getRight(); + + ListStoresRequest.Builder gotRequest = null; + ListStoresResponse.Builder gotResponse = null; + try { + String requestJson = logObj.getAsJsonObject("request").toString(); + gotRequest = ListStoresRequest.newBuilder(); + protoJSONParser.merge(requestJson, gotRequest); + + String responseJson = logObj.getAsJsonObject("response").toString(); + gotResponse = ListStoresResponse.newBuilder(); + protoJSONParser.merge(responseJson, gotResponse); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + + assertThat(gotRequest.build(), equalTo(request)); + assertThat(gotResponse.build(), equalTo(response)); + }); + } + + /** + * Filter and Parse out Message Audit Logs from the given logsStrings for the given method name + */ + private List parseMessageJsonLogObjects(List logsStrings, String methodName) { + JsonParser jsonParser = new JsonParser(); + // copy to prevent concurrent modification. + return logsStrings.stream() + .map(logJSON -> jsonParser.parse(logJSON).getAsJsonObject()) + // Filter to only include message audit logs + .filter( + logObj -> + logObj + .getAsJsonPrimitive("kind") + .getAsString() + .equals(AuditLogEntryKind.MESSAGE.toString()) + // filter by method name to ensure logs from other tests do not interfere with + // test + && logObj.get("method").getAsString().equals(methodName)) + .collect(Collectors.toList()); + } +} diff --git a/core/src/test/java/feast/core/logging/TestLogAppender.java b/core/src/test/java/feast/core/logging/TestLogAppender.java new file mode 100644 index 00000000000..b1908c0db2c --- /dev/null +++ b/core/src/test/java/feast/core/logging/TestLogAppender.java @@ -0,0 +1,68 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.core.logging; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; +import lombok.Getter; +import org.apache.logging.log4j.core.Appender; +import org.apache.logging.log4j.core.Core; +import org.apache.logging.log4j.core.Filter; +import org.apache.logging.log4j.core.Layout; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +import org.apache.logging.log4j.core.config.plugins.Plugin; +import org.apache.logging.log4j.core.config.plugins.PluginAttribute; +import org.apache.logging.log4j.core.config.plugins.PluginElement; +import org.apache.logging.log4j.core.config.plugins.PluginFactory; +import org.apache.logging.log4j.core.layout.PatternLayout; + +/** Test Log Appender used for collecting logs for testing logging. */ +@Plugin( + name = "TestLogAppender", + category = Core.CATEGORY_NAME, + elementType = Appender.ELEMENT_TYPE) +@Getter +public class TestLogAppender extends AbstractAppender { + private List logs; + + protected TestLogAppender(String name, Filter filter, Layout layout) { + super(name, filter, layout, false, new Property[] {}); + logs = new ArrayList<>(); + } + + @Override + public void append(LogEvent event) { + getLogs().add(event.getMessage().toString()); + } + + @PluginFactory + public static TestLogAppender createAppender( + @PluginAttribute("name") String name, + @PluginElement("Layout") Layout layout, + @PluginElement("Filter") final Filter filter) { + if (name == null) { + return null; + } + if (layout == null) { + layout = PatternLayout.createDefaultLayout(); + } + return new TestLogAppender(name, filter, layout); + } +} diff --git a/core/src/test/resources/log4j2.xml b/core/src/test/resources/log4j2.xml new file mode 100644 index 00000000000..9f785952e0c --- /dev/null +++ b/core/src/test/resources/log4j2.xml @@ -0,0 +1,53 @@ + + + + + + + %d{yyyy-MM-dd HH:mm:ss.SSS} %5p ${hostName} --- [%15.15t] %-40.40c{1.} : %m%n%ex + + + {"time":"%d{yyyy-MM-dd'T'HH:mm:ssXXX}","hostname":"${hostName}","severity":"%p","message":%m}%n%ex + + + + + + + + + + + + + + + + + + + + + + + + + + + +