完成告警切换的功能
This commit is contained in:
zhangxin10
parent
fb2d7d554c
commit
3d5a652d23
|
|
@ -0,0 +1,88 @@
|
|||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<groupId>com.ai.cloud</groupId>
|
||||
<artifactId>skywalking-alarm</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<name>skywalking-alarm</name>
|
||||
<url>http://maven.apache.org</url>
|
||||
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
<repositories>
|
||||
<repository>
|
||||
<id>Company</id>
|
||||
<url>http://10.1.228.199:18081/nexus/content/groups/public/</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<version>4.12</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-core</artifactId>
|
||||
<version>2.4.1</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>redis.clients</groupId>
|
||||
<artifactId>jedis</artifactId>
|
||||
<version>2.8.0</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.sun.mail</groupId>
|
||||
<artifactId>javax.mail</artifactId>
|
||||
<version>1.5.4</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.freemarker</groupId>
|
||||
<artifactId>freemarker</artifactId>
|
||||
<version>2.3.23</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.curator</groupId>
|
||||
<artifactId>curator-framework</artifactId>
|
||||
<version>2.8.0</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.zookeeper</groupId>
|
||||
<artifactId>zookeeper</artifactId>
|
||||
<version>3.4.0</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.curator</groupId>
|
||||
<artifactId>curator-recipes</artifactId>
|
||||
<version>2.8.0</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.google.code.gson</groupId>
|
||||
<artifactId>gson</artifactId>
|
||||
<version>2.2.2</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>mysql</groupId>
|
||||
<artifactId>mysql-connector-java</artifactId>
|
||||
<version>5.1.37</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<source>1.7</source>
|
||||
<target>1.7</target>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
|
|
@ -0,0 +1,135 @@
|
|||
package com.ai.cloud.skywalking.alarm;
|
||||
|
||||
import com.ai.cloud.skywalking.alarm.dao.AlarmMessageDao;
|
||||
import com.ai.cloud.skywalking.alarm.model.ApplicationInfo;
|
||||
import com.ai.cloud.skywalking.alarm.model.UserInfo;
|
||||
import com.ai.cloud.skywalking.alarm.zk.ZKUtil;
|
||||
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
|
||||
public class AlarmMessageProcessThread extends Thread {
|
||||
private int rank;
|
||||
private String threadId;
|
||||
private Logger logger = LogManager.getLogger(AlarmMessageProcessThread.class);
|
||||
private boolean isChanged = false;
|
||||
private List<InterProcessMutex> locks;
|
||||
private List<UserInfo> toBeProcessUsers;
|
||||
|
||||
public AlarmMessageProcessThread() {
|
||||
threadId = UUID.randomUUID().toString();
|
||||
registerServer(threadId);
|
||||
rank = ballot(threadId);
|
||||
ZKUtil.watch(AlarmServerRegisterWatcher.getInstance());
|
||||
toBeProcessUsers = getToBeProcessUsers();
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
lockAllUsers();
|
||||
while (true) {
|
||||
for (UserInfo userInfo : toBeProcessUsers) {
|
||||
for (ApplicationInfo applicationInfo : userInfo.getApplicationInfos()) {
|
||||
System.out.println(threadId + applicationInfo.getAppId());
|
||||
}
|
||||
}
|
||||
|
||||
if (isChanged) {
|
||||
unlockAllUsers();
|
||||
rank = ballot(threadId);
|
||||
toBeProcessUsers = getToBeProcessUsers();
|
||||
lockAllUsers();
|
||||
isChanged = false;
|
||||
}
|
||||
|
||||
try {
|
||||
Thread.sleep(1000L);
|
||||
} catch (InterruptedException e) {
|
||||
logger.error("Sleep failed", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void unlockAllUsers() {
|
||||
for (InterProcessMutex lock : locks) {
|
||||
try {
|
||||
lock.release();
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to release lock[{}]", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void lockAllUsers() {
|
||||
locks = new ArrayList<InterProcessMutex>(toBeProcessUsers.size());
|
||||
InterProcessMutex tmpLock;
|
||||
for (UserInfo userInfo : toBeProcessUsers) {
|
||||
tmpLock = ZKUtil.getProcessUserLock(userInfo.getUserId());
|
||||
locks.add(tmpLock);
|
||||
try {
|
||||
tmpLock.acquire();
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to lock ");
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private List<UserInfo> getToBeProcessUsers() {
|
||||
List<UserInfo> userInfos = AlarmMessageDao.selectAllUserInfo();
|
||||
List<String> allServerIds = ZKUtil.selectAllServerIds();
|
||||
int step = (int) Math.ceil(userInfos.size() * 1.0 / allServerIds.size());
|
||||
int start = rank * step;
|
||||
int end = (rank + 1) * step;
|
||||
if (end > userInfos.size()) {
|
||||
return new ArrayList<UserInfo>();
|
||||
}
|
||||
|
||||
List<UserInfo> toBeProcessUsers = userInfos.subList(start, end);
|
||||
for (UserInfo userInfo : toBeProcessUsers) {
|
||||
userInfo.setApplicationInfos(AlarmMessageDao.selectAllApplicationsByUserId(userInfo.getUserId()));
|
||||
}
|
||||
return toBeProcessUsers;
|
||||
}
|
||||
|
||||
|
||||
private void registerServer(String serverId) {
|
||||
InterProcessMutex registerLock = ZKUtil.getRegisterLock();
|
||||
try {
|
||||
registerLock.acquire();
|
||||
ZKUtil.register(serverId);
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to lock.", e);
|
||||
} finally {
|
||||
if (registerLock != null) {
|
||||
try {
|
||||
registerLock.release();
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to release lock.", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private int ballot(String threadId) {
|
||||
List<String> serverIds = ZKUtil.selectAllServerIds();
|
||||
int rank = 0;
|
||||
for (String tmpServerId : serverIds) {
|
||||
if (tmpServerId.hashCode() < threadId.hashCode()) {
|
||||
rank++;
|
||||
}
|
||||
}
|
||||
return rank;
|
||||
}
|
||||
|
||||
|
||||
public void setChanged(boolean changed) {
|
||||
isChanged = changed;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,44 @@
|
|||
package com.ai.cloud.skywalking.alarm;
|
||||
|
||||
import com.ai.cloud.skywalking.alarm.conf.Config;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class AlarmProcessServer {
|
||||
|
||||
private static Logger logger = LogManager.getLogger(AlarmProcessServer.class);
|
||||
|
||||
private static List<AlarmMessageProcessThread> processThreads =
|
||||
new ArrayList<AlarmMessageProcessThread>();
|
||||
|
||||
public static void main(String[] main) {
|
||||
logger.info("Begin to start alarm process server....");
|
||||
logger.info("Begin to start process thread...");
|
||||
AlarmMessageProcessThread tmpThread;
|
||||
for (int i = 0; i < Config.Server.PROCESS_THREAD_SIZE; i++) {
|
||||
tmpThread = new AlarmMessageProcessThread();
|
||||
tmpThread.start();
|
||||
processThreads.add(tmpThread);
|
||||
}
|
||||
logger.info("Successfully launched {} processing threads.", Config.Server.PROCESS_THREAD_SIZE);
|
||||
logger.info("Begin to start user inspector thread....");
|
||||
new UserInfoInspector().start();
|
||||
logger.info("Start user inspector thread success....");
|
||||
logger.info("Alarm process server successfully started.");
|
||||
while (true) {
|
||||
try {
|
||||
Thread.sleep(Config.Server.DAEMON_THREAD_WAITE_INTERVAL);
|
||||
} catch (InterruptedException e) {
|
||||
logger.error("Sleep failed", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public static List<AlarmMessageProcessThread> getProcessThreads() {
|
||||
return processThreads;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
package com.ai.cloud.skywalking.alarm;
|
||||
|
||||
import com.ai.cloud.skywalking.alarm.zk.ZKUtil;
|
||||
import org.apache.curator.framework.api.CuratorWatcher;
|
||||
import org.apache.zookeeper.WatchedEvent;
|
||||
import org.apache.zookeeper.Watcher;
|
||||
|
||||
public class AlarmServerRegisterWatcher implements CuratorWatcher {
|
||||
|
||||
private static AlarmServerRegisterWatcher watcher = new AlarmServerRegisterWatcher();
|
||||
|
||||
@Override
|
||||
public void process(WatchedEvent watchedEvent) throws Exception {
|
||||
if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
|
||||
for (AlarmMessageProcessThread thread : AlarmProcessServer.getProcessThreads()) {
|
||||
thread.setChanged(true);
|
||||
}
|
||||
}
|
||||
|
||||
ZKUtil.watch(getInstance());
|
||||
}
|
||||
|
||||
public static AlarmServerRegisterWatcher getInstance() {
|
||||
return watcher;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
package com.ai.cloud.skywalking.alarm;
|
||||
|
||||
import com.ai.cloud.skywalking.alarm.dao.AlarmMessageDao;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
public class UserInfoInspector extends Thread {
|
||||
|
||||
private Logger logger = LogManager.getLogger(UserInfoInspector.class);
|
||||
|
||||
private int preUserSize;
|
||||
|
||||
public UserInfoInspector() {
|
||||
preUserSize = AlarmMessageDao.selectUserCount();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
int currentUserSize;
|
||||
while (true) {
|
||||
try {
|
||||
Thread.sleep(10 * 1000L);
|
||||
} catch (InterruptedException e) {
|
||||
logger.error("Sleep Failed", e);
|
||||
}
|
||||
|
||||
currentUserSize = AlarmMessageDao.selectUserCount();
|
||||
|
||||
if (currentUserSize != preUserSize) {
|
||||
logger.info("Total user has been changed. Notice all process thread to change process date.");
|
||||
for (AlarmMessageProcessThread thread : AlarmProcessServer.getProcessThreads()) {
|
||||
thread.setChanged(true);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,58 @@
|
|||
package com.ai.cloud.skywalking.alarm.conf;
|
||||
|
||||
public class Config {
|
||||
|
||||
public static class Server {
|
||||
public static int PROCESS_THREAD_SIZE = 2;
|
||||
|
||||
public static long DAEMON_THREAD_WAITE_INTERVAL = 50000L;
|
||||
|
||||
}
|
||||
|
||||
public static class ZKPath {
|
||||
|
||||
public static String CONNECT_STR = "127.0.0.1:2181";
|
||||
public static int CONNECT_TIMEOUT = 1000;
|
||||
public static int RETRY_TIMEOUT = 1000;
|
||||
public static int RETRY_TIMES = 3;
|
||||
|
||||
public static String NODE_PREFIX = "/skywalking";
|
||||
|
||||
public static String SERVER_REGISTER_LOCK_PATH = "/alarm-server/register";
|
||||
public static String REGISTER_SERVER_PATH = "/alarm-server/servers";
|
||||
public static String USER_REGISTER_LOCK_PATH = "/alarm-server/users";
|
||||
}
|
||||
|
||||
public static class DB {
|
||||
public static String PASSWORD = "devrdbusr13";
|
||||
public static String USER_NAME = "devrdbusr13";
|
||||
public static String DRIVER_CLASS = "com.mysql.jdbc.Driver";
|
||||
public static String URL = "jdbc:mysql://10.1.228.200:31306/test";
|
||||
}
|
||||
|
||||
public static class Alarm {
|
||||
|
||||
public static String REDIS_SERVER = "127.0.0.1:6379";
|
||||
|
||||
public static int REDIS_MAX_IDLE = 10;
|
||||
|
||||
public static int REDIS_MIN_IDLE = 1;
|
||||
|
||||
public static int REDIS_MAX_TOTAL = 20;
|
||||
|
||||
public static boolean ALARM_OFF_FLAG = false;
|
||||
}
|
||||
|
||||
public static class MailSender {
|
||||
|
||||
public static String HOST = "";
|
||||
|
||||
public static String TRANSPORT_PROTOCOL = "";
|
||||
|
||||
public static boolean SMTP_AUTH = true;
|
||||
|
||||
public static String USER_NAME = "";
|
||||
|
||||
public static String PASSWORD = "";
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,85 @@
|
|||
package com.ai.cloud.skywalking.alarm.conf;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.lang.reflect.Field;
|
||||
import java.lang.reflect.Modifier;
|
||||
import java.util.LinkedList;
|
||||
import java.util.Properties;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class ConfigInitializer {
|
||||
private static Logger logger = Logger.getLogger(ConfigInitializer.class.getName());
|
||||
|
||||
public static void initialize() {
|
||||
InputStream inputStream = ConfigInitializer.class.getResourceAsStream("/sky-walking.auth");
|
||||
if (inputStream == null) {
|
||||
logger.log(Level.ALL, "No provider sky-walking certification documents, buried point won't work");
|
||||
} else {
|
||||
try {
|
||||
Properties properties = new Properties();
|
||||
properties.load(inputStream);
|
||||
initNextLevel(properties, Config.class, new ConfigDesc());
|
||||
} catch (IllegalAccessException e) {
|
||||
logger.log(Level.ALL, "Parsing certification file failed, buried won't work");
|
||||
} catch (IOException e) {
|
||||
logger.log(Level.ALL, "Failed to read the certification file, buried won't work");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void initNextLevel(Properties properties, Class<?> recentConfigType, ConfigDesc parentDesc) throws NumberFormatException, IllegalArgumentException, IllegalAccessException {
|
||||
for (Field field : recentConfigType.getFields()) {
|
||||
if (Modifier.isPublic(field.getModifiers()) && Modifier.isStatic(field.getModifiers())) {
|
||||
String configKey = (parentDesc + "." +
|
||||
field.getName()).toLowerCase();
|
||||
String value = properties.getProperty(configKey);
|
||||
if (value != null) {
|
||||
if (field.getType().equals(int.class))
|
||||
field.set(null, Integer.valueOf(value));
|
||||
if (field.getType().equals(String.class))
|
||||
field.set(null, value);
|
||||
if (field.getType().equals(long.class))
|
||||
field.set(null, Long.valueOf(value));
|
||||
if (field.getType().equals(boolean.class))
|
||||
field.set(null, Boolean.valueOf(value));
|
||||
}
|
||||
}
|
||||
}
|
||||
for (Class<?> innerConfiguration : recentConfigType.getClasses()) {
|
||||
parentDesc.append(innerConfiguration.getSimpleName());
|
||||
initNextLevel(properties, innerConfiguration, parentDesc);
|
||||
parentDesc.removeLastDesc();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class ConfigDesc {
|
||||
private LinkedList<String> descs = new LinkedList<String>();
|
||||
|
||||
void append(String currentDesc) {
|
||||
descs.addLast(currentDesc);
|
||||
}
|
||||
|
||||
void removeLastDesc() {
|
||||
descs.removeLast();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
if (descs.size() == 0) {
|
||||
return "";
|
||||
}
|
||||
StringBuilder ret = new StringBuilder(descs.getFirst());
|
||||
boolean first = true;
|
||||
for (String desc : descs) {
|
||||
if (first) {
|
||||
first = false;
|
||||
continue;
|
||||
}
|
||||
ret.append(".").append(desc);
|
||||
}
|
||||
return ret.toString();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,121 @@
|
|||
package com.ai.cloud.skywalking.alarm.dao;
|
||||
|
||||
import com.ai.cloud.skywalking.alarm.conf.Config;
|
||||
import com.ai.cloud.skywalking.alarm.model.ApplicationInfo;
|
||||
import com.ai.cloud.skywalking.alarm.model.UserInfo;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
import java.sql.*;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class AlarmMessageDao {
|
||||
|
||||
private static Logger logger = LogManager.getLogger(AlarmMessageDao.class);
|
||||
private static Connection con;
|
||||
|
||||
static {
|
||||
try {
|
||||
Class.forName(Config.DB.DRIVER_CLASS);
|
||||
} catch (ClassNotFoundException e) {
|
||||
logger.error("Failed to found DB driver class.", e);
|
||||
System.exit(-1);
|
||||
}
|
||||
|
||||
try {
|
||||
con = DriverManager.getConnection(Config.DB.URL, Config.DB.USER_NAME, Config.DB.PASSWORD);
|
||||
} catch (SQLException e) {
|
||||
logger.error("Failed to connect DB", e);
|
||||
System.exit(-1);
|
||||
}
|
||||
}
|
||||
|
||||
public static List<UserInfo> selectAllUserInfo() {
|
||||
List<UserInfo> result = new ArrayList<UserInfo>();
|
||||
try {
|
||||
PreparedStatement ps = con.prepareStatement("SELECT user_info.uid FROM user_info WHERE sts = ?");
|
||||
ps.setString(1, "A");
|
||||
ResultSet rs = ps.executeQuery();
|
||||
UserInfo userInfo;
|
||||
while (rs.next()) {
|
||||
userInfo = new UserInfo(rs.getString("uid"));
|
||||
result.add(userInfo);
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
logger.error("Failed to select all user info", e);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
public static int selectUserCount() {
|
||||
try {
|
||||
PreparedStatement ps = con.prepareStatement("SELECT count(user_info.uid) as totalNumber FROM user_info WHERE sts = ?");
|
||||
ps.setString(1, "A");
|
||||
ResultSet rs = ps.executeQuery();
|
||||
rs.next();
|
||||
return rs.getInt("totalNumber");
|
||||
} catch (SQLException e) {
|
||||
logger.error("Failed to select all user info", e);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
public static List<ApplicationInfo> selectAllApplicationsByUserId(String userId) {
|
||||
List<ApplicationInfo> result = new ArrayList<ApplicationInfo>();
|
||||
try {
|
||||
PreparedStatement ps = con.prepareStatement("SELECT alarm_rule.app_id, alarm_rule.uid,alarm_rule.is_global, alarm_rule.todo_type," +
|
||||
" alarm_rule.config_args FROM alarm_rule WHERE uid = ? AND sts = ?");
|
||||
ps.setString(1, userId);
|
||||
ps.setString(2, "A");
|
||||
ResultSet rs = ps.executeQuery();
|
||||
ApplicationInfo globalConfig = null;
|
||||
ApplicationInfo tmpApplication;
|
||||
while (rs.next()) {
|
||||
if ("1".equals(rs.getString("is_global"))) {
|
||||
globalConfig = new ApplicationInfo();
|
||||
globalConfig.setConfigArgs(rs.getString("config_args"));
|
||||
continue;
|
||||
}
|
||||
tmpApplication = new ApplicationInfo();
|
||||
tmpApplication.setAppId(rs.getString("app_id"));
|
||||
tmpApplication.setUId(rs.getString("uid"));
|
||||
tmpApplication.setConfigArgs(rs.getString("config_args"));
|
||||
tmpApplication.setToDoType(rs.getString("todo_type"));
|
||||
result.add(tmpApplication);
|
||||
}
|
||||
|
||||
if (globalConfig == null) {
|
||||
//
|
||||
throw new IllegalArgumentException("Can not found the global config");
|
||||
}
|
||||
|
||||
List<ApplicationInfo> allApplication = new ArrayList<ApplicationInfo>();
|
||||
ps = con.prepareStatement("SELECT application_info.app_id, application_info.uid FROM application_info WHERE uid = ? AND sts = ?");
|
||||
ps.setString(1, userId);
|
||||
ps.setString(2, "A");
|
||||
rs = ps.executeQuery();
|
||||
ApplicationInfo applicationInfo;
|
||||
while (rs.next()) {
|
||||
applicationInfo = new ApplicationInfo();
|
||||
applicationInfo.setAppId(rs.getString("app_id"));
|
||||
applicationInfo.setUId(rs.getString("uid"));
|
||||
allApplication.add(applicationInfo);
|
||||
}
|
||||
|
||||
allApplication.removeAll(result);
|
||||
|
||||
for (ApplicationInfo app : allApplication) {
|
||||
app.setConfigArgs(globalConfig.getConfigArgs());
|
||||
app.setToDoType(globalConfig.getToDoType());
|
||||
result.add(app);
|
||||
}
|
||||
|
||||
} catch (SQLException e) {
|
||||
logger.error("Failed to query applications.", e);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,70 @@
|
|||
package com.ai.cloud.skywalking.alarm.model;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
|
||||
public class ApplicationInfo {
|
||||
private String appId;
|
||||
private String configArgs;
|
||||
private String UId;
|
||||
private String toDoType;
|
||||
private ConfigArgsDescriber configArgsDescriber;
|
||||
|
||||
public ApplicationInfo() {
|
||||
}
|
||||
|
||||
public String getAppId() {
|
||||
return appId;
|
||||
}
|
||||
|
||||
public void setAppId(String appId) {
|
||||
this.appId = appId;
|
||||
}
|
||||
|
||||
public void setConfigArgs(String configArgs) {
|
||||
this.configArgs = configArgs;
|
||||
configArgsDescriber = new Gson().fromJson(configArgs, ConfigArgsDescriber.class);
|
||||
}
|
||||
|
||||
public ConfigArgsDescriber getConfigArgsDescriber() {
|
||||
return configArgsDescriber;
|
||||
}
|
||||
|
||||
public String getConfigArgs() {
|
||||
return configArgs;
|
||||
}
|
||||
|
||||
public void setUId(String UId) {
|
||||
this.UId = UId;
|
||||
}
|
||||
|
||||
public String getUId() {
|
||||
return UId;
|
||||
}
|
||||
|
||||
public void setToDoType(String toDoType) {
|
||||
this.toDoType = toDoType;
|
||||
}
|
||||
|
||||
public String getToDoType() {
|
||||
return toDoType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) return true;
|
||||
if (!(o instanceof ApplicationInfo)) return false;
|
||||
|
||||
ApplicationInfo that = (ApplicationInfo) o;
|
||||
|
||||
if (getAppId() != null ? !getAppId().equals(that.getAppId()) : that.getAppId() != null) return false;
|
||||
return !(getUId() != null ? !getUId().equals(that.getUId()) : that.getUId() != null);
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int result = getAppId() != null ? getAppId().hashCode() : 0;
|
||||
result = 31 * result + (getUId() != null ? getUId().hashCode() : 0);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
package com.ai.cloud.skywalking.alarm.model;
|
||||
|
||||
public class ConfigArgsDescriber {
|
||||
private int period;
|
||||
|
||||
public int getPeriod() {
|
||||
return period;
|
||||
}
|
||||
|
||||
public void setPeriod(int period) {
|
||||
this.period = period;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
package com.ai.cloud.skywalking.alarm.model;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public class UserInfo {
|
||||
private String userId;
|
||||
|
||||
private List<ApplicationInfo> applicationInfos;
|
||||
|
||||
public UserInfo(String userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
public String getUserId() {
|
||||
return userId;
|
||||
}
|
||||
|
||||
public List<ApplicationInfo> getApplicationInfos() {
|
||||
return applicationInfos;
|
||||
}
|
||||
|
||||
public void setApplicationInfos(List<ApplicationInfo> applicationInfos) {
|
||||
this.applicationInfos = applicationInfos;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,128 @@
|
|||
package com.ai.cloud.skywalking.alarm.redis;
|
||||
|
||||
import com.ai.cloud.skywalking.alarm.conf.Config;
|
||||
import com.ai.cloud.skywalking.alarm.model.ApplicationInfo;
|
||||
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
import redis.clients.jedis.Jedis;
|
||||
import redis.clients.jedis.JedisPool;
|
||||
import redis.clients.jedis.exceptions.JedisConnectionException;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class RedisUtil {
|
||||
private static Logger logger = LogManager.getLogger(RedisUtil.class);
|
||||
private static JedisPool jedisPool;
|
||||
private static String[] config;
|
||||
private static RedisInspector connector = new RedisInspector();
|
||||
private static Object lock = new Object();
|
||||
|
||||
static {
|
||||
GenericObjectPoolConfig genericObjectPoolConfig = buildGenericObjectPoolConfig();
|
||||
String redisServerConfig = Config.Alarm.REDIS_SERVER;
|
||||
if (redisServerConfig == null || redisServerConfig.length() <= 0) {
|
||||
logger.error("Redis server is not setting. Switch off alarm module. ");
|
||||
} else {
|
||||
config = redisServerConfig.split(":");
|
||||
if (config.length != 2) {
|
||||
logger.error("Redis server address is illegal setting, need to be 'ip:port'. Switch off alarm module. ");
|
||||
Config.Alarm.ALARM_OFF_FLAG = true;
|
||||
} else {
|
||||
jedisPool = new JedisPool(genericObjectPoolConfig, config[0],
|
||||
Integer.valueOf(config[1]));
|
||||
// Test connect redis.
|
||||
Jedis jedis = null;
|
||||
try {
|
||||
jedis = jedisPool.getResource();
|
||||
} catch (Exception e) {
|
||||
handleFailedToConnectRedisServerException(e);
|
||||
logger.error("can't connect to redis["
|
||||
+ Config.Alarm.REDIS_SERVER + "]", e);
|
||||
} finally {
|
||||
if (jedis != null) {
|
||||
jedis.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static List<String> getAlarmMessage(ApplicationInfo applicationInfo) {
|
||||
Jedis jedis = null;
|
||||
try {
|
||||
jedis = jedisPool.getResource();
|
||||
return new ArrayList<String>(jedis.hgetAll(generateAlarmKey(applicationInfo)).values());
|
||||
} catch (Exception e) {
|
||||
handleFailedToConnectRedisServerException(e);
|
||||
logger.error("Failed to set data.", e);
|
||||
} finally {
|
||||
if (jedis != null) {
|
||||
jedis.close();
|
||||
}
|
||||
}
|
||||
|
||||
return new ArrayList<String>();
|
||||
}
|
||||
|
||||
private static String generateAlarmKey(ApplicationInfo applicationInfo) {
|
||||
return applicationInfo.getUId() + "-" + applicationInfo.getAppId() + "-"
|
||||
+ ((System.currentTimeMillis() / (10000 * 6))
|
||||
- applicationInfo.getConfigArgsDescriber().getPeriod());
|
||||
}
|
||||
|
||||
private static GenericObjectPoolConfig buildGenericObjectPoolConfig() {
|
||||
GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
|
||||
genericObjectPoolConfig.setTestOnBorrow(true);
|
||||
genericObjectPoolConfig.setMaxIdle(Config.Alarm.REDIS_MAX_IDLE);
|
||||
genericObjectPoolConfig.setMinIdle(Config.Alarm.REDIS_MIN_IDLE);
|
||||
genericObjectPoolConfig.setMaxTotal(Config.Alarm.REDIS_MAX_TOTAL);
|
||||
return genericObjectPoolConfig;
|
||||
}
|
||||
|
||||
private static void handleFailedToConnectRedisServerException(Exception e) {
|
||||
if (e instanceof JedisConnectionException) {
|
||||
// 发生连接不上Redis
|
||||
if (connector == null || !connector.isAlive()) {
|
||||
synchronized (lock) {
|
||||
if (!connector.isAlive()) {
|
||||
// 启动巡检线程
|
||||
connector.start();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static class RedisInspector extends Thread {
|
||||
@Override
|
||||
public void run() {
|
||||
logger.debug("Connecting to redis....");
|
||||
Jedis jedis;
|
||||
while (true) {
|
||||
try {
|
||||
jedisPool = new JedisPool(buildGenericObjectPoolConfig(),
|
||||
config[0], Integer.valueOf(config[1]));
|
||||
jedis = jedisPool.getResource();
|
||||
jedis.get("ok");
|
||||
break;
|
||||
} catch (Exception e) {
|
||||
if (e instanceof JedisConnectionException) {
|
||||
try {
|
||||
Thread.sleep(5000L);
|
||||
} catch (InterruptedException e1) {
|
||||
logger.error("Sleep failed", e);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
logger.debug("Connected to redis success. Open alarm function.");
|
||||
Config.Alarm.ALARM_OFF_FLAG = false;
|
||||
// 清理当前线程
|
||||
connector = null;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,96 @@
|
|||
package com.ai.cloud.skywalking.alarm.zk;
|
||||
|
||||
import com.ai.cloud.skywalking.alarm.AlarmServerRegisterWatcher;
|
||||
import com.ai.cloud.skywalking.alarm.conf.Config;
|
||||
import org.apache.curator.RetryPolicy;
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.curator.framework.CuratorFrameworkFactory;
|
||||
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
|
||||
import org.apache.curator.retry.ExponentialBackoffRetry;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class ZKUtil {
|
||||
|
||||
private static Logger logger = LogManager.getLogger(ZKUtil.class);
|
||||
|
||||
private static CuratorFramework client;
|
||||
|
||||
static {
|
||||
try {
|
||||
RetryPolicy retryPolicy = new ExponentialBackoffRetry(Config.ZKPath.RETRY_TIMEOUT,
|
||||
Config.ZKPath.RETRY_TIMES);
|
||||
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().
|
||||
connectString(Config.ZKPath.CONNECT_STR)
|
||||
.connectionTimeoutMs(Config.ZKPath.CONNECT_TIMEOUT).retryPolicy(retryPolicy);
|
||||
client = builder.build();
|
||||
client.start();
|
||||
|
||||
createIfNotExists(getRegisterLockPath());
|
||||
createIfNotExists(getUserLockPathPrefix());
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to connect zookeeper.", e);
|
||||
System.exit(-1);
|
||||
}
|
||||
}
|
||||
|
||||
public static InterProcessMutex getRegisterLock() {
|
||||
return new InterProcessMutex(client, getRegisterLockPath());
|
||||
}
|
||||
|
||||
public static InterProcessMutex getProcessUserLock(String uid) {
|
||||
return new InterProcessMutex(client, getUserLockPath(uid));
|
||||
}
|
||||
|
||||
public static void register(String serverId) {
|
||||
try {
|
||||
client.create().creatingParentsIfNeeded().forPath(getRegisterServerPathPrefix() + "/" + serverId);
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to register server", e);
|
||||
}
|
||||
}
|
||||
|
||||
public static List<String> selectAllServerIds() {
|
||||
try {
|
||||
return client.getChildren().forPath(getRegisterServerPathPrefix());
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to get children of Path[" + getRegisterServerPathPrefix() + "].", e);
|
||||
}
|
||||
return new ArrayList<String>();
|
||||
}
|
||||
|
||||
public static void watch(AlarmServerRegisterWatcher instance) {
|
||||
try {
|
||||
client.getChildren().usingWatcher(instance).forPath(getRegisterServerPathPrefix());
|
||||
} catch (Exception e) {
|
||||
logger.error("Failed to get children for path[" + getRegisterServerPathPrefix() + "].", e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static String getRegisterLockPath() {
|
||||
return Config.ZKPath.NODE_PREFIX + Config.ZKPath.SERVER_REGISTER_LOCK_PATH;
|
||||
}
|
||||
|
||||
private static String getUserLockPath(String uid) {
|
||||
return getUserLockPathPrefix() + "/" + uid;
|
||||
}
|
||||
|
||||
private static String getUserLockPathPrefix() {
|
||||
return Config.ZKPath.NODE_PREFIX + Config.ZKPath.USER_REGISTER_LOCK_PATH;
|
||||
}
|
||||
|
||||
|
||||
private static void createIfNotExists(String path) throws Exception {
|
||||
if (client.checkExists().forPath(path) == null) {
|
||||
client.create().creatingParentsIfNeeded().forPath(path);
|
||||
}
|
||||
}
|
||||
|
||||
private static String getRegisterServerPathPrefix() {
|
||||
return Config.ZKPath.NODE_PREFIX + Config.ZKPath.REGISTER_SERVER_PATH;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
<?xml version="1.0" encoding="UTF-8" ?>
|
||||
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
|
||||
|
||||
<log4j:configuration>
|
||||
|
||||
<appender name="CONSOLE" class="org.apache.log4j.ConsoleAppender">
|
||||
<layout class="org.apache.log4j.PatternLayout">
|
||||
<param name="ConversionPattern"
|
||||
value="%d - %c -%-4r [%t] %-5p %x - %m%n" />
|
||||
</layout>
|
||||
</appender>
|
||||
|
||||
<root>
|
||||
<priority value="debug" />
|
||||
<appender-ref ref="CONSOLE" />
|
||||
</root>
|
||||
|
||||
|
||||
</log4j:configuration>
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Configuration status="debug">
|
||||
<Appenders>
|
||||
<Console name="Console" target="SYSTEM_OUT">
|
||||
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
|
||||
</Console>
|
||||
</Appenders>
|
||||
<Loggers>
|
||||
<Root level="debug">
|
||||
<AppenderRef ref="Console"/>
|
||||
</Root>
|
||||
</Loggers>
|
||||
</Configuration>
|
||||
Loading…
Reference in New Issue