nprasad2077/doris
1
0 Fork 0
mirror of https://github.com/apache/doris.git synced 2026-11-04 07:31:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

[chore](fe) Remove the standalone auditloader plugin from fe_plugins (#68563)

Browse Source 
### 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()`.
This commit is contained in:
Calvin Kirs authored and GitHub committed 2026-09-29 09:58:12 +08:00
1 parent 225bcd3018
commit 83d300d279
13 files changed
+4 -935
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

fe_plugins/auditloader/build.sh
-38
View File
@@ -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"
fe_plugins/auditloader/pom.xml
-87
View File
@@ -1,87 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.doris</groupId>
<artifactId>fe-plugins</artifactId>
<version>1.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>auditloader</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>fe-core</artifactId>
<version>${doris.version}</version>
</dependency>
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>fe-common</artifactId>
<version>${doris.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-api -->
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-core -->
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-slf4j-impl -->
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/log4j/log4j -->
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</dependency>
</dependencies>
<build>
<finalName>auditloader</finalName>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4.1</version>
<configuration>
<appendAssemblyId>false</appendAssemblyId>
<descriptors>
<descriptor>src/main/assembly/zip.xml</descriptor>
</descriptors>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
fe_plugins/auditloader/src/main/assembly/plugin.conf
-56
View File
@@ -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
fe_plugins/auditloader/src/main/assembly/plugin.properties
-23
View File
@@ -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
fe_plugins/auditloader/src/main/assembly/zip.xml
-43
View File
@@ -1,43 +0,0 @@
<?xml version="1.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.
-->
<assembly>
<id>plugin</id>
<formats>
<format>zip</format>
</formats>
<includeBaseDirectory>false</includeBaseDirectory>
<fileSets>
<fileSet>
<directory>target</directory>
<includes>
<include>*.jar</include>
</includes>
<outputDirectory>/</outputDirectory>
</fileSet>
<fileSet>
<directory>src/main/assembly</directory>
<includes>
<include>plugin.properties</include>
<include>plugin.conf</include>
</includes>
<outputDirectory>/</outputDirectory>
</fileSet>
</fileSets>
</assembly>
fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/AuditLoaderPlugin.java
-369
View File
@@ -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<AuditEvent> 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<String, String> 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<String, String> entry : pluginInfoProperties.entrySet()) {
props.setProperty(entry.getKey(), entry.getValue());
}
final Map<String, String> 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<String, String> 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()));
}
}
fe_plugins/auditloader/src/main/java/org/apache/doris/plugin/audit/custom/DorisStreamLoader.java
-199
View File
@@ -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();
}
}
}
fe_plugins/pom.xml
-1
View File
@@ -64,7 +64,6 @@ under the License.
</mailingLists>
<modules>
<module>auditdemo</module>
<module>auditloader</module>
<module>trino-converter</module>
<module>sparksql-converter</module>
</modules>
pytest/deploy/deploy.py
+2 -7
View File
@@ -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__':
pytest/deploy/plugin_auditload.conf
-47
View File
@@ -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=
pytest/deploy/prepare_package.py
+1 -4
View File
@@ -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 .')
pytest/deploy/process_config_file.py
-9
View File
@@ -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__':
pytest/deploy/start.py
+1 -52
View File
@@ -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__':