[INLONG-9928][Audit] Audit-service HA election through mysql (apache#…

…9929)

inlong-audit/audit-service/src/main/java/config/ConfigConstants.java

@@ -101,4 +101,17 @@ public class ConfigConstants {
  public static final String KEY_DAILY_SUMMARY_BEFORE_TIMES = "daily.summary.before.times";
  public static final int DEFAULT_DAILY_SUMMARY_BEFORE_TIMES = 2;
 
+ // HA selector config
+ public static final String KEY_RELEASE_LEADER_INTERVAL = "release.leader.interval";
+ public static final int DEFAULT_RELEASE_LEADER_INTERVAL = 40;
+ public static final String KEY_SELECTOR_THREAD_POOL_SIZE = "selector.thread.pool.size";
+ public static final int DEFAULT_SELECTOR_THREAD_POOL_SIZE = 3;
+
+ // HikariConfig
+ public static final String CACHE_PREP_STMTS = "cachePrepStmts";
+ public static final String PREP_STMT_CACHE_SIZE = "prepStmtCacheSize";
+ public static final String PREP_STMT_CACHE_SQL_LIMIT = "prepStmtCacheSqlLimit";
+
+ public static final int MAX_INIT_COUNT = 2;
+ public static final int RANDOM_BOUND = 10;
 }

inlong-audit/audit-service/src/main/java/config/SqlConstants.java

@@ -0,0 +1,37 @@
+/*
+ * 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 config;
+
+/**
+ * Sql constants
+ */
+public class SqlConstants {
+
+ // HA selector sql
+ public static final String SELECTOR_SQL =
+ "insert ignore into {0} (service_id, leader_id, last_seen_active) values (''{1}'', ''{2}'', now()) on duplicate key update leader_id = if(last_seen_active < now() - interval # second, values(leader_id), leader_id),last_seen_active = if(leader_id = values(leader_id), values(last_seen_active), last_seen_active)";
+ public static final String REPLACE_LEADER_SQL =
+ "replace into {0} ( service_id, leader_id, last_seen_active ) values (''{1}'', ''#'', now())";
+ public static final String RELEASE_SQL = "delete from {0} where service_id=''{1}'' and leader_id= ''{2}''";
+ public static final String IS_LEADER_SQL =
+ "select count(*) as is_leader from {0} where service_id=''{1}'' and leader_id=''{2}''";
+ public static final String SEARCH_CURRENT_LEADER_SQL =
+ "select leader_id as leader from {0} where service_id=''{1}''";
+ public static final String SELECT_TEST_SQL = "SELECT 1 ";
+
+}

inlong-audit/audit-service/src/main/java/elector/ElectorChangeListenerImpl.java

@@ -0,0 +1,34 @@
+/*
+ * 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 elector;
+
+import elector.api.SelectorChangeListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Elector change listener impl
+ */
+public class ElectorChangeListenerImpl implements SelectorChangeListener {
+
+ private static final Logger logger = LoggerFactory.getLogger(ElectorChangeListenerImpl.class);
+
+ public void leaderChanged(boolean currentNodeIsLeader) {
+ logger.info("LeaderChanged {}:", currentNodeIsLeader);
+ }
+}

inlong-audit/audit-service/src/main/java/elector/api/Selector.java

@@ -38,5 +38,5 @@ public abstract class Selector {
 
  public abstract boolean rebuildSelectorDBSource();
 
- public abstract boolean close();
+ public abstract void close();
 }

inlong-audit/audit-service/src/main/java/elector/api/SelectorFactory.java

@@ -0,0 +1,30 @@
+/*
+ * 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 elector.api;
+
+import elector.impl.SelectorImpl;
+
+/**
+ * Selector factory
+ */
+public class SelectorFactory {
+
+ public static Selector getNewElector(SelectorConfig electorConfig) {
+ return new SelectorImpl(electorConfig);
+ }
+}