This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 6d3a131701 [INLONG-10441][DataProxy] Supports obtaining Audit-Proxy
through InLong Manager (#10442)
6d3a131701 is described below
commit 6d3a13170135d5c1830f60dc2c509015ca995d66
Author: Goson Zhang <[email protected]>
AuthorDate: Wed Jun 19 09:43:33 2024 +0800
[INLONG-10441][DataProxy] Supports obtaining Audit-Proxy through InLong
Manager (#10442)
Co-authored-by: gosonzhang <[email protected]>
Co-authored-by: Charles Zhang <[email protected]>
---
inlong-dataproxy/conf/common.properties | 2 ++
.../dataproxy/config/CommonConfigHolder.java | 23 ++++++++++++++++++++++
.../inlong/dataproxy/metrics/audit/AuditUtils.java | 12 +++++++++--
.../src/test/resources/common.properties | 2 ++
4 files changed, 37 insertions(+), 2 deletions(-)
diff --git a/inlong-dataproxy/conf/common.properties
b/inlong-dataproxy/conf/common.properties
index 6d9e399867..269bf5c274 100644
--- a/inlong-dataproxy/conf/common.properties
+++ b/inlong-dataproxy/conf/common.properties
@@ -39,5 +39,7 @@ online.metric.prometheus.http.port=9081
# whether to enable audit
audit.enable=true
+# whether to enable audit proxy address discovery by the Manager
+audit.proxys.discovery.manager.enable=false
# audit proxy address
audit.proxys=127.0.0.1:10081
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
index a788702489..c1434d36fd 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
@@ -128,6 +128,8 @@ public class CommonConfigHolder {
// Audit fields
private static final String KEY_ENABLE_AUDIT = "audit.enable";
private static final boolean VAL_DEF_ENABLE_AUDIT = true;
+ private static final String KEY_AUDIT_PROXYS_DISCOVERY_MANAGER_ENABLE =
"audit.proxys.discovery.manager.enable";
+ private static final boolean VAL_DEF_AUDIT_PROXYS_DISCOVERY_MANAGER_ENABLE
= false;
private static final String KEY_AUDIT_PROXYS = "audit.proxys";
@Deprecated
private static final String KEY_AUDIT_FILE_PATH = "audit.filePath";
@@ -209,6 +211,7 @@ public class CommonConfigHolder {
private long metaConfigSyncInvlMs = VAL_DEF_CONFIG_SYNC_INTERVAL_MS;
private long metaConfigWastAlarmMs =
VAL_DEF_META_CONFIG_SYNC_WAST_ALARM_MS;
private boolean enableAudit = VAL_DEF_ENABLE_AUDIT;
+ private boolean enableAuditProxysDiscoveryFromManager =
VAL_DEF_AUDIT_PROXYS_DISCOVERY_MANAGER_ENABLE;
private final HashSet<String> auditProxys = new HashSet<>();
private String auditFilePath = VAL_DEF_AUDIT_FILE_PATH;
private int auditMaxCacheRows = VAL_DEF_AUDIT_MAX_CACHE_ROWS;
@@ -337,6 +340,10 @@ public class CommonConfigHolder {
return enableAudit;
}
+ public boolean isEnableAuditProxysDiscoveryFromManager() {
+ return enableAuditProxysDiscoveryFromManager;
+ }
+
public boolean isEnableFileMetric() {
return enableFileMetric;
}
@@ -657,6 +664,11 @@ public class CommonConfigHolder {
if (StringUtils.isNotEmpty(tmpValue)) {
this.enableAudit = "TRUE".equalsIgnoreCase(tmpValue.trim());
}
+ // read whether discovery audit proxys from manager
+ tmpValue = this.props.get(KEY_AUDIT_PROXYS_DISCOVERY_MANAGER_ENABLE);
+ if (StringUtils.isNotEmpty(tmpValue)) {
+ this.enableAuditProxysDiscoveryFromManager =
"TRUE".equalsIgnoreCase(tmpValue.trim());
+ }
// read audit proxys
tmpValue = this.props.get(KEY_AUDIT_PROXYS);
if (StringUtils.isNotBlank(tmpValue)) {
@@ -668,6 +680,15 @@ public class CommonConfigHolder {
this.auditProxys.add(tmpIPPort.trim());
}
}
+ // check auditProxys configure
+ if (this.enableAudit) {
+ if (!this.enableAuditProxysDiscoveryFromManager &&
this.auditProxys.isEmpty()) {
+ LOG.error("{}'s {} must be configured when {} is true and {}
is false, exist!",
+ COMMON_CONFIG_FILE_NAME, KEY_AUDIT_PROXYS,
KEY_ENABLE_AUDIT,
+ KEY_AUDIT_PROXYS_DISCOVERY_MANAGER_ENABLE);
+ System.exit(2);
+ }
+ }
// read audit file path
tmpValue = compatGetValue(this.props,
KEY_AUDIT_FILE_PATHV2, KEY_AUDIT_FILE_PATH);
@@ -774,6 +795,8 @@ public class CommonConfigHolder {
.append("metaConfigSyncInvlMs", metaConfigSyncInvlMs)
.append("metaConfigWastAlarmMs", metaConfigWastAlarmMs)
.append("enableAudit", enableAudit)
+ .append("enableAuditProxysDiscoveryFromManager",
+ enableAuditProxysDiscoveryFromManager)
.append("auditProxys", auditProxys)
.append("auditFilePath", auditFilePath)
.append("auditMaxCacheRows", auditMaxCacheRows)
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/metrics/audit/AuditUtils.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/metrics/audit/AuditUtils.java
index ec81ff36ff..fc2fc8db2e 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/metrics/audit/AuditUtils.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/metrics/audit/AuditUtils.java
@@ -19,6 +19,7 @@ package org.apache.inlong.dataproxy.metrics.audit;
import org.apache.inlong.audit.AuditIdEnum;
import org.apache.inlong.audit.AuditOperator;
+import org.apache.inlong.audit.entity.AuditComponent;
import org.apache.inlong.audit.util.AuditConfig;
import org.apache.inlong.common.enums.MessageWrapType;
import org.apache.inlong.common.msg.AttributeConstants;
@@ -49,8 +50,15 @@ public class AuditUtils {
public static void initAudit() {
if (CommonConfigHolder.getInstance().isEnableAudit()) {
// AuditProxy
- AuditOperator.getInstance().setAuditProxy(
- CommonConfigHolder.getInstance().getAuditProxys());
+ if
(CommonConfigHolder.getInstance().isEnableAuditProxysDiscoveryFromManager()) {
+
AuditOperator.getInstance().setAuditProxy(AuditComponent.DATAPROXY,
+
CommonConfigHolder.getInstance().getManagerHosts().get(0),
+
CommonConfigHolder.getInstance().getManagerAuthSecretId(),
+
CommonConfigHolder.getInstance().getManagerAuthSecretKey());
+ } else {
+ AuditOperator.getInstance().setAuditProxy(
+ CommonConfigHolder.getInstance().getAuditProxys());
+ }
// AuditConfig
AuditConfig auditConfig = new AuditConfig(
CommonConfigHolder.getInstance().getAuditFilePath(),
diff --git
a/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
b/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
index 4077a082ad..6e422cdd51 100644
--- a/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
+++ b/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
@@ -34,3 +34,5 @@ id2topic.unconfigured.default.topics= test1 test2 test3
msg.send.failure.retry.enable = true
# max retries after sent failure
msg.max.retries=2
+# whether to enable audit proxy address discovery by the Manager
+audit.proxys.discovery.manager.enable=true
