This is an automated email from the ASF dual-hosted git repository.
doleyzi 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 5fa60262c7 [INLONG-10387][Audit] Audit SDK supports obtaining
Audit-Proxy capabilities through InLong Manager (#10398)
5fa60262c7 is described below
commit 5fa60262c79be84744619e5be261b8c869cc5139
Author: doleyzi <[email protected]>
AuthorDate: Fri Jun 14 15:29:51 2024 +0800
[INLONG-10387][Audit] Audit SDK supports obtaining Audit-Proxy
capabilities through InLong Manager (#10398)
[INLONG-10387][Audit] Audit SDK supports obtaining Audit-Proxy
capabilities through InLong Manager (#10398)
---
inlong-audit/audit-common/pom.xml | 4 +
.../audit/{entities => entity}/AuditComponent.java | 14 ++-
.../apache/inlong/audit/entity}/AuditProxy.java | 19 +++-
.../apache/inlong/audit/entity/CommonResponse.java | 91 ++++++++++++++++++
.../org/apache/inlong/audit/utils/HttpUtils.java | 106 +++++++++++++++++++++
.../apache/inlong/audit/heartbeat/Heartbeat.java | 2 +-
inlong-audit/audit-sdk/README.md | 48 ++++++++--
.../java/org/apache/inlong/audit/AuditIdEnum.java | 10 +-
.../org/apache/inlong/audit/AuditReporterImpl.java | 44 ++++++---
.../org/apache/inlong/audit/entity/AuditType.java | 6 +-
.../org/apache/inlong/audit/send/ProxyManager.java | 88 ++++++++++++++++-
.../apache/inlong/audit/cache/AuditProxyCache.java | 2 +-
.../apache/inlong/audit/service/ApiService.java | 4 +-
13 files changed, 405 insertions(+), 33 deletions(-)
diff --git a/inlong-audit/audit-common/pom.xml
b/inlong-audit/audit-common/pom.xml
index 70f0d1911a..ee0fbf81e1 100644
--- a/inlong-audit/audit-common/pom.xml
+++ b/inlong-audit/audit-common/pom.xml
@@ -42,6 +42,10 @@
<groupId>io.netty</groupId>
<artifactId>netty-buffer</artifactId>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entities/AuditComponent.java
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/AuditComponent.java
similarity index 81%
rename from
inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entities/AuditComponent.java
rename to
inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/AuditComponent.java
index d7e077e3fa..b0d2e17b54 100644
---
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entities/AuditComponent.java
+++
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/AuditComponent.java
@@ -15,17 +15,29 @@
* limitations under the License.
*/
-package org.apache.inlong.audit.entities;
+package org.apache.inlong.audit.entity;
public enum AuditComponent {
AGENT("Agent"), DATAPROXY("DataProxy"), SORT("Sort"),
COMMON_AUDIT("Common");
private final String component;
+ /**
+ * Constructor for the enum.
+ *
+ * @param component the name of the component
+ */
+
AuditComponent(String component) {
this.component = component;
}
+ /**
+ * Returns the name of the component.
+ *
+ * @return the name of the component
+ */
+
public String getComponent() {
return component;
}
diff --git
a/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/entities/AuditProxy.java
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/AuditProxy.java
similarity index 71%
rename from
inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/entities/AuditProxy.java
rename to
inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/AuditProxy.java
index c594ae2ffe..8dc2e6f7e8 100644
---
a/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/entities/AuditProxy.java
+++
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/AuditProxy.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.audit.entities;
+package org.apache.inlong.audit.entity;
import lombok.AllArgsConstructor;
import lombok.Data;
@@ -24,6 +24,23 @@ import lombok.Data;
@AllArgsConstructor
public class AuditProxy {
+ /**
+ * The host of the audit proxy.
+ */
private String host;
+
+ /**
+ * The port of the audit proxy.
+ */
private int port;
+
+ /**
+ * Returns a string representation of the audit proxy.
+ *
+ * @return a string representation of the audit proxy
+ */
+ @Override
+ public String toString() {
+ return String.format("%s:%d", host, port);
+ }
}
diff --git
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/CommonResponse.java
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/CommonResponse.java
new file mode 100644
index 0000000000..190a1ac2df
--- /dev/null
+++
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/CommonResponse.java
@@ -0,0 +1,91 @@
+/*
+ * 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.inlong.audit.entity;
+
+import com.google.gson.Gson;
+import lombok.Data;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+
+/**
+ * Class representing a common response.
+ */
+@Data
+public class CommonResponse<T> {
+
+ /**
+ * Gson instance for JSON serialization and deserialization.
+ */
+ private static final Gson gson = new Gson();
+
+ /**
+ * Error message of the response.
+ */
+ private String errMsg;
+
+ /**
+ * Success status of the response.
+ */
+ private boolean success;
+
+ /**
+ * Data of the response.
+ */
+ private List<T> data;
+
+ /**
+ * Converts a JSON string to a CommonResponse object.
+ *
+ * @param json the JSON string
+ * @param clazz the class of the data
+ * @return a CommonResponse object
+ */
+ public static CommonResponse fromJson(String json, Class clazz) {
+ Type objectType = type(CommonResponse.class, clazz);
+ return gson.fromJson(json, objectType);
+ }
+
+ /**
+ * Returns a parameterized type.
+ *
+ * @param raw the raw type
+ * @param args the actual type arguments
+ * @return a parameterized type
+ */
+ private static ParameterizedType type(final Class raw, final Type... args)
{
+ return new ParameterizedType() {
+
+ @Override
+ public Type getRawType() {
+ return raw;
+ }
+
+ @Override
+ public Type[] getActualTypeArguments() {
+ return args;
+ }
+
+ @Override
+ public Type getOwnerType() {
+ return null;
+ }
+ };
+ }
+}
diff --git
a/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/utils/HttpUtils.java
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/utils/HttpUtils.java
new file mode 100644
index 0000000000..34c60dc66b
--- /dev/null
+++
b/inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/utils/HttpUtils.java
@@ -0,0 +1,106 @@
+/*
+ * 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.inlong.audit.utils;
+
+import org.apache.inlong.common.util.BasicAuth;
+
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.ssl.SSLContextBuilder;
+import org.apache.http.util.EntityUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.SSLContext;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class HttpUtils {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(HttpUtils.class);
+ private static final String PARAM_COMPONENT = "component";
+ private static HttpClient httpClient = null;
+
+ static {
+ try {
+ SSLContext sslContext = SSLContextBuilder.create()
+ .loadTrustMaterial(new TrustSelfSignedStrategy())
+ .build();
+
+ httpClient = HttpClientBuilder.create()
+ .setSSLContext(sslContext)
+ .build();
+ } catch (Exception e) {
+ LOGGER.error("Error initializing SSL context or HTTP client", e);
+ }
+ }
+
+ public static Map<String, String> getAuthHeader(String secretId, String
secretKey) {
+ Map<String, String> header = new HashMap<>();
+ try {
+ header.put(BasicAuth.BASIC_AUTH_HEADER,
+ BasicAuth.genBasicAuthCredential(secretId, secretKey));
+ } catch (Exception e) {
+ LOGGER.error("Get auth header error", e);
+ }
+ return header;
+ }
+
+ public static String httpGet(String component, String url, String
secretId, String secretKey, int timeoutMs) {
+ if (httpClient == null) {
+ LOGGER.error("httpClient is null");
+ return null;
+ }
+ try {
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectTimeout(timeoutMs)
+ .setConnectionRequestTimeout(timeoutMs)
+ .setSocketTimeout(timeoutMs)
+ .build();
+ URIBuilder uriBuilder = new URIBuilder(url);
+ uriBuilder.addParameter(PARAM_COMPONENT, component);
+ String finalUrl = uriBuilder.build().toString();
+
+ HttpGet request = new HttpGet(finalUrl);
+ request.setConfig(requestConfig);
+
+ Map<String, String> authHeaders = getAuthHeader(secretId,
secretKey);
+ for (Map.Entry<String, String> entry : authHeaders.entrySet()) {
+ request.addHeader(entry.getKey(), entry.getValue());
+ }
+
+ try (CloseableHttpResponse response = (CloseableHttpResponse)
httpClient.execute(request)) {
+ String responseStr =
EntityUtils.toString(response.getEntity());
+ LOGGER.info("Http response: {}", responseStr);
+ if (responseStr != null && !responseStr.isEmpty()
+ && response.getStatusLine().getStatusCode() == 200) {
+ return responseStr;
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error("Send get request has exception", e);
+ }
+ return null;
+ }
+}
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/heartbeat/Heartbeat.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/heartbeat/Heartbeat.java
index ce921f6051..e4f23a6ed8 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/heartbeat/Heartbeat.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/heartbeat/Heartbeat.java
@@ -39,7 +39,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
-import static org.apache.inlong.audit.entities.AuditComponent.COMMON_AUDIT;
+import static org.apache.inlong.audit.entity.AuditComponent.COMMON_AUDIT;
public class Heartbeat {
diff --git a/inlong-audit/audit-sdk/README.md b/inlong-audit/audit-sdk/README.md
index 0d31329ef6..2689dbb463 100644
--- a/inlong-audit/audit-sdk/README.md
+++ b/inlong-audit/audit-sdk/README.md
@@ -12,21 +12,59 @@ which can ensure that each module is reconciled in
accordance with the unified a
## Usage
### Configure Audit Proxy Addresses
-The Audit SDK will summarize the results according to the cycle
+- The Audit SDK will summarize the results according to the cycle
and send them to the ip:port list set by the interface.
+- There are two ways to set the address of the Audit Proxy, configure the
address directly or get the address from the manager. Please choose one of the
methods.
+#### Configure Audit Proxy Addresses by fixed ip:port
If the ip:port of the AuditProxy is fixed, then this interface needs to be
called once.
-If the AuditProxy changes in real time, then the business program needs to
call this interface periodically to update
+If the AuditProxy changes in real time, then the business program needs to
call this interface periodically to update.
```java
- HashSet<String> ipPortList=new HashSet<>();
+ HashSet<String> ipPortList = new HashSet<>();
ipPortList.add("0.0.0.0:54041");
AuditOperator.getInstance().setAuditProxy(ipPortList);
```
+#### Configure Audit Proxy Addresses by InLong Manager
+By configuring the InLong Manager's address, module information, and manager
certification information,
+The Audit SDK will automatically fetch the Manager to obtain the address of
the Audit Proxy.
+```java
+ String host = "127.0.0.1:8083"; // The manager address
+ String secretId = "*****"; // Secret id
+ String secretKey = "******"; // Secret key
+
AuditOperator.getInstance().setAuditProxy(AuditComponent,host,secretId,secretKey);
+```
+- Explain of AuditComponent
+```java
+public enum AuditComponent {
+
+ AGENT("Agent"), DATAPROXY("DataProxy"), SORT("Sort"),
COMMON_AUDIT("Common");
+ private final String component;
+
+ /**
+ * Constructor for the enum.
+ *
+ * @param component the name of the component
+ */
+
+ AuditComponent(String component) {
+ this.component = component;
+ }
+
+ /**
+ * Returns the name of the component.
+ *
+ * @return the name of the component
+ */
+
+ public String getComponent() {
+ return component;
+ }
+}
+```
### Add Audit Data
Call the add method for statistics, where the auditID parameter uniquely
identifies an audit object,
inlongGroupID,inlongStreamID,logTime are audit dimensions, count is the number
of items, size is the size, and logTime
is milliseconds.
-
#### Example for Agent to Add Audit Data
```java
AuditOperator.getInstance().add(auditID, auditTag, inlongGroupID,
inlongStreamID, logTime,
@@ -39,7 +77,6 @@ The scenario of supplementary recording of agent data, so the
version number par
count, size, auditVersion);
```
The scenario of supplementary recording of DataProxy data, so the version
number parameter needs to be passed in.
-
#### Example for Sort Flink to Add Audit Data
```java
AuditReporterImpl auditReporter=new AuditReporterImpl();
@@ -58,7 +95,6 @@ The scenario of supplementary recording of DataProxy data, so
the version number
logTime, count, size, auditVersion)
```
In order to ensure the accuracy of auditing, each operator needs to create an
auditAuditReporterImpl instance.
-
- Explain of AuditDimensions
| parameter | description
|
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditIdEnum.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditIdEnum.java
index ca437fb035..8186e98181 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditIdEnum.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditIdEnum.java
@@ -40,7 +40,7 @@ import static org.apache.inlong.audit.entity.AuditType.MYSQL;
import static org.apache.inlong.audit.entity.AuditType.POSTGRES;
import static org.apache.inlong.audit.entity.AuditType.SDK;
import static org.apache.inlong.audit.entity.AuditType.STARROCKS;
-import static org.apache.inlong.audit.entity.AuditType.TUBE;
+import static org.apache.inlong.audit.entity.AuditType.TUBEMQ;
import static org.apache.inlong.audit.entity.FlowType.INPUT;
import static org.apache.inlong.audit.entity.FlowType.OUTPUT;
@@ -88,11 +88,11 @@ public enum AuditIdEnum {
SORT_POSTGRES_INPUT(27, INPUT, POSTGRES, "Received Audit Metrics for Sort
Postgres"),
SORT_POSTGRES_OUTPUT(28, OUTPUT, POSTGRES, "Sent Audit Metrics for Sort
Postgres"),
- SORT_BINLOG_INPUT(29, INPUT, BINLOG, "Received Audit Metrics for Sort
Binlog"),
- SORT_BINLOG_OUTPUT(30, OUTPUT, BINLOG, "Sent Audit Metrics for Sort
Binlog"),
+ SORT_BINLOG_INPUT(35, INPUT, BINLOG, "Received Audit Metrics for Sort
Binlog"),
+ SORT_BINLOG_OUTPUT(36, OUTPUT, BINLOG, "Sent Audit Metrics for Sort
Binlog"),
- SORT_TUBE_INPUT(33, INPUT, TUBE, "Received Audit Metrics for Sort Tube"),
- SORT_TUBE_OUTPUT(34, OUTPUT, TUBE, "Sent Audit Metrics for Sort Tube"),
+ SORT_TUBE_INPUT(33, INPUT, TUBEMQ, "Received Audit Metrics for Sort
TubeMQ"),
+ SORT_TUBE_OUTPUT(34, OUTPUT, TUBEMQ, "Sent Audit Metrics for Sort TubeMQ"),
SORT_MYSQL_INPUT(35, INPUT, MYSQL, "Received Audit Metrics for Sort
MySQL"),
SORT_MYSQL_OUTPUT(36, OUTPUT, MYSQL, "Sent Audit Metrics for Sort MySQL"),
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
index 745821b512..d1748a5b4b 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
@@ -17,6 +17,7 @@
package org.apache.inlong.audit;
+import org.apache.inlong.audit.entity.AuditComponent;
import org.apache.inlong.audit.entity.AuditInformation;
import org.apache.inlong.audit.entity.AuditMetric;
import org.apache.inlong.audit.entity.FlowType;
@@ -78,7 +79,7 @@ public class AuditReporterImpl implements Serializable {
private final ScheduledExecutorService timeoutExecutor =
Executors.newSingleThreadScheduledExecutor();
private int packageId = 1;
private int dataId = 0;
- private boolean initialized = false;
+ private volatile boolean initialized = false;
private SenderManager manager;
private AtomicInteger flushStat = new AtomicInteger(0);
private AuditConfig auditConfig = null;
@@ -189,18 +190,24 @@ public class AuditReporterImpl implements Serializable {
* Set AuditProxy from the ip
*/
public void setAuditProxy(HashSet<String> ipPortList) {
- try {
- GLOBAL_LOCK.lockInterruptibly();
- if (!initialized) {
- init();
- initialized = true;
- }
- ProxyManager.getInstance().setAuditProxy(ipPortList);
- } catch (InterruptedException e) {
- LOGGER.error(e.getMessage());
- } finally {
- GLOBAL_LOCK.unlock();
+ checkInitStatus();
+ ProxyManager.getInstance().setAuditProxy(ipPortList);
+ }
+
+ /**
+ * Set AuditProxy from the manager host
+ */
+ public void setAuditProxy(AuditComponent component, String managerHost,
String secretId, String secretKey) {
+ checkInitStatus();
+ ProxyManager.getInstance().setManagerConfig(component, managerHost,
secretId, secretKey);
+ }
+
+ private synchronized void checkInitStatus() {
+ if (initialized) {
+ return;
}
+ init();
+ initialized = true;
}
/**
@@ -591,4 +598,17 @@ public class AuditReporterImpl implements Serializable {
public int getStartAuditIdForMetric() {
return AuditManagerUtils.getStartAuditIdForMetric();
}
+
+ public void setManagerTimeout(int timeoutMs) {
+
+ ProxyManager.getInstance().setManagerTimeout(timeoutMs);
+ }
+
+ public void setAutoUpdateAuditProxy(boolean autoUpdateAuditProxy) {
+
ProxyManager.getInstance().setAutoUpdateAuditProxy(autoUpdateAuditProxy);
+ }
+
+ public void setUpdateInterval(int updateInterval) {
+ ProxyManager.getInstance().setUpdateInterval(updateInterval);
+ }
}
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/entity/AuditType.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/entity/AuditType.java
index 0af03844d6..3b195b8214 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/entity/AuditType.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/entity/AuditType.java
@@ -32,9 +32,9 @@ public enum AuditType {
DORIS("Doris"),
KUDU("Kudu"),
POSTGRES("Postgres"),
- BINLOG("Binlog"),
- TUBE("Tube"),
- MYSQL("MySQL"),
+ BINLOG("MYSQL_BINLOG"),
+ TUBEMQ("TubeMQ"),
+ MYSQL("MYSQL_SQL"),
HDFS("HDFS");
private final String auditType;
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/ProxyManager.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/ProxyManager.java
index a3dfcd4e75..159839d2fc 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/ProxyManager.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/ProxyManager.java
@@ -17,6 +17,11 @@
package org.apache.inlong.audit.send;
+import org.apache.inlong.audit.entity.AuditComponent;
+import org.apache.inlong.audit.entity.AuditProxy;
+import org.apache.inlong.audit.entity.CommonResponse;
+import org.apache.inlong.audit.utils.HttpUtils;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -25,12 +30,25 @@ import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
public class ProxyManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(ProxyManager.class);
private static final ProxyManager instance = new ProxyManager();
private final List<String> currentIpPorts = new CopyOnWriteArrayList<>();
+ private final ScheduledExecutorService timer =
Executors.newSingleThreadScheduledExecutor();
+ private final static String GET_AUDIT_PROXY_API_PATH =
"/inlong/manager/openapi/audit/getAuditProxy";
+ private int timeoutMs = 10000;
+ private boolean autoUpdateAuditProxy = false;
+ private int updateInterval = 60000;
+ private String auditProxyApiUrl;
+ private AuditComponent component;
+ private String secretId;
+ private String secretKey;
+ private volatile boolean timerStarted = false;
private ProxyManager() {
}
@@ -42,13 +60,81 @@ public class ProxyManager {
/**
* update config
*/
- public void setAuditProxy(HashSet<String> ipPortList) {
+ public synchronized void setAuditProxy(HashSet<String> ipPortList) {
if (!ipPortList.equals(new HashSet<>(currentIpPorts))) {
currentIpPorts.clear();
currentIpPorts.addAll(ipPortList);
}
}
+ public synchronized void setManagerConfig(AuditComponent component, String
managerHost, String secretId,
+ String secretKey) {
+ if (!managerHost.endsWith("/")) {
+ managerHost = managerHost + "/";
+ }
+ if (!(managerHost.startsWith("http://";) ||
managerHost.startsWith("https://";))) {
+ managerHost = "http://"; + managerHost;
+ }
+ auditProxyApiUrl = String.format("%s%s", managerHost,
GET_AUDIT_PROXY_API_PATH);
+ LOGGER.info("Audit Proxy API URL: {}", auditProxyApiUrl);
+
+ this.component = component;
+ this.secretId = secretId;
+ this.secretKey = secretKey;
+
+ updateAuditProxy();
+
+ if (autoUpdateAuditProxy) {
+ startTimer();
+ LOGGER.info("Auto update from manager");
+ }
+ }
+
+ private void updateAuditProxy() {
+ String response = HttpUtils.httpGet(component.getComponent(),
auditProxyApiUrl, secretId, secretKey, timeoutMs);
+ if (response == null) {
+ LOGGER.error("Response is null: {} {} {} ",
component.getComponent(), auditProxyApiUrl, secretId,
+ secretKey);
+ return;
+ }
+ CommonResponse<AuditProxy> commonResponse =
+ CommonResponse.fromJson(response, AuditProxy.class);
+ if (commonResponse == null) {
+ LOGGER.error("No data in the response: {} {} {} {}",
component.getComponent(), auditProxyApiUrl, secretId,
+ secretKey);
+ return;
+ }
+ HashSet<String> proxyList = new HashSet<>();
+ for (AuditProxy auditProxy : commonResponse.getData()) {
+ proxyList.add(auditProxy.toString());
+ }
+ setAuditProxy(proxyList);
+ LOGGER.info("Get audit proxy from manager: {}", proxyList);
+ }
+
+ private synchronized void startTimer() {
+ if (timerStarted) {
+ return;
+ }
+ timer.scheduleWithFixedDelay(this::updateAuditProxy,
+ 0,
+ updateInterval,
+ TimeUnit.MILLISECONDS);
+ timerStarted = true;
+ }
+
+ public void setManagerTimeout(int timeoutMs) {
+ this.timeoutMs = timeoutMs;
+ }
+
+ public void setAutoUpdateAuditProxy(boolean autoUpdateAuditProxy) {
+ this.autoUpdateAuditProxy = autoUpdateAuditProxy;
+ }
+
+ public void setUpdateInterval(int updateInterval) {
+ this.updateInterval = updateInterval;
+ }
+
public InetSocketAddress getInetSocketAddress() {
if (currentIpPorts.isEmpty()) {
return null;
diff --git
a/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/cache/AuditProxyCache.java
b/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/cache/AuditProxyCache.java
index b51c2b2ff1..f05525096c 100644
---
a/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/cache/AuditProxyCache.java
+++
b/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/cache/AuditProxyCache.java
@@ -18,8 +18,8 @@
package org.apache.inlong.audit.cache;
import org.apache.inlong.audit.config.Configuration;
-import org.apache.inlong.audit.entities.AuditProxy;
import org.apache.inlong.audit.entities.JdbcConfig;
+import org.apache.inlong.audit.entity.AuditProxy;
import org.apache.inlong.audit.utils.JdbcUtils;
import com.github.benmanes.caffeine.cache.Cache;
diff --git
a/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/ApiService.java
b/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/ApiService.java
index 8693539282..3861ff60a2 100644
---
a/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/ApiService.java
+++
b/inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/service/ApiService.java
@@ -25,10 +25,10 @@ import org.apache.inlong.audit.cache.RealTimeQuery;
import org.apache.inlong.audit.cache.TenMinutesCache;
import org.apache.inlong.audit.config.Configuration;
import org.apache.inlong.audit.entities.ApiType;
-import org.apache.inlong.audit.entities.AuditComponent;
import org.apache.inlong.audit.entities.AuditCycle;
-import org.apache.inlong.audit.entities.AuditProxy;
import org.apache.inlong.audit.entities.StatData;
+import org.apache.inlong.audit.entity.AuditComponent;
+import org.apache.inlong.audit.entity.AuditProxy;
import org.apache.inlong.audit.heartbeat.ProxyHeartbeat;
import com.google.common.util.concurrent.RateLimiter;
