Merge branch 'apache:master' into INLONG-10312

bin/init-config.sh

@@ -53,8 +53,6 @@ init_inlong_agent() {
  cd $INLONG_HOME/inlong-agent/conf
  $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() {
@@ -77,14 +75,16 @@ init_inlong_audit() {
  $SED_COMMAND 's#jdbc:mysql://.*apache_inlong_audit#'''jdbc:mysql://${spring_datasource_hostname}:${spring_datasource_port}/apache_inlong_audit'''#g' audit-service.properties
  $SED_COMMAND 's/mysql.username=.*/'''mysql.username=${spring_datasource_username}'''/g' audit-service.properties
  $SED_COMMAND 's/mysql.password=.*/'''mysql.password=${spring_datasource_password}'''/g' audit-service.properties
+ $SED_COMMAND 's/audit.proxy.address.agent=.*/'''audit.proxy.address.agent=${audit_service_ip}:${audit_proxy_port}'''/g' audit-service.properties
+ $SED_COMMAND 's/audit.proxy.address.dataproxy=.*/'''audit.proxy.address.dataproxy=${audit_service_ip}:${audit_proxy_port}'''/g' audit-service.properties
+ $SED_COMMAND 's/audit.proxy.address.sort=.*/'''audit.proxy.address.sort=${audit_service_ip}:${audit_proxy_port}'''/g' audit-service.properties
 }
 
 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
+ $SED_COMMAND "s/audit.proxys.discovery.manager.enable=.*$/audit.proxys.discovery.manager.enable=true/g" common.properties
 }
 
 init_inlong_manager() {
@@ -96,14 +96,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

@@ -60,7 +60,7 @@ services:
  - ZK_URL=tubemq-server:2181
  - FLINK_HOST=jobmanager
  - FLINK_PORT=8081
- - AUDIT_PROXY_URL=audit:10081
+ - AUDIT_QUERY_URL=audit:10080
 
  dashboard:
  image: inlong/dashboard:${VERSION_TAG}
@@ -83,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
@@ -99,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
 
@@ -111,14 +109,16 @@ services:
  mysql:
  condition: service_healthy
  environment:
- - JDBC_URL=mysql:3306
- - USERNAME=root
- - PASSWORD=inlong
+ - AUDIT_JDBC_URL=mysql:3306
+ - AUDIT_JDBC_USERNAME=root
+ - AUDIT_JDBC_PASSWORD=inlong
  - MANAGER_OPENAPI_IP=manager
  - MANAGER_OPENAPI_PORT=8083
+ - AUDIT_PROXY_ADDRESS=audit:10081
  # pulsar or kafka
  - MQ_TYPE=pulsar
  ports:
+ - "10080:10080"
  - "10081:10081"
 
  # flink jobmanager

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/AgentConstants.java

@@ -68,9 +68,6 @@ public class AgentConstants {
  public static final String AUDIT_ENABLE = "audit.enable";
  public static final boolean DEFAULT_AUDIT_ENABLE = true;
 
- public static final String AUDIT_KEY_PROXYS = "audit.proxys";
- public static final String DEFAULT_AUDIT_PROXYS = "";
-
  public static final String AGENT_HISTORY_PATH = "agent.history.path";
  public static final String DEFAULT_AGENT_HISTORY_PATH = ".history";
 

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/metrics/audit/AuditUtils.java

@@ -20,17 +20,9 @@
 import org.apache.inlong.agent.conf.AgentConfiguration;
 import org.apache.inlong.audit.AuditOperator;
 import org.apache.inlong.audit.entity.AuditComponent;
-import org.apache.inlong.audit.util.AuditConfig;
-
-import org.apache.commons.lang3.StringUtils;
-
-import java.util.Collections;
-import java.util.HashSet;
 
 import static org.apache.inlong.agent.constant.AgentConstants.AUDIT_ENABLE;
-import static org.apache.inlong.agent.constant.AgentConstants.AUDIT_KEY_PROXYS;
 import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_AUDIT_ENABLE;
-import static org.apache.inlong.agent.constant.AgentConstants.DEFAULT_AUDIT_PROXYS;
 import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_ADDR;
 import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_AUTH_SECRET_ID;
 import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_AUTH_SECRET_KEY;
@@ -42,10 +34,6 @@
  */
 public class AuditUtils {
 
- public static final String AUDIT_KEY_FILE_PATH = "audit.filePath";
- public static final String AUDIT_DEFAULT_FILE_PATH = "/data/inlong/audit/";
- public static final String AUDIT_KEY_MAX_CACHE_ROWS = "audit.maxCacheRows";
- public static final int AUDIT_DEFAULT_MAX_CACHE_ROWS = 2000000;
  public static int AUDIT_ID_AGENT_READ_SUCCESS = 3;
  public static int AUDIT_ID_AGENT_SEND_SUCCESS = 4;
  public static int AUDIT_ID_AGENT_READ_FAILED = 524291;
@@ -79,25 +67,8 @@ public static void initAudit() {
  AgentConfiguration conf = AgentConfiguration.getAgentConf();
  IS_AUDIT = conf.getBoolean(AUDIT_ENABLE, DEFAULT_AUDIT_ENABLE);
  if (IS_AUDIT) {
- if (conf.hasKey(AUDIT_KEY_PROXYS)) {
- // AuditProxy
- String strIpPorts = conf.get(AUDIT_KEY_PROXYS, DEFAULT_AUDIT_PROXYS);
- HashSet<String> proxySet = new HashSet<>();
- if (!StringUtils.isBlank(strIpPorts)) {
- String[] ipPorts = strIpPorts.split("\\s+");
- Collections.addAll(proxySet, ipPorts);
- }
- AuditOperator.getInstance().setAuditProxy(proxySet);
- } else {
- AuditOperator.getInstance().setAuditProxy(AuditComponent.AGENT, conf.get(AGENT_MANAGER_ADDR),
- conf.get(AGENT_MANAGER_AUTH_SECRET_ID), conf.get(AGENT_MANAGER_AUTH_SECRET_KEY));
- }
-
- // AuditConfig
- String filePath = conf.get(AUDIT_KEY_FILE_PATH, AUDIT_DEFAULT_FILE_PATH);
- int maxCacheRow = conf.getInt(AUDIT_KEY_MAX_CACHE_ROWS, AUDIT_DEFAULT_MAX_CACHE_ROWS);
- AuditConfig auditConfig = new AuditConfig(filePath, maxCacheRow);
- AuditOperator.getInstance().setAuditConfig(auditConfig);
+ AuditOperator.getInstance().setAuditProxy(AuditComponent.AGENT, conf.get(AGENT_MANAGER_ADDR),
+ conf.get(AGENT_MANAGER_AUTH_SECRET_ID), conf.get(AGENT_MANAGER_AUTH_SECRET_KEY));
  }
  }
 

inlong-agent/agent-docker/Dockerfile

@@ -31,9 +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
 ENV AGENT_FETCH_INTERVAL=10
 ENV AGENT_HEARTBEAT_INTERVAL=10

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

@@ -24,8 +24,6 @@ sed -i "s/agent.local.ip=.*$/agent.local.ip=$local_ip/g" "${file_path}/conf/agen
 sed -i "s/agent.fetcher.interval=.*$/agent.fetcher.interval=$AGENT_FETCH_INTERVAL/g" "${file_path}/conf/agent.properties"
 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-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/store/RocksDBStoreImpl.java

@@ -82,8 +82,6 @@ private RocksDB initEnv(String childPath) {
  File finalPath = new File(parentPath, childPath);
  storePath = finalPath.getAbsolutePath();
  RocksDB.loadLibrary();
- final Options options = new Options();
- options.setCreateIfMissing(true);
  try {
  boolean result = finalPath.mkdirs();
  LOGGER.info("create directory {}, result is {}", finalPath, result);

inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/entity/AuditComponent.java

@@ -19,7 +19,7 @@
 
 public enum AuditComponent {
 
- AGENT("Agent"), DATAPROXY("DataProxy"), SORT("Sort"), COMMON_AUDIT("Common");
+ AGENT("Agent"), DATAPROXY("DataProxy"), SORT("Sort");
  private final String component;
 
  /**

inlong-audit/audit-common/src/main/java/org/apache/inlong/audit/utils/ThreadUtils.java

@@ -0,0 +1,33 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ThreadUtils {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(HttpUtils.class);
+ public static void sleep(long milliseconds) {
+ try {
+ Thread.sleep(milliseconds);
+ } catch (InterruptedException e) {
+ LOGGER.error("Sleep was interrupted", e);
+ }
+ }
+}

inlong-audit/audit-docker/Dockerfile

@@ -41,6 +41,9 @@ ENV AUDIT_JDBC_URL=127.0.0.1:3306
 ENV AUDIT_JDBC_USERNAME=root
 ENV AUDIT_JDBC_PASSWORD=inlong
 
+# Audit Proxy host
+ENV AUDIT_PROXY_ADDRESS=127.0.0.1:10081
+
 # jvm
 ENV AUDIT_JVM_HEAP_OPTS="-XX:+UseContainerSupport -XX:InitialRAMPercentage=40.0 -XX:MaxRAMPercentage=80.0 -XX:-UseAdaptiveSizePolicy"
 WORKDIR /opt/inlong-audit

inlong-audit/audit-docker/audit-docker.sh

@@ -61,6 +61,9 @@ sed -i "s/audit.store.jdbc.password=.*$/audit.store.jdbc.password=${AUDIT_JDBC_P
 sed -i "s/mysql.jdbc.url=.*$/mysql.jdbc.url=jdbc:mysql:\/\/${AUDIT_JDBC_URL}\/${AUDIT_DBNAME}/g" "${service_conf_file}"
 sed -i "s/mysql.jdbc.username=.*$/mysql.jdbc.username=${AUDIT_JDBC_USERNAME}/g" "${service_conf_file}"
 sed -i "s/mysql.jdbc.password=.*$/mysql.jdbc.password=${AUDIT_JDBC_PASSWORD}/g" "${service_conf_file}"
+sed -i "s/audit.proxy.address.agent=.*$/audit.proxy.address.agent = ${AUDIT_PROXY_ADDRESS}/g" "${service_conf_file}"
+sed -i "s/audit.proxy.address.dataproxy=.*$/audit.proxy.address.dataproxy = ${AUDIT_PROXY_ADDRESS}/g" "${service_conf_file}"
+sed -i "s/audit.proxy.address.sort=.*$/audit.proxy.address.sort = ${AUDIT_PROXY_ADDRESS}/g" "${service_conf_file}"
 
 # Whether the database table exists. If it does not exist, initialize the database and skip if it exists.
 if [[ "${AUDIT_JDBC_URL}" =~ (.+):([0-9]+) ]]; then