From 83d300d27934762430c8176cab7e2d41b6022996 Mon Sep 17 00:00:00 2001 From: Calvin Kirs Date: Tue, 29 Sep 2026 09:58:12 +0800 Subject: [PATCH] [chore](fe) Remove the standalone auditloader plugin from fe_plugins (#68563) ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: `fe_plugins/auditloader` is the old standalone audit loader plugin, which had to be built with `build-plugin.sh` and installed with `INSTALL PLUGIN`. FE now ships a built-in audit loader (`org.apache.doris.plugin.audit.AuditLoader`), registered by `PluginMgr` at startup, which writes audit events into `__internal_schema.audit_log` and is switched by the global variable `enable_audit_plugin`. Nothing in FE or the build depends on the standalone copy any more, and it has drifted from the built-in one. This PR removes the module and its entry in `fe_plugins/pom.xml`. The legacy `pytest/deploy` scripts had an optional path that installed this plugin. When `output/audit_loader/auditloader.zip` existed, they rendered `plugin_auditload.conf`, unpacked the zip into the FE directory, created `doris_audit_db__.doris_audit_tbl__` and ran `INSTALL PLUGIN`. No build produces that zip, so the path never ran. This PR removes that path, `plugin_auditload.conf`, and the `deploy_audit` parameter it threaded through `prepare_palo_package()` and `start_palo()`. --- fe_plugins/auditloader/build.sh | 38 -- fe_plugins/auditloader/pom.xml | 87 ----- .../auditloader/src/main/assembly/plugin.conf | 56 --- .../src/main/assembly/plugin.properties | 23 -- .../auditloader/src/main/assembly/zip.xml | 43 -- .../audit/custom/AuditLoaderPlugin.java | 369 ------------------ .../audit/custom/DorisStreamLoader.java | 199 ---------- fe_plugins/pom.xml | 1 - pytest/deploy/deploy.py | 9 +- pytest/deploy/plugin_auditload.conf | 47 --- pytest/deploy/prepare_package.py | 5 +- pytest/deploy/process_config_file.py | 9 - pytest/deploy/start.py | 53 +-- 13 files changed, 4 insertions(+), 935 deletions(-) delete mode 100755 fe_plugins/auditloader/build.sh delete mode 100644 fe_plugins/auditloader/pom.xml delete mode 100755 fe_plugins/auditloader/src/main/assembly/plugin.conf delete mode 100755 fe_plugins/auditloader/src/main/assembly/plugin.properties delete mode 100644 fe_plugins/auditloader/src/main/assembly/zip.xml delete mode 100755 fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/AuditLoaderPlugin.java delete mode 100644 fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/DorisStreamLoader.java delete mode 100755 pytest/deploy/plugin_auditload.conf diff --git a/fe_plugins/auditloader/build.sh b/fe_plugins/auditloader/build.sh deleted file mode 100755 index 051fd0bc859..00000000000 --- a/fe_plugins/auditloader/build.sh +++ /dev/null @@ -1,38 +0,0 @@ -#!/usr/bin/env bash -# 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. - -set -eo pipefail - -ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" &>/dev/null && pwd)" - -export DORIS_HOME="${ROOT}/../.." - -. "${DORIS_HOME}/env.sh" - -export AUDITLOADER_HOME="${ROOT}" - -"${MVN_CMD}" clean package -DskipTests - -echo "Install auditloader..." - -AUDITLOADER_OUTPUT="${AUDITLOADER_HOME}/output" -rm -rf "${AUDITLOADER_OUTPUT}" -mkdir "${AUDITLOADER_OUTPUT}" -cp "${AUDITLOADER_HOME}/target/auditloader.zip" "${AUDITLOADER_HOME}/output"/ - -echo "Build Auditloader Finished" diff --git a/fe_plugins/auditloader/pom.xml b/fe_plugins/auditloader/pom.xml deleted file mode 100644 index 204822b414d..00000000000 --- a/fe_plugins/auditloader/pom.xml +++ /dev/null @@ -1,87 +0,0 @@ - - - - - org.apache.doris - fe-plugins - 1.0-SNAPSHOT - ../pom.xml - - 4.0.0 - auditloader - jar - - - org.apache.doris - fe-core - ${doris.version} - - - org.apache.doris - fe-common - ${doris.version} - - - - org.apache.logging.log4j - log4j-api - - - - org.apache.logging.log4j - log4j-core - - - - org.apache.logging.log4j - log4j-slf4j-impl - - - - log4j - log4j - - - - auditloader - - - maven-assembly-plugin - 2.4.1 - - false - - src/main/assembly/zip.xml - - - - - make-assembly - package - - single - - - - - - - diff --git a/fe_plugins/auditloader/src/main/assembly/plugin.conf b/fe_plugins/auditloader/src/main/assembly/plugin.conf deleted file mode 100755 index aec8724fd96..00000000000 --- a/fe_plugins/auditloader/src/main/assembly/plugin.conf +++ /dev/null @@ -1,56 +0,0 @@ -# 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. - -### plugin configuration - -# The max size of a batch, default is 50MB -max_batch_size=52428800 - -# The max interval of batch loaded, default is 60 seconds -max_batch_interval_sec=60 - -# the max stmt length to be loaded in audit table, default is 4096 -max_stmt_length=4096 - -# the capacity of audit queue, default is 1000 -max_queue_size=1000 - -# Doris FE host for loading the audit, default is 127.0.0.1:8030. -# this should be the host port for stream load -frontend_host_port=127.0.0.1:8030 - -# Database of the audit table -database=doris_audit_db__ - -# Audit table name, to save the audit log data. -audit_log_table=doris_audit_log_tbl__ - -# Audit table name, to save the slow log data. -slow_log_table=doris_slow_log_tbl__ - -# Whether import slow logs into a separate slow table, default is false -enable_slow_log=false - -# Doris user. This user must have LOAD_PRIV to the audit table. -user=root - -# Doris user's password -password= - -# Use doris cluster token for stream load authorization, if true, user and password will be ignored. -use_auth_token=false - diff --git a/fe_plugins/auditloader/src/main/assembly/plugin.properties b/fe_plugins/auditloader/src/main/assembly/plugin.properties deleted file mode 100755 index eb7ba9c29b0..00000000000 --- a/fe_plugins/auditloader/src/main/assembly/plugin.properties +++ /dev/null @@ -1,23 +0,0 @@ -# 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. - -name=AuditLoader -type=AUDIT -description=load audit log to olap load, and user can view the statistic of queries -version=1.0.0 -java.version=1.8.0 -classname=org.apache.doris.plugin.audit.custom.AuditLoaderPlugin diff --git a/fe_plugins/auditloader/src/main/assembly/zip.xml b/fe_plugins/auditloader/src/main/assembly/zip.xml deleted file mode 100644 index 515e68751b0..00000000000 --- a/fe_plugins/auditloader/src/main/assembly/zip.xml +++ /dev/null @@ -1,43 +0,0 @@ - - - - plugin - - zip - - false - - - target - - *.jar - - / - - - src/main/assembly - - plugin.properties - plugin.conf - - / - - - diff --git a/fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/AuditLoaderPlugin.java b/fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/AuditLoaderPlugin.java deleted file mode 100755 index a3c850c654a..00000000000 --- a/fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/AuditLoaderPlugin.java +++ /dev/null @@ -1,369 +0,0 @@ -// 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.plugin.audit.custom; - -import org.apache.doris.common.Config; -import org.apache.doris.catalog.Env; -import org.apache.doris.plugin.AuditPlugin; -import org.apache.doris.plugin.Plugin; -import org.apache.doris.plugin.PluginContext; -import org.apache.doris.plugin.PluginException; -import org.apache.doris.plugin.PluginInfo; - -import com.google.common.collect.Queues; -import org.apache.doris.plugin.audit.AuditEvent; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - -import java.io.IOException; -import java.io.InputStream; -import java.nio.ByteBuffer; -import java.nio.CharBuffer; -import java.nio.charset.Charset; -import java.nio.charset.CharsetDecoder; -import java.nio.charset.CodingErrorAction; -import java.nio.file.FileSystems; -import java.nio.file.Files; -import java.nio.file.Path; -import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneId; -import java.time.format.DateTimeFormatter; -import java.util.Map; -import java.util.Properties; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.TimeUnit; -import java.util.function.Function; -import java.util.stream.Collectors; - -/* - * This plugin will load audit log to specified doris table at specified interval - */ -public class AuditLoaderPlugin extends Plugin implements AuditPlugin { - private final static Logger LOG = LogManager.getLogger(AuditLoaderPlugin.class); - - private static final DateTimeFormatter DATETIME_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS") - .withZone(ZoneId.systemDefault()); - - private StringBuilder auditLogBuffer = new StringBuilder(); - private StringBuilder slowLogBuffer = new StringBuilder(); - private long lastLoadTimeAuditLog = 0; - private long lastLoadTimeSlowLog = 0; - - private BlockingQueue auditEventQueue; - private DorisStreamLoader streamLoader; - private Thread loadThread; - - private AuditLoaderConf conf; - private volatile boolean isClosed = false; - private volatile boolean isInit = false; - - @Override - public void init(PluginInfo info, PluginContext ctx) throws PluginException { - super.init(info, ctx); - - synchronized (this) { - if (isInit) { - return; - } - this.lastLoadTimeAuditLog = System.currentTimeMillis(); - this.lastLoadTimeSlowLog = System.currentTimeMillis(); - - loadConfig(ctx, info.getProperties()); - this.auditEventQueue = Queues.newLinkedBlockingDeque(conf.maxQueueSize); - this.streamLoader = new DorisStreamLoader(conf); - this.loadThread = new Thread(new LoadWorker(this.streamLoader), "audit loader thread"); - this.loadThread.start(); - - isInit = true; - } - } - - private void loadConfig(PluginContext ctx, Map pluginInfoProperties) throws PluginException { - Path pluginPath = FileSystems.getDefault().getPath(ctx.getPluginPath()); - if (!Files.exists(pluginPath)) { - throw new PluginException("plugin path does not exist: " + pluginPath); - } - - Path confFile = pluginPath.resolve("plugin.conf"); - if (!Files.exists(confFile)) { - throw new PluginException("plugin conf file does not exist: " + confFile); - } - - final Properties props = new Properties(); - try (InputStream stream = Files.newInputStream(confFile)) { - props.load(stream); - } catch (IOException e) { - throw new PluginException(e.getMessage()); - } - - for (Map.Entry entry : pluginInfoProperties.entrySet()) { - props.setProperty(entry.getKey(), entry.getValue()); - } - - final Map properties = props.stringPropertyNames().stream() - .collect(Collectors.toMap(Function.identity(), props::getProperty)); - conf = new AuditLoaderConf(); - conf.init(properties); - conf.feIdentity = ctx.getFeIdentity(); - } - - @Override - public void close() throws IOException { - super.close(); - isClosed = true; - if (loadThread != null) { - try { - loadThread.join(); - } catch (InterruptedException e) { - LOG.debug("encounter exception when closing the audit loader", e); - } - } - } - - public boolean eventFilter(AuditEvent.EventType type) { - return type == AuditEvent.EventType.AFTER_QUERY; - } - - public void exec(AuditEvent event) { - try { - auditEventQueue.add(event); - } catch (Exception e) { - // In order to ensure that the system can run normally, here we directly - // discard the current audit_event. If this problem occurs frequently, - // improvement can be considered. - LOG.debug("encounter exception when putting current audit batch, discard current audit event", e); - } - } - - private void assembleAudit(AuditEvent event) { - if (conf.enableSlowLog && event.queryTime > Config.qe_slow_log_ms) { - fillLogBuffer(event, slowLogBuffer); - } - fillLogBuffer(event, auditLogBuffer); - } - - private void fillLogBuffer(AuditEvent event, StringBuilder logBuffer) { - logBuffer.append(event.queryId).append("\t"); - logBuffer.append(longToTimeString(event.timestamp)).append("\t"); - logBuffer.append(event.clientIp).append("\t"); - logBuffer.append(event.user).append("\t"); - logBuffer.append(event.ctl).append("\t"); - logBuffer.append(event.db).append("\t"); - logBuffer.append(event.state).append("\t"); - logBuffer.append(event.errorCode).append("\t"); - logBuffer.append(event.errorMessage).append("\t"); - logBuffer.append(event.queryTime).append("\t"); - logBuffer.append(event.scanBytes).append("\t"); - logBuffer.append(event.scanRows).append("\t"); - logBuffer.append(event.returnRows).append("\t"); - logBuffer.append(event.stmtId).append("\t"); - logBuffer.append(event.stmtType).append("\t"); - logBuffer.append(event.isQuery ? 1 : 0).append("\t"); - logBuffer.append(event.feIp).append("\t"); - logBuffer.append(event.cpuTimeMs).append("\t"); - logBuffer.append(event.sqlHash).append("\t"); - logBuffer.append(event.sqlDigest).append("\t"); - logBuffer.append(event.peakMemoryBytes).append("\t"); - logBuffer.append(event.workloadGroup).append("\t"); - // trim the query to avoid too long - // use `getBytes().length` to get real byte length - String stmt = truncateByBytes(event.stmt).replace("\n", " ") - .replace("\t", " ") - .replace("\r", " "); - if (LOG.isDebugEnabled()) { - LOG.debug("receive audit event with stmt: {}", stmt); - } - logBuffer.append(stmt).append("\n"); - } - - private String truncateByBytes(String str) { - int maxLen = Math.min(conf.max_stmt_length, str.getBytes().length); - if (maxLen >= str.getBytes().length) { - return str; - } - Charset utf8Charset = Charset.forName("UTF-8"); - CharsetDecoder decoder = utf8Charset.newDecoder(); - byte[] sb = str.getBytes(utf8Charset); - ByteBuffer buffer = ByteBuffer.wrap(sb, 0, maxLen); - CharBuffer charBuffer = CharBuffer.allocate(maxLen); - decoder.onMalformedInput(CodingErrorAction.IGNORE); - decoder.decode(buffer, charBuffer, true); - decoder.flush(charBuffer); - return new String(charBuffer.array(), 0, charBuffer.position()); - } - - private void loadIfNecessary(DorisStreamLoader loader, boolean slowLog) { - StringBuilder logBuffer = slowLog ? slowLogBuffer : auditLogBuffer; - long lastLoadTime = slowLog ? lastLoadTimeSlowLog : lastLoadTimeAuditLog; - long currentTime = System.currentTimeMillis(); - - if (logBuffer.length() >= conf.maxBatchSize || currentTime - lastLoadTime >= conf.maxBatchIntervalSec * 1000) { - // begin to load - try { - String token = ""; - if (conf.use_auth_token) { - try { - // Acquire token from master - token = Env.getCurrentEnv().getLoadManager().getTokenManager().acquireToken(); - } catch (Exception e) { - LOG.error("Failed to get auth token: {}", e); - } - } - DorisStreamLoader.LoadResponse response = loader.loadBatch(logBuffer, slowLog, token); - LOG.debug("audit loader response: {}", response); - } catch (Exception e) { - LOG.debug("encounter exception when putting current audit batch, discard current batch", e); - } finally { - // make a new string builder to receive following events. - resetLogBufferAndLastLoadTime(currentTime, slowLog); - } - } - - return; - } - - private void resetLogBufferAndLastLoadTime(long currentTime, boolean slowLog) { - if (slowLog) { - this.slowLogBuffer = new StringBuilder(); - lastLoadTimeSlowLog = currentTime; - } else { - this.auditLogBuffer = new StringBuilder(); - lastLoadTimeAuditLog = currentTime; - } - - return; - } - - public static class AuditLoaderConf { - public static final String PROP_MAX_BATCH_SIZE = "max_batch_size"; - public static final String PROP_MAX_BATCH_INTERVAL_SEC = "max_batch_interval_sec"; - public static final String PROP_MAX_QUEUE_SIZE = "max_queue_size"; - public static final String PROP_FRONTEND_HOST_PORT = "frontend_host_port"; - public static final String PROP_USER = "user"; - public static final String PROP_PASSWORD = "password"; - public static final String PROP_DATABASE = "database"; - public static final String PROP_TABLE = "table"; - public static final String PROP_AUDIT_LOG_TABLE = "audit_log_table"; - public static final String PROP_SLOW_LOG_TABLE = "slow_log_table"; - public static final String PROP_ENABLE_SLOW_LOG = "enable_slow_log"; - // the max stmt length to be loaded in audit table. - public static final String MAX_STMT_LENGTH = "max_stmt_length"; - public static final String USE_AUTH_TOKEN = "use_auth_token"; - - public long maxBatchSize = 50 * 1024 * 1024; - public long maxBatchIntervalSec = 60; - public int maxQueueSize = 1000; - public String frontendHostPort = "127.0.0.1:8030"; - public String user = "root"; - public String password = ""; - public String database = "doris_audit_db__"; - public String auditLogTable = "doris_audit_log_tbl__"; - public String slowLogTable = "doris_slow_log_tbl__"; - public boolean enableSlowLog = false; - // the identity of FE which run this plugin - public String feIdentity = ""; - public int max_stmt_length = 4096; - // auth_token is not used by default - public boolean use_auth_token = false; - - - public void init(Map properties) throws PluginException { - try { - if (properties.containsKey(PROP_MAX_BATCH_SIZE)) { - maxBatchSize = Long.valueOf(properties.get(PROP_MAX_BATCH_SIZE)); - } - if (properties.containsKey(PROP_MAX_BATCH_INTERVAL_SEC)) { - maxBatchIntervalSec = Long.valueOf(properties.get(PROP_MAX_BATCH_INTERVAL_SEC)); - } - if (properties.containsKey(PROP_MAX_QUEUE_SIZE)) { - maxQueueSize = Integer.valueOf(properties.get(PROP_MAX_QUEUE_SIZE)); - } - if (properties.containsKey(PROP_FRONTEND_HOST_PORT)) { - frontendHostPort = properties.get(PROP_FRONTEND_HOST_PORT); - } - if (properties.containsKey(PROP_USER)) { - user = properties.get(PROP_USER); - } - if (properties.containsKey(PROP_PASSWORD)) { - password = properties.get(PROP_PASSWORD); - } - if (properties.containsKey(PROP_DATABASE)) { - database = properties.get(PROP_DATABASE); - } - // If plugin.conf is not changed, the audit logs are imported to previous table - if (properties.containsKey(PROP_TABLE)) { - auditLogTable = properties.get(PROP_TABLE); - } - if (properties.containsKey(PROP_AUDIT_LOG_TABLE)) { - auditLogTable = properties.get(PROP_AUDIT_LOG_TABLE); - } - if (properties.containsKey(PROP_SLOW_LOG_TABLE)) { - slowLogTable = properties.get(PROP_SLOW_LOG_TABLE); - } - if (properties.containsKey(PROP_ENABLE_SLOW_LOG)) { - enableSlowLog = Boolean.valueOf(properties.get(PROP_ENABLE_SLOW_LOG)); - } - if (properties.containsKey(MAX_STMT_LENGTH)) { - max_stmt_length = Integer.parseInt(properties.get(MAX_STMT_LENGTH)); - } - if (properties.containsKey(USE_AUTH_TOKEN)) { - use_auth_token = Boolean.valueOf(properties.get(USE_AUTH_TOKEN)); - } - } catch (Exception e) { - throw new PluginException(e.getMessage()); - } - } - } - - private class LoadWorker implements Runnable { - private DorisStreamLoader loader; - - public LoadWorker(DorisStreamLoader loader) { - this.loader = loader; - } - - public void run() { - while (!isClosed) { - try { - AuditEvent event = auditEventQueue.poll(5, TimeUnit.SECONDS); - if (event != null) { - assembleAudit(event); - // process slow audit logs - if (conf.enableSlowLog) { - loadIfNecessary(loader, true); - } - // process all audit logs - loadIfNecessary(loader, false); - } - } catch (InterruptedException ie) { - LOG.debug("encounter exception when loading current audit batch", ie); - } catch (Exception e) { - LOG.error("run audit logger error:", e); - } - } - } - } - - public static String longToTimeString(long timeStamp) { - if (timeStamp <= 0L) { - return "1900-01-01 00:00:00.000"; - } - return DATETIME_FORMAT.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(timeStamp), ZoneId.systemDefault())); - } -} diff --git a/fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/DorisStreamLoader.java b/fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/DorisStreamLoader.java deleted file mode 100644 index b5ea0277986..00000000000 --- a/fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/DorisStreamLoader.java +++ /dev/null @@ -1,199 +0,0 @@ -// 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.plugin.audit.custom; - -import org.apache.doris.catalog.InternalSchema; - -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - -import java.io.BufferedOutputStream; -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStreamReader; -import java.net.HttpURLConnection; -import java.net.URL; -import java.nio.charset.StandardCharsets; -import java.util.Base64; -import java.util.Calendar; -import java.util.stream.Collectors; - -public class DorisStreamLoader { - private final static Logger LOG = LogManager.getLogger(DorisStreamLoader.class); - private static String loadUrlPattern = "http://%s/api/%s/%s/_stream_load?"; - private String hostPort; - private String db; - private String auditLogTbl; - private String slowLogTbl; - private String user; - private String passwd; - private String auditLogLoadUrlStr; - private String slowLogLoadUrlStr; - private String authEncoding; - private String feIdentity; - - public DorisStreamLoader(AuditLoaderPlugin.AuditLoaderConf conf) { - this.hostPort = conf.frontendHostPort; - this.db = conf.database; - this.auditLogTbl = conf.auditLogTable; - this.slowLogTbl = conf.slowLogTable; - this.user = conf.user; - this.passwd = conf.password; - - this.auditLogLoadUrlStr = String.format(loadUrlPattern, hostPort, db, auditLogTbl); - this.slowLogLoadUrlStr = String.format(loadUrlPattern, hostPort, db, slowLogTbl); - this.authEncoding = Base64.getEncoder().encodeToString(String.format("%s:%s", user, passwd).getBytes(StandardCharsets.UTF_8)); - // currently, FE identity is FE's IP, so we replace the "." in IP to make it suitable for label - this.feIdentity = conf.feIdentity.replaceAll("\\.", "_"); - } - - private HttpURLConnection getConnection(String urlStr, String label, String clusterToken) throws IOException { - URL url = new URL(urlStr); - HttpURLConnection conn = (HttpURLConnection) url.openConnection(); - conn.setInstanceFollowRedirects(false); - conn.setRequestMethod("PUT"); - conn.setRequestProperty("token", clusterToken); - conn.setRequestProperty("Authorization", "Basic " + authEncoding); - conn.addRequestProperty("Expect", "100-continue"); - conn.addRequestProperty("Content-Type", "text/plain; charset=UTF-8"); - - conn.addRequestProperty("label", label); - conn.addRequestProperty("max_filter_ratio", "1.0"); - conn.addRequestProperty("columns", - InternalSchema.AUDIT_SCHEMA.stream().map(c -> c.getName()).collect(Collectors.joining(","))); - conn.setDoOutput(true); - conn.setDoInput(true); - - return conn; - } - - private String toCurl(HttpURLConnection conn) { - StringBuilder sb = new StringBuilder("curl -v "); - sb.append("-X ").append(conn.getRequestMethod()).append(" \\\n "); - sb.append("-H \"").append("Authorization\":").append("\"Basic " + authEncoding).append("\" \\\n "); - sb.append("-H \"").append("Expect\":").append("\"100-continue\" \\\n "); - sb.append("-H \"").append("Content-Type\":").append("\"text/plain; charset=UTF-8\" \\\n "); - sb.append("-H \"").append("max_filter_ratio\":").append("\"1.0\" \\\n "); - sb.append("-H \"").append("columns\":") - .append("\"" + InternalSchema.AUDIT_SCHEMA.stream().map(c -> c.getName()).collect( - Collectors.joining(",")) + "\" \\\n "); - sb.append("\"").append(conn.getURL()).append("\""); - return sb.toString(); - } - - private String getContent(HttpURLConnection conn) { - BufferedReader br = null; - StringBuilder response = new StringBuilder(); - String line; - try { - if (100 <= conn.getResponseCode() && conn.getResponseCode() <= 399) { - br = new BufferedReader(new InputStreamReader(conn.getInputStream())); - } else { - br = new BufferedReader(new InputStreamReader(conn.getErrorStream())); - } - while ((line = br.readLine()) != null) { - response.append(line); - } - } catch (IOException e) { - LOG.warn("get content error,", e); - } - - return response.toString(); - } - - public LoadResponse loadBatch(StringBuilder sb, boolean slowLog, String clusterToken) { - Calendar calendar = Calendar.getInstance(); - String label = String.format("_log_%s%02d%02d_%02d%02d%02d_%s", - calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1, calendar.get(Calendar.DAY_OF_MONTH), - calendar.get(Calendar.HOUR_OF_DAY), calendar.get(Calendar.MINUTE), calendar.get(Calendar.SECOND), - feIdentity); - - HttpURLConnection feConn = null; - HttpURLConnection beConn = null; - try { - // build request and send to fe - if (slowLog) { - label = "slow" + label; - feConn = getConnection(slowLogLoadUrlStr, label, clusterToken); - } else { - label = "audit" + label; - feConn = getConnection(auditLogLoadUrlStr, label, clusterToken); - } - int status = feConn.getResponseCode(); - // fe send back http response code TEMPORARY_REDIRECT 307 and new be location - if (status != 307) { - throw new Exception("status is not TEMPORARY_REDIRECT 307, status: " + status - + ", response: " + getContent(feConn) + ", request is: " + toCurl(feConn)); - } - String location = feConn.getHeaderField("Location"); - if (location == null) { - throw new Exception("redirect location is null"); - } - // build request and send to new be location - beConn = getConnection(location, label, clusterToken); - // send data to be - try (BufferedOutputStream bos = new BufferedOutputStream(beConn.getOutputStream())) { - bos.write(sb.toString().getBytes()); - } - - // get respond - status = beConn.getResponseCode(); - String respMsg = beConn.getResponseMessage(); - String response = getContent(beConn); - - LOG.info("AuditLoader plugin load with label: {}, response code: {}, msg: {}, content: {}", - label, status, respMsg, response); - - return new LoadResponse(status, respMsg, response); - - } catch (Exception e) { - e.printStackTrace(); - String err = "failed to load audit via AuditLoader plugin with label: " + label; - LOG.warn(err, e); - return new LoadResponse(-1, e.getMessage(), err); - } finally { - if (feConn != null) { - feConn.disconnect(); - } - if (beConn != null) { - beConn.disconnect(); - } - } - } - - public static class LoadResponse { - public int status; - public String respMsg; - public String respContent; - - public LoadResponse(int status, String respMsg, String respContent) { - this.status = status; - this.respMsg = respMsg; - this.respContent = respContent; - } - - @Override - public String toString() { - StringBuilder sb = new StringBuilder(); - sb.append("status: ").append(status); - sb.append(", resp msg: ").append(respMsg); - sb.append(", resp content: ").append(respContent); - return sb.toString(); - } - } -} diff --git a/fe_plugins/pom.xml b/fe_plugins/pom.xml index ef2ade6f676..6c7ab239687 100644 --- a/fe_plugins/pom.xml +++ b/fe_plugins/pom.xml @@ -64,7 +64,6 @@ under the License. auditdemo - auditloader trino-converter sparksql-converter diff --git a/pytest/deploy/deploy.py b/pytest/deploy/deploy.py index 6b95559c2e7..beeacabdfe4 100644 --- a/pytest/deploy/deploy.py +++ b/pytest/deploy/deploy.py @@ -21,7 +21,6 @@ This module provide Palo deployer. Date: 2015/10/07 17:23:06 """ -import os import init_env import backup import config_be @@ -36,10 +35,6 @@ import time def deploy_palo(): """deploy palo """ - if os.path.exists('./output/audit_loader/auditloader.zip'): - deploy_audit = True - else: - deploy_audit = False # 集群停止，备份 backup.backup_palo() # 初始化fe/be目录等 @@ -48,13 +43,13 @@ def deploy_palo(): # 更新conf文件 process_config_file.process_palo_conf() # 准备部署表 - prepare_package.prepare_palo_package(deploy_audit=deploy_audit) + prepare_package.prepare_palo_package() # hadoop_mkdir.create_hdfs_dir() # 分发部署包到fe/be机器 distribute.distribute_package() time.sleep(100) # config_be.config_be() - start.start_palo(init_state=True, deploy_audit=deploy_audit) + start.start_palo(init_state=True) if __name__ == '__main__': diff --git a/pytest/deploy/plugin_auditload.conf b/pytest/deploy/plugin_auditload.conf deleted file mode 100755 index 14fdd32b983..00000000000 --- a/pytest/deploy/plugin_auditload.conf +++ /dev/null @@ -1,47 +0,0 @@ -# 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. - -### plugin configuration - -# The max size of a batch, default is 50MB -max_batch_size=52428800 - -# The max interval of batch loaded, default is 60 seconds -max_batch_interval_sec=60 - -# the max stmt length to be loaded in audit table, default is 4096 -max_stmt_length=4096 - -# the capacity of audit queue, default is 1000 -max_queue_size=1000 - -# Doris FE host for loading the audit, default is 127.0.0.1:8030. -# this should be the host port for stream load -frontend_host_port=127.0.0.1:8030 - -# Database of the audit table -database=doris_audit_db__ - -# Audit table name, to save the audit data. -table=doris_audit_tbl__ - -# Doris user. This user must have LOAD_PRIV to the audit table. -user=root - -# Doris user's password -password= - diff --git a/pytest/deploy/prepare_package.py b/pytest/deploy/prepare_package.py index 9a16c2ca5c4..e96aa85d0cc 100644 --- a/pytest/deploy/prepare_package.py +++ b/pytest/deploy/prepare_package.py @@ -25,7 +25,7 @@ import os import env_config -def prepare_palo_package(deploy_audit=False): +def prepare_palo_package(): """prepare palo package """ os.system('mv fe.conf.out output/fe/conf/fe.conf') @@ -36,9 +36,6 @@ def prepare_palo_package(deploy_audit=False): 'ulimit -c unlimited\\n nohup/g" output/be/bin/start_be.sh') os.system('sed -i "s/limit3 -c 0 -n/limit3 -n/g" output/be/bin/start_be.sh') os.system('mkdir -p output/be/var/pull_load > /dev/null 2>&1') - if deploy_audit: - os.system('unzip -q -u -d output/fe/plugin_auditloader output/audit_loader/auditloader.zip') - os.system('mv plugin_auditload.conf.out output/fe/plugin_auditloader/plugin.conf') os.system('cd output;tar -cf - fe | gzip --fast > fe.tar.gz; cd -') os.system('cd output;tar -cf - be | gzip --fast > be.tar.gz; cd -') os.system('mv output/fe.tar.gz .') diff --git a/pytest/deploy/process_config_file.py b/pytest/deploy/process_config_file.py index d0d71ed6715..6f4084a527f 100644 --- a/pytest/deploy/process_config_file.py +++ b/pytest/deploy/process_config_file.py @@ -81,20 +81,11 @@ def process_be_conf(): pass -def process_auditload_conf(): - """""" - os.system('cp plugin_auditload.conf plugin_auditload.conf.out') - os.system('sed -i "s/password=.*/password=%s/g" plugin_auditload.conf.out' % env_config.fe_password) - os.system('sed -i "s/frontend_host_port=.*/frontend_host_port=%s:%s/g" plugin_auditload.conf.out' \ - % (env_config.master, env_config.fe_query_port - 1000)) - - def process_palo_conf(): """process palo conf """ process_fe_conf() process_be_conf() - process_auditload_conf() if __name__ == '__main__': diff --git a/pytest/deploy/start.py b/pytest/deploy/start.py index 7c979d9d334..535bd3cc8de 100644 --- a/pytest/deploy/start.py +++ b/pytest/deploy/start.py @@ -172,56 +172,7 @@ def add_brokers(): os.system(cmd) -def add_auditload_plugin(): - """add audit load plugin""" - sql = """ - create table doris_audit_db__.doris_audit_tbl__ \ - ( \ - query_id varchar(48) comment 'Unique query id', \ - \`time\` datetime not null comment 'Query start time', \ - client_ip varchar(32) comment 'Client IP', \ - user varchar(64) comment 'User name', \ - db varchar(96) comment 'Database of this query', \ - state varchar(8) comment 'Query result state. EOF, ERR, OK', \ - query_time bigint comment 'Query execution time in millisecond', \ - scan_bytes bigint comment 'Total scan bytes of this query', \ - scan_rows bigint comment 'Total scan rows of this query', \ - return_rows bigint comment 'Returned rows of this query', \ - stmt_id int comment 'An incremental id of statement', \ - is_query tinyint comment 'Is this statemt a query. 1 or 0', \ - frontend_ip varchar(32) comment 'Frontend ip of executing this statement', \ - cpu_time_ms bigint comment 'Total scan cpu time in millisecond of this query', \ - sql_hash varchar(48) comment 'Hash value for this query', \ - sql_digest varchar(48) comment 'Sql digest of this query, will be empty if not a slow query', \ - peak_memory_bytes bigint comment 'Peak memory bytes used on all backends of this query', \ - stmt string comment 'The original statement, trimed if longer than 2G ' \ - ) engine=OLAP \ - duplicate key(query_id, \`time\`, client_ip) \ - partition by range(\`time\`) () \ - distributed by hash(query_id) buckets 1 \ - properties( \ - 'dynamic_partition.time_unit' = 'HOUR', \ - 'dynamic_partition.start' = '-48', \ - 'dynamic_partition.end' = '3', \ - 'dynamic_partition.prefix' = 'p', \ - 'dynamic_partition.buckets' = '1', \ - 'dynamic_partition.enable' = 'true', \ - 'replication_num' = '3' \ - ); - """ - cmd = 'mysql -h %s -P%s -uroot -p%s -e "%s"' % (env_config.master, env_config.fe_query_port, - env_config.fe_password, "create database doris_audit_db__") - os.system(cmd) - cmd = 'mysql -h %s -P%s -uroot -p%s -e "%s"' % (env_config.master, env_config.fe_query_port, - env_config.fe_password, sql) - os.system(cmd) - sql = "INSTALL PLUGIN FROM '%s/fe/plugin_auditloader'" % env_config.fe_path - cmd = 'mysql -h %s -P%s -uroot -p%s -e "%s"' % (env_config.master, env_config.fe_query_port, - env_config.fe_password, sql) - os.system(cmd) - - -def start_palo(init_state=False, deploy_audit=False): +def start_palo(init_state=False): """start palo """ start_master() @@ -235,8 +186,6 @@ def start_palo(init_state=False, deploy_audit=False): start_other_fe() start_be() time.sleep(5) - if deploy_audit: - add_auditload_plugin() if __name__ == '__main__':