[INLONG-10472][Manager] Use audit SDK to obtain audit proxy URL (#10473)

bin/init-config.sh

@@ -54,7 +54,6 @@ init_inlong_agent() {
   $SED_COMMAND "s|agent.local.ip=.*|agent.local.ip=${local_ip}|g" agent.properties
   $SED_COMMAND "s|agent.manager.addr=.*|agent.manager.addr=http://${manager_server_hostname}:${manager_server_port}|g" agent.properties
   $SED_COMMAND "s/audit.enable=.*$/audit.enable=true/g" agent.properties
-  $SED_COMMAND 's/audit.proxys=.*/'''audit.proxys=${audit_proxy_ip}:${audit_proxy_port}'''/g' agent.properties
 }
 
 init_inlong_audit() {
@@ -83,7 +82,6 @@ init_inlong_dataproxy() {
   echo "Init dataproxy configuration parameters"
   cd $INLONG_HOME/inlong-dataproxy/conf
   $SED_COMMAND 's/manager.hosts=.*/'''manager.hosts=${manager_server_hostname}:${manager_server_port}'''/g' common.properties
-  $SED_COMMAND 's/audit.proxys=.*/'''audit.proxys=${audit_proxy_ip}:${audit_proxy_port}'''/g' common.properties
   $SED_COMMAND "s/audit.enable=.*$/audit.enable=true/g" common.properties
 }
 
@@ -96,14 +94,12 @@ init_inlong_manager() {
     $SED_COMMAND 's#jdbc:mysql://.*apache_inlong_manager#'''jdbc:mysql://${spring_datasource_hostname}:${spring_datasource_port}/apache_inlong_manager'''#g' application-dev.properties
     $SED_COMMAND 's/spring.datasource.druid.username=.*/'''spring.datasource.druid.username=${spring_datasource_username}'''/g' application-dev.properties
     $SED_COMMAND 's/spring.datasource.druid.password=.*/'''spring.datasource.druid.password=${spring_datasource_password}'''/g' application-dev.properties
-    $SED_COMMAND 's/metrics.audit.proxy.hosts=.*/'''metrics.audit.proxy.hosts=${audit_proxy_ip}:${audit_proxy_port}'''/g' application-dev.properties
     $SED_COMMAND 's/audit.query.url=.*/'''audit.query.url=${audit_service_ip}:${audit_service_port}'''/g' application-dev.properties
   fi
   if [ $spring_profiles_active == "prod" ]; then
     $SED_COMMAND 's#jdbc:mysql://.*apache_inlong_manager#'''jdbc:mysql://${spring_datasource_hostname}:${spring_datasource_port}/apache_inlong_manager'''#g' application-prod.properties
     $SED_COMMAND 's/spring.datasource.druid.username=.*/'''spring.datasource.druid.username=${spring_datasource_username}'''/g' application-prod.properties
     $SED_COMMAND 's/spring.datasource.druid.password=.*/'''spring.datasource.druid.password=${spring_datasource_password}'''/g' application-prod.properties
-    $SED_COMMAND 's/metrics.audit.proxy.hosts=.*/'''metrics.audit.proxy.hosts=${audit_proxy_ip}:${audit_proxy_port}'''/g' application-prod.properties
     $SED_COMMAND 's/audit.query.url=.*/'''audit.query.url=${audit_service_ip}:${audit_service_port}'''/g' application-prod.properties
   fi
   echo "Init inlong manager flink plugin configuration"

bin/inlong-daemon

@@ -83,10 +83,10 @@ start_inlong_audit() {
   if [ "${mq_type}" = "kafka" ]; then
     bash +x ./bin/proxy-start.sh kafka
   fi
+  wait_port_to_listen audit ${audit_proxy_port}
   echo "start audit store"
   bash +x ./bin/store-start.sh
   # wait to start
-  wait_port_to_listen audit ${audit_proxy_port}
   echo "start audit service"
   bash +x ./bin/service-start.sh
   # wait to start

conf/inlong.conf

@@ -53,7 +53,5 @@ dataproxy_port=46801
 audit_service_ip=127.0.0.1
 # audit service Port
 audit_service_port=10080
-# audit proxy IP
-audit_proxy_ip=127.0.0.1
 # audit proxy Port
 audit_proxy_port=10081

docker/docker-compose/docker-compose.yml

@@ -61,7 +61,6 @@ services:
       - FLINK_HOST=jobmanager
       - FLINK_PORT=8081
       - AUDIT_QUERY_URL=audit:10080
-      - AUDIT_PROXY_URL=audit:10081
 
   dashboard:
     image: inlong/dashboard:${VERSION_TAG}
@@ -84,7 +83,6 @@ services:
     environment:
       - MANAGER_OPENAPI_IP=manager
       - MANAGER_OPENAPI_PORT=8083
-      - AUDIT_PROXY_URL=audit:10081
       # pulsar or kafka
       - MQ_TYPE=pulsar
       - ETH_NAME=eth0
@@ -100,7 +98,6 @@ services:
       - MANAGER_OPENAPI_PORT=8083
       - DATAPROXY_IP=dataproxy
       - DATAPROXY_PORT=46801
-      - AUDIT_PROXY_URL=audit:10081
     volumes:
       - ./collect-data:/data/collect-data
 

inlong-agent/agent-docker/Dockerfile

@@ -31,7 +31,6 @@ ENV MANAGER_OPENAPI_IP=127.0.0.1
 ENV MANAGER_OPENAPI_PORT=8082
 ENV DATAPROXY_IP=127.0.0.1
 ENV DATAPROXY_PORT=46801
-ENV AUDIT_PROXY_URL=127.0.0.1:10081
 # enable audit, true or false
 ENV AUDIT_ENABLE=true
 ENV ETH_NETWORK=eth0

inlong-agent/agent-docker/agent-docker.sh

@@ -25,7 +25,6 @@ sed -i "s/agent.fetcher.interval=.*$/agent.fetcher.interval=$AGENT_FETCH_INTERVA
 sed -i "s/agent.heartbeat.interval=.*$/agent.heartbeat.interval=$AGENT_HEARTBEAT_INTERVAL/g" "${file_path}/conf/agent.properties"
 sed -i "s/agent.manager.addr=.*$/agent.manager.addr=http:\/\/$MANAGER_OPENAPI_IP:$MANAGER_OPENAPI_PORT/g" "${file_path}/conf/agent.properties"
 sed -i "s/audit.enable=.*$/audit.enable=$AUDIT_ENABLE/g" "${file_path}/conf/agent.properties"
-sed -i "s/audit.proxys=.*$/audit.proxys=$AUDIT_PROXY_URL/g" "${file_path}/conf/agent.properties"
 sed -i "s/agent.cluster.tag=.*$/agent.cluster.tag=$CLUSTER_TAG/g" "${file_path}/conf/agent.properties"
 sed -i "s/agent.cluster.name=.*$/agent.cluster.name=$CLUSTER_NAME/g" "${file_path}/conf/agent.properties"
 sed -i "s/agent.cluster.inCharges=.*$/agent.cluster.inCharges=$CLUSTER_IN_CHARGES/g" "${file_path}/conf/agent.properties"

inlong-dataproxy/dataproxy-docker/Dockerfile

@@ -30,7 +30,6 @@ ENV MANAGER_OPENAPI_PORT=8083
 ENV ETH_NAME=eth0
 # enable audit, true or false
 ENV AUDIT_ENABLE=true
-ENV AUDIT_PROXY_URL=127.0.0.1:10081
 # pulsar or tubemq or kafka
 ENV MQ_TYPE=pulsar
 ENV CLUSTER_TAG=default_cluster

inlong-dataproxy/dataproxy-docker/dataproxy-docker.sh

@@ -34,7 +34,6 @@ fi
 
 sed -i "s/manager.hosts=.*$/manager.hosts=${MANAGER_OPENAPI_IP}:${MANAGER_OPENAPI_PORT}/g" "${common_conf_file}"
 sed -i "s/audit.enable=.*$/audit.enable=${AUDIT_ENABLE}/g" "${common_conf_file}"
-sed -i "s/audit.proxys=.*$/audit.proxys=${AUDIT_PROXY_URL}/g" "${common_conf_file}"
 sed -i "s/localhost.*$/${local_ip}/g" "${mq_conf_file}"
 sed -i "s/proxy.cluster.tag=.*$/proxy.cluster.tag=${CLUSTER_TAG}/g" "${common_conf_file}"
 sed -i "s/proxy.cluster.name=.*$/proxy.cluster.name=${CLUSTER_NAME}/g" "${common_conf_file}"

inlong-manager/manager-docker/Dockerfile

@@ -52,7 +52,6 @@ ADD target/sort-connector-kafka-v1.15-${VERSION}.jar /opt/inlong-sort/connectors
 ADD target/sort-connector-mysql-cdc-v1.15-${VERSION}.jar /opt/inlong-sort/connectors/
 # audit
 ENV AUDIT_QUERY_URL=127.0.0.1:10080
-ENV AUDIT_PROXY_URL=127.0.0.1:10081
 ADD manager-docker.sh bin/
 RUN chmod +x bin/manager-docker.sh
 CMD ["bin/manager-docker.sh"]

inlong-manager/manager-docker/manager-docker.sh

@@ -42,7 +42,6 @@ sed -i "s/data.cleansing.before.days=.*$/data.cleansing.before.days=${CLEANSING_
 sed -i "s/data.cleansing.batchSize=.*$/data.cleansing.batchSize=${CLEANSING_BATCHSIZE}/g" "${conf_file}"
 # for audit data
 sed -i "s/audit.query.url=.*$/audit.query.url=${AUDIT_QUERY_URL}/g" "${conf_file}"
-sed -i "s/metrics.audit.proxy.hosts=.*$/metrics.audit.proxy.hosts=${AUDIT_PROXY_URL}/g" "${conf_file}"
 # flink-sort-plugin.properties
 sed -i "s/flink.rest.address=.*$/flink.rest.address=${FLINK_HOST}/g" "${flink_conf_file}"
 sed -i "s/flink.rest.port=.*$/flink.rest.port=${FLINK_PORT}/g" "${flink_conf_file}"

inlong-manager/manager-pojo/pom.xml

@@ -32,6 +32,17 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.inlong</groupId>
+            <artifactId>audit-sdk</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-simple</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.apache.inlong</groupId>
             <artifactId>manager-common</artifactId>

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/audit/AuditProxyResponse.java

@@ -17,9 +17,9 @@
 
 package org.apache.inlong.manager.pojo.audit;
 
-import lombok.AllArgsConstructor;
+import org.apache.inlong.audit.entity.AuditProxy;
+
 import lombok.Data;
-import lombok.NoArgsConstructor;
 
 import java.util.List;
 
@@ -30,12 +30,4 @@ public class AuditProxyResponse {
     private String errMsg;
     private List<AuditProxy> data;
 
-    @Data
-    @NoArgsConstructor
-    @AllArgsConstructor
-    public static class AuditProxy {
-
-        private String host;
-        private int port;
-    }
 }

inlong-manager/manager-service/pom.xml

@@ -469,17 +469,6 @@
             <groupId>redis.clients</groupId>
             <artifactId>jedis</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.apache.inlong</groupId>
-            <artifactId>audit-sdk</artifactId>
-            <version>${project.version}</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.slf4j</groupId>
-                    <artifactId>slf4j-simple</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
         <dependency>
             <groupId>org.apache.inlong</groupId>
             <artifactId>sdk-common</artifactId>

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/node/AgentClusterNodeInstallOperator.java

@@ -58,7 +58,6 @@ public class AgentClusterNodeInstallOperator implements InlongClusterNodeInstall
     public static final String AGENT_MANAGER_ADDR = "agent.manager.addr";
     public static final String AGENT_CLUSTER_NAME = "agent.cluster.name";
     public static final String AGENT_CLUSTER_TAG = "agent.cluster.tag";
-    public static final String AUDIT_PROXYS_URL = "audit.proxys";
     public static final String AGENT_LOCAL_IP = "agent.local.ip";
 
     @Autowired
@@ -72,8 +71,6 @@ public class AgentClusterNodeInstallOperator implements InlongClusterNodeInstall
     @Autowired
     private UserEntityMapper userEntityMapper;
 
-    @Value("${metrics.audit.proxy.hosts:127.0.0.1:10081}")
-    private String auditProxyUrl;
     @Value("${agent.install.path:inlong/inlong-installer/}")
     private String agentInstallPath;
     @Value("${manager.url:127.0.0.1:8083}")
@@ -112,7 +109,6 @@ public boolean install(ClusterNodeRequest clusterNodeRequest, String operator) {
             configMap.put(AGENT_MANAGER_AUTH_SECRET_KEY, secretKey);
             configMap.put(AGENT_CLUSTER_TAG, clusterEntity.getClusterTags());
             configMap.put(AGENT_CLUSTER_NAME, clusterEntity.getName());
-            configMap.put(AUDIT_PROXYS_URL, auditProxyUrl);
             commandExecutor.modifyConfig(request, configMap, confFile);
             String startCmd = agentInstallPath + INSTALLER_START_CMD;
             commandExecutor.execRemote(request, startCmd);

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/AuditService.java

@@ -18,8 +18,8 @@
 package org.apache.inlong.manager.service.core;
 
 import org.apache.inlong.audit.entity.AuditInformation;
+import org.apache.inlong.audit.entity.AuditProxy;
 import org.apache.inlong.common.enums.IndicatorType;
-import org.apache.inlong.manager.pojo.audit.AuditProxyResponse.AuditProxy;
 import org.apache.inlong.manager.pojo.audit.AuditRequest;
 import org.apache.inlong.manager.pojo.audit.AuditVO;
 

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java

@@ -19,6 +19,7 @@
 
 import org.apache.inlong.audit.AuditOperator;
 import org.apache.inlong.audit.entity.AuditInformation;
+import org.apache.inlong.audit.entity.AuditProxy;
 import org.apache.inlong.audit.entity.FlowType;
 import org.apache.inlong.common.enums.IndicatorType;
 import org.apache.inlong.manager.common.consts.InlongConstants;
@@ -35,7 +36,6 @@
 import org.apache.inlong.manager.dao.mapper.StreamSinkEntityMapper;
 import org.apache.inlong.manager.dao.mapper.StreamSourceEntityMapper;
 import org.apache.inlong.manager.pojo.audit.AuditProxyResponse;
-import org.apache.inlong.manager.pojo.audit.AuditProxyResponse.AuditProxy;
 import org.apache.inlong.manager.pojo.audit.AuditRequest;
 import org.apache.inlong.manager.pojo.audit.AuditVO;
 import org.apache.inlong.manager.pojo.user.LoginUserUtils;

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/SortFlinkConfigOperator.java

@@ -17,6 +17,8 @@
 
 package org.apache.inlong.manager.service.resource.sort;
 
+import org.apache.inlong.audit.entity.AuditComponent;
+import org.apache.inlong.audit.entity.AuditProxy;
 import org.apache.inlong.common.enums.IndicatorType;
 import org.apache.inlong.manager.common.consts.InlongConstants;
 import org.apache.inlong.manager.common.consts.SinkType;
@@ -40,12 +42,12 @@
 import org.apache.inlong.sort.protocol.node.transform.TransformNode;
 import org.apache.inlong.sort.protocol.transformation.relation.NodeRelation;
 
+import com.google.common.base.Joiner;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import java.util.ArrayList;
@@ -69,8 +71,6 @@ public class SortFlinkConfigOperator implements SortConfigOperator {
     private static final Logger LOGGER = LoggerFactory.getLogger(SortFlinkConfigOperator.class);
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
-    @Value("${metrics.audit.proxy.hosts:127.0.0.1}")
-    private String auditHost;
     @Autowired
     private StreamSourceService sourceService;
     @Autowired
@@ -298,8 +298,10 @@ private void addToStreamExt(InlongStreamInfo streamInfo, String value) {
     private void addAuditId(Map<String, Object> properties, String type, IndicatorType indicatorType) {
         try {
             String auditId = auditService.getAuditId(type, indicatorType);
+            List<AuditProxy> auditProxyList = auditService.getAuditProxy(AuditComponent.SORT.getComponent());
             properties.putIfAbsent("metrics.audit.key", auditId);
-            properties.putIfAbsent("metrics.audit.proxy.hosts", auditHost);
+            properties.putIfAbsent("metrics.audit.proxy.hosts",
+                    Joiner.on(InlongConstants.AMPERSAND).join(auditProxyList));
         } catch (Exception e) {
             LOGGER.error("Current type ={} is not set auditId", type);
         }

inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/AuditController.java

@@ -18,7 +18,7 @@
 package org.apache.inlong.manager.web.controller;
 
 import org.apache.inlong.audit.entity.AuditInformation;
-import org.apache.inlong.manager.pojo.audit.AuditProxyResponse.AuditProxy;
+import org.apache.inlong.audit.entity.AuditProxy;
 import org.apache.inlong.manager.pojo.audit.AuditRequest;
 import org.apache.inlong.manager.pojo.audit.AuditVO;
 import org.apache.inlong.manager.pojo.common.Response;

inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/AuditController.java

@@ -17,9 +17,9 @@
 
 package org.apache.inlong.manager.web.controller.openapi;
 
+import org.apache.inlong.audit.entity.AuditProxy;
 import org.apache.inlong.common.pojo.audit.AuditConfig;
 import org.apache.inlong.common.pojo.audit.AuditConfigRequest;
-import org.apache.inlong.manager.pojo.audit.AuditProxyResponse.AuditProxy;
 import org.apache.inlong.manager.pojo.audit.AuditRequest;
 import org.apache.inlong.manager.pojo.audit.AuditVO;
 import org.apache.inlong.manager.pojo.common.Response;

inlong-manager/manager-web/src/main/resources/application-dev.properties

@@ -93,9 +93,6 @@ group.deleted.batch.size=100
 # If turned on, the groups could be deleted periodically.
 group.deleted.enabled=false
 
-# Audit Proxy Address
-metrics.audit.proxy.hosts=127.0.0.1:10081
-
 # Tencent cloud log service endpoint, The Operator cls resource by it
 cls.manager.endpoint=127.0.0.1
 

inlong-manager/manager-web/src/main/resources/application-prod.properties

@@ -92,9 +92,6 @@ group.deleted.batch.size=100
 # If turned on, the groups could be deleted periodically.
 group.deleted.enabled=false
 
-# Audit Proxy Address
-metrics.audit.proxy.hosts=127.0.0.1:10081
-
 # Tencent cloud log service endpoint, The Operator cls resource by it
 cls.manager.endpoint=127.0.0.1
 

inlong-manager/manager-web/src/main/resources/application-test.properties

@@ -93,9 +93,6 @@ group.deleted.batch.size=100
 # If turned on, the groups could be deleted periodically.
 group.deleted.enabled=false
 
-# Audit Proxy Address
-metrics.audit.proxy.hosts=127.0.0.1:10081
-
 # Tencent cloud log service endpoint, The Operator cls resource by it
 cls.manager.endpoint=127.0.0.1