1.Remove all server side module. Prepare to use `AKKA` as new cluster engine.

2.Release the prototype of ‘TraceSegment’.
This commit is contained in:
wusheng 2017-02-17 10:31:01 +08:00
parent 4cce700631
commit 1326f5452b
1371 changed files with 177 additions and 159004 deletions

View File

@ -6,7 +6,7 @@
<groupId>com.a.eye</groupId>
<artifactId>skywalking</artifactId>
<version>2.2-2017</version>
<version>3.0-2017</version>
<licenses>
<license>
@ -29,10 +29,7 @@
<modules>
<module>skywalking-commons</module>
<module>skywalking-alarm</module>
<module>skywalking-webui</module>
<module>skywalking-sniffer</module>
<module>skywalking-storage-center</module>
<module>skywalking-application-toolkit</module>
</modules>
<packaging>pom</packaging>

View File

@ -1,57 +0,0 @@
#!/bin/sh
# OS specific support. $var _must_ be set to either true or false.
cygwin=false
os400=false
darwin=false
case "`uname`" in
CYGWIN*) cygwin=true;;
OS400*) os400=true;;
Darwin*) darwin=true;;
esac
# resolve links - $0 may be a softlink
SW_ALARM_ALARM_SERVER_BIN="$0"
while [ -h "$SW_ALARM_ALARM_SERVER_BIN" ]; do
ls=`ls -ld "$SW_ALARM_ALARM_SERVER_BIN"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
SW_ALARM_ALARM_SERVER_BIN="$link"
else
SW_ALARM_ALARM_SERVER_BIN=`dirname "$SW_ALARM_ALARM_SERVER_BIN"`/"$link"
fi
done
# Get standard environment variables
SW_ALARM_ALARM_SERVER_BIN_DIR=`dirname "$SW_ALARM_ALARM_SERVER_BIN"`
SW_ALARM_ALARM_PREFIX="${SW_ALARM_ALARM_SERVER_BIN_DIR}/.."
SW_ALARM_ALARM_LOG_DIR="${SW_ALARM_ALARM_SERVER_BIN_DIR}/../logs"
SW_ALARM_ALARM_CFG_DIR="${SW_ALARM_ALARM_SERVER_BIN_DIR}/../config"
if [ ! -d "${SW_ALARM_ALARM_LOG_DIR}" ]; then
echo "create directory $SW_ALARM_ALARM_LOG_DIR"
mkdir -p ${SW_ALARM_ALARM_LOG_DIR}
fi
#echo $SW_ALARM_ALARM_SERVER_BIN_DIR
#set java home
if [ "$JAVA_HOME" != "" ]; then
JAVA="$JAVA_HOME/bin/java"
else
JAVA=java
fi
CLASSPATH="$SW_ALARM_ALARM_CFG_DIR:$CLASSPATH"
for i in "${SW_ALARM_ALARM_SERVER_BIN_DIR}"/../libs/*.jar
do
CLASSPATH="$i:$CLASSPATH"
done
echo "CLASSPATH=$CLASSPATH"
JAVA_OPTS="$JAVA_OPTS -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false"
$JAVA ${JAVA_OPTS} -classpath $CLASSPATH com.a.eye.skywalking.alarm.AlarmProcessServer >> ${SW_ALARM_ALARM_LOG_DIR}/sw-alarm-server.log & 2>&1&

View File

@ -1,25 +0,0 @@
FROM openjdk:8u111-jdk
ENV ALARM_COORDINATE_ZK_ADDRESS=localhost:2181 \
MYSQL_SERVER=localhost:3306 \
MYSQL_USER=root \
MYSQL_PASSWORD=root \
REDIS_SERVER=localhost:6379 \
ALARM_MAIL_HOST=smtp.mail.com \
MAIL_SSL_ENABLE=true \
MAIL_USER_NAME=skywalking \
MAIL_PASSWORD=skywalking \
MAIL_SENDER_MAIL=skywalking@mail.com \
WEBUI_DEPLOY_ADDRESS=localhost:8080 \
WEBUI_CONTEXT_NAME=skywalking
ADD skywalking-alarm.tar /usr/local/
COPY config.properties /usr/local/skywalking-alarm/config
COPY sw-alarm-server.sh /usr/local/skywalking-alarm/bin
COPY docker-entrypoint.sh /
RUN chmod +x /docker-entrypoint.sh && chmod +x /usr/local/skywalking-alarm/bin/sw-alarm-server.sh
ENTRYPOINT ["/docker-entrypoint.sh"]
CMD ["/usr/local/skywalking-alarm/bin/sw-alarm-server.sh"]

View File

@ -1,81 +0,0 @@
#告警处理线程数
server.process_thread_size=2
#守护线程等待周期
server.daemon_thread_wait_interval=50000
#告警规则生效周期
server.alarm_rule_activity_interval=3600000
#处理线程等待周期
processthread.thread_wait_interval=5000
#zookeeper连接地址
zkpath.connect_str={ALARM_COORDINATE_ZK_ADDRESS}
#zookeeper连接超时时间(单位:毫秒)
zkpath.connect_timeout=1000
#zookeeper重试次数
zkpath.retry_times=3
#zookeeper重试超时时间(单位:毫秒)
zkpath.retry_timeout=1000
#告警规则的path前缀
zkpath.node_prefix=/skywalking
#告警服务的处理线程注册地址
zkpath.register_server_path=/skywalking/alarm-server/register-servers
#告警服务的协调器的竞争锁地址
zkpath.coordinator_path=/skywalking/alarm-server/coordinator/lock
#尝试成为协调器的竞争锁周期(单位:秒)
coordinator.retry_get_coordinator_lock_interval=5
#尝试成为协调器的等待时间(单位:毫秒)
coordinator.retry_become_coordinator_wait_time=10000
#检查是否需要重新分配的周期(单位:毫秒)
coordinator.check_redistribute_interval=5000
#检查所有处理线程的周期(单位:毫秒)
coordinator.check_all_process_thread_interval=500
#数据库驱动类
db.driver_class=com.mysql.jdbc.Driver
#数据库连接地址
db.url=jdbc:mysql://{MYSQL_SERVER}/test
#数据库用户名
db.user_name={MYSQL_USER}
#数据库密码
db.password={MYSQL_PASSWORD}
#数据库连接最大空闲数
db.max_idle=1
#数据库最大连接数
db.max_pool_size=20
#数据连接超时时间
db.connect_timeout=10000
#告警信息存在的redis服务器地址
alarm.redis_server={REDIS_SERVER}
#redis的最大空闲连接数
alarm.redis_max_idle=20
#redis的最小空闲连接数
alarm.redis_min_idle=1
#redis最大连接数
alarm.redis_max_total=50
#是否关闭告警发送
alarm.alarm_off_flag=false
alarm.checker.turn_on_exception_checker=true
#告警检查器:执行时间超时告警检查
alarm.checker.turn_on_execute_time_checker=true
#邮件发送配置id
mailsenderinfo.mail_host={ALARM_MAIL_HOST}
mailsenderinfo.transport_protocol=smtp
mailsenderinfo.smtp_auth=true
mailsenderinfo.smtp_start_ssl_enable={MAIL_SSL_ENABLE}
mailsenderinfo.username={MAIL_USER_NAME}
mailsenderinfo.password={MAIL_PASSWORD}
mailsenderinfo.sender={MAIL_SENDER_MAIL}
mailsenderinfo.ssl_enable={MAIL_SSL_ENABLE}
#邮件模板配置id
templateinfo.portal_url=http://{WEBUI_DEPLOY_ADDRESS}/{WEBUI_APPLICATION_CONTEXT}
#尝试获取轮询锁的周期(单位:毫秒)
inspectthread.retry_get_inspect_lock_interval=10000
#尝试检查用户列表的周期(单位:毫秒)
inspectthread.check_user_list_interval=300000

View File

@ -1,42 +0,0 @@
#!/bin/bash
echo "replace ALARM_COORDINATE_ZK_ADDRESS with $ALARM_COORDINATE_ZK_ADDRESS"
eval sed -i -e 's/\{ALARM_COORDINATE_ZK_ADDRESS\}/$ALARM_COORDINATE_ZK_ADDRESS/' /usr/local/skywalking-alarm/config/config.properties
echo "replace MYSQL_SERVER with $MYSQL_SERVER"
eval sed -i -e 's/\{MYSQL_SERVER\}/$MYSQL_SERVER/' /usr/local/skywalking-alarm/config/config.properties
echo "replace MYSQL_USER with $MYSQL_USER"
eval sed -i -e 's/\{MYSQL_USER\}/$MYSQL_USER/' /usr/local/skywalking-alarm/config/config.properties
echo "replace MYSQL_PASSWORD with $MYSQL_PASSWORD"
eval sed -i -e 's/\{MYSQL_PASSWORD\}/$MYSQL_PASSWORD/' /usr/local/skywalking-alarm/config/config.properties
echo "replace REDIS_SERVER with $REDIS_SERVER"
eval sed -i -e 's/\{REDIS_SERVER\}/$REDIS_SERVER/' /usr/local/skywalking-alarm/config/config.properties
echo "replcae ALARM_MAIL_HOST with $ALARM_MAIL_HOST"
eval sed -i -e 's/\{ALARM_MAIL_HOST\}/$ALARM_MAIL_HOST/' /usr/local/skywalking-alarm/config/config.properties
echo "replace MAIL_SSL_ENABLE with $MAIL_SSL_ENABLE"
eval sed -i -e 's/\{MAIL_SSL_ENABLE\}/$MAIL_SSL_ENABLE/' /usr/local/skywalking-alarm/config/config.properties
echo "replace MAIL_USER_NAME with $MAIL_USER_NAME"
eval sed -i -e 's/\{MAIL_USER_NAME\}/$MAIL_USER_NAME/' /usr/local/skywalking-alarm/config/config.properties
echo "replace MAIL_PASSWORD with $MAIL_PASSWORD"
eval sed -i -e 's/\{MAIL_PASSWORD\}/$MAIL_PASSWORD/' /usr/local/skywalking-alarm/config/config.properties
echo "replace MAIL_SENDER_MAIL with $MAIL_SENDER_MAIL"
eval sed -i -e 's/\{MAIL_SENDER_MAIL\}/$MAIL_SENDER_MAIL/' /usr/local/skywalking-alarm/config/config.properties
echo "replace WEBUI_DEPLOY_ADDRESS with $WEBUI_DEPLOY_ADDRESS"
eval sed -i -e 's/\{WEBUI_DEPLOY_ADDRESS\}/$WEBUI_DEPLOY_ADDRESS/' /usr/local/skywalking-alarm/config/config.properties
echo "replace WEBUI_CONTEXT_NAME with $WEBUI_APPLICATION_CONTEXT"
eval sed -i -e 's/\{WEBUI_APPLICATION_CONTEXT\}/$WEBUI_APPLICATION_CONTEXT/' /usr/local/skywalking-alarm/config/config.properties
mkdir -p /usr/local/skywalking-alarm/logs
exec "$@"

View File

@ -1,57 +0,0 @@
#!/bin/sh
# OS specific support. $var _must_ be set to either true or false.
cygwin=false
os400=false
darwin=false
case "`uname`" in
CYGWIN*) cygwin=true;;
OS400*) os400=true;;
Darwin*) darwin=true;;
esac
# resolve links - $0 may be a softlink
SW_ALARM_ALARM_SERVER_BIN="$0"
while [ -h "$SW_ALARM_ALARM_SERVER_BIN" ]; do
ls=`ls -ld "$SW_ALARM_ALARM_SERVER_BIN"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
SW_ALARM_ALARM_SERVER_BIN="$link"
else
SW_ALARM_ALARM_SERVER_BIN=`dirname "$SW_ALARM_ALARM_SERVER_BIN"`/"$link"
fi
done
# Get standard environment variables
SW_ALARM_ALARM_SERVER_BIN_DIR=`dirname "$SW_ALARM_ALARM_SERVER_BIN"`
SW_ALARM_ALARM_PREFIX="${SW_ALARM_ALARM_SERVER_BIN_DIR}/.."
SW_ALARM_ALARM_LOG_DIR="${SW_ALARM_ALARM_SERVER_BIN_DIR}/../logs"
SW_ALARM_ALARM_CFG_DIR="${SW_ALARM_ALARM_SERVER_BIN_DIR}/../config"
if [ ! -d "${SW_ALARM_ALARM_LOG_DIR}" ]; then
echo "create directory $SW_ALARM_ALARM_LOG_DIR"
mkdir -p ${SW_ALARM_ALARM_LOG_DIR}
fi
#echo $SW_ALARM_ALARM_SERVER_BIN_DIR
#set java home
if [ "$JAVA_HOME" != "" ]; then
JAVA="$JAVA_HOME/bin/java"
else
JAVA=java
fi
CLASSPATH="$SW_ALARM_ALARM_CFG_DIR:$CLASSPATH"
for i in "${SW_ALARM_ALARM_SERVER_BIN_DIR}"/../libs/*.jar
do
CLASSPATH="$i:$CLASSPATH"
done
echo "CLASSPATH=$CLASSPATH"
JAVA_OPTS="$JAVA_OPTS -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false"
$JAVA ${JAVA_OPTS} -classpath $CLASSPATH com.a.eye.skywalking.alarm.AlarmProcessServer

View File

@ -1,131 +0,0 @@
<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>
<parent>
<artifactId>skywalking</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
</parent>
<artifactId>skywalking-alarm</artifactId>
<packaging>jar</packaging>
<name>skywalking-alarm</name>
<url>http://maven.apache.org</url>
<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>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>2.4.3</version>
</dependency>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-util</artifactId>
<version>2.2-2017</version>
</dependency>
</dependencies>
<build>
<finalName>skywalking-alarm</finalName>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<executions>
<execution>
<id>make-zip</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
<configuration>
<descriptors>
<descriptor>src/main/assembly/assembly.xml</descriptor>
</descriptors>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<excludes>
<exclude>*.properties</exclude>
<exclude>*.xml</exclude>
</excludes>
</configuration>
</plugin>
<plugin>
<groupId>com.spotify</groupId>
<artifactId>docker-maven-plugin</artifactId>
<version>${docker.plugin.version}</version>
<configuration>
<skipDocker>false</skipDocker>
<dockerDirectory>${project.basedir}/docker</dockerDirectory>
<imageName>skywalking/skywalking-alarm</imageName>
<imageTags>
<imageTag>2.2-2017</imageTag>
</imageTags>
<resources>
<resource>
<targetPath>/</targetPath>
<directory>${project.build.directory}</directory>
<include>${build.finalName}.tar</include>
</resource>
</resources>
</configuration>
</plugin>
</plugins>
</build>
</project>

View File

@ -1,36 +0,0 @@
<assembly
xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd">
<id></id>
<formats>
<format>tar</format>
</formats>
<dependencySets>
<dependencySet>
<outputDirectory>/libs</outputDirectory>
<scope>runtime</scope>
</dependencySet>
</dependencySets>
<fileSets>
<fileSet>
<directory>${project.basedir}/bin</directory>
<outputDirectory>/bin</outputDirectory>
<includes>
<include>sw-alarm-server.sh</include>
</includes>
<fileMode>0755</fileMode>
</fileSet>
<fileSet>
<directory>src/main/resources</directory>
<outputDirectory>/config</outputDirectory>
</fileSet>
<fileSet>
<directory>${project.build.directory}</directory>
<outputDirectory>/libs</outputDirectory>
<includes>
<include>${build.finalName}.jar</include>
</includes>
</fileSet>
</fileSets>
</assembly>

View File

@ -1,166 +0,0 @@
package com.a.eye.skywalking.alarm;
import com.a.eye.skywalking.alarm.conf.Config;
import com.a.eye.skywalking.alarm.dao.AlarmMessageDao;
import com.a.eye.skywalking.alarm.model.AlarmRule;
import com.a.eye.skywalking.alarm.model.ProcessThreadStatus;
import com.a.eye.skywalking.alarm.model.ProcessThreadValue;
import com.a.eye.skywalking.alarm.procesor.AlarmMessageProcessor;
import com.a.eye.skywalking.alarm.util.ZKUtil;
import com.a.eye.skywalking.alarm.model.UserInfo;
import com.a.eye.skywalking.alarm.util.ProcessUtil;
import com.google.gson.Gson;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
public class AlarmMessageProcessThread extends Thread {
private Logger logger = LogManager.getLogger(AlarmMessageProcessThread.class);
private String threadId;
private ProcessThreadStatus status;
private List<String> processUserIds;
private CoordinatorStatusWatcher watcher = new CoordinatorStatusWatcher();
private Map<UserInfo, List<AlarmRule>> cacheRules = new HashMap<UserInfo, List<AlarmRule>>();
private static AlarmMessageProcessor processor = new AlarmMessageProcessor();
public AlarmMessageProcessThread() {
// 初始化生成ThreadId
threadId = UUID.randomUUID().toString();
this.setDaemon(true);
}
@Override
public void run() {
long interval = 0L;
//注册服务(默认为空闲状态)
registerProcessThread(threadId, ProcessThreadStatus.FREE);
while (true) {
try {
//检查是否为忙碌状态
if (status == ProcessThreadStatus.BUSY) {
if (interval > Config.Server.ALARM_RULE_ACTIVITY_INTERVAL) {
//重新加载用户配置
cacheProcessUser(processUserIds);
interval = 0;
}
//处理告警信息
for (Map.Entry<UserInfo, List<AlarmRule>> entry : cacheRules.entrySet()) {
for (AlarmRule rule : entry.getValue()) {
processor.process(entry.getKey(), rule);
}
}
}
//检查是否分配线程的状态(重新分配状态)
if (status == ProcessThreadStatus.REDISTRIBUTING) {
logger.info("The current thread[{}] state will change for the idle state", currentThread().getName());
// 修改自身状态(空闲状态)
status = ProcessThreadStatus.FREE;
ProcessUtil.changeProcessThreadStatus(threadId, ProcessThreadStatus.FREE);
//清空缓存数据
clearCacheProcessUser(cacheRules);
}
//检查分配线程的状态(分配完成状态)
if (status == ProcessThreadStatus.REDISTRIBUTE_SUCCESS) {
logger.info("The current thread[{}] state will change for the busy", currentThread().getName());
// 获取待处理的用户
processUserIds = acquireProcessedUsers();
// 缓存数据
cacheProcessUser(processUserIds);
interval = 0;
// 修改自身状态 (忙碌状态)
status = ProcessThreadStatus.BUSY;
ProcessUtil.changeProcessThreadStatus(threadId, ProcessThreadStatus.BUSY);
}
try {
Thread.sleep(Config.ProcessThread.THREAD_WAIT_INTERVAL);
interval += Config.ProcessThread.THREAD_WAIT_INTERVAL;
} catch (InterruptedException e) {
logger.error("Sleep failed.", e);
}
} catch (Exception e) {
logger.error("Failed to process data.", e);
}
}
}
private void clearCacheProcessUser(Map<UserInfo, List<AlarmRule>> cacheRules) {
cacheRules.clear();
}
private void cacheProcessUser(List<String> processUserIds) throws SQLException {
UserInfo tmpUserInfo;
List<AlarmRule> alarmRules;
// 清理原有的CacheRule
cacheRules.clear();
for (String userId : processUserIds) {
tmpUserInfo = AlarmMessageDao.selectUser(userId);
if (tmpUserInfo == null) {
continue;
}
alarmRules = AlarmMessageDao.selectAlarmRulesByUserId(userId);
cacheRules.put(tmpUserInfo, alarmRules);
}
}
private List<String> acquireProcessedUsers() throws Exception {
String path = Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId;
String value = ZKUtil.getPathData(path);
ProcessThreadValue processThreadValue = new Gson().fromJson(value, ProcessThreadValue.class);
return processThreadValue.getDealUserIds();
}
private void registerProcessThread(String threadId, ProcessThreadStatus status) {
try {
String registerPath = Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId;
ProcessThreadValue initValue = new ProcessThreadValue();
initValue.setStatus(status.getValue());
ZKUtil.getZkClient().create().creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL).forPath
(registerPath, new Gson().toJson(initValue).getBytes());
this.status = status;
ZKUtil.getPathDataWithWatch(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId, watcher);
} catch (Exception e) {
logger.error("Failed to register process thread.", e);
}
}
private class CoordinatorStatusWatcher implements CuratorWatcher {
public void process(WatchedEvent watchedEvent) throws Exception {
if (watchedEvent.getType() == Watcher.Event.EventType.NodeDataChanged) {
String value = ZKUtil.getPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId);
ProcessThreadValue processThreadValue = new Gson().fromJson(value, ProcessThreadValue.class);
status = ProcessThreadStatus.convert(processThreadValue.getStatus());
}
try {
ZKUtil.getPathDataWithWatch(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId, watcher);
} catch (Exception e) {
logger.error("Failed to Watcher path [{}]", e);
}
}
}
}

View File

@ -1,66 +0,0 @@
package com.a.eye.skywalking.alarm;
import com.a.eye.skywalking.alarm.conf.Config;
import com.a.eye.skywalking.alarm.util.ZKUtil;
import com.a.eye.skywalking.util.ConfigInitializer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
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) throws IOException, IllegalAccessException {
logger.info("Begin to start alarm process server....");
logger.info("Begin to initialize configuration");
initializeParam();
logger.info("Finished to initialize configuration");
if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH)) {
ZKUtil.createPath(Config.ZKPath.REGISTER_SERVER_PATH);
}
new UserInfoCoordinator().start();
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);
new UsersChangedDetectionThread().start();
logger.info("Successfully launched the thread that inspect the number of user");
logger.info("Alarm process server successfully started.");
while (true) {
try {
Thread.sleep(Config.Server.DAEMON_THREAD_WAIT_INTERVAL);
} catch (InterruptedException e) {
logger.error("Sleep failed", e);
}
}
}
private static void initializeParam() throws IllegalAccessException, IOException {
Properties properties = new Properties();
try {
properties.load(AlarmProcessServer.class.getResourceAsStream("/config.properties"));
ConfigInitializer.initialize(properties, Config.class);
} catch (IllegalAccessException e) {
logger.error("Initialize the collect server configuration failed", e);
throw e;
} catch (IOException e) {
logger.error("Initialize the collect server configuration failed", e);
throw e;
}
}
}

View File

@ -1,247 +0,0 @@
package com.a.eye.skywalking.alarm;
import com.a.eye.skywalking.alarm.dao.AlarmMessageDao;
import com.a.eye.skywalking.alarm.conf.Config;
import com.a.eye.skywalking.alarm.model.ProcessThreadStatus;
import com.a.eye.skywalking.alarm.model.ProcessThreadValue;
import com.a.eye.skywalking.alarm.util.ProcessUtil;
import com.a.eye.skywalking.alarm.util.ZKUtil;
import com.google.gson.Gson;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
public class UserInfoCoordinator extends Thread {
private Logger logger = LogManager.getLogger(UserInfoCoordinator.class);
private static boolean redistributing;
private RegisterServerWatcher watcher = new RegisterServerWatcher();
private InterProcessMutex lock = new InterProcessMutex(
ZKUtil.getZkClient(), Config.ZKPath.COORDINATOR_PATH);
private boolean isCoordinator = false;
public UserInfoCoordinator() {
this.setDaemon(true);
}
@Override
public void run() {
while (true) {
try {
if (!isCoordinator) {
while (!retryBecomeCoordinator()) {
try {
Thread.sleep(Config.Coordinator.RETRY_BECOME_COORDINATOR_WAIT_TIME);
} catch (Exception e) {
logger.error("Sleep Failed.", e);
}
}
isCoordinator = true;
watcherRegisterServerPath();
redistributing = true;
}
// 检查是否有新服务注册或者在重分配过程做有新处理线程启动了
if (!redistributing) {
try {
Thread.sleep(Config.Coordinator.CHECK_REDISTRIBUTE_INTERVAL);
} catch (InterruptedException e) {
logger.error("Sleep error", e);
}
continue;
}
redistributing = false;
// 获取当前所有的注册的处理线程
List<String> registeredThreads = acquireAllRegisteredThread();
// 修改状态 (开始重新分配状态
changeStatus(registeredThreads,
ProcessThreadStatus.REDISTRIBUTING);
// 检查所有的服务是否都处于空闲状态
int retryTimes = 0;
while (!checkAllProcessStatus(registeredThreads,
ProcessThreadStatus.FREE)) {
try {
Thread.sleep(Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL);
retryTimes++;
} catch (InterruptedException e) {
logger.error("Sleep failed", e);
}
if (retryTimes > 1000) {
logger.warn("checking all processors are free, waiting {}ms", Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL * retryTimes);
retryTimes = 0;
}
}
// 查询当前有多少用户
List<String> users = AlarmMessageDao.selectAllUserIds();
logger.info("Query to the {} pending user from the database", users.size());
// 将用户重新分配给服务
List<String> realRedistributeThread = allocationUser(
registeredThreads, users);
logger.info("All users are assigned to {} processing threads", realRedistributeThread.size());
// 修改状态(分配完成)
changeStatus(realRedistributeThread,
ProcessThreadStatus.REDISTRIBUTE_SUCCESS);
logger.info("Change state of {} processing threads to idle state", realRedistributeThread.size());
// 检查所有的服务是否都处于忙碌状态
while (!checkAllProcessStatus(realRedistributeThread,
ProcessThreadStatus.BUSY)) {
try {
Thread.sleep(Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL);
} catch (InterruptedException e) {
logger.error("Sleep failed", e);
}
if (retryTimes > 1000) {
logger.warn("checking all processors are busy, waiting {}ms", Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL * retryTimes);
retryTimes = 0;
}
}
} catch (Exception e) {
logger.error("Failed to coordinate, retry. ", e);
releaseCoordinator();
isCoordinator = false;
}
}
}
private boolean retryBecomeCoordinator() {
try {
return lock.acquire(
Config.Coordinator.RETRY_GET_COORDINATOR_LOCK_INTERVAL,
TimeUnit.SECONDS);
} catch (Exception e) {
logger.error("Failed to acquire lock .", e);
return false;
}
}
private void releaseCoordinator() {
if (lock != null && lock.isAcquiredInThisProcess()) {
try {
lock.release();
} catch (Exception e1) {
logger.error("Failed to release lock.", e1);
}
}
}
private List<String> allocationUser(List<String> registeredThreads,
List<String> userIds) throws Exception {
List<String> realRedistributeThread = new ArrayList<String>();
Set<String> sortThreadIds = new HashSet<String>(registeredThreads);
int step = (int) Math.ceil(userIds.size() * 1.0 / sortThreadIds.size());
int start = 0;
int end = step;
if (end > userIds.size()) {
end = userIds.size();
}
for (String thread : sortThreadIds) {
if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/"
+ thread))
throw new RuntimeException("Process thread[" + thread + "] is not exists");
String value = ZKUtil
.getPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/"
+ thread);
ProcessThreadValue value1 = new Gson().fromJson(value,
ProcessThreadValue.class);
value1.setDealUserIds(userIds.subList(start, end));
ZKUtil.setPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/"
+ thread, new Gson().toJson(value1));
// 实际重新分配的线程Id
realRedistributeThread.add(thread);
start = end;
end += step;
if (start >= userIds.size()) {
break;
}
if (end > userIds.size()) {
end = userIds.size();
}
}
return realRedistributeThread;
}
private List<String> acquireAllRegisteredThread() throws Exception {
return ZKUtil.getChildren(Config.ZKPath.REGISTER_SERVER_PATH);
}
private boolean checkAllProcessStatus(List<String> registeredThreadIds,
ProcessThreadStatus status) throws Exception {
String registerPathPrefix = Config.ZKPath.REGISTER_SERVER_PATH + "/";
for (String threadId : registeredThreadIds) {
if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/"
+ threadId))
throw new RuntimeException("Process thread[" + threadId + "] is not exists");
if (getProcessThreadStatus(registerPathPrefix, threadId) != status) {
return false;
}
}
return true;
}
private ProcessThreadStatus getProcessThreadStatus(
String registerPathPrefix, String threadId) throws Exception {
if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId))
return ProcessThreadStatus.FREE;
String value = ZKUtil.getPathData(registerPathPrefix + threadId);
if (value == null || value.length() == 0)
return ProcessThreadStatus.FREE;
ProcessThreadValue value1 = new Gson().fromJson(value,
ProcessThreadValue.class);
return ProcessThreadStatus.convert(value1.getStatus());
}
private void changeStatus(List<String> registeredThreadIds,
ProcessThreadStatus status) throws Exception {
for (String threadId : registeredThreadIds) {
ProcessUtil.changeProcessThreadStatus(threadId, status);
}
}
public class RegisterServerWatcher implements CuratorWatcher {
public void process(WatchedEvent watchedEvent) {
if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
redistributing = true;
}
watcherRegisterServerPath();
}
}
private void watcherRegisterServerPath() {
try {
ZKUtil.getChildrenWithWatcher(Config.ZKPath.REGISTER_SERVER_PATH,
watcher);
} catch (Exception e) {
logger.error("Failed to set watcher for get children", e);
}
}
public static void activateRedistribute() {
redistributing = true;
}
}

View File

@ -1,61 +0,0 @@
package com.a.eye.skywalking.alarm;
import com.a.eye.skywalking.alarm.conf.Config;
import com.a.eye.skywalking.alarm.dao.AlarmMessageDao;
import com.a.eye.skywalking.alarm.util.MD5Encryption;
import com.a.eye.skywalking.alarm.util.ZKUtil;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.sql.SQLException;
import java.util.concurrent.TimeUnit;
public class UsersChangedDetectionThread extends Thread {
private boolean isInspector = false;
private InterProcessMutex inspectorLock = new InterProcessMutex(ZKUtil.getZkClient(),
Config.ZKPath.INSPECTOR_LOCK_PATH);
private String userIdsEncryptedStr;
private Logger logger = LogManager.getLogger(UsersChangedDetectionThread.class);
public UsersChangedDetectionThread() {
this.setDaemon(true);
}
public void run() {
while (true) {
try {
// 探寻是否成为探测者
if (!isInspector) {
while (!inspectorLock.acquire(5, TimeUnit.SECONDS)) {
Thread.sleep(Config.InspectThread.RETRY_GET_INSPECT_LOCK_INTERVAL);
}
isInspector = true;
userIdsEncryptedStr = MD5Encryption.getEncryption(AlarmMessageDao.selectAllUserIds().toString());
}
// 判断上次用户数量是否发生变化
if (!checkUserNumber()) {
Thread.sleep(Config.InspectThread.CHECK_USER_LIST_INTERVAL);
continue;
}
logger.info("The number of users has changed, activated redistribution task");
// 激活重分配
UserInfoCoordinator.activateRedistribute();
} catch (Exception e) {
logger.error("Failed to inspect number of user.", e);
}
}
}
private boolean checkUserNumber() throws SQLException {
String currentUserIdsEncryptedStr = MD5Encryption.getEncryption(AlarmMessageDao.selectAllUserIds().toString());
if (userIdsEncryptedStr.equals(currentUserIdsEncryptedStr)) {
return false;
}
userIdsEncryptedStr = currentUserIdsEncryptedStr;
return true;
}
}

View File

@ -1,108 +0,0 @@
package com.a.eye.skywalking.alarm.conf;
public class Config {
public static class Server {
public static int PROCESS_THREAD_SIZE = 1;
public static long DAEMON_THREAD_WAIT_INTERVAL = 50000L;
public static long ALARM_RULE_ACTIVITY_INTERVAL = 3600000L;
}
public static class ProcessThread {
public static long THREAD_WAIT_INTERVAL = 60 * 1000L;
}
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 REGISTER_SERVER_PATH = "/skywalking/alarm-server/register-servers";
public static String COORDINATOR_PATH = "/skywalking/alarm-server/coordinator/lock";
public static String INSPECTOR_LOCK_PATH = "/skywalking/alarm-server/user-inspector/lock";
}
public static class Coordinator {
// 单位:()
public static long RETRY_GET_COORDINATOR_LOCK_INTERVAL = 5;
public static long RETRY_BECOME_COORDINATOR_WAIT_TIME = 10 * 1000L;
// 单位:(毫秒)
public static long CHECK_REDISTRIBUTE_INTERVAL = 5 * 1000;
// 单位:(毫秒)
public static long CHECK_ALL_PROCESS_THREAD_INTERVAL = 500L;
}
public static class DB {
public static String PASSWORD = "root";
public static String USER_NAME = "root";
public static String DRIVER_CLASS = "com.mysql.jdbc.Driver";
public static String URL = "jdbc:mysql://127.0.0.1:3307/test";
public static int MAX_IDLE = 1;
public static int MAX_POOL_SIZE = 20;
public static long CONNECT_TIMEOUT = 10000;
}
public static class InspectThread {
public static long RETRY_GET_INSPECT_LOCK_INTERVAL = 10000;
public static long CHECK_USER_LIST_INTERVAL = 5 * 60 * 1000;
}
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 MailSenderInfo {
public static String MAIL_HOST = "mail.qq.com";
public static String TRANSPORT_PROTOCOL = "smtp";
public static boolean SMTP_AUTH = true;
public static boolean SSL_ENABLE = false;
public static String USERNAME = "skywalking@foxmail.com";
public static String PASSWORD = "tcaeuhuslkjlebjg";
public static String SENDER = "skywalking@foxmail.com";
}
public static class TemplateInfo {
public static String PORTAL_URL = "http://127.0.0.1:8080/skywalking";
}
}

View File

@ -1,134 +0,0 @@
package com.a.eye.skywalking.alarm.dao;
import com.a.eye.skywalking.alarm.model.AlarmRule;
import com.a.eye.skywalking.alarm.model.ApplicationInfo;
import com.a.eye.skywalking.alarm.model.UserInfo;
import com.a.eye.skywalking.alarm.util.DBConnectUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class AlarmMessageDao {
private static Logger logger = LogManager.getLogger(AlarmMessageDao.class);
public static List<String> selectAllUserIds() throws SQLException {
List<String> result = new ArrayList<String>();
Connection connection = DBConnectUtil.getConnection();
try {
PreparedStatement ps = connection.prepareStatement("SELECT user_info.uid FROM user_info WHERE sts = ?");
ps.setString(1, "A");
ResultSet rs = ps.executeQuery();
while (rs.next()) {
result.add(rs.getString("uid"));
}
} catch (SQLException e) {
logger.error("Failed to select all user info", e);
throw e;
} finally {
if (connection != null) {
connection.close();
}
}
return result;
}
public static UserInfo selectUser(String userId) throws SQLException {
UserInfo userInfo = null;
Connection connection = DBConnectUtil.getConnection();
try {
PreparedStatement ps = connection.prepareStatement("SELECT user_info.uid,user_info.user_name FROM user_info WHERE sts = ? AND uid = ?");
ps.setString(1, "A");
ps.setString(2, userId);
ResultSet rs = ps.executeQuery();
rs.next();
userInfo = new UserInfo(rs.getString("uid"));
userInfo.setUserName(rs.getString("user_name"));
} catch (SQLException e) {
logger.error("Failed to select all user info", e);
throw e;
} finally {
if (connection != null) {
connection.close();
}
}
return userInfo;
}
public static List<AlarmRule> selectAlarmRulesByUserId(String userId) throws SQLException {
List<AlarmRule> rules = new ArrayList<AlarmRule>();
Connection connection = DBConnectUtil.getConnection();
try {
PreparedStatement ps = connection.
prepareStatement("SELECT alarm_rule.app_id,alarm_rule.rule_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();
AlarmRule tmpAlarmRule = null;
ApplicationInfo tmpApplication;
Map<String, AlarmRule> rulesMap = new HashMap<String, AlarmRule>();
while (rs.next()) {
tmpAlarmRule = new AlarmRule(rs.getString("uid"), rs.getString("rule_id"));
tmpAlarmRule.setConfigArgs(rs.getString("config_args"));
tmpAlarmRule.setTodoType(rs.getString("todo_type"));
if ("1".equals(rs.getString("is_global"))) {
rulesMap.put("*", tmpAlarmRule);
continue;
}
tmpApplication = new ApplicationInfo();
tmpApplication.setAppId(rs.getString("app_id"));
tmpApplication.setUId(rs.getString("uid"));
tmpAlarmRule.getApplicationInfos().add(tmpApplication);
rulesMap.put(rs.getString("app_id"), tmpAlarmRule);
}
List<ApplicationInfo> allApplication = new ArrayList<ApplicationInfo>();
ps = connection
.prepareStatement("SELECT application_info.app_id, application_info.uid, app_code 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"));
applicationInfo.setAppCode(rs.getString("app_code"));
allApplication.add(applicationInfo);
}
for (ApplicationInfo app : allApplication) {
tmpAlarmRule = rulesMap.get(app.getAppId());
if (tmpAlarmRule != null) {
tmpAlarmRule.getApplicationInfos().get(0).setAppCode(app.getAppCode());
} else {
rulesMap.get("*").getApplicationInfos().add(app);
}
}
rules.addAll(rulesMap.values());
} catch (SQLException e) {
logger.error("Failed to query applications.", e);
throw e;
} finally {
if (connection != null) {
connection.close();
}
}
return rules;
}
}

View File

@ -1,48 +0,0 @@
package com.a.eye.skywalking.alarm.model;
import java.util.Date;
public class AlarmMessage {
private String traceid;
private String exceptionMsg;
private Date date;
public AlarmMessage(String traceid, String exceptionMsg) {
super();
this.traceid = traceid;
this.exceptionMsg = exceptionMsg;
this.date = this.getDate(traceid);
}
public String getTraceid() {
return traceid;
}
public String getExceptionMsg() {
return exceptionMsg;
}
public Date getDate() {
return date;
}
public void setDate(Date date) {
this.date = date;
}
private Date getDate(String traceid) {
Date date;
String[] traceidPartitions = traceid.split("\\.");
try {
String timeStr = traceidPartitions[2];
Long timeStamp = Long.parseLong(timeStr);
date = new Date(timeStamp);
} catch (Exception e) {
return null;
}
return date;
}
}

View File

@ -1,93 +0,0 @@
package com.a.eye.skywalking.alarm.model;
import com.a.eye.skywalking.alarm.util.RedisUtil;
import com.google.gson.Gson;
import redis.clients.jedis.Jedis;
import java.util.ArrayList;
import java.util.List;
public class AlarmRule {
private ConfigArgsDescriber configArgsDescriber;
private String configArgs;
private String todoType;
private String ruleId;
private boolean global = false;
private long previousFireTimeM;
private String uid;
public AlarmRule(String uid, String ruleId) {
this.ruleId = ruleId;
this.uid = uid;
previousFireTimeM = getPreviousFireTime(uid, ruleId);
}
public void setConfigArgs(String configArgs) {
this.configArgs = configArgs;
configArgsDescriber = new Gson().fromJson(configArgs, ConfigArgsDescriber.class);
}
private List<ApplicationInfo> applicationInfos = new ArrayList<ApplicationInfo>();
public List<ApplicationInfo> getApplicationInfos() {
return applicationInfos;
}
public void setApplicationInfos(List<ApplicationInfo> applicationInfos) {
this.applicationInfos = applicationInfos;
}
public ConfigArgsDescriber getConfigArgsDescriber() {
return configArgsDescriber;
}
public void setConfigArgsDescriber(ConfigArgsDescriber configArgsDescriber) {
this.configArgsDescriber = configArgsDescriber;
}
public void setTodoType(String todoType) {
this.todoType = todoType;
}
public String getTodoType() {
return todoType;
}
public String getRuleId() {
return ruleId;
}
public void setRuleId(String ruleId) {
this.ruleId = ruleId;
}
public void setGlobal(boolean global) {
this.global = global;
}
public boolean isGlobal() {
return global;
}
public long getPreviousFireTimeM() {
return previousFireTimeM;
}
public void setPreviousFireTimeM(long previousFireTimeM) {
this.previousFireTimeM = previousFireTimeM;
}
private static long getPreviousFireTime(final String userId, final String ruleId) {
return RedisUtil.execute(new RedisUtil.Executable<Long>() {
public Long exe(Jedis client) {
String previousTime = client.get(userId + "-" + ruleId);
if (previousTime == null || previousTime.length() <= 0) {
return System.currentTimeMillis() / (10000 * 6);
}
return Long.valueOf(previousTime);
}
});
}
}

View File

@ -1,47 +0,0 @@
package com.a.eye.skywalking.alarm.model;
public class AlarmType {
//告警类型名称
private String type;
//告警标签
private String label;
//告警描述
private String desc;
public AlarmType(String type, String label, String desc) {
this.type = type;
this.label = label;
this.desc = desc;
}
@Override
public String toString() {
return "AlarmType [type=" + type + ", label=" + label + ", desc=" + desc + "]";
}
public String getType() {
return type;
}
public void setType(String type) {
this.type = type;
}
public String getLabel() {
return label;
}
public void setLabel(String label) {
this.label = label;
}
public String getDesc() {
return desc;
}
public void setDesc(String desc) {
this.desc = desc;
}
}

View File

@ -1,56 +0,0 @@
package com.a.eye.skywalking.alarm.model;
public class ApplicationInfo {
private String appId;
private String UId;
private String toDoType;
private String appCode;
public ApplicationInfo() {
}
public String getAppId() {
return appId;
}
public void setAppId(String appId) {
this.appId = appId;
}
public void setUId(String UId) {
this.UId = UId;
}
public String getUId() {
return UId;
}
@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;
}
public void setAppCode(String appCode) {
this.appCode = appCode;
}
public String getAppCode() {
return appCode;
}
}

View File

@ -1,33 +0,0 @@
package com.a.eye.skywalking.alarm.model;
public class ConfigArgsDescriber {
private int period;
private MailInfo mailInfo;
private UrlInfo urlInfo;
public int getPeriod() {
return period;
}
public void setPeriod(int period) {
this.period = period;
}
public MailInfo getMailInfo() {
return mailInfo;
}
public void setMailInfo(MailInfo mailInfo) {
this.mailInfo = mailInfo;
}
public UrlInfo getUrlInfo() {
return urlInfo;
}
public void setUrlInfo(UrlInfo urlInfo) {
this.urlInfo = urlInfo;
}
}

View File

@ -1,24 +0,0 @@
package com.a.eye.skywalking.alarm.model;
public class MailInfo {
private String[] mailTo;
private String[] mailCc;
public String[] getMailTo() {
return mailTo;
}
public void setMailTo(String[] mailTo) {
this.mailTo = mailTo;
}
public String[] getMailCc() {
return mailCc;
}
public void setMailCc(String[] mailCc) {
this.mailCc = mailCc;
}
}

View File

@ -1,37 +0,0 @@
package com.a.eye.skywalking.alarm.model;
public enum ProcessThreadStatus {
REDISTRIBUTING(1), REDISTRIBUTE_SUCCESS(2), FREE(0), BUSY(3);
private int value;
ProcessThreadStatus(int value) {
this.value = value;
}
public int getValue() {
return value;
}
public static ProcessThreadStatus convert(int value) {
ProcessThreadStatus status;
switch (value) {
case 0:
status = FREE;
break;
case 1:
status = REDISTRIBUTING;
break;
case 2:
status = REDISTRIBUTE_SUCCESS;
break;
case 3:
status = BUSY;
break;
default:
throw new IllegalArgumentException("Coordinator status illegal");
}
return status;
}
}

View File

@ -1,24 +0,0 @@
package com.a.eye.skywalking.alarm.model;
import java.util.List;
public class ProcessThreadValue {
private int status;
private List<String> dealUserIds;
public int getStatus() {
return status;
}
public void setStatus(int status) {
this.status = status;
}
public List<String> getDealUserIds() {
return dealUserIds;
}
public void setDealUserIds(List<String> dealUserIds) {
this.dealUserIds = dealUserIds;
}
}

View File

@ -1,25 +0,0 @@
package com.a.eye.skywalking.alarm.model;
public class UrlInfo {
private String urlCall;
private String method;
public String getUrlCall() {
return urlCall;
}
public void setUrlCall(String urlCall) {
this.urlCall = urlCall;
}
public String getMethod() {
return method;
}
public void setMethod(String method) {
this.method = method;
}
}

View File

@ -1,24 +0,0 @@
package com.a.eye.skywalking.alarm.model;
public class UserInfo {
private String userId;
private String userName;
public UserInfo(String userId) {
this.userId = userId;
}
public String getUserId() {
return userId;
}
public String getUserName() {
return userName;
}
public void setUserName(String userName) {
this.userName = userName;
}
}

View File

@ -1,219 +0,0 @@
package com.a.eye.skywalking.alarm.procesor;
import java.io.IOException;
import java.io.StringWriter;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import com.a.eye.skywalking.alarm.model.AlarmRule;
import com.a.eye.skywalking.alarm.model.AlarmType;
import com.a.eye.skywalking.alarm.model.ApplicationInfo;
import com.a.eye.skywalking.alarm.model.MailInfo;
import com.a.eye.skywalking.alarm.util.MailSender;
import com.a.eye.skywalking.alarm.util.RedisUtil;
import com.a.eye.skywalking.alarm.util.TemplateConfigurationUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import redis.clients.jedis.Jedis;
import com.a.eye.skywalking.alarm.model.AlarmMessage;
import com.a.eye.skywalking.alarm.model.UserInfo;
import com.a.eye.skywalking.alarm.util.AlarmTypeUtil;
import freemarker.template.Template;
import freemarker.template.TemplateException;
public class AlarmMessageProcessor {
private static Logger logger = LogManager
.getLogger(AlarmMessageProcessor.class);
private static final String TYPE_OF_EXCEPTION_WARNING = "default";
static List<AlarmType> alarmTypeList;
static Template t;
static {
alarmTypeList = AlarmTypeUtil.getAlarmTypeList();
}
public void process(UserInfo userInfo, AlarmRule rule)
throws TemplateException, IOException, SQLException {
Map<String, List<AlarmMessage>> warningMap = new HashMap<String, List<AlarmMessage>>();
Set<String> warningMessageKeys = new HashSet<String>();
long currentFireMinuteTime = System.currentTimeMillis() / (1000 * 60);
long warningTimeWindowSize = currentFireMinuteTime
- rule.getPreviousFireTimeM();
// 获取待发送数据
if (warningTimeWindowSize >= rule.getConfigArgsDescriber().getPeriod()) {
for (AlarmType alarmType : alarmTypeList) {
String type = alarmType.getType();
List<AlarmMessage> warningObjects = new ArrayList<AlarmMessage>();
for (ApplicationInfo applicationInfo : rule.getApplicationInfos()) {
for (int period = 0; period < warningTimeWindowSize; period++) {
Long currentMinuteTime = currentFireMinuteTime - period
- 1;
String alarmKey = userInfo.getUserName() + "-"
+ applicationInfo.getAppCode() + "-"
+ currentMinuteTime;
if (!TYPE_OF_EXCEPTION_WARNING.equals(type)) {
alarmKey += "-" + type;
}
warningMessageKeys.add(alarmKey);
setAlarmMessages(alarmKey, warningObjects);
}
}
if (warningObjects.size() > 0) {
warningMap.put(type, warningObjects);
}
}
// 发送告警数据
int warningSize = this.getWarningSize(warningMap);
int exceptionAlarmSize = this.getExceptionSize(warningMap);
if (warningSize > 0) {
if ("0".equals(rule.getTodoType())) {
logger.info(
"A total of {} alarm information needs to be sent {}",
warningSize, rule.getConfigArgsDescriber()
.getMailInfo().getMailTo());
// 发送邮件
String subjects = generateSubject(userInfo.getUserName(),
exceptionAlarmSize, warningSize,
rule.getPreviousFireTimeM(), currentFireMinuteTime);
Map<String, Object> parameter = new HashMap<String, Object>();
parameter.put("alarmTypeList", alarmTypeList);
parameter.put("warningMap", warningMap);
parameter.put("name", userInfo.getUserName());
parameter.put("startDate", new SimpleDateFormat(
"yyyy-MM-dd HH:mm:ss").format(new Date(rule
.getPreviousFireTimeM() * 10000 * 6)));
parameter.put("endDate", new SimpleDateFormat(
"yyyy-MM-dd HH:mm:ss").format(new Date(
currentFireMinuteTime * 10000 * 6)));
String mailContext = generateContent(parameter);
if (mailContext.length() > 0) {
MailInfo mailInfo = rule.getConfigArgsDescriber()
.getMailInfo();
MailSender.send(mailInfo.getMailTo(),
mailInfo.getMailCc(), mailContext, subjects);
}
}
}
// 清理数据
for (String toBeRemovedKey : warningMessageKeys) {
expiredAlarmMessage(toBeRemovedKey);
}
// 修改-保存上次处理时间
dealPreviousFireTime(userInfo, rule, currentFireMinuteTime);
}
}
private void dealPreviousFireTime(UserInfo userInfo, AlarmRule rule,
long currentFireMinuteTime) {
rule.setPreviousFireTimeM(currentFireMinuteTime);
savePreviousFireTime(userInfo.getUserId(), rule.getRuleId(),
currentFireMinuteTime);
}
private String generateSubject(String userName, int exceptionAlarmSize,
int count, long startTime, long endTime) {
String title = (exceptionAlarmSize > 0 ? "[Error]" : "[Warning]")
+ " Dear "
+ userName
+ ", Skywalking has collected "
+ (exceptionAlarmSize > 0 ? exceptionAlarmSize
+ " system exception message(s) " : "");
if (count > exceptionAlarmSize) {
title += exceptionAlarmSize > 0 ? "and " : "" + (count - exceptionAlarmSize) + " system warning message(s) ";
}
title += "between "
+ new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(
startTime * 10000 * 6))
+ " and "
+ new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(
endTime * 10000 * 6));
return title;
}
private void expiredAlarmMessage(final String key) {
RedisUtil.execute(new RedisUtil.Executable<Long>() {
public Long exe(Jedis client) {
return client.expire(key, 0);
}
});
}
private void savePreviousFireTime(final String userId, final String ruleId,
final long currentFireMinuteTime) {
RedisUtil.execute(new RedisUtil.Executable<Long>() {
public Long exe(Jedis client) {
return client.hset(userId, ruleId,
String.valueOf(currentFireMinuteTime));
}
});
}
private void setAlarmMessages(final String key,
final Collection<AlarmMessage> warningTracingIds) {
RedisUtil.execute(new RedisUtil.Executable<Object>() {
public Collection<String> exe(Jedis client) {
Map<String, String> result = client.hgetAll(key);
if (result != null) {
for (String traceid : result.keySet()) {
warningTracingIds.add(new AlarmMessage(traceid, result
.get(traceid)));
}
}
return null;
}
});
}
private String generateContent(Map parameter) throws IOException,
TemplateException, SQLException {
if (t == null) {
t = TemplateConfigurationUtil.getConfiguration().getTemplate(
"mail-template.ftl");
}
StringWriter out = new StringWriter();
t.process(parameter, out);
return out.getBuffer().toString();
}
private int getExceptionSize(Map<String, List<AlarmMessage>> warningMap) {
if (warningMap.containsKey(TYPE_OF_EXCEPTION_WARNING)) {
return warningMap.get(TYPE_OF_EXCEPTION_WARNING).size();
}
return 0;
}
private int getWarningSize(Map<String, List<AlarmMessage>> warningMap) {
int result = 0;
for (Entry<String, List<AlarmMessage>> entry : warningMap.entrySet()) {
if (entry.getValue() != null) {
result += entry.getValue().size();
}
}
return result;
}
}

View File

@ -1,22 +0,0 @@
package com.a.eye.skywalking.alarm.util;
import com.a.eye.skywalking.alarm.model.AlarmType;
import java.util.ArrayList;
import java.util.List;
public class AlarmTypeUtil {
private static List<AlarmType> alarmTypeList;
public static List<AlarmType> getAlarmTypeList() {
if (alarmTypeList == null || alarmTypeList.isEmpty()) {
alarmTypeList = new ArrayList<AlarmType>();
alarmTypeList.add(new AlarmType("default", "exception", "System Exception"));
alarmTypeList.add(new AlarmType("ExecuteTime-PossibleError", "remark", "Excution Time > 5s"));
alarmTypeList.add(new AlarmType("ExecuteTime-Warning", "remark", "Excution Time > 500ms"));
}
return alarmTypeList;
}
}

View File

@ -1,41 +0,0 @@
package com.a.eye.skywalking.alarm.util;
import com.a.eye.skywalking.alarm.conf.Config;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.sql.Connection;
import java.sql.SQLException;
public class DBConnectUtil {
private static Logger logger = LogManager.getLogger(DBConnectUtil.class);
private static HikariDataSource hikariDataSource;
public static Connection getConnection() {
if (hikariDataSource == null) {
HikariConfig config = new HikariConfig();
config.setJdbcUrl(Config.DB.URL);
config.setUsername(Config.DB.USER_NAME);
config.setPassword(Config.DB.PASSWORD);
config.setDriverClassName(Config.DB.DRIVER_CLASS);
config.addDataSourceProperty("cachePrepStmts", "true");
config.addDataSourceProperty("prepStmtCacheSize", "250");
config.setMinimumIdle(Config.DB.MAX_IDLE);
config.setMaximumPoolSize(Config.DB.MAX_POOL_SIZE);
config.setConnectionTimeout(Config.DB.CONNECT_TIMEOUT);
config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
hikariDataSource = new HikariDataSource(config);
}
try {
return hikariDataSource.getConnection();
} catch (SQLException e) {
logger.error("Failed to get connection", e);
throw new RuntimeException("Cannot get connection.");
}
}
}

View File

@ -1,35 +0,0 @@
package com.a.eye.skywalking.alarm.util;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
public class MD5Encryption {
private static Logger logger = LogManager.getLogger(DBConnectUtil.class);
private MD5Encryption() {
}
public static String getEncryption(String originString) {
String result = null;
if (originString != null) {
try {
MessageDigest md = MessageDigest.getInstance("MD5");
byte bytes[] = md.digest(originString.getBytes());
for (int i = 0; i < bytes.length; i++) {
String str = Integer.toHexString(bytes[i] & 0xFF);
if (str.length() == 1) {
str += "F";
}
result += str;
}
} catch (NoSuchAlgorithmException e) {
logger.error("No such algorithmException.", e);
return originString;
}
}
return result.toUpperCase();
}
}

View File

@ -1,89 +0,0 @@
package com.a.eye.skywalking.alarm.util;
import com.a.eye.skywalking.alarm.conf.Config;
import com.sun.mail.util.MailSSLSocketFactory;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import javax.mail.*;
import javax.mail.internet.AddressException;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class MailSender {
private static Logger logger = LogManager.getLogger(MailSender.class);
private static MailSender sender = new MailSender();
private String mailSender;
private Properties config;
private MailSender() {
try {
config = new Properties();
config.setProperty("mail.transport.protocol", Config.MailSenderInfo.TRANSPORT_PROTOCOL);
config.setProperty("mail.smtp.auth", String.valueOf(Config.MailSenderInfo.SMTP_AUTH));
if (Config.MailSenderInfo.SSL_ENABLE) {
MailSSLSocketFactory sf = new MailSSLSocketFactory();
sf.setTrustAllHosts(true);
config.put("mail.smtp.ssl.enable", "true");
config.put("mail.smtp.ssl.socketFactory", sf);
}
} catch (Exception e) {
logger.error("Failed to load mail sender info.", e);
System.exit(-1);
}
mailSender = Config.MailSenderInfo.SENDER;
}
public static void send(String[] recipientAccounts, String[] ccList, String content, String title) {
Session session = Session.getInstance(sender.config);
Transport ts = null;
try {
ts = session.getTransport();
ts.connect(Config.MailSenderInfo.MAIL_HOST, Config.MailSenderInfo.USERNAME, Config.MailSenderInfo.PASSWORD);
MimeMessage message = new MimeMessage(session);
message.setFrom(new InternetAddress(sender.mailSender));
InternetAddress[] recipientAccountArray = new InternetAddress[recipientAccounts.length];
for (int i = 0; i < recipientAccounts.length; i++) {
recipientAccountArray[i] = new InternetAddress(recipientAccounts[i]);
}
message.addRecipients(Message.RecipientType.TO, recipientAccountArray);
if (ccList != null && ccList.length > 0) {
List<InternetAddress> ccAccountArray = new ArrayList<InternetAddress>();
for (int i = 0; i < ccList.length; i++) {
if (ccList[i] != null && ccList[i].length() > 0)
ccAccountArray.add(new InternetAddress(ccList[i]));
}
if (ccAccountArray.size() > 0) {
message.addRecipients(Message.RecipientType.CC, ccAccountArray.toArray(new InternetAddress[ccAccountArray.size()]));
}
}
message.setSubject(title);
message.setContent(content, "text/html;charset=UTF-8");
ts.sendMessage(message, message.getAllRecipients());
} catch (AddressException e) {
logger.error("Recipient Account is not correct.", e);
} catch (NoSuchProviderException e) {
logger.error("Failed to send mail.", e);
} catch (MessagingException e) {
logger.error("Failed to send mail.", e);
} finally {
if (ts != null) {
try {
ts.close();
} catch (MessagingException e) {
logger.error("Failed to close transport.", e);
}
}
}
}
}

View File

@ -1,18 +0,0 @@
package com.a.eye.skywalking.alarm.util;
import com.a.eye.skywalking.alarm.conf.Config;
import com.a.eye.skywalking.alarm.model.ProcessThreadStatus;
import com.a.eye.skywalking.alarm.model.ProcessThreadValue;
import com.google.gson.Gson;
public class ProcessUtil {
public static void changeProcessThreadStatus(String threadId, ProcessThreadStatus status) throws Exception {
String path = Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId;
String value = ZKUtil.getPathData(path);
ProcessThreadValue newValue = new Gson().fromJson(value, ProcessThreadValue.class);
newValue.setStatus(status.getValue());
ZKUtil.setPathData(path, new Gson().toJson(newValue));
}
}

View File

@ -1,63 +0,0 @@
package com.a.eye.skywalking.alarm.util;
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 com.a.eye.skywalking.alarm.conf.Config;
public class RedisUtil {
private static Logger logger = LogManager.getLogger(RedisUtil.class);
private static JedisPool jedisPool;
private static String[] config;
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]));
}
}
}
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 Jedis getRedisClient() {
return jedisPool.getResource();
}
public static <R> R execute(Executable<R> exec) {
Jedis client = null;
try {
client = RedisUtil.getRedisClient();
return exec.exe(client);
} finally {
if (client != null) {
client.close();
}
}
}
public interface Executable<R> {
R exe(Jedis client);
}
}

View File

@ -1,30 +0,0 @@
package com.a.eye.skywalking.alarm.util;
import java.io.File;
import java.io.IOException;
import java.sql.SQLException;
import com.a.eye.skywalking.alarm.conf.Config;
import freemarker.template.Configuration;
import freemarker.template.TemplateModelException;
import freemarker.template.Version;
public class TemplateConfigurationUtil {
private static Configuration cfg;
public static Configuration getConfiguration() throws SQLException, TemplateModelException, IOException {
if (cfg == null) {
cfg = new Configuration(new Version("2.3.23"));
cfg.setDefaultEncoding("UTF-8");
cfg.setSharedVariable("portalAddr", Config.TemplateInfo.PORTAL_URL);
//获取资源路径
String classPath = cfg.getClass().getResource("/").getFile().toString();
cfg.setDirectoryForTemplateLoading(new File(classPath));
}
return cfg;
}
}

View File

@ -1,81 +0,0 @@
package com.a.eye.skywalking.alarm.util;
import com.a.eye.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.api.CuratorWatcher;
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 org.apache.zookeeper.CreateMode;
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();
} catch (Exception e) {
logger.error("Failed to connect zookeeper.", e);
System.exit(-1);
}
}
public static CuratorFramework getZkClient() {
return client;
}
public static InterProcessMutex getLock(String path) {
return new InterProcessMutex(client, path);
}
public static String getPathData(String path) throws Exception {
return new String(client.getData().forPath(path));
}
public static String getPathDataWithWatch(String path, CuratorWatcher watcher) throws Exception {
return new String(client.getData().usingWatcher(watcher).forPath(path));
}
public static void setPathData(String path, String value) throws Exception {
client.setData().forPath(path, value.getBytes());
}
public static List<String> getChildren(String registerServerPath) throws Exception {
return client.getChildren().forPath(registerServerPath);
}
public static List<String> getChildrenWithWatcher(String registerServerPath, CuratorWatcher watcher) throws Exception {
return client.getChildren().usingWatcher(watcher).forPath(registerServerPath);
}
public static void createPath(String path) {
try {
client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
} catch (Exception e) {
logger.error("Failed to create path.");
}
}
public static boolean exists(String registerServerPath) {
try {
return client.checkExists().forPath(registerServerPath) != null;
} catch (Exception e) {
logger.error("Failed check exists for path");
}
return false;
}
}

View File

@ -1,77 +0,0 @@
# The work thread num.
server.process_thread_size=2
# The interval of daemon thread check.
server.daemon_thread_wait_interval=50000
# The expired time of alarm rule caches
server.alarm_rule_activity_interval=3600000
# The interval of work threads work.
processthread.thread_wait_interval=5000
#zookeeper connect url
zkpath.connect_str=127.0.0.1:2181
#zookeeper connect timeout(ms)
zkpath.connect_timeout=1000
#zookeeper re-connect times
zkpath.retry_times=3
#zookeeper re-connect timeout(ms)
zkpath.retry_timeout=1000
#zookeeper prefix path of store alarm rules
zkpath.node_prefix=/skywalking
#zookeeper path of register servers
zkpath.register_server_path=/skywalking/alarm-server/register-servers
#zookeeper path of coordinator
zkpath.coordinator_path=/skywalking/alarm-server/coordinator/lock
#The interval of coordinator acquire lock. (s)
coordinator.retry_get_coordinator_lock_interval=5
#The timeout of coordinator acquire lock. (ms)
coordinator.retry_become_coordinator_wait_time=10000
#The interval of coordinator checks redistribute. (ms)
coordinator.check_redistribute_interval=5000
#The interval of coordinator checks all process threads status. (ms)
coordinator.check_all_process_thread_interval=500
#database driver
db.driver_class=com.mysql.jdbc.Driver
#database connect url
db.url=jdbc:mysql://localhost:3306/test
#database username
db.user_name=root
#database password
db.password=root
#database connection pool max idle.
db.max_idle=1
#database connection pool max size.
db.max_pool_size=20
#timeout of database connection.
db.connect_timeout=10000
#redis addr
alarm.redis_server=127.0.0.1:6379
#redis client pool max idle.
alarm.redis_max_idle=20
#redis client pool min idle.
alarm.redis_min_idle=1
#redis client pool max size.
alarm.redis_max_total=50
#switch of alarm.
alarm.alarm_off_flag=false
#mai info
mailsenderinfo.mail_host=smtp.qq.com
mailsenderinfo.transport_protocol=smtp
mailsenderinfo.smtp_auth=true
mailsenderinfo.smtp_start_ssl_enable=false
mailsenderinfo.username=skywalking
mailsenderinfo.password=skywalking
mailsenderinfo.sender=skywalking@foxmail.com
mailsenderinfo.ssl_enable=true
#the url of sky-walking webui.
templateinfo.portal_url=http://127.0.0.1:8080/skywalking
#The interval of re-get inspect lock. (ms)
inspectthread.retry_get_inspect_lock_interval=10000
#The interval of checking users. (ms)
inspectthread.check_user_list_interval=300000

View File

@ -1,21 +0,0 @@
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration>
<appender name="rollingFile"
class="org.apache.log4j.RollingFileAppender">
<param name="File" value="../log/skywalking-server-log4j.log"/>
<param name="ImmediateFlush" value="true"/>
<param name="Threshold" value="INFO"></param>
<param name="Append" value="true"></param>
<param name="MaxFileSize" value="10240KB"></param>
<param name="MaxBackupIndex" value="10"></param>
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="%d - %c -%-4r [%t] %-5p %x - %m%n"></param>
</layout>
</appender>
<root>
<priority value="INFO" />
<appender-ref ref="rollingFile" />
</root>
</log4j:configuration>

View File

@ -1,26 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="DEBUG">
<Properties>
<Property name="log-path">../</Property>
</Properties>
<Appenders>
<RollingFile name="RollingFile" fileName="${log-path}/skywalking-server-log4j2.log"
filePattern="${log-path}/skywalking-server-log4j2-%d{yyyy-MM-dd}-%i.log" >
<PatternLayout>
<pattern>%d - %c -%-4r [%t] %-5p %x - %m%n</pattern>
</PatternLayout>
<Policies>
<SizeBasedTriggeringPolicy size="102400KB" />
</Policies>
<DefaultRolloverStrategy max="30"/>
</RollingFile>
</Appenders>
<Loggers>
<Logger name="root" level="INFO" additivity="false">
<appender-ref ref="RollingFile" level="INFO"/>
</Logger>
<Root level="INFO" additivity="false">
<AppenderRef ref="RollingFile"/>
</Root>
</Loggers>
</Configuration>

View File

@ -1,122 +0,0 @@
<!DOCTYPE html PUBLIC"-//W3C//DTD XHTML 1.0 Transitional//EN""http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns=\"http://www.w3.org/1999/xhtml\">
<head>
<meta http-equiv="Content-Type"content="text/html;charset=utf-8">
<style type="text/css">
li{text-indent:2em}
body {
color: #4f6b72;0
background: #E6EAE9;
}
#mainTable {
text-align:left;
font:12px/15px simsun;
line-height:20px;
color:#000;
background: #fff;
padding:30px 20px;
border:20px solid #fff;
margin:0 auto;
border-collapse:collapse;
border-spacing:0;
}
a {
color: #c75f3e;
}
.greetings {
font-family:Microsoft YaHei,SimHei,Arial;
}
#mainTd {
margin:0;
padding:0;
}
#mainDiv {
padding:25px;
font-size:14px;
line-height:25px;
margin:5px 0;
overflow:hidden;
}
#dataTable {
border-collapse:collapse;
border:none;
}
#dataTableHead {
display: none
}
#dataTable td {
border-bottom: 1px solid #C1DAD7;
background: #fff;
font-size:11px;
padding: 6px 6px 6px 12px;
color: #4f6b72;
}
#dataTable td.alt {
border-right: 1px solid #C1DAD7;
background: #F5FAFA;
color: #797268;
}
th.spec {
border: 1px solid #C1DAD7;
background: #CAE8EA ;
font: bold 10px "Trebuchet MS", Verdana, Arial, Helvetica, sans-serif;
padding: 6px 6px 6px 12px;
}
th.specalt {
border: 1px solid #C1DAD7;
background: #f5fafa;
font: bold 10px "Trebuchet MS", Verdana, Arial, Helvetica, sans-serif;
color: #797268;
padding: 6px 6px 6px 12px;
}
.type td p {
font: bold 14px "Trebuchet MS", Verdana, Arial, Helvetica, sans-serif;
}
</style>
<title>
templete1_welcome
</title>
</head>
<body>
<div id="mainDiv">
<p class="greetings">Dear ${name!}:</p>
<p class="greetings">&nbsp&nbsp&nbsp&nbspHere are the alarm message(s) between <b>${startDate!} </b> and <b> ${endDate!}</b> :
</p>
<table id="dataTable" width="80%">
<tr id="dataTableHead">
<td width="15%"></td>
<td></td>
</tr>
<#list alarmTypeList as alarmType>
<#if warningMap?exists>
<#if warningMap[alarmType.type]?exists>
<tr class="type">
<td class="typeTd" colspan="2" style="padding: 6px 6px 6px 6px;"><p>${alarmType.desc}</p></td>
</tr>
<#list warningMap[alarmType.type] as element>
<tr>
<th class="spec"><p>traceid</p></th>
<td class="alt"><a href="${(portalAddr + element.traceid)!}">${element.traceid!}</a><#if element.date?exists>&nbsp&nbsp&nbsp&nbsp<span>${element.date?string("yyyy-MM-dd HH:mm:ss ")}</span></#if></td>
</tr>
<tr>
<th class="specalt"><p>${(alarmType.label)}</p></th>
<td class="alt"><p>${element.exceptionMsg}</p></td>
</tr>
</#list>
</#if>
</#if>
</#list>
</table>
</body>
</html>

View File

@ -3,7 +3,7 @@
<parent>
<artifactId>skywalking</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>skywalking-application-toolkit</artifactId>

View File

@ -5,7 +5,7 @@
<parent>
<artifactId>skywalking-application-toolkit</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>

View File

@ -5,7 +5,7 @@
<parent>
<artifactId>skywalking-application-toolkit</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>

View File

@ -5,7 +5,7 @@
<parent>
<artifactId>skywalking-application-toolkit</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>

View File

@ -5,7 +5,7 @@
<parent>
<artifactId>skywalking-application-toolkit</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>

View File

@ -4,7 +4,7 @@
<parent>
<artifactId>skywalking-application-toolkit</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>

View File

@ -3,7 +3,7 @@
<parent>
<artifactId>skywalking</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@ -11,8 +11,8 @@
<packaging>pom</packaging>
<modules>
<module>skywalking-trace</module>
<module>skywalking-logging</module>
<module>skywalking-registry</module>
<module>skywalking-health-report</module>
<module>skywalking-util</module>
</modules>

View File

@ -3,7 +3,7 @@
<parent>
<artifactId>skywalking-commons</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>

View File

@ -3,7 +3,7 @@
<parent>
<artifactId>skywalking-commons</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>

View File

@ -5,7 +5,7 @@
<parent>
<artifactId>skywalking-logging</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>jar</packaging>

View File

@ -5,7 +5,7 @@
<parent>
<artifactId>skywalking-logging</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>jar</packaging>

View File

@ -1,43 +0,0 @@
<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">
<parent>
<artifactId>skywalking-commons</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>skywalking-registry</artifactId>
<packaging>jar</packaging>
<name>skywalking-registry</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-logging-api</artifactId>
<version>2.2-2017</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.8</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<version>2.8.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.10</version>
</dependency>
</dependencies>
</project>

View File

@ -1,40 +0,0 @@
package com.a.eye.skywalking.registry;
import com.a.eye.skywalking.registry.api.Center;
import com.a.eye.skywalking.registry.api.RegistryCenter;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.ServiceLoader;
/**
* Created by xin on 2016/11/10.
*/
public class RegistryCenterFactory {
public static RegistryCenterFactory INSTANCE = new RegistryCenterFactory();
private Map<String, RegistryCenter> registryCenter = new HashMap<String, RegistryCenter>();
private RegistryCenterFactory() {
ServiceLoader<RegistryCenter> loaders = ServiceLoader.load(RegistryCenter.class);
Iterator<RegistryCenter> iterator = loaders.iterator();
while (iterator.hasNext()) {
RegistryCenter center = iterator.next();
Center centerInfo = center.getClass().getAnnotation(Center.class);
if (centerInfo == null) {
continue;
}
registryCenter.put(centerInfo.type(), center);
}
}
public RegistryCenter getRegistryCenter(String type) {
return registryCenter.get(type);
}
}

