diff --git a/fe/be-java-extensions/lakesoul-scanner/pom.xml b/fe/be-java-extensions/lakesoul-scanner/pom.xml index 32f853bc76cfc2..3c195eaae56804 100644 --- a/fe/be-java-extensions/lakesoul-scanner/pom.xml +++ b/fe/be-java-extensions/lakesoul-scanner/pom.xml @@ -61,6 +61,11 @@ under the License. + + org.mockito + mockito-inline + test + diff --git a/fe/be-java-extensions/lakesoul-scanner/src/main/java/org/apache/doris/lakesoul/LakeSoulJniScanner.java b/fe/be-java-extensions/lakesoul-scanner/src/main/java/org/apache/doris/lakesoul/LakeSoulJniScanner.java index 0ff491b246a867..914b508d2e9d19 100644 --- a/fe/be-java-extensions/lakesoul-scanner/src/main/java/org/apache/doris/lakesoul/LakeSoulJniScanner.java +++ b/fe/be-java-extensions/lakesoul-scanner/src/main/java/org/apache/doris/lakesoul/LakeSoulJniScanner.java @@ -69,7 +69,7 @@ public void open() throws IOException { withAllocator(nativeIOReader.getAllocator()); nativeIOReader.setBatchSize(batchSize); - LOG.info("opening LakeSoulJniScanner with params={}", params); + LOG.info("opening LakeSoulJniScanner, queryId={}, batchSize={}", params.get("query_id"), batchSize); // add files for (String file : params.get(LakeSoulUtils.FILE_NAMES).split(LakeSoulUtils.LIST_DELIM)) { diff --git a/fe/be-java-extensions/lakesoul-scanner/src/test/java/org/apache/doris/lakesoul/LakeSoulJniScannerTest.java b/fe/be-java-extensions/lakesoul-scanner/src/test/java/org/apache/doris/lakesoul/LakeSoulJniScannerTest.java new file mode 100644 index 00000000000000..c75d8860e3412a --- /dev/null +++ b/fe/be-java-extensions/lakesoul-scanner/src/test/java/org/apache/doris/lakesoul/LakeSoulJniScannerTest.java @@ -0,0 +1,122 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 +// +// http://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 org.apache.doris.lakesoul; + +import com.dmetasoul.lakesoul.lakesoul.io.NativeIOReader; +import com.lakesoul.shaded.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.MockedConstruction; +import org.mockito.Mockito; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class LakeSoulJniScannerTest { + @Test + public void testOpenDoesNotLogCredentialsAtInfo() throws Exception { + assertSafeOpen(Level.INFO, "test-query-id"); + } + + @Test + public void testOpenDoesNotLogCredentialsAtDebug() throws Exception { + assertSafeOpen(Level.DEBUG, "test-query-id"); + } + + @Test + public void testOpenWithoutQueryIdDoesNotLogCredentials() throws Exception { + assertSafeOpen(Level.DEBUG, null); + } + + private void assertSafeOpen(Level level, String queryId) throws Exception { + Map options = new HashMap<>(); + options.put(LakeSoulUtils.FS_S3A_ACCESS_KEY, "FAKE_ACCESS_KEY"); + options.put(LakeSoulUtils.FS_S3A_SECRET_KEY, "FAKE_SECRET_KEY"); + options.put(LakeSoulUtils.FS_S3A_ENDPOINT, "https://object-store.invalid"); + options.put(LakeSoulUtils.FS_S3A_PATH_STYLE_ACCESS, "true"); + Map params = new HashMap<>(); + if (queryId != null) { + params.put("query_id", queryId); + } + params.put(LakeSoulUtils.OPTIONS, new ObjectMapper().writeValueAsString(options)); + params.put(LakeSoulUtils.FILE_NAMES, "s3://test/file?signature=FAKE_SIGNATURE"); + params.put(LakeSoulUtils.SCHEMA_JSON, "{\"fields\":[]}"); + params.put(LakeSoulUtils.REQUIRED_FIELDS, ""); + params.put("future_auth_option", "FAKE_FUTURE_TOKEN"); + Map originalParams = new HashMap<>(params); + IOException stopBeforeNativeIo = new IOException("stop before native initialization"); + + Logger logger = (Logger) LogManager.getLogger(LakeSoulJniScanner.class); + Level originalLevel = logger.getLevel(); + boolean originalAdditive = logger.isAdditive(); + CapturingAppender appender = new CapturingAppender(); + appender.start(); + logger.addAppender(appender); + logger.setLevel(level); + logger.setAdditive(false); + try (MockedConstruction readers = Mockito.mockConstruction( + NativeIOReader.class, (reader, context) -> + Mockito.doThrow(stopBeforeNativeIo).when(reader).initializeReader())) { + LakeSoulJniScanner scanner = new LakeSoulJniScanner(128, params); + Assert.assertSame(stopBeforeNativeIo, Assert.assertThrows(IOException.class, scanner::open)); + Assert.assertEquals(1, readers.constructed().size()); + NativeIOReader reader = readers.constructed().get(0); + Mockito.verify(reader).setBatchSize(128); + for (Map.Entry option : options.entrySet()) { + Mockito.verify(reader).setObjectStoreOption(option.getKey(), option.getValue()); + } + Assert.assertEquals(originalParams, params); + + String logs = String.join("\n", appender.messages); + Assert.assertTrue(logs.contains("opening LakeSoulJniScanner")); + Assert.assertFalse("Credentials must not appear in scanner logs", logs.contains("FAKE_")); + Assert.assertTrue(logs.contains("queryId=" + queryId)); + Assert.assertTrue(logs.contains("batchSize=128")); + Assert.assertFalse(logs.contains("fs.s3a.")); + Assert.assertFalse(logs.contains("options=")); + Assert.assertFalse(logs.contains("future_auth_option")); + } finally { + logger.removeAppender(appender); + appender.stop(); + logger.setLevel(originalLevel); + logger.setAdditive(originalAdditive); + } + } + + private static class CapturingAppender extends AbstractAppender { + private final List messages = new ArrayList<>(); + + CapturingAppender() { + super("lakesoul-scanner-test-capture", null, null, false, Property.EMPTY_ARRAY); + } + + @Override + public void append(LogEvent event) { + messages.add(event.getMessage().getFormattedMessage()); + } + } +} diff --git a/fe/be-java-extensions/lakesoul-scanner/src/test/resources/log4j2-test.xml b/fe/be-java-extensions/lakesoul-scanner/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000000000..cd0ccaeb80ae4c --- /dev/null +++ b/fe/be-java-extensions/lakesoul-scanner/src/test/resources/log4j2-test.xml @@ -0,0 +1,25 @@ + + + + + + + + diff --git a/fe/be-java-extensions/pom.xml b/fe/be-java-extensions/pom.xml index c0b2aaebb166c7..5611b74e5c148a 100644 --- a/fe/be-java-extensions/pom.xml +++ b/fe/be-java-extensions/pom.xml @@ -51,4 +51,14 @@ under the License. 1 + + + + lakesoul-scanner-tests + + lakesoul-scanner + + + + diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lakesoul/source/LakeSoulScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lakesoul/source/LakeSoulScanNode.java index a1e287432deb7b..cfa917327f0052 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lakesoul/source/LakeSoulScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lakesoul/source/LakeSoulScanNode.java @@ -181,9 +181,6 @@ private void setLakeSoulParams(TFileRangeDesc rangeDesc, LakeSoulSplit lakeSoulS options.put(LakeSoulUtils.SUBSTRAIT_PREDICATE, SubstraitUtil.encodeBase64String(predicate)); } Map catalogProps = getCatalog().getProperties(); - if (LOG.isDebugEnabled()) { - LOG.debug("{}", catalogProps); - } if (catalogProps.get("AWS_ENDPOINT") != null) { options.put(LakeSoulUtils.FS_S3A_ENDPOINT, catalogProps.get("AWS_ENDPOINT")); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lakesoul/source/LakeSoulScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lakesoul/source/LakeSoulScanNodeTest.java new file mode 100644 index 00000000000000..07d99d966163a9 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lakesoul/source/LakeSoulScanNodeTest.java @@ -0,0 +1,110 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 +// +// http://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 org.apache.doris.datasource.lakesoul.source; + +import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.TableFormatType; +import org.apache.doris.datasource.lakesoul.LakeSoulUtils; +import org.apache.doris.thrift.TFileRangeDesc; + +import com.alibaba.fastjson.JSON; +import com.dmetasoul.lakesoul.meta.LakeSoulOptions; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class LakeSoulScanNodeTest { + @Test + public void testScanOptionsPreserveCredentialsWithoutLoggingThem() throws Exception { + Map catalogProps = new HashMap<>(); + catalogProps.put("AWS_ENDPOINT", "https://object-store.invalid"); + catalogProps.put("AWS_ACCESS_KEY", "FAKE_ACCESS_KEY"); + catalogProps.put("AWS_SECRET_KEY", "FAKE_SECRET_KEY"); + catalogProps.put("AWS_REGION", "test-region"); + catalogProps.put("future_auth_option", "FAKE_FUTURE_TOKEN"); + Map originalProps = new HashMap<>(catalogProps); + ExternalCatalog catalog = Mockito.mock(ExternalCatalog.class); + Mockito.when(catalog.getProperties()).thenReturn(catalogProps); + LakeSoulScanNode scanNode = Mockito.mock(LakeSoulScanNode.class, Mockito.CALLS_REAL_METHODS); + Mockito.doReturn(catalog).when(scanNode).getCatalog(); + scanNode.readType = LakeSoulOptions.ReadType$.MODULE$.FULL_READ(); + LakeSoulSplit split = new LakeSoulSplit(Collections.singletonList("s3://test/file"), + Collections.emptyList(), Collections.emptyMap(), "{\"fields\":[]}", + 0, 0, 0, new String[0], Collections.emptyList()); + split.setTableFormatType(TableFormatType.LAKESOUL); + TFileRangeDesc range = new TFileRangeDesc(); + + Logger logger = (Logger) LogManager.getLogger(LakeSoulScanNode.class); + Level originalLevel = logger.getLevel(); + boolean originalAdditive = logger.isAdditive(); + CapturingAppender appender = new CapturingAppender(); + appender.start(); + logger.addAppender(appender); + logger.setLevel(Level.DEBUG); + logger.setAdditive(false); + // Predicate translation is independent of the catalog options being tested. + try (MockedStatic ignored = Mockito.mockStatic(LakeSoulUtils.class)) { + scanNode.setScanParams(range, split); + Map options = JSON.parseObject( + range.getTableFormatParams().getLakesoulParams().getOptions()); + Assert.assertEquals("FAKE_ACCESS_KEY", options.get(LakeSoulUtils.FS_S3A_ACCESS_KEY)); + Assert.assertEquals("FAKE_SECRET_KEY", options.get(LakeSoulUtils.FS_S3A_SECRET_KEY)); + Assert.assertEquals("https://object-store.invalid", options.get(LakeSoulUtils.FS_S3A_ENDPOINT)); + Assert.assertEquals("test-region", options.get(LakeSoulUtils.FS_S3A_REGION)); + Assert.assertEquals(originalProps, catalogProps); + + Assert.assertFalse(appender.messages.isEmpty()); + String logs = String.join("\n", appender.messages); + Assert.assertFalse("Catalog credentials must not appear in logs", logs.contains("FAKE_")); + Assert.assertFalse(logs.contains("AWS_ACCESS_KEY")); + Assert.assertFalse(logs.contains("AWS_SECRET_KEY")); + Assert.assertFalse(logs.contains("future_auth_option")); + } finally { + logger.removeAppender(appender); + appender.stop(); + logger.setLevel(originalLevel); + logger.setAdditive(originalAdditive); + } + } + + private static class CapturingAppender extends AbstractAppender { + private final List messages = new ArrayList<>(); + + CapturingAppender() { + super("lakesoul-scan-node-test-capture", null, null, false, Property.EMPTY_ARRAY); + } + + @Override + public void append(LogEvent event) { + messages.add(event.getMessage().getFormattedMessage()); + } + } +}