View File

@ -1,41 +0,0 @@
package com.a.eye.skywalking.registry;
import com.a.eye.skywalking.registry.api.RegistryNode;
import java.util.ArrayList;
import java.util.List;
public class RegistryNodeManager {
private List<String> connectionURLOfPreviousChanged;
public RegistryNodeManager() {
connectionURLOfPreviousChanged = new ArrayList<String>();
}
public List<RegistryNode> calculateChangeOfConnectionURL(final List<String> currentConnectionURL) {
List<RegistryNode> connectionURLS = new ArrayList<RegistryNode>();
for (String URL : currentConnectionURL) {
if (!connectionURLOfPreviousChanged.contains(URL)) {
connectionURLS.add(new RegistryNode(URL, RegistryNode.ChangeType.ADDED));
}
}
for (String URL : connectionURLOfPreviousChanged) {
if (!currentConnectionURL.contains(URL)) {
connectionURLS.add(new RegistryNode(URL, RegistryNode.ChangeType.REMOVED));
}
}
connectionURLOfPreviousChanged = new ArrayList<String>(currentConnectionURL);
return connectionURLS;
}
public List<String> getConnectionURLOfPreviousChanged() {
return new ArrayList<String>(connectionURLOfPreviousChanged);
}
public void clear() {
connectionURLOfPreviousChanged = new ArrayList<String>();
}
}

View File

@ -1,9 +0,0 @@
package com.a.eye.skywalking.registry.api;
import java.lang.annotation.*;
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
public @interface Center {
String type() default CenterType.DEFAULT_CENTER_TYPE;
}

View File

@ -1,8 +0,0 @@
package com.a.eye.skywalking.registry.api;
/**
* Created by wusheng on 2016/11/10.
*/
public interface CenterType {
String DEFAULT_CENTER_TYPE = "zookeeper";
}

View File

@ -1,7 +0,0 @@
package com.a.eye.skywalking.registry.api;
import java.util.List;
public interface NotifyListener {
void notify(List<RegistryNode> registryNodes);
}

View File

@ -1,36 +0,0 @@
package com.a.eye.skywalking.registry.api;
import java.util.Properties;
/**
* 主要用于注册中心的维护
*/
public interface RegistryCenter {
/**
* 主要用于storage启动注册使用将自身IP和端口注册到注册中心
*
*
* @param path 格式为:/storage_list/192.168.0.1:3400
*/
void register(String path);
/**
* 主要用于routing节点在启动完成之后读取和监听stroage节点列表
*
* @param path
* @param listener
*/
void subscribe(String path, NotifyListener listener);
/**
* 在注册和订阅之前需要先启动注册中心
*
* @param centerConfig 配置参数
*/
void start(Properties centerConfig);
void stop();
}

View File

@ -1,27 +0,0 @@
package com.a.eye.skywalking.registry.api;
/**
* Created by xin on 2016/12/1.
*/
public class RegistryNode {
private String node;
private ChangeType changeType;
public RegistryNode(String node, ChangeType eventType) {
this.node = node;
this.changeType = eventType;
}
public enum ChangeType {
ADDED, REMOVED
}
public String getNode() {
return node;
}
public ChangeType getChangeType() {
return changeType;
}
}

View File

@ -1,10 +0,0 @@
package com.a.eye.skywalking.registry.api;
/**
* Created by xin on 2016/12/19.
*/
public class RegistryNodeFailedException extends RuntimeException {
public RegistryNodeFailedException(String message) {
super(message);
}
}

View File

@ -1,83 +0,0 @@
package com.a.eye.skywalking.registry.assist;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.Enumeration;
import java.util.regex.Pattern;
/**
* Created by xin on 2016/11/12.
*/
public class NetUtils {
private static ILog logger = LogManager.getLogger(NetUtils.class);
public static final String LOCALHOST = "127.0.0.1";
private static final String[] ANY_HOST = {"0.0.0.0", "::0"};
private static final Pattern IP_PATTERN = Pattern.compile("\\d{1,3}(\\.\\d{1,3}){3,5}$");
public static InetAddress getLocalAddress() {
InetAddress localAddress = null;
try {
localAddress = InetAddress.getLocalHost();
if (isValidAddress(localAddress)) {
return localAddress;
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
try {
// 获取所有的网卡
Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
if (interfaces != null) {
while (interfaces.hasMoreElements()) {
try {
NetworkInterface network = interfaces.nextElement();
// 遍历网卡中所有绑定的地址
Enumeration<InetAddress> addresses = network.getInetAddresses();
if (addresses != null) {
while (addresses.hasMoreElements()) {
try {
InetAddress address = addresses.nextElement();
// 判断地址是否为合法的IP地址
if (isValidAddress(address)) {
return address;
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
}
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
}
}
} catch (Throwable e) {
logger.warn("Failed to get ip address.", e);
}
return localAddress;
}
private static boolean isValidAddress(InetAddress address) {
if (address == null || address.isLoopbackAddress())
return false;
String name = address.getHostAddress();
// 不能是0.0.0.0 也不能是127.0.0.1 并且还得符合IP的正则
return (name != null && !isAnyHost(name) && !LOCALHOST.equals(name) && IP_PATTERN.matcher
(name).matches());
}
public static boolean isAnyHost(String host){
for (String s : ANY_HOST) {
if (s.equals(host)){
return true;
}
}
return false;
}
}

View File

@ -1,50 +0,0 @@
package com.a.eye.skywalking.registry.impl.zookeeper;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import java.util.Properties;
/**
* Created by xin on 2016/11/10.
*/
public class ZookeeperConfig {
private static ILog logger = LogManager.getLogger(ZookeeperConfig.class);
public static final String CONNECT_URL = "CONNECT_URL";
public static final String AUTH_SCHEMA = "AUTH_SCHEMA";
public static final String AUTH_INFO = "AUTH_INFO";
private String connectURL;
private String autSchema;
private byte[] auth;
public ZookeeperConfig(Properties config) {
this.connectURL = config.getProperty(CONNECT_URL);
if (this.connectURL == null || this.connectURL.length() == 0) {
throw new IllegalArgumentException("Connect url cannot be null");
}
this.autSchema = config.getProperty(AUTH_SCHEMA);
String authString = config.getProperty(AUTH_INFO);
if (authString != null) {
this.auth = authString.getBytes();
}
logger.info("connection url: {} \n auth schema : {} \n auth info : {} ", connectURL, autSchema, authString);
}
public boolean hasAuthInfo() {
return (this.autSchema != null && this.autSchema.length() > 0) && (this.auth != null && this.auth.length > 0);
}
public String getConnectURL() {
return connectURL;
}
public String getAutSchema() {
return autSchema;
}
public byte[] getAuth() {
return auth;
}
}

View File

@ -1,68 +0,0 @@
package com.a.eye.skywalking.registry.impl.zookeeper;
import com.a.eye.skywalking.registry.RegistryNodeManager;
import com.a.eye.skywalking.registry.api.Center;
import com.a.eye.skywalking.registry.api.CenterType;
import com.a.eye.skywalking.registry.api.NotifyListener;
import com.a.eye.skywalking.registry.api.RegistryCenter;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import java.util.List;
import java.util.Properties;
@Center(type = CenterType.DEFAULT_CENTER_TYPE)
public class ZookeeperRegistryCenter implements RegistryCenter {
private ZkClient client;
private RegistryNodeManager nodeManager = new RegistryNodeManager();
@Override
public void register(String path) {
String[] pathSegment = path.split("/");
StringBuilder createPath = new StringBuilder();
for (int i = 0; i < pathSegment.length - 1; i++) {
if (pathSegment[i] == null || pathSegment[i].length() == 0)
continue;
createPath.append("/" + pathSegment[i]);
if (!exists(createPath.toString())) {
client.createPersistent(createPath.toString());
}
}
client.createEphemeral(createPath.append("/" + pathSegment[pathSegment.length - 1]).toString());
}
@Override
public void subscribe(String path, final NotifyListener listener) {
List<String> children = client.subscribeChildChanges(path, new IZkChildListener() {
@Override
public void handleChildChange(String parentPath, List<String> children) throws Exception {
listener.notify(nodeManager.calculateChangeOfConnectionURL(children));
}
});
if (children != null && children.size() > 0)
listener.notify(nodeManager.calculateChangeOfConnectionURL(children));
}
private boolean exists(String path) {
return client.exists(path);
}
@Override
public void start(Properties centerConfig) {
ZookeeperConfig config = new ZookeeperConfig(centerConfig);
client = new ZkClient(config.getConnectURL(), 60 * 1000);
if (config.hasAuthInfo()) {
client.addAuthInfo(config.getAutSchema(), config.getAuth());
}
}
@Override
public void stop() {
nodeManager.clear();
}
}

View File

@ -1 +0,0 @@
com.a.eye.skywalking.registry.impl.zookeeper.ZookeeperRegistryCenter

View File

@ -1,55 +0,0 @@
package com.a.eye.skywalking.registry;
import com.a.eye.skywalking.registry.api.RegistryNode;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
/**
* Created by xin on 2016/12/1.
*/
public class RegistryNodeManagerTest {
RegistryNodeManager connectionURLManager = new RegistryNodeManager();
private List<RegistryNode> connectionURLS;
private List<String> url;
@Before
public void setUp() {
url = new ArrayList<String>() {{
add("127.0.0.1:34000");
add("127.0.0.1:35000");
}};
connectionURLS = connectionURLManager.calculateChangeOfConnectionURL(url);
}
@Test
public void calculateInitCandition() throws Exception {
Assert.assertEquals(connectionURLS.size(), 2);
for (RegistryNode connectionURL : connectionURLS) {
Assert.assertEquals(connectionURL.getChangeType(), RegistryNode.ChangeType.ADDED);
}
Assert.assertEquals(2, connectionURLManager.getConnectionURLOfPreviousChanged().size());
}
@Test
public void calculateContainAdd() throws Exception {
url.add("127.0.0.1:36000");
url.add("127.0.0.1:37000");
connectionURLS = connectionURLManager.calculateChangeOfConnectionURL(url);
Assert.assertEquals(4, connectionURLManager.getConnectionURLOfPreviousChanged().size());
Assert.assertEquals(2, connectionURLS.size());
for (RegistryNode connectionURL : connectionURLS) {
Assert.assertEquals(connectionURL.getChangeType(), RegistryNode.ChangeType.ADDED);
}
}
}

View File

@ -1,95 +0,0 @@
package com.a.eye.skywalking.registry.impl.zookeeper;
import com.a.eye.skywalking.registry.RegistryCenterFactory;
import com.a.eye.skywalking.registry.api.CenterType;
import com.a.eye.skywalking.registry.api.NotifyListener;
import com.a.eye.skywalking.registry.api.RegistryCenter;
import com.a.eye.skywalking.registry.api.RegistryNode;
import junit.framework.TestSuite;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.curator.test.TestingServer;
import org.apache.zookeeper.KeeperException;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.List;
import java.util.Properties;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class ZookeeperRegistryCenterTest extends TestSuite {
private TestingServer zkTestServer;
private ZkClient zkClient;
private RegistryCenter registryCenter;
@Before
public void setUp() throws Exception {
zkTestServer = new TestingServer(42181, true);
zkClient = new ZkClient("127.0.0.1:42181", 2000);
registryCenter = RegistryCenterFactory.INSTANCE.getRegistryCenter(CenterType.DEFAULT_CENTER_TYPE);
Properties config = new Properties();
config.put(ZookeeperConfig.CONNECT_URL, "127.0.0.1:42181");
registryCenter.start(config);
}
@After
public void tearDown() throws Exception {
zkTestServer.getTempDirectory().delete();
zkTestServer.stop();
registryCenter.stop();
}
@Test
public void subscribeNodeTest() throws InterruptedException {
final StringBuilder addUrl = new StringBuilder();
registryCenter.subscribe("/skywalking/storage", new NotifyListener() {
@Override
public void notify(List<RegistryNode> registryNodes) {
for (RegistryNode url : registryNodes) {
addUrl.append(url.getNode() + ",");
}
}
});
registryCenter.register("/skywalking/storage/127.0.0.1:9400");
Thread.sleep(100L);
assertEquals(addUrl.deleteCharAt(addUrl.length() - 1).toString(), "127.0.0.1:9400");
}
@Test
public void subscribeNodeAfterNodeRegistryTest() throws InterruptedException {
registryCenter.register("/skywalking/storage/127.0.0.1:9400");
final StringBuilder addUrl = new StringBuilder();
registryCenter.subscribe("/skywalking/storage", new NotifyListener() {
@Override
public void notify(List<RegistryNode> registryNodes) {
for (RegistryNode url : registryNodes) {
addUrl.append(url.getNode() + ",");
}
}
});
Thread.sleep(100L);
assertEquals(addUrl.deleteCharAt(addUrl.length() - 1).toString(), "127.0.0.1:9400");
}
@Test
public void registryNodeTest() throws IOException, InterruptedException, KeeperException {
registryCenter.register("/skywalking/storage/test");
assertTrue(zkClient.exists("/skywalking/storage/test"));
}
@Test(expected =ZkNodeExistsException.class)
public void registryTwoTmpPath(){
zkClient.createEphemeral("/testa");
zkClient.createEphemeral("/testa");
}
}

View File

@ -1,13 +0,0 @@
# log4j properties #
# logger #
log4j.rootLogger=DEBUG,CONSOLE
log4j.logger.org=ON
#log4j.logger.org.systemgo.devframework=DEBUG
# Console Appender #
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.Target=System.out
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d %-5p %c{1}:%L - %m%n

View File

@ -1,13 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="debug">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d - %c -%-4r [%t] %-5p %x - %m%n"/>
</Console>
</Appenders>
<Loggers>
<Root level="debug">
<AppenderRef ref="Console"/>
</Root>
</Loggers>
</Configuration>

View File

@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<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">
<parent>
<artifactId>skywalking-commons</artifactId>
<groupId>com.a.eye</groupId>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>skywalking-trace</artifactId>
</project>

View File

@ -0,0 +1,24 @@
package com.a.eye.skywalking.trace;
/**
* Span is a concept from OpenTracing Spec, also from Google Dapper Paper.
* Traces in OpenTracing are defined implicitly by their Spans.
*
* [Span A] (the root span)
* |
* +------+------+
* | |
* [Span B] [Span C] (Span C is a `ChildOf` Span A)
* | |
* [Span D] +---+-------+
* | |
* [Span E] [Span F] >>> [Span G] >>> [Span H]
*
*
*
* (Span G `FollowsFrom` Span F)
*
* Created by wusheng on 2017/2/17.
*/
public class Span {
}

View File

@ -0,0 +1,81 @@
package com.a.eye.skywalking.trace;
import java.util.LinkedList;
import java.util.List;
/**
* {@link TraceSegment} is a segment or fragment of the distributed trace.
* {@see https://github.com/opentracing/specification/blob/master/specification.md#the-opentracing-data-model}
* A {@link
* TraceSegment} means the segment, which exists in current {@link Thread}. And the distributed trace is formed by multi
* {@link TraceSegment}s, because the distributed trace crosses multi-processes, multi-threads.
*
* Created by wusheng on 2017/2/17.
*/
public class TraceSegment {
/**
* The id of this trace segment.
* Every segment has its unique-global-id.
*/
private String traceSegmentId;
/**
* The start time of this trace segment.
*/
private long startTime;
/**
* The end time of this trace segment.
*/
private long endTime;
/**
* The primary ref of the parent trace segment.
* Use {@link TraceSegmentRef}, we can link this trace segment to the primary parent segment.
*/
private TraceSegmentRef primaryRef;
/**
* The refs of other parent trace segments, except the primary one.
* For most RPC call, {@link #refs} stay in null,
* but if this segment is a start span of batch process, the segment faces multi parents,
* at this moment, we use this {@link #refs} to link them.
*/
private List<TraceSegmentRef> refs;
/**
* The spans belong to this trace segment.
* They all have finished.
* All active spans are hold and controlled by "skywalking-api" module.
*/
private List<Span> spans;
/**
* Create a trace segment, by given segmentId.
* This segmentId is generated by TraceSegmentRef, AKA tracer/agent module.
*
* @param segmentId {@link #traceSegmentId}
*/
public TraceSegment(String segmentId) {
this.traceSegmentId = segmentId;
this.startTime = System.currentTimeMillis();
}
/**
* Establish the link between this segment to his parents.
* The first time, you {@link #ref(TraceSegmentRef)} to parent, it is affirmed as {@link #primaryRef}.
* And others are affirmed as {@link #refs}.
*
* @param refSegment {@link TraceSegmentRef}
*/
public void ref(TraceSegmentRef refSegment){
if(primaryRef == null){
primaryRef = refSegment;
}else {
if (refs == null) {
refs = new LinkedList<TraceSegmentRef>();
}
refs.add(refSegment);
}
}
}

View File

@ -0,0 +1,9 @@
package com.a.eye.skywalking.trace;
/**
*
*
* Created by wusheng on 2017/2/17.
*/
public class TraceSegmentRef {
}

View File

@ -3,7 +3,7 @@
<parent>
<artifactId>skywalking-commons</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@ -17,11 +17,4 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-network</artifactId>
<version>2.2-2017</version>
</dependency>
</dependencies>
</project>

View File

@ -1,39 +0,0 @@
package com.a.eye.skywalking.util;
import com.a.eye.skywalking.network.grpc.TraceId;
/**
* Created by xin on 2016/12/8.
*/
public class TraceIdUtil {
public static String formatTraceId(TraceId traceId) {
StringBuilder traceIdBuilder = new StringBuilder();
for (Long segment : traceId.getSegmentsList()) {
traceIdBuilder.append(segment).append(".");
}
return traceIdBuilder.substring(0, traceIdBuilder.length() - 1).toString();
}
public static TraceId toTraceId(String traceId) {
String[] traceIdSegment = traceId.split("\\.");
TraceId.Builder builder = TraceId.newBuilder();
for (String segment : traceIdSegment) {
builder = builder.addSegments(Long.parseLong(segment));
}
return builder.build();
}
public static boolean isIllegalTraceId(String traceId){
if (StringUtil.isEmpty(traceId)){
return true;
}
return traceId.split("\\.").length != 6;
}
}

View File

@ -1,196 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<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">
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<modelVersion>4.0.0</modelVersion>
<artifactId>skywalking-network</artifactId>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<shade.package>com.a.eye.skywalking.network.dependencies</shade.package>
<!-- Google dependence-->
<shade.com.google.source>com.google</shade.com.google.source>
<shade.com.google.target>${shade.package}.com.google</shade.com.google.target>
<!-- GRPC -->
<shade.io.grpc.source>io.grpc</shade.io.grpc.source>
<shade.io.grpc.target>${shade.package}.io.grpc</shade.io.grpc.target>
<!--netty -->
<shade.io.netty.source>io.netty</shade.io.netty.source>
<shade.io.netty.target>${shade.package}.io.netty</shade.io.netty.target>
</properties>
<dependencies>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty</artifactId>
<version>1.0.2</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<version>1.0.2</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<version>1.0.2</version>
</dependency>
</dependencies>
<build>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.4.1.Final</version>
</extension>
</extensions>
<plugins>
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<version>0.5.0</version>
<configuration>
<protocArtifact>com.google.protobuf:protoc:3.0.2:exe:${os.detected.classifier}</protocArtifact>
<pluginId>grpc-java</pluginId>
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.0.2:exe:${os.detected.classifier}</pluginArtifact>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<shadedArtifactAttached>false</shadedArtifactAttached>
<createDependencyReducedPom>true</createDependencyReducedPom>
<createSourcesJar>true</createSourcesJar>
<shadeSourcesContent>true</shadeSourcesContent>
<relocations>
<relocation>
<pattern>${shade.com.google.source}</pattern>
<shadedPattern>${shade.com.google.target}</shadedPattern>
</relocation>
<relocation>
<pattern>${shade.io.grpc.source}</pattern>
<shadedPattern>${shade.io.grpc.target}</shadedPattern>
</relocation>
<relocation>
<pattern>${shade.io.netty.source}</pattern>
<shadedPattern>${shade.io.netty.target}</shadedPattern>
</relocation>
</relocations>
<artifactSet>
<includes>
<include>io.grpc:grpc-*:jar:*</include>
<inclued>io.netty:netty-*:jar:*</inclued>
<inclued>com.google.protobuf:*:jar:*</inclued>
<inclued>com.google.code.gson:gson:jar:*</inclued>
<include>com.google.guava:guava:jar:*</include>
</includes>
</artifactSet>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
<!-- add this plugin to fix shade-plugin bug: not rename provider file name. -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-antrun-plugin</artifactId>
<version>1.8</version>
<executions>
<execution>
<id>unpack</id>
<phase>package</phase>
<configuration>
<target>
<echo message="unjar" />
<unzip src="${project.build.directory}/${artifactId}-${version}.jar" dest="${project.build.directory}/unpacked/" />
<echo message="rename service providers in META-INF/services" />
<move todir="${project.build.directory}/unpacked/META-INF/services" includeemptydirs="false">
<fileset dir="${project.build.directory}/unpacked/META-INF/services"/>
<mapper type="glob" from="*" to="${shade.package}.*"/>
</move>
<echo message="jar back" />
<jar destfile="${project.build.directory}/${artifactId}-${version}.jar" basedir="${project.build.directory}/unpacked" />
</target>
</configuration>
<goals>
<goal>run</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
<version>1.9.1</version>
<executions>
<execution>
<id>add-source</id>
<phase>generate-sources</phase>
<goals>
<goal>add-source</goal>
</goals>
<configuration>
<sources>
<source>${project.basedir}/src/main/gen-java</source>
</sources>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<version>2.4.3</version>
<configuration>
<encoding>${project.build.sourceEncoding}</encoding>
</configuration>
</plugin>
<plugin>
<!-- 源码插件 -->
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-source-plugin</artifactId>
<!-- 发布时自动将源码同时发布的配置 -->
<executions>
<execution>
<id>attach-sources</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
<version>2.4</version>
</plugin>
</plugins>
</build>
</project>

View File

@ -1,31 +0,0 @@
package com.a.eye.skywalking.network;
import com.a.eye.skywalking.network.grpc.client.SpanStorageClient;
import com.a.eye.skywalking.network.grpc.client.TraceSearchClient;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
public class Client {
private ManagedChannel channel;
public Client(String ip, int port) {
channel = ManagedChannelBuilder.forAddress(ip, port).usePlaintext(true).build();
}
public SpanStorageClient newSpanStorageClient(StorageClientListener listener) {
return new SpanStorageClient(channel, listener);
}
public TraceSearchClient newTraceSearchClient() {
return new TraceSearchClient(channel);
}
public void shutdown() {
channel.shutdownNow();
}
public boolean isShutdown() {
return channel.isShutdown() || channel.isTerminated();
}
}

View File

@ -1,71 +0,0 @@
package com.a.eye.skywalking.network;
import com.a.eye.skywalking.network.grpc.server.AsyncTraceSearchServer;
import com.a.eye.skywalking.network.grpc.server.SpanStorageServer;
import com.a.eye.skywalking.network.grpc.server.TraceSearchServer;
import com.a.eye.skywalking.network.listener.server.SpanStorageServerListener;
import com.a.eye.skywalking.network.listener.server.TraceSearchListener;
import io.grpc.netty.NettyServerBuilder;
import io.netty.channel.nio.NioEventLoopGroup;
import java.io.IOException;
import java.net.InetSocketAddress;
public class Server {
private io.grpc.Server server;
private Server(io.grpc.Server server) {
this.server = server;
}
public void start() throws IOException, InterruptedException {
server.start();
// 当JVM停止之后Server也需要停止
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
Server.this.stop();
}
});
}
public void stop() {
if (server != null) {
server.shutdown();
}
}
public static TransferServiceBuilder newBuilder(String host, int port) {
return new TransferServiceBuilder(host, port);
}
public static class TransferServiceBuilder {
private TransferServiceBuilder(String host, int port) {
serverBuilder = NettyServerBuilder.forAddress(new InetSocketAddress
(host, port));
serverBuilder.maxConcurrentCallsPerConnection(4);
}
private NettyServerBuilder serverBuilder;
public Server build() {
return new Server(serverBuilder.bossEventLoopGroup(new NioEventLoopGroup(1))
.workerEventLoopGroup(new NioEventLoopGroup()).build());
}
public TransferServiceBuilder addSpanStorageService(SpanStorageServerListener spanStorageServerListener) {
serverBuilder.addService(new SpanStorageServer(spanStorageServerListener));
return this;
}
public TransferServiceBuilder addTraceSearchService(TraceSearchListener traceSearchListener) {
serverBuilder.addService(new TraceSearchServer(traceSearchListener));
return this;
}
public TransferServiceBuilder addAsyncTraceSearchService(TraceSearchListener traceSearchListener){
serverBuilder.addService(new AsyncTraceSearchServer(traceSearchListener));
return this;
}
}
}

View File

@ -1,74 +0,0 @@
package com.a.eye.skywalking.network.grpc.client;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.SendResult;
import com.a.eye.skywalking.network.grpc.SpanStorageServiceGrpc;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import io.grpc.ManagedChannel;
import io.grpc.stub.CallStreamObserver;
import io.grpc.stub.StreamObserver;
import java.util.List;
import java.util.concurrent.locks.LockSupport;
public class SpanStorageClient {
private final SpanStorageServiceGrpc.SpanStorageServiceStub spanStorageStub;
private final StorageClientListener listener;
public SpanStorageClient(ManagedChannel channel, StorageClientListener listener) {
this.spanStorageStub = SpanStorageServiceGrpc.newStub(channel);
this.listener = listener;
}
public void sendRequestSpan(List<RequestSpan> requestSpan) {
StreamObserver<RequestSpan> requestSpanStreamObserver = spanStorageStub.storageRequestSpan(new StreamObserver<SendResult>() {
@Override
public void onNext(SendResult sendResult) {
}
@Override
public void onError(Throwable throwable) {
listener.onError(throwable);
}
@Override
public void onCompleted() {
listener.onBatchFinished();
}
});
for (RequestSpan span : requestSpan) {
requestSpanStreamObserver.onNext(span);
}
requestSpanStreamObserver.onCompleted();
}
public void sendACKSpan(List<AckSpan> ackSpan) {
StreamObserver<AckSpan> ackSpanStreamObserver = spanStorageStub.storageACKSpan(new StreamObserver<SendResult>() {
@Override
public void onNext(SendResult sendResult) {
}
@Override
public void onError(Throwable throwable) {
listener.onError(throwable);
}
@Override
public void onCompleted() {
listener.onBatchFinished();
}
});
for (AckSpan span : ackSpan) {
ackSpanStreamObserver.onNext(span);
}
ackSpanStreamObserver.onCompleted();
}
}

View File

@ -1,49 +0,0 @@
package com.a.eye.skywalking.network.grpc.client;
import com.a.eye.skywalking.network.grpc.AsyncTraceSearchServiceGrpc;
import com.a.eye.skywalking.network.grpc.QueryTask;
import com.a.eye.skywalking.network.grpc.SearchResult;
import com.a.eye.skywalking.network.grpc.TraceSearchServiceGrpc;
import com.a.eye.skywalking.network.listener.client.SearchClientListener;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
/**
* Created by wusheng on 2016/11/26.
*/
public class TraceSearchClient {
private final AsyncTraceSearchServiceGrpc.AsyncTraceSearchServiceStub traceSearchServiceStub;
private final TraceSearchServiceGrpc.TraceSearchServiceBlockingStub traceSearchServiceBlockingStub;
public TraceSearchClient(ManagedChannel channel) {
this.traceSearchServiceStub = AsyncTraceSearchServiceGrpc.newStub(channel);
this.traceSearchServiceBlockingStub = TraceSearchServiceGrpc.newBlockingStub(channel);
}
public void search(QueryTask queryTask, final SearchClientListener listener){
StreamObserver<SearchResult> serverStreamObserver = new StreamObserver<SearchResult>() {
@Override
public void onNext(SearchResult searchResult) {
listener.onReturn(searchResult);
}
@Override
public void onError(Throwable throwable) {
listener.onError(throwable);
}
@Override
public void onCompleted() {
listener.onFinished();
}
};
StreamObserver<QueryTask> searchResult = traceSearchServiceStub.search(serverStreamObserver);
searchResult.onNext(queryTask);
searchResult.onCompleted();
}
public SearchResult search(QueryTask queryTask){
return traceSearchServiceBlockingStub.search(queryTask);
}
}

View File

@ -1,51 +0,0 @@
package com.a.eye.skywalking.network.grpc.server;
import com.a.eye.skywalking.network.grpc.AsyncTraceSearchServiceGrpc;
import com.a.eye.skywalking.network.grpc.QueryTask;
import com.a.eye.skywalking.network.grpc.SearchResult;
import com.a.eye.skywalking.network.grpc.Span;
import com.a.eye.skywalking.network.listener.server.TraceSearchListener;
import io.grpc.stub.StreamObserver;
import java.util.List;
/**
* Created by xin on 2016/11/15.
*/
public class AsyncTraceSearchServer extends AsyncTraceSearchServiceGrpc.AsyncTraceSearchServiceImplBase {
private TraceSearchListener searchListener;
public AsyncTraceSearchServer(TraceSearchListener searchListener) {
this.searchListener = searchListener;
}
@Override
public StreamObserver<QueryTask> search(final StreamObserver<SearchResult> responseObserver) {
return new StreamObserver<QueryTask>() {
private List<Span> spans;
@Override
public void onNext(QueryTask value) {
spans = searchListener.search(value.getTraceId());
}
@Override
public void onError(Throwable t) {
SearchResult.Builder builder = SearchResult.newBuilder();
responseObserver.onNext(builder.build());
responseObserver.onCompleted();
}
@Override
public void onCompleted() {
SearchResult.Builder builder = SearchResult.newBuilder();
if(spans != null) {
builder = builder.addAllSpans(spans);
}
responseObserver.onNext(builder.build());
responseObserver.onCompleted();
}
};
}
}

View File

@ -1,63 +0,0 @@
package com.a.eye.skywalking.network.grpc.server;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.SendResult;
import com.a.eye.skywalking.network.grpc.SpanStorageServiceGrpc;
import com.a.eye.skywalking.network.listener.server.SpanStorageServerListener;
import io.grpc.stub.StreamObserver;
public class SpanStorageServer extends SpanStorageServiceGrpc.SpanStorageServiceImplBase {
private SpanStorageServerListener listener;
public SpanStorageServer(SpanStorageServerListener listener) {
this.listener = listener;
}
@Override
public StreamObserver<AckSpan> storageACKSpan(final StreamObserver<SendResult> responseObserver) {
return new StreamObserver<AckSpan>() {
@Override
public void onNext(AckSpan value) {
listener.storage(value);
}
@Override
public void onError(Throwable t) {
responseObserver.onNext(SendResult.newBuilder().setResult(false).build());
responseObserver.onCompleted();
}
@Override
public void onCompleted() {
responseObserver.onNext(SendResult.newBuilder().setResult(true).build());
responseObserver.onCompleted();
}
};
}
@Override
public StreamObserver<RequestSpan> storageRequestSpan(final StreamObserver<SendResult> responseObserver) {
return new StreamObserver<RequestSpan>() {
@Override
public void onNext(RequestSpan value) {
listener.storage(value);
}
@Override
public void onError(Throwable t) {
responseObserver.onNext(SendResult.newBuilder().setResult(false).build());
responseObserver.onCompleted();
}
@Override
public void onCompleted() {
responseObserver.onNext(SendResult.newBuilder().setResult(true).build());
responseObserver.onCompleted();
}
};
}
}

View File

@ -1,27 +0,0 @@
package com.a.eye.skywalking.network.grpc.server;
import com.a.eye.skywalking.network.grpc.*;
import com.a.eye.skywalking.network.listener.server.TraceSearchListener;
import io.grpc.stub.StreamObserver;
import java.util.List;
/**
* Created by xin on 2016/11/12.
*/
public class TraceSearchServer extends TraceSearchServiceGrpc.TraceSearchServiceImplBase {
private TraceSearchListener traceSearchListener;
public TraceSearchServer(TraceSearchListener traceSearchListener) {
this.traceSearchListener = traceSearchListener;
}
@Override
public void search(QueryTask request, StreamObserver<SearchResult> responseObserver) {
List<Span> spans = traceSearchListener.search(request.getTraceId());
responseObserver.onNext(SearchResult.newBuilder().addAllSpans(spans).build());
responseObserver.onCompleted();
}
}

View File

@ -1,14 +0,0 @@
package com.a.eye.skywalking.network.listener.client;
import com.a.eye.skywalking.network.grpc.SearchResult;
/**
* Created by wusheng on 2016/12/3.
*/
public interface SearchClientListener {
void onError(Throwable throwable);
void onReturn(SearchResult result);
void onFinished();
}

View File

@ -1,10 +0,0 @@
package com.a.eye.skywalking.network.listener.client;
/**
* Created by wusheng on 2016/11/27.
*/
public interface StorageClientListener {
void onError(Throwable throwable);
void onBatchFinished();
}

View File

@ -1,10 +0,0 @@
package com.a.eye.skywalking.network.listener.server;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
public interface SpanStorageServerListener {
boolean storage(RequestSpan requestSpan);
boolean storage(AckSpan ackSpan);
}

View File

@ -1,10 +0,0 @@
package com.a.eye.skywalking.network.listener.server;
import com.a.eye.skywalking.network.grpc.Span;
import com.a.eye.skywalking.network.grpc.TraceId;
import java.util.List;
public interface TraceSearchListener{
List<Span> search(TraceId traceId);
}

View File

@ -1,43 +0,0 @@
package com.a.eye.skywalking.network.model;
/**
* Created by xin on 2016/12/22.
*/
public enum Tag {
VIEW_POINT("viewpoint", 1), BUSINESS_KEY("business_key", 2), CALL_TYPE("call.type", 3),
SPAN_TYPE("type", 4), CALL_DESC("call.desc", 5), USER_NAME("username", 6),
ADDRESS("hostname", 7), PROCESS_NO("process_no", 8), APPLICATION_CODE("application_code", 9),
STATUS("error.status", 10), EXCEPTION_STACK("error.exception_stack", 11);
private String keyName;
private int keyValue;
Tag(String keyName, int keyValue) {
this.keyName = keyName;
this.keyValue = keyValue;
}
public static Tag convert(String value) {
String[] valueSegment = value.split("_");
if (valueSegment.length != 2) {
throw new IllegalArgumentException("Failed to convert to tag[ " + value + "]");
}
switch (Integer.parseInt(valueSegment[1])) {
case 1:
return VIEW_POINT;
case 2:
return BUSINESS_KEY;
default:
throw new IllegalArgumentException("Cannot find the tag by keyValue[" + valueSegment[1] + "]");
}
}
public String toString() {
return keyName + "_" + keyValue;
}
public String key() {
return toString();
}
}

View File

@ -1,18 +0,0 @@
syntax = "proto3";
option java_multiple_files = true;
option java_package = "com.a.eye.skywalking.network.grpc";
import "Spans.proto";
service SpanStorageService {
rpc storageACKSpan (stream AckSpan) returns (SendResult) {
};
rpc storageRequestSpan (stream RequestSpan) returns (SendResult) {
};
}
message SendResult {
// 0 1
bool result = 1;
}

View File

@ -1,46 +0,0 @@
syntax = "proto3";
option java_multiple_files = true;
option java_package = "com.a.eye.skywalking.network.grpc";
message AckSpan {
TraceId traceId = 1;
string parentLevel = 2;
int32 levelId = 3;
int64 cost = 4;
int32 routeKey = 5;
map<string, string> tags = 6;
}
message RequestSpan {
TraceId traceId = 1;
string parentLevel = 2;
int32 levelId = 3;
int64 startTimestamp = 4;
int32 routeKey = 5;
map<string, string> tags = 6;
}
message TraceId {
repeated int64 segments = 1;
}
message Span {
TraceId traceId = 1;
string parentLevelId=2;
int32 levelId = 3;
string viewpoint = 4;
int64 startTimestamp = 5;
int64 cost = 6;
int32 statusCode = 7;
string exceptionStack = 8;
string spanTypeDesc = 9;
string callType = 10;
int32 spanType = 11;
string applicationCode = 12;
string username = 13;
string businessKey = 14;
int32 processNo = 15;
string address = 16;
}

View File

@ -1,24 +0,0 @@
syntax = "proto3";
option java_multiple_files = true;
option java_package = "com.a.eye.skywalking.network.grpc";
import "Spans.proto";
service AsyncTraceSearchService {
rpc search (stream QueryTask) returns (stream SearchResult) {
};
}
service TraceSearchService {
rpc search (QueryTask) returns (SearchResult) {
};
}
message QueryTask {
TraceId traceId = 2;
}
message SearchResult {
repeated Span spans = 2;
}

View File

@ -5,7 +5,7 @@
<parent>
<artifactId>skywalking</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@ -27,7 +27,7 @@
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-util</artifactId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</dependency>
</dependencies>
</project>

View File

@ -3,7 +3,7 @@
<parent>
<artifactId>skywalking-sniffer</artifactId>
<groupId>com.a.eye</groupId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>skywalking-agent</artifactId>

View File

@ -5,7 +5,7 @@
<parent>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-sniffer</artifactId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
<artifactId>skywalking-agent</artifactId>

View File

@ -5,7 +5,7 @@
<parent>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-sniffer</artifactId>
<version>2.2-2017</version>
<version>3.0-2017</version>
</parent>
@ -28,8 +28,8 @@
</dependency>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-network</artifactId>
<version>2.2-2017</version>
<artifactId>skywalking-trace</artifactId>
<version>3.0-2017</version>
</dependency>
<dependency>
<groupId>com.lmax</groupId>

View File

@ -1,30 +0,0 @@
package com.a.eye.skywalking.api;
import com.a.eye.skywalking.conf.Config;
import com.a.eye.skywalking.context.CurrentThreadSpanStack;
import com.a.eye.skywalking.model.Span;
public final class BusinessKeyAppender {
private BusinessKeyAppender() {
// Non
}
/**
* 在当前埋点上下文中设置业务级信息
*/
public static void setBusinessKey2Trace(String businessKey) {
Span spanData = CurrentThreadSpanStack.peek();
if (spanData == null) {
return;
}
if (businessKey.length() <= Config.BuriedPoint.BUSINESSKEY_MAX_LENGTH) {
spanData.setBusinessKey(businessKey);
return;
}
spanData.setBusinessKey(businessKey.substring(0, Config.BuriedPoint.BUSINESSKEY_MAX_LENGTH));
}
}

View File

@ -1,40 +0,0 @@
package com.a.eye.skywalking.api;
public interface IBuriedPointType {
String getTypeName();
CallType getCallType();
enum CallType {
LOCAL('L'),
SYNC('S'),
ASYNC('A');
private char value;
CallType(char value) {
this.value = value;
}
public static CallType convert(String id) {
char v = id.charAt(0);
switch (v) {
case 'L':
return LOCAL;
case 'S':
return SYNC;
case 'A':
return ASYNC;
default:
throw new IllegalStateException("Failed to convert callType[" + id + "]");
}
}
@Override
public String toString() {
return String.valueOf(value);
}
}
}

View File

@ -1,54 +0,0 @@
package com.a.eye.skywalking.api;
import com.a.eye.skywalking.context.CurrentThreadSpanStack;
import com.a.eye.skywalking.model.RefContext;
import com.a.eye.skywalking.model.Span;
import static com.a.eye.skywalking.util.TraceIdUtil.formatTraceId;
public final class Tracing {
/**
* Get the traceId of current trace context.
*
* @return traceId, if it exists, or empty {@link String}.
*/
public static String getTraceId() {
Span spanData = CurrentThreadSpanStack.peek();
if (spanData == null) {
return "";
}
return formatTraceId(spanData.getTraceId());
}
/**
* Get the current span of current trace.
*
* @return span. if it exists, or null
*/
public static Span getCurrentSpan() {
Span spanData = CurrentThreadSpanStack.peek();
return spanData;
}
/**
* the span will be tagged with the given key and value pair
*
* @param span
* @param tagKey key of tag
* @param tagValue value of tag
*/
public static void tag(Span span, String tagKey, String tagValue) {
span.tag(tagKey, tagValue);
}
/**
* init the ref context
*
* @param refContext ref context
*/
public static void initRefContext(RefContext refContext){
CurrentThreadSpanStack.initRefContext(refContext);
}
}

View File

@ -1,185 +0,0 @@
package com.a.eye.skywalking.client;
import com.a.eye.skywalking.conf.Config;
import com.a.eye.skywalking.disruptor.ack.SendAckSpanEventHandler;
import com.a.eye.skywalking.disruptor.request.SendRequestSpanEventHandler;
import com.a.eye.skywalking.health.report.HealthCollector;
import com.a.eye.skywalking.health.report.HeathReading;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.network.Client;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.client.SpanStorageClient;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
/**
* Created by wusheng on 2016/11/27.
*/
public class Agent2RoutingClient extends Thread {
private static ILog logger = LogManager.getLogger(Agent2RoutingClient.class);
private List<ServerAddr> addrList;
private Client client;
private SpanStorageClient spanStorageClient;
private NetworkListener listener;
private SendRequestSpanEventHandler requestSpanDataSupplier = null;
private SendAckSpanEventHandler ackSpanDataSupplier = null;
private volatile boolean connected = false;
public static Agent2RoutingClient INSTANCE = new Agent2RoutingClient();
public Agent2RoutingClient() {
String[] serverList = Config.SkyWalking.SERVERS.split(",");
addrList = new ArrayList<>(serverList.length);
for (String server : serverList) {
String[] addrSegments = server.split(":");
if (addrSegments.length != 2) {
throw new IllegalArgumentException("server addr should like ip:port, illegal addr:" + server);
}
addrList.add(new ServerAddr(addrSegments[0], addrSegments[1]));
}
listener = new NetworkListener();
onReady();
}
public void onReady() {
this.connect();
this.start();
}
public void setRequestSpanDataSupplier(SendRequestSpanEventHandler requestSpanDataSupplier) {
this.requestSpanDataSupplier = requestSpanDataSupplier;
}
public void setAckSpanDataSupplier(SendAckSpanEventHandler ackSpanDataSupplier) {
this.ackSpanDataSupplier = ackSpanDataSupplier;
}
private void connect() {
try {
if (client != null && !client.isShutdown()) {
client.shutdown();
}
int addrIdx = new Random().nextInt(addrList.size());
ServerAddr addr = addrList.get(addrIdx);
client = new Client(addr.ip, addr.port);
spanStorageClient = client.newSpanStorageClient(listener);
connected = true;
} catch (Exception e) {
HealthCollector.getCurrentHeathReading("Agent2RoutingClient").updateData(HeathReading.ERROR, "connect to routing node failure.");
}
}
@Override
public void run() {
while (true) {
try {
while (requestSpanDataSupplier != null && ackSpanDataSupplier != null && connected && !client.isShutdown()) {
List<RequestSpan> requestData = this.requestSpanDataSupplier.getBufferData();
List<AckSpan> ackData = this.ackSpanDataSupplier.getBufferData();
boolean hasData = false;
if (requestData.size() > 0) {
hasData = true;
listener.begin();
spanStorageClient.sendRequestSpan(requestData);
listener.wait2Confirm();
}
if (ackData.size() > 0) {
hasData = true;
listener.begin();
spanStorageClient.sendACKSpan(ackData);
listener.wait2Confirm();
}
if (!hasData) {
try {
Thread.sleep(10 * 1000L);
} catch (InterruptedException e) {
}
}
}
try {
Thread.sleep(30 * 1000L);
} catch (InterruptedException e) {
}
this.connect();
} catch (Throwable e) {
logger.error("unexpected failure.", e);
}
}
}
class NetworkListener implements StorageClientListener {
private volatile boolean batchFinished = false;
void begin() {
batchFinished = false;
}
boolean isBatchFinished() {
return batchFinished;
}
@Override
public void onError(Throwable throwable) {
batchFinished = true;
HealthCollector.getCurrentHeathReading("Agent2RoutingClient").updateData(HeathReading.ERROR, "send data to routing node failure.");
}
@Override
public void onBatchFinished() {
batchFinished = true;
HealthCollector.getCurrentHeathReading("Agent2RoutingClient").updateData(HeathReading.INFO, "batch send data to routing node.");
}
void wait2Confirm() {
// wait 20s, most.
int countDown = 100 * 20;
while (!listener.isBatchFinished()) {
try {
Thread.sleep(10L);
if (countDown-- < 0) {
batchFinished = true;
}
} catch (InterruptedException e) {
}
}
}
}
class ServerAddr {
String ip;
Integer port;
public ServerAddr(String ip, String port) {
this.ip = ip;
try {
this.port = Integer.parseInt(port);
} catch (NumberFormatException e) {
throw new IllegalArgumentException("server addr should like ip:port, illegal port:" + port);
}
}
}
}

View File

@ -1,142 +0,0 @@
package com.a.eye.skywalking.context;
import com.a.eye.skywalking.model.RefContext;
import com.a.eye.skywalking.model.Span;
import java.util.ArrayList;
import java.util.List;
/**
* Core in-process propagation context accessor.
* You can push, peek, pop {@link Span}s.
* In processing, every span be created and pushed into the context when begin
* This context based on stack structure. see the stack on {@link SpanNodeStack}
*
* @author wusheng
*/
public class CurrentThreadSpanStack {
private static ThreadLocal<SpanNodeStack> nodes = new ThreadLocal<SpanNodeStack>();
private CurrentThreadSpanStack() {
}
public static void push(Span span) {
if (nodes.get() == null) {
nodes.set(new SpanNodeStack());
}
nodes.get().push(span);
}
public static Span peek() {
if (nodes.get() == null) {
return null;
}
return nodes.get().peek();
}
public static Span pop() {
if (nodes.get() == null) {
return null;
}
return nodes.get().pop();
}
public static void initRefContext(RefContext refContext) {
if (nodes.get() == null){
nodes.set(new SpanNodeStack());
}
nodes.get().initRefContext(refContext);
}
static class SpanNodeStack {
/**
* This {@link RefContext} contain the trace context from another processor.
*/
private RefContext refContext = null;
/**
* The depth of call stack should less than 20, in most cases.
* The depth is calculated by span, not class or the depth of java stack.
*/
private List<SpanNode> spans = new ArrayList<SpanNode>(20);
public Span pop() {
Span span = spans.remove(getTopElementIdx()).getData();
if (!isEmpty()) {
spans.get(getTopElementIdx()).incrementNextSubSpanLevelId();
}else{
refContext = null;
}
return span;
}
public void push(Span span) {
if (!isEmpty()) {
listPush(new SpanNode(span, spans.get(getTopElementIdx()).getNextSubSpanLevelId()));
} else {
if (refContext != null) {
span.setTraceId(refContext.getTraceId());
span.setParentLevel(refContext.getParentLevelId());
span.setLevelId(0);
}
listPush(new SpanNode(span));
}
}
public Span peek() {
if (spans.isEmpty()) {
return null;
}
return spans.get(getTopElementIdx()).getData();
}
private int getTopElementIdx() {
return spans.size() - 1;
}
private boolean isEmpty() {
return spans.isEmpty();
}
private void listPush(SpanNode spanNode) {
spans.add(spans.size(), spanNode);
}
public void initRefContext(RefContext refContext) {
this.refContext = refContext;
}
}
static class SpanNode {
private Span data;
private int nextSubSpanLevelId = 0;
public SpanNode(Span data) {
this.data = data;
}
public SpanNode(Span data, int levelId) {
this.data = data;
this.data.setLevelId(levelId);
}
public Span getData() {
return data;
}
public int getNextSubSpanLevelId() {
return nextSubSpanLevelId;
}
public void incrementNextSubSpanLevelId() {
this.nextSubSpanLevelId++;
}
}
}

View File

@ -0,0 +1,7 @@
package com.a.eye.skywalking.context;
/**
* Created by wusheng on 2017/2/17.
*/
public class TraceContext {
}

View File

@ -1,93 +0,0 @@
package com.a.eye.skywalking.model;
import com.a.eye.skywalking.api.Tracing;
import com.a.eye.skywalking.network.grpc.TraceId;
import static com.a.eye.skywalking.conf.Constants.CONTEXT_DATA_SEGMENT_SPILT_CHAR;
import static com.a.eye.skywalking.util.TraceIdUtil.formatTraceId;
public class ContextData {
private TraceId traceId;
private String parentLevel;
private int levelId;
private int routeKey;
ContextData() {
}
public ContextData(TraceId traceId, String parentLevelId, int routeKey) {
this.traceId = traceId;
this.parentLevel = parentLevelId;
this.routeKey = routeKey;
}
public ContextData(Span span) {
this.traceId = span.getTraceId();
this.parentLevel = span.getParentLevel();
this.levelId = span.getLevelId();
this.routeKey = span.getRouteKey();
}
public ContextData(String contextDataStr) {
// 反序列化参数
String[] value = contextDataStr.split(CONTEXT_DATA_SEGMENT_SPILT_CHAR);
if (value == null || value.length != 4) {
throw new IllegalArgumentException("illegal context");
}
String traceIdStr = value[0];
String[] traceIdSegments = traceIdStr.split("\\.");
if(traceIdSegments == null || traceIdSegments.length != 6){
throw new IllegalArgumentException("illegal traceid in context");
}
TraceId.Builder traceIdBuilder = TraceId.newBuilder();
int i = 0;
for (String traceIdSegment : traceIdSegments) {
try {
traceIdBuilder.addSegments(Long.parseLong(traceIdSegment));
}catch(NumberFormatException e){
throw new IllegalArgumentException("illegal traceid seg[" + i + "] in context", e);
}
i++;
}
this.traceId = traceIdBuilder.build();
this.parentLevel = value[1].trim();
this.levelId = Integer.valueOf(value[2]);
this.routeKey = Integer.parseInt(value[3]);
}
public TraceId getTraceId() {
return traceId;
}
public String getParentLevel() {
return parentLevel;
}
public int getLevelId() {
return levelId;
}
public int getRouteKey(){
return this.routeKey;
}
@Override
public String toString() {
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.append(formatTraceId(traceId));
stringBuilder.append(CONTEXT_DATA_SEGMENT_SPILT_CHAR);
if (parentLevel == null || parentLevel.length() == 0) {
stringBuilder.append(" ");
} else {
stringBuilder.append(parentLevel);
}
stringBuilder.append(CONTEXT_DATA_SEGMENT_SPILT_CHAR);
stringBuilder.append(levelId);
stringBuilder.append(CONTEXT_DATA_SEGMENT_SPILT_CHAR);
stringBuilder.append(routeKey);
return stringBuilder.toString();
}
}

View File

@ -1,6 +0,0 @@
package com.a.eye.skywalking.model;
public class EmptyContextData extends ContextData {
public EmptyContextData() {
}
}

View File

@ -1,95 +0,0 @@
package com.a.eye.skywalking.model;
import com.a.eye.skywalking.api.IBuriedPointType;
import com.a.eye.skywalking.util.StringUtil;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class Identification {
private String viewPoint;
private String businessKey;
private String spanTypeDesc;
private String callType;
private long startTimestamp;
private Map<String, String> tags = new HashMap<String, String>();
public Identification() {
//Non
}
public String getViewPoint() {
return viewPoint;
}
public String getSpanTypeDesc() {
return spanTypeDesc;
}
public String getCallType() {
return callType;
}
public String getBusinessKey() {
return businessKey;
}
public long getStartTimestamp() {
return startTimestamp;
}
public Map<String, String> getTags() {
if (tags == null){
return Collections.EMPTY_MAP;
}
return tags;
}
public static IdentificationBuilder newBuilder() {
return new IdentificationBuilder();
}
public static class IdentificationBuilder {
private Identification identification;
IdentificationBuilder() {
identification = new Identification();
}
public Identification build() {
return identification;
}
public IdentificationBuilder viewPoint(String viewPoint) {
identification.viewPoint = viewPoint;
return this;
}
public IdentificationBuilder businessKey(String businessKey) {
identification.businessKey = businessKey;
return this;
}
public IdentificationBuilder spanType(IBuriedPointType spanType) {
if (StringUtil.isEmpty(spanType.getTypeName())) {
throw new IllegalArgumentException("Span Type name cannot be null");
}
identification.spanTypeDesc = spanType.getTypeName();
identification.callType = spanType.getCallType().toString();
return this;
}
public IdentificationBuilder startTime(long startTime) {
identification.startTimestamp = startTime;
return this;
}
public IdentificationBuilder tag(String tagKey, String tagValue) {
identification.tags.put(tagKey, tagValue);
return this;
}
}
}

Some files were not shown because too many files have changed in this diff Show